This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new e211d267a0 [#10685] feat(iceberg-rest): Support vended credentials on
planTableScan endpoint (#10800)
e211d267a0 is described below
commit e211d267a0ef31d4844dfe2165a8c33bc8ed90c6
Author: Sachin Ranjalkar <[email protected]>
AuthorDate: Mon Jul 13 07:01:01 2026 +0530
[#10685] feat(iceberg-rest): Support vended credentials on planTableScan
endpoint (#10800)
### What changes were proposed in this pull request?
Add `X-Iceberg-Access-Delegation` header support to the `planTableScan`
endpoint and return storage credentials in the response when
`vended-credentials` is requested. Mirrors the existing
credential-vending flow from `createTable`/`loadTable`/`registerTable`
(#10684).
Changes:
- `IcebergTableOperations.planTableScan`: read
`@HeaderParam(X_ICEBERG_ACCESS_DELEGATION)`, compute
`isCredentialVending`, build `IcebergRequestContext` with credential
flag
- `IcebergTableOperationExecutor.planTableScan`: resolve
`CredentialPrivilege` and pass credential-vending flag through the
dispatcher to `CatalogWrapperForREST`
- `CatalogWrapperForREST.planTableScan`: add 4-arg overload that accepts
`requestCredentialVending` and `CredentialPrivilege`; inject credentials
via `PlanTableScanResponse.Builder.withCredentials()` *after* cache
lookup so cached plans remain credential-free
- `CatalogWrapperForREST.injectScanCredentials`: reuses the
already-loaded `Table` object from scan planning (no redundant
`loadTable`), catches only `ServiceUnavailableException`
- `CatalogWrapperForREST.getCredentialFromTable`: extracts credentials
from a `Table` object directly, following the same location/property
logic as `getCredential(TableMetadata, ...)`
- `FederatedCatalogWrapper.planTableScan`: overrides to fetch
credentials from the remote REST catalog instead of the local credential
manager
-
`TestCatalogWrapperForREST.testPlanTableScanCacheDoesNotLeakCredentials`:
verifies cached plans don't leak credentials to non-vended requests and
that vended requests on cache hits get fresh credentials
- `TestIcebergTableOperations`: three new tests — credential vending (no
header / local / S3 with prefix verification), remote-signing rejection
(406), invalid header (400)
### Why are the changes needed?
The Iceberg REST spec defines `X-Iceberg-Access-Delegation` as a valid
header on `planTableScan` and `CompletedPlanningResult` includes a
`storage-credentials` field. Currently, clients performing server-side
scan planning must make a separate `GET .../credentials` call to obtain
storage access credentials before reading the data files returned in the
scan plan.
Fixes #10685
### Does this PR introduce _any_ user-facing change?
Yes. The `planTableScan` REST endpoint now accepts the
`X-Iceberg-Access-Delegation` header and returns vended credentials in
the response `storage-credentials` field when requested. Backward
compatible -- clients that do not send the header get existing behavior.
### How was this patch tested?
Added unit tests:
-
`TestCatalogWrapperForREST.testPlanTableScanCacheDoesNotLeakCredentials`
-- vended request returns credentials, non-vended cache hit returns
none, vended cache hit returns fresh credentials
- `TestIcebergTableOperations.testPlanTableScanWithCredentialVending` --
no vending without header, no vending for `file://` location, vending
present for `s3://` location with correct credential type and prefix
-
`TestIcebergTableOperations.testPlanTableScanRemoteSigningNotSupported`
-- 406 response
- `TestIcebergTableOperations.testPlanTableScanInvalidAccessDelegation`
-- 400 response
All existing tests pass (no regressions).
---
.../iceberg/service/CatalogWrapperForREST.java | 115 ++++++--
.../iceberg/service/FederatedCatalogWrapper.java | 124 +++++++++
.../iceberg/service/IcebergRESTUtils.java | 44 +++
.../dispatcher/IcebergTableOperationExecutor.java | 6 +-
.../service/rest/IcebergTableOperations.java | 13 +-
.../iceberg/service/TestCatalogWrapperForREST.java | 305 +++++++++++++++++++++
.../service/rest/TestIcebergTableOperations.java | 110 ++++++++
7 files changed, 682 insertions(+), 35 deletions(-)
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index e02c6d4f55..a2749325ae 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -39,6 +39,7 @@ import
org.apache.gravitino.credential.CatalogCredentialManager;
import org.apache.gravitino.credential.Credential;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
+import org.apache.gravitino.credential.CredentialPropertyUtils;
import org.apache.gravitino.credential.PathBasedCredentialContext;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
@@ -48,6 +49,7 @@ import org.apache.gravitino.storage.GCSProperties;
import org.apache.gravitino.utils.ClassUtils;
import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.IncrementalAppendScan;
import org.apache.iceberg.Scan;
@@ -373,7 +375,19 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
/**
- * Plan table scan and return scan tasks.
+ * Plan table scan without credential vending.
+ *
+ * @param tableIdentifier The table identifier.
+ * @param scanRequest The scan request parameters.
+ * @return PlanTableScanResponse with status=COMPLETED and file scan tasks.
+ */
+ public PlanTableScanResponse planTableScan(
+ TableIdentifier tableIdentifier, PlanTableScanRequest scanRequest) {
+ return planTableScan(tableIdentifier, scanRequest, false,
CredentialPrivilege.READ);
+ }
+
+ /**
+ * Plan table scan and optionally inject vended storage credentials.
*
* <p>This method performs server-side scan planning to optimize query
performance by reducing
* client-side metadata loading and enabling parallel task execution.
@@ -384,18 +398,27 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
* different from asynchronous mode (SUBMITTED status) where a plan ID is
returned for later
* retrieval.
*
+ * <p>When {@code requestCredentialVending} is true and the table is
eligible (non-local,
+ * non-HDFS), storage credentials are injected directly into the response
using the table already
+ * loaded for scan planning -- avoiding a redundant {@code loadTable} call.
+ *
* <p>Referenced from Iceberg PR #13400 for scan planning implementation.
*
* @param tableIdentifier The table identifier.
* @param scanRequest The scan request parameters including filters,
projections, snapshot-id,
* etc.
+ * @param requestCredentialVending whether the client requested credential
vending
+ * @param privilege the credential privilege level for vending
* @return PlanTableScanResponse with status=COMPLETED and file scan tasks.
* @throws IllegalArgumentException if scan request validation fails
* @throws org.apache.gravitino.exceptions.NoSuchTableException if table
doesn't exist
* @throws RuntimeException for other scan planning failures
*/
public PlanTableScanResponse planTableScan(
- TableIdentifier tableIdentifier, PlanTableScanRequest scanRequest) {
+ TableIdentifier tableIdentifier,
+ PlanTableScanRequest scanRequest,
+ boolean requestCredentialVending,
+ CredentialPrivilege privilege) {
LOG.debug(
"Planning scan for table: {}, snapshotId: {}, startSnapshotId: {},
endSnapshotId: {}, select: {}, caseSensitive: {}",
@@ -410,43 +433,47 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
Table table = getCatalog().loadTable(tableIdentifier);
Optional<PlanTableScanResponse> cachedResponse =
scanPlanCache.get(ScanPlanCacheKey.create(tableIdentifier, table,
scanRequest));
+
+ PlanTableScanResponse response;
if (cachedResponse.isPresent()) {
LOG.info("Using cached scan plan for table: {}", tableIdentifier);
- return cachedResponse.get();
- }
+ response = cachedResponse.get();
+ } else {
+ List<FileScanTask> fileScanTasks = new ArrayList<>();
+
+ try (CloseableIterable<FileScanTask> scanTasks =
+ createFilePlanScanTasks(table, tableIdentifier, scanRequest)) {
+ for (FileScanTask fileScanTask : scanTasks) {
+ fileScanTasks.add(fileScanTask);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to close scan task iterator for table: {}",
tableIdentifier, e);
+ throw new RuntimeException("Failed to plan scan tasks: " +
e.getMessage(), e);
+ }
- List<FileScanTask> fileScanTasks = new ArrayList<>();
+ if (fileScanTasks.isEmpty()) {
+ LOG.info(
+ "Scan planning returned no tasks for table: {}. Table may be
empty or fully filtered.",
+ tableIdentifier);
+ }
- try (CloseableIterable<FileScanTask> scanTasks =
- createFilePlanScanTasks(table, tableIdentifier, scanRequest)) {
- for (FileScanTask fileScanTask : scanTasks) {
- fileScanTasks.add(fileScanTask);
+ try {
+ response = buildCompletedPlanTableScanResponse(table, fileScanTasks);
+ } catch (Exception e) {
+ LOG.error("Failed to build scan plan response for table: {}",
tableIdentifier, e);
+ throw new RuntimeException(
+ String.format(
+ "Failed to build scan plan response for table: %s. Error:
%s",
+ tableIdentifier, e.getMessage()),
+ e);
}
- } catch (IOException e) {
- LOG.error("Failed to close scan task iterator for table: {}",
tableIdentifier, e);
- throw new RuntimeException("Failed to plan scan tasks: " +
e.getMessage(), e);
- }
- if (fileScanTasks.isEmpty()) {
- LOG.info(
- "Scan planning returned no tasks for table: {}. Table may be empty
or fully filtered.",
- tableIdentifier);
+ scanPlanCache.put(ScanPlanCacheKey.create(tableIdentifier, table,
scanRequest), response);
}
- PlanTableScanResponse response;
- try {
- response = buildCompletedPlanTableScanResponse(table, fileScanTasks);
- } catch (Exception e) {
- LOG.error("Failed to build scan plan response for table: {}",
tableIdentifier, e);
- throw new RuntimeException(
- String.format(
- "Failed to build scan plan response for table: %s. Error: %s",
- tableIdentifier, e.getMessage()),
- e);
+ if (requestCredentialVending &&
!isLocalOrHdfsLocation(table.location())) {
+ response = injectScanCredentials(tableIdentifier, table, response,
privilege);
}
-
- // Cache the scan plan response
- scanPlanCache.put(ScanPlanCacheKey.create(tableIdentifier, table,
scanRequest), response);
return response;
} catch (IllegalArgumentException e) {
@@ -462,6 +489,34 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
}
+ /**
+ * Inject vended credentials into a scan response using the already-loaded
table, avoiding a
+ * redundant {@code loadTable} call. Follows the same eligibility logic as
{@link
+ * #shouldGenerateCredential} and the same credential generation as {@link
#getCredential}.
+ */
+ private PlanTableScanResponse injectScanCredentials(
+ TableIdentifier tableIdentifier,
+ Table table,
+ PlanTableScanResponse response,
+ CredentialPrivilege privilege) {
+ try {
+ validateCredentialLocation(table.location());
+ TableMetadata metadata = ((BaseTable) table).operations().current();
+ Credential credential = getCredential(metadata, privilege);
+ Map<String, String> config =
+ new
HashMap<>(CredentialPropertyUtils.toIcebergProperties(credential));
+ config.putAll(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), tableIdentifier,
config));
+ org.apache.iceberg.rest.credentials.Credential restCred =
+ IcebergRESTUtils.toRESTCredential(table.location(), config);
+ return IcebergRESTUtils.copyWithCredentials(response,
Collections.singletonList(restCred));
+ } catch (ServiceUnavailableException e) {
+ LOG.warn("Failed to generate scan credentials for table: {}",
tableIdentifier, e);
+ return response;
+ }
+ }
+
/**
* Builds a synchronous COMPLETED scan plan response for Iceberg 1.11+ REST
clients only.
*
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
index d11da80c98..b4128ff724 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.iceberg.service;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
@@ -57,6 +58,7 @@ import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.rest.CatalogHandlers;
import org.apache.iceberg.rest.ErrorHandlers;
import org.apache.iceberg.rest.HTTPClient;
+import org.apache.iceberg.rest.ParserContext;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.rest.RESTClient;
import org.apache.iceberg.rest.RESTUtil;
@@ -66,10 +68,12 @@ import org.apache.iceberg.rest.auth.AuthManagers;
import org.apache.iceberg.rest.auth.AuthSession;
import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.PlanTableScanResponse;
/**
* A {@link CatalogWrapperForREST} for a federated Iceberg REST catalog (the
underlying catalog is a
@@ -153,6 +157,37 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
catalogCredentialManager.catalogName(), identifier, upstream);
}
+ /**
+ * Delegates scan planning to the remote REST catalog instead of executing
it locally.
+ *
+ * <p>When credential vending is requested, the {@code
X-Iceberg-Access-Delegation:
+ * vended-credentials} header is forwarded so the remote catalog returns
credentials inline. Any
+ * upstream credential refresh endpoints are rewritten to point at this IRC
instance.
+ *
+ * @param tableIdentifier the table to scan.
+ * @param scanRequest the scan request parameters.
+ * @param requestCredentialVending whether the client requested vended
credentials.
+ * @param privilege ignored; the remote REST catalog decides what to vend.
+ * @return the scan response from the remote catalog, with rewritten
credential refresh endpoints.
+ */
+ @Override
+ public PlanTableScanResponse planTableScan(
+ TableIdentifier tableIdentifier,
+ PlanTableScanRequest scanRequest,
+ boolean requestCredentialVending,
+ CredentialPrivilege privilege) {
+ Table table = getCatalog().loadTable(tableIdentifier);
+ PlanTableScanResponse response =
+ getRESTTablePlanScan(
+ (RESTCatalog) getCatalog(),
+ tableIdentifier,
+ scanRequest,
+ requestCredentialVending,
+ table.specs());
+ return IcebergRESTUtils.rewriteScanPlanCredentials(
+ catalogCredentialManager.catalogName(), tableIdentifier, response);
+ }
+
private static LoadCredentialsResponse getRESTTableCredentials(
RESTCatalog restCatalog, TableIdentifier identifier) {
Map<String, String> properties = Maps.newHashMap(restCatalog.properties());
@@ -207,6 +242,95 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
}
}
+ /**
+ * Sends a {@code POST {table}/plan} request to the remote REST catalog.
+ *
+ * <p>Follows the same HTTP client lifecycle as {@link
#getRESTTableCredentials}. When credential
+ * vending is requested, the {@code X-Iceberg-Access-Delegation:
vended-credentials} header is
+ * included so the remote catalog returns credentials inline in the plan
response.
+ *
+ * <p>The Iceberg response deserializer requires pre-loaded partition specs
to parse {@code
+ * file-scan-tasks}. These are supplied via a {@link ParserContext} built
from the caller-provided
+ * {@code specsById} map (typically obtained from a prior {@code loadTable}
call).
+ *
+ * @param restCatalog the underlying REST catalog whose properties supply
the URI and auth config.
+ * @param identifier the table to plan.
+ * @param scanRequest the scan request parameters.
+ * @param requestCredentialVending whether to include the access-delegation
header.
+ * @param specsById partition specs for the table, needed for response
deserialization.
+ * @return the plan response from the remote catalog.
+ */
+ private static PlanTableScanResponse getRESTTablePlanScan(
+ RESTCatalog restCatalog,
+ TableIdentifier identifier,
+ PlanTableScanRequest scanRequest,
+ boolean requestCredentialVending,
+ Map<Integer, PartitionSpec> specsById) {
+ Map<String, String> properties = Maps.newHashMap(restCatalog.properties());
+ String planPath =
ResourcePaths.forCatalogProperties(properties).planTableScan(identifier);
+
+ Map<String, String> headers =
+ requestCredentialVending
+ ? ImmutableMap.of("X-Iceberg-Access-Delegation",
"vended-credentials")
+ : Collections.emptyMap();
+
+ ParserContext parserContext =
+ ParserContext.builder()
+ .add("specsById", specsById)
+ .add("caseSensitive", scanRequest.caseSensitive())
+ .build();
+
+ AuthManager authManager = null;
+ RESTClient client = null;
+ AuthSession authSession = null;
+ try {
+ authManager = AuthManagers.loadAuthManager(restCatalog.name(),
properties);
+ client =
+ HTTPClient.builder(properties)
+ .uri(properties.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(properties))
+ .build();
+ authSession = authManager.catalogSession(client, properties);
+ return client
+ .withAuthSession(authSession)
+ .post(
+ planPath,
+ scanRequest,
+ PlanTableScanResponse.class,
+ headers,
+ ErrorHandlers.planErrorHandler(),
+ ignored -> {},
+ parserContext);
+ } finally {
+ if (authSession != null) {
+ try {
+ authSession.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close auth session when planning table scan for
table: {}", identifier, e);
+ }
+ }
+
+ if (client != null) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close REST client when planning table scan for table:
{}", identifier, e);
+ }
+ }
+
+ if (authManager != null) {
+ try {
+ authManager.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close auth manager when planning table scan for
table: {}", identifier, e);
+ }
+ }
+ }
+ }
+
/**
* Federation-aware {@code createTable}: creates the table on the underlying
(remote) catalog and
* extracts client-facing FileIO/credential properties from {@code
table.io()}.
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
index 9a4d74091d..34e80d2453 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
@@ -58,6 +58,7 @@ import org.apache.iceberg.rest.responses.ErrorResponse;
import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.PlanTableScanResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -232,6 +233,49 @@ public class IcebergRESTUtils {
return toRESTCredential(prefix, ImmutableMap.copyOf(filteredConfig));
}
+ /**
+ * Rewrites credentials in a {@link PlanTableScanResponse} so their {@code
+ * refresh-credentials-endpoint} entries point at this IRC instance instead
of the upstream
+ * catalog. Returns the response unchanged when no credentials are present.
+ *
+ * @param catalogName IRC catalog name used to build refresh paths
+ * @param tableIdentifier table receiving the credentials
+ * @param response the scan plan response returned by the upstream REST
catalog
+ * @return the scan plan response with IRC-local refresh endpoints on any
credentials
+ */
+ @SuppressWarnings("deprecation")
+ public static PlanTableScanResponse rewriteScanPlanCredentials(
+ String catalogName, TableIdentifier tableIdentifier,
PlanTableScanResponse response) {
+ if (response.credentials() == null || response.credentials().isEmpty()) {
+ return response;
+ }
+ List<org.apache.iceberg.rest.credentials.Credential> rewritten = new
ArrayList<>();
+ for (org.apache.iceberg.rest.credentials.Credential cred :
response.credentials()) {
+ rewritten.add(rewriteCredential(catalogName, tableIdentifier,
cred.prefix(), cred.config()));
+ }
+ return copyWithCredentials(response, rewritten);
+ }
+
+ /**
+ * Copies a {@link PlanTableScanResponse} with replacement credentials.
+ *
+ * <p>TODO: Remove when PlanTableScanResponse supports a native
copy/toBuilder method.
+ */
+ @SuppressWarnings("deprecation")
+ static PlanTableScanResponse copyWithCredentials(
+ PlanTableScanResponse response,
+ List<org.apache.iceberg.rest.credentials.Credential> credentials) {
+ return PlanTableScanResponse.builder()
+ .withPlanStatus(response.planStatus())
+ .withPlanId(response.planId())
+ .withPlanTasks(response.planTasks())
+ .withFileScanTasks(response.fileScanTasks())
+ .withSpecsById(response.specsById())
+ .withErrorResponse(response.errorResponse())
+ .withCredentials(credentials)
+ .build();
+ }
+
public static <T> Response ok(T t) {
return
Response.status(Response.Status.OK).entity(t).type(MediaType.APPLICATION_JSON).build();
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
index 6307013315..954d2a8c27 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
@@ -222,9 +222,13 @@ public class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
IcebergRequestContext context,
TableIdentifier tableIdentifier,
PlanTableScanRequest scanRequest) {
+ CredentialPrivilege privilege = CredentialPrivilege.READ;
+ if (context.requestCredentialVending()) {
+ privilege = getCredentialPrivilege(context, tableIdentifier);
+ }
return icebergCatalogWrapperManager
.getCatalogWrapper(context.catalogName())
- .planTableScan(tableIdentifier, scanRequest);
+ .planTableScan(tableIdentifier, scanRequest,
context.requestCredentialVending(), privilege);
}
@Override
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java
index 029722cfd8..4bedb7c41e 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java
@@ -524,16 +524,21 @@ public class IcebergTableOperations {
@Encoded() @PathParam("namespace") @AuthorizationMetadata(type =
EntityType.SCHEMA)
String namespace,
@Encoded() @PathParam("table") @AuthorizationMetadata(type =
EntityType.TABLE) String table,
- PlanTableScanRequest scanRequest) {
+ PlanTableScanRequest scanRequest,
+ @HeaderParam(X_ICEBERG_ACCESS_DELEGATION) String accessDelegation) {
+ boolean isCredentialVending = isCredentialVending(accessDelegation);
String catalogName = IcebergRESTUtils.getCatalogName(prefix);
Namespace icebergNS =
RESTUtil.decodeNamespace(namespace,
IcebergRESTUtils.NAMESPACE_SEPARATOR_URLENCODED_UTF_8);
String tableName = RESTUtil.decodeString(table);
LOG.info(
- "Plan table scan, catalog: {}, namespace: {}, table: {}",
+ "Plan table scan, catalog: {}, namespace: {}, table: {}, "
+ + "accessDelegation: {}, isCredentialVending: {}",
catalogName,
icebergNS,
- tableName);
+ tableName,
+ accessDelegation,
+ isCredentialVending);
try {
return Utils.doAs(
@@ -541,7 +546,7 @@ public class IcebergTableOperations {
() -> {
TableIdentifier tableIdentifier = TableIdentifier.of(icebergNS,
tableName);
IcebergRequestContext context =
- new IcebergRequestContext(httpServletRequest(), catalogName);
+ new IcebergRequestContext(httpServletRequest(), catalogName,
isCredentialVending);
PlanTableScanResponse scanResponse =
tableOperationDispatcher.planTableScan(context,
tableIdentifier, scanRequest);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
index 2607eae063..713091a9e4 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
@@ -47,6 +47,7 @@ import
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.iceberg.service.cache.LocalScanPlanCache;
import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BaseTransaction;
@@ -72,15 +73,19 @@ import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.ResolvingFileIO;
import org.apache.iceberg.io.StorageCredential;
import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.rest.PlanStatus;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.rest.auth.AuthProperties;
import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.PlanTableScanResponse;
+import org.apache.iceberg.rest.responses.PlanTableScanResponseParser;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -417,6 +422,306 @@ public class TestCatalogWrapperForREST {
Assertions.assertEquals("s3://bucket/wh/db/tbl",
response.credentials().get(0).prefix());
}
+ @Test
+ void testPlanTableScanCacheDoesNotLeakCredentials() {
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE,
+ IcebergConstants.SCAN_PLAN_CACHE_IMPL,
+ LocalScanPlanCache.class.getName()));
+
+ CatalogWrapperForREST wrapper = new CatalogWrapperForREST("cache-test",
config);
+ Namespace namespace = Namespace.of("db");
+ Catalog catalog = wrapper.getCatalog();
+ ((SupportsNamespaces) catalog).createNamespace(namespace);
+ TableIdentifier tableId = TableIdentifier.of(namespace, "tbl");
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ catalog.createTable(
+ tableId,
+ schema,
+ PartitionSpec.unpartitioned(),
+ "s3://bucket/db/tbl",
+ Collections.emptyMap());
+
+ PlanTableScanRequest request = PlanTableScanRequest.builder().build();
+
+ PlanTableScanResponse vended1 =
+ wrapper.planTableScan(tableId, request, true,
CredentialPrivilege.READ);
+ Assertions.assertFalse(
+ vended1.credentials().isEmpty(), "Vended request should return
credentials");
+
+ PlanTableScanResponse nonVended =
+ wrapper.planTableScan(tableId, request, false,
CredentialPrivilege.READ);
+ Assertions.assertTrue(
+ nonVended.credentials().isEmpty(),
+ "Non-vended request should not return credentials from cache");
+
+ PlanTableScanResponse vended2 =
+ wrapper.planTableScan(tableId, request, true,
CredentialPrivilege.READ);
+ Assertions.assertFalse(
+ vended2.credentials().isEmpty(),
+ "Vended request on cache hit should return fresh credentials");
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ void testFederatedPlanTableScanDelegatesToRemote() throws Exception {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ String expectedPath = "/v1/upstream/namespaces/db/tables/tbl/plan";
+
+ org.apache.iceberg.rest.credentials.Credential cred =
+ IcebergRESTUtils.toRESTCredential(
+ "s3://bucket/db/tbl/",
+ ImmutableMap.of(
+ "s3.access-key-id", "upstream-key",
+ "s3.secret-access-key", "upstream-secret",
+ "s3.session-token", "upstream-token",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+ PlanTableScanResponse upstreamResponse =
+ PlanTableScanResponse.builder()
+ .withPlanStatus(PlanStatus.COMPLETED)
+ .withSpecsById(ImmutableMap.of(0, PartitionSpec.unpartitioned()))
+ .withCredentials(Collections.singletonList(cred))
+ .build();
+ String upstreamJson = PlanTableScanResponseParser.toJson(upstreamResponse);
+
+ AtomicReference<String> requestPath = new AtomicReference<>();
+ AtomicReference<String> requestMethod = new AtomicReference<>();
+ AtomicReference<String> accessDelegationHeader = new AtomicReference<>();
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ requestPath.set(exchange.getRequestURI().getPath());
+ requestMethod.set(exchange.getRequestMethod());
+ accessDelegationHeader.set(
+
exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation"));
+ byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+ Table mockTable = mock(Table.class);
+ when(mockTable.specs()).thenReturn(ImmutableMap.of(0,
PartitionSpec.unpartitioned()));
+ when(restCatalog.loadTable(table)).thenReturn(mockTable);
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ PlanTableScanRequest scanRequest =
PlanTableScanRequest.builder().build();
+ PlanTableScanResponse response =
+ wrapper.planTableScan(table, scanRequest, true,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(expectedPath, requestPath.get());
+ Assertions.assertEquals("POST", requestMethod.get());
+ Assertions.assertEquals("vended-credentials",
accessDelegationHeader.get());
+ Assertions.assertFalse(response.credentials().isEmpty());
+ Credential credential = response.credentials().get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-key",
credential.config().get("s3.access-key-id"));
+ Assertions.assertEquals(
+ "v1/local/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ void testFederatedPlanTableScanNoCredentials() throws Exception {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ PlanTableScanResponse upstreamResponse =
+ PlanTableScanResponse.builder()
+ .withPlanStatus(PlanStatus.COMPLETED)
+ .withSpecsById(ImmutableMap.of(0, PartitionSpec.unpartitioned()))
+ .build();
+ String upstreamJson = PlanTableScanResponseParser.toJson(upstreamResponse);
+
+ AtomicReference<String> accessDelegationHeader = new AtomicReference<>();
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ accessDelegationHeader.set(
+
exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation"));
+ byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+ Table mockTable = mock(Table.class);
+ when(mockTable.specs()).thenReturn(ImmutableMap.of(0,
PartitionSpec.unpartitioned()));
+ when(restCatalog.loadTable(table)).thenReturn(mockTable);
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ PlanTableScanRequest scanRequest =
PlanTableScanRequest.builder().build();
+ PlanTableScanResponse response =
+ wrapper.planTableScan(table, scanRequest, false,
CredentialPrivilege.READ);
+
+ Assertions.assertNull(
+ accessDelegationHeader.get(),
+ "X-Iceberg-Access-Delegation header should not be sent without
credential vending");
+ Assertions.assertTrue(
+ response.credentials() == null || response.credentials().isEmpty(),
+ "Non-vended request should not return credentials");
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testFederatedPlanTableScanOnFailure() throws Exception {
+ TableIdentifier tableId = TableIdentifier.of(Namespace.of("db"), "tbl");
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.sendResponseHeaders(500, -1);
+ exchange.close();
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+ Table mockTable = mock(Table.class);
+ when(mockTable.specs()).thenReturn(ImmutableMap.of(0,
PartitionSpec.unpartitioned()));
+ when(restCatalog.loadTable(tableId)).thenReturn(mockTable);
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ PlanTableScanRequest scanRequest =
PlanTableScanRequest.builder().build();
+ Assertions.assertThrows(
+ ServiceFailureException.class,
+ () -> wrapper.planTableScan(tableId, scanRequest, true,
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testFederatedPlanTableScanNoSuchTable() throws Exception {
+ TableIdentifier tableId = TableIdentifier.of(Namespace.of("db"),
"missing");
+ String errorJson =
+ "{\"error\":{\"message\":\"Table not
found\",\"type\":\"NoSuchTableException\","
+ + "\"code\":404,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(404, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+ Table mockTable = mock(Table.class);
+ when(mockTable.specs()).thenReturn(ImmutableMap.of(0,
PartitionSpec.unpartitioned()));
+ when(restCatalog.loadTable(tableId)).thenReturn(mockTable);
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ PlanTableScanRequest scanRequest =
PlanTableScanRequest.builder().build();
+ Assertions.assertThrows(
+ NoSuchTableException.class,
+ () -> wrapper.planTableScan(tableId, scanRequest, true,
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
@Test
void testValidateCredentialLocation() {
Assertions.assertDoesNotThrow(
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
index 14d2782b4f..0abaa7305c 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
@@ -1115,4 +1115,114 @@ public class TestIcebergTableOperations extends
IcebergNamespaceTestBase {
NameIdentifier.of("metalake", "catalog", "schema"));
Assertions.assertEquals(-1, event.resultCount());
}
+
+ @ParameterizedTest
+
@MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+ void testPlanTableScanWithCredentialVending(Namespace namespace) {
+ verifyCreateNamespaceSucc(namespace);
+ PlanTableScanRequest emptyRequest = PlanTableScanRequest.builder().build();
+
+ // scan without credential vending -- no storage-credentials in response
+ String tableName = "scan_cred_no_header";
+ verifyCreateTableSucc(namespace, tableName);
+ Response noCredResponse = doPlanTableScan(namespace, tableName,
emptyRequest);
+ Assertions.assertEquals(Status.OK.getStatusCode(),
noCredResponse.getStatus());
+ try {
+ JsonNode noCredJson =
JsonUtil.mapper().readTree(noCredResponse.readEntity(String.class));
+ Assertions.assertFalse(
+ noCredJson.has("storage-credentials"),
+ "Response should not have storage-credentials without header");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ // scan with credential vending on local table -- no storage-credentials
+ String localTableName = "scan_cred_local";
+ Response localCreateResponse =
+ doCreateTableWithCredentialVending(
+ namespace, localTableName, "file:///tmp/" + localTableName);
+ Assertions.assertEquals(Status.OK.getStatusCode(),
localCreateResponse.getStatus());
+ Response localScanResponse =
+ doPlanTableScanWithCredentialVending(namespace, localTableName,
emptyRequest);
+ Assertions.assertEquals(Status.OK.getStatusCode(),
localScanResponse.getStatus());
+ try {
+ JsonNode localScanJson =
+
JsonUtil.mapper().readTree(localScanResponse.readEntity(String.class));
+ Assertions.assertFalse(
+ localScanJson.has("storage-credentials"),
+ "Local table should not have storage-credentials");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ // scan with credential vending on S3 table -- should have
storage-credentials
+ String s3TableName = "scan_cred_s3";
+ String s3Location = "s3://dummy-bucket/" + s3TableName;
+ Response s3CreateResponse =
+ doCreateTableWithCredentialVending(namespace, s3TableName, s3Location);
+ Assertions.assertEquals(Status.OK.getStatusCode(),
s3CreateResponse.getStatus());
+ Response s3ScanResponse =
+ doPlanTableScanWithCredentialVending(namespace, s3TableName,
emptyRequest);
+ Assertions.assertEquals(Status.OK.getStatusCode(),
s3ScanResponse.getStatus());
+ try {
+ JsonNode s3ScanJson =
JsonUtil.mapper().readTree(s3ScanResponse.readEntity(String.class));
+ Assertions.assertTrue(
+ s3ScanJson.has("storage-credentials"), "S3 table should have
storage-credentials");
+ JsonNode credentials = s3ScanJson.get("storage-credentials");
+ Assertions.assertTrue(credentials.isArray() && credentials.size() > 0);
+ JsonNode firstCred = credentials.get(0);
+ Assertions.assertTrue(firstCred.has("config"));
+ Assertions.assertEquals(
+ DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE,
+ firstCred.get("config").get(Credential.CREDENTIAL_TYPE).asText());
+ Assertions.assertTrue(firstCred.has("prefix"), "Credential should have a
prefix field");
+ Assertions.assertEquals(
+ s3Location,
+ firstCred.get("prefix").asText(),
+ "Credential prefix should match the table location");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @ParameterizedTest
+
@MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+ void testPlanTableScanRemoteSigningNotSupported(Namespace namespace) {
+ verifyCreateNamespaceSucc(namespace);
+ verifyCreateTableSucc(namespace, "scan_remote_signing");
+ PlanTableScanRequest request = PlanTableScanRequest.builder().build();
+ Response response =
+ getTableClientBuilder(namespace,
Optional.of("scan_remote_signing/plan"))
+ .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION,
"remote-signing")
+ .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+ Assertions.assertEquals(406, response.getStatus());
+ String errorBody = response.readEntity(String.class);
+ Assertions.assertTrue(
+ errorBody.contains("remote signing") ||
errorBody.contains("remote-signing"),
+ "Error message should mention remote signing: " + errorBody);
+ }
+
+ @ParameterizedTest
+
@MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+ void testPlanTableScanInvalidAccessDelegation(Namespace namespace) {
+ verifyCreateNamespaceSucc(namespace);
+ verifyCreateTableSucc(namespace, "scan_invalid_delegation");
+ PlanTableScanRequest request = PlanTableScanRequest.builder().build();
+ Response response =
+ getTableClientBuilder(namespace,
Optional.of("scan_invalid_delegation/plan"))
+ .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION,
"invalid-value")
+ .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+ Assertions.assertEquals(400, response.getStatus());
+ String errorBody = response.readEntity(String.class);
+ Assertions.assertTrue(
+ errorBody.contains("vended-credentials") &&
errorBody.contains("illegal"),
+ "Error message should mention valid values: " + errorBody);
+ }
+
+ private Response doPlanTableScanWithCredentialVending(
+ Namespace ns, String tableName, PlanTableScanRequest request) {
+ return getTableClientBuilder(ns, Optional.of(tableName + "/plan"))
+ .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION,
"vended-credentials")
+ .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+ }
}
