[#11580] feat(iceberg): Support credential vending in planTableScan endpoint

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java

@@ -511,23 +511,28 @@ public Response planTableScan(
       @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);
     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(
           httpRequest,
           () -> {
             TableIdentifier tableIdentifier = TableIdentifier.of(icebergNS, tableName);
             IcebergRequestContext context =
-                new IcebergRequestContext(httpServletRequest(), catalogName);
+                new IcebergRequestContext(httpServletRequest(), catalogName, isCredentialVending);
 
             PlanTableScanResponse scanResponse =
                 tableOperationDispatcher.planTableScan(context, tableIdentifier, scanRequest);

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java

@@ -201,6 +201,66 @@ void testPlanTableScanTableNotFound(Namespace namespace) {
         dummyEventListener.popPostEvent() instanceof IcebergPlanTableScanFailureEvent);
   }
 
+  @ParameterizedTest
+  @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+  void testPlanTableScanWithCredentialVending(Namespace namespace) {
+    verifyCreateNamespaceSucc(namespace);
+    verifyCreateTableSucc(namespace, "plan_scan_cred_vending", true);
+
+    dummyEventListener.clearEvent();
+    TableMetadata metadata = getTableMeta(namespace, "plan_scan_cred_vending");
+    Long snapshotId = metadata.ref(SnapshotRef.MAIN_BRANCH).snapshotId();
+
+    PlanTableScanRequest request = buildPlanTableScanRequest(snapshotId);
+    Response response =
+        doPlanTableScanWithAccessDelegation(
+            namespace, "plan_scan_cred_vending", request, "vended-credentials");
+    Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+
+    Assertions.assertTrue(dummyEventListener.popPreEvent() instanceof IcebergPlanTableScanPreEvent);
+    Assertions.assertTrue(dummyEventListener.popPostEvent() instanceof IcebergPlanTableScanEvent);
+  }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+  void testPlanTableScanRemoteSigningNotSupported(Namespace namespace) {
+    verifyCreateNamespaceSucc(namespace);
+    verifyCreateTableSucc(namespace, "plan_scan_remote_signing", true);
+
+    TableMetadata metadata = getTableMeta(namespace, "plan_scan_remote_signing");
+    Long snapshotId = metadata.ref(SnapshotRef.MAIN_BRANCH).snapshotId();
+
+    PlanTableScanRequest request = buildPlanTableScanRequest(snapshotId);
+    Response response =
+        doPlanTableScanWithAccessDelegation(
+            namespace, "plan_scan_remote_signing", request, "remote-signing");
+    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, "plan_scan_invalid_delegation", true);
+
+    TableMetadata metadata = getTableMeta(namespace, "plan_scan_invalid_delegation");
+    Long snapshotId = metadata.ref(SnapshotRef.MAIN_BRANCH).snapshotId();
+
+    PlanTableScanRequest request = buildPlanTableScanRequest(snapshotId);
+    Response response =
+        doPlanTableScanWithAccessDelegation(
+            namespace, "plan_scan_invalid_delegation", request, "invalid-value");
+    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);
+  }
+
   @ParameterizedTest
   @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
   void testPlanTableScanWithIncrementalAppendScanValidRange(Namespace namespace) {
@@ -498,6 +558,14 @@ private Response doPlanTableScan(Namespace ns, String tableName, PlanTableScanRe
     return builder.post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
   }
 
+  private Response doPlanTableScanWithAccessDelegation(
+      Namespace ns, String tableName, PlanTableScanRequest request, String accessDelegation) {
+    Invocation.Builder builder =
+        getTableClientBuilder(ns, Optional.of(tableName + "/plan"))
+            .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, accessDelegation);
+    return builder.post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+  }
+
   private Response doUpdateTable(Namespace ns, String name, TableMetadata base) {
     TableMetadata newMetadata = base.updateSchema(newTableSchema);
     List<MetadataUpdate> metadataUpdates = newMetadata.changes();