[#11284] feat(iceberg-rest): Add the fetch scan tasks endpoint and its plumbing

core/src/main/java/org/apache/gravitino/audit/AuditLog.java

@@ -262,6 +262,8 @@ enum Operation {
 
     PLAN_TABLE_SCAN,
 
+    FETCH_SCAN_TASKS,
+
     PURGE_TABLE,
 
     LIST_TABLE,

core/src/main/java/org/apache/gravitino/audit/v2/CompatibilityUtils.java

@@ -55,6 +55,7 @@ public class CompatibilityUtils {
           .put(OperationType.LOAD_TABLE, Operation.LOAD_TABLE)
           .put(OperationType.LOAD_TABLE_CREDENTIAL, Operation.LOAD_TABLE_CREDENTIAL)
           .put(OperationType.PLAN_TABLE_SCAN, Operation.PLAN_TABLE_SCAN)
+          .put(OperationType.FETCH_SCAN_TASKS, Operation.FETCH_SCAN_TASKS)
           .put(OperationType.LIST_TABLE, Operation.LIST_TABLE)
           .put(OperationType.ALTER_TABLE, Operation.ALTER_TABLE)
           .put(OperationType.RENAME_TABLE, Operation.RENAME_TABLE)

core/src/main/java/org/apache/gravitino/listener/api/event/OperationType.java

@@ -27,6 +27,7 @@ public enum OperationType {
   LOAD_TABLE,
   LOAD_TABLE_CREDENTIAL,
   PLAN_TABLE_SCAN,
+  FETCH_SCAN_TASKS,
   LIST_TABLE,
   ALTER_TABLE,
   RENAME_TABLE,

core/src/test/java/org/apache/gravitino/audit/v2/TestCompatibilityUtils.java

@@ -62,6 +62,7 @@ public void testToAuditLogOperation() {
       {OperationType.LOAD_TABLE, Operation.LOAD_TABLE},
       {OperationType.LOAD_TABLE_CREDENTIAL, Operation.LOAD_TABLE_CREDENTIAL},
       {OperationType.PLAN_TABLE_SCAN, Operation.PLAN_TABLE_SCAN},
+      {OperationType.FETCH_SCAN_TASKS, Operation.FETCH_SCAN_TASKS},
       {OperationType.TABLE_EXISTS, Operation.TABLE_EXISTS},
       {OperationType.LIST_TABLE, Operation.LIST_TABLE},
       {OperationType.RENAME_TABLE, Operation.RENAME_TABLE},

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

@@ -59,13 +59,16 @@
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchPlanTaskException;
 import org.apache.iceberg.exceptions.ServiceUnavailableException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.rest.CatalogHandlers;
 import org.apache.iceberg.rest.PlanStatus;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -484,6 +487,34 @@ public PlanTableScanResponse planTableScan(
     }
   }
 
+  /**
+   * Fetch the scan tasks covered by a {@code plan-task} previously handed out by {@link
+   * #planTableScan}, completing the second step of the Iceberg REST two-step scan planning
+   * protocol.
+   *
+   * <p>{@link #planTableScan} currently returns every file scan task inline and hands out no {@code
+   * plan-tasks}, so no plan task presented here was issued by this server and every request is
+   * rejected as unknown. The endpoint is deliberately not advertised in {@code /v1/config} while
+   * that is the case, so a spec-compliant client never reaches it. Batching a plan into plan tasks,
+   * and redeeming them here, follows in a later change.
+   *
+   * @param tableIdentifier the table the plan task belongs to.
+   * @param request the request carrying the {@code plan-task}.
+   * @return the file scan tasks the plan task covers.
+   * @throws org.apache.iceberg.exceptions.NoSuchTableException if the table doesn't exist.
+   * @throws NoSuchPlanTaskException if the plan task was not issued for this table.
+   */
+  public FetchScanTasksResponse fetchScanTasks(
+      TableIdentifier tableIdentifier, FetchScanTasksRequest request) {
+    // Validate the table exists first, so a bad table reports 404 for the table rather than
+    // masking it as an unknown plan task. Consistent with planTableScan behavior.
+    getCatalog().loadTable(tableIdentifier);
+
+    LOG.info("Rejecting unknown plan task '{}' for table {}", request.planTask(), tableIdentifier);
+    throw new NoSuchPlanTaskException(
+        "Plan task %s was not issued for table %s", request.planTask(), tableIdentifier);
+  }
+
   /**
    * 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

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

@@ -37,6 +37,7 @@
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchIcebergTableException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchPlanTaskException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
@@ -72,6 +73,7 @@ public class IcebergExceptionMapper implements ExceptionMapper<Exception> {
           .put(NoSuchTableException.class, 404)
           .put(NoSuchIcebergTableException.class, 404)
           .put(NoSuchCatalogException.class, 404)
+          .put(NoSuchPlanTaskException.class, 404)
           .put(UnsupportedOperationException.class, 406)
           .put(NoSuchViewException.class, 404)
           .put(AlreadyExistsException.class, 409)

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

@@ -31,6 +31,9 @@
 import org.apache.gravitino.listener.api.event.IcebergDropTableEvent;
 import org.apache.gravitino.listener.api.event.IcebergDropTableFailureEvent;
 import org.apache.gravitino.listener.api.event.IcebergDropTablePreEvent;
+import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksEvent;
+import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksPreEvent;
 import org.apache.gravitino.listener.api.event.IcebergListTableEvent;
 import org.apache.gravitino.listener.api.event.IcebergListTableFailureEvent;
 import org.apache.gravitino.listener.api.event.IcebergListTablePreEvent;
@@ -56,9 +59,11 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -305,6 +310,36 @@ public PlanTableScanResponse planTableScan(
     return planTableScanResponse;
   }
 
+  /**
+   * Fetch the scan tasks for a {@code plan-task} returned by a prior scan plan.
+   *
+   * @param context Iceberg REST request context information.
+   * @param tableIdentifier The Iceberg table identifier.
+   * @param request The request carrying the {@code plan-task}.
+   * @return A FetchScanTasksResponse containing the scan tasks for that plan task
+   */
+  @Override
+  public FetchScanTasksResponse fetchScanTasks(
+      IcebergRequestContext context,
+      TableIdentifier tableIdentifier,
+      FetchScanTasksRequest request) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRESTUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), tableIdentifier);
+    eventBus.dispatchEvent(new IcebergFetchScanTasksPreEvent(context, gravitinoNameIdentifier));
+    FetchScanTasksResponse fetchScanTasksResponse;
+    try {
+      fetchScanTasksResponse =
+          icebergTableOperationDispatcher.fetchScanTasks(context, tableIdentifier, request);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new IcebergFetchScanTasksFailureEvent(context, gravitinoNameIdentifier, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(new IcebergFetchScanTasksEvent(context, gravitinoNameIdentifier));
+    return fetchScanTasksResponse;
+  }
+
   @Override
   public Optional<String> getTableMetadataLocation(
       IcebergRequestContext context, TableIdentifier tableIdentifier) {

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

@@ -38,9 +38,11 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -197,6 +199,23 @@ public PlanTableScanResponse planTableScan(
     return dispatcher.planTableScan(context, tableIdentifier, scanRequest);
   }
 
+  /**
+   * Fetch the scan tasks for a {@code plan-task} returned by a prior scan plan. Read-only, so no
+   * hooks are needed and the call is passed straight through.
+   *
+   * @param context Iceberg REST request context information.
+   * @param tableIdentifier The Iceberg table identifier.
+   * @param request The request carrying the {@code plan-task}.
+   * @return A FetchScanTasksResponse containing the scan tasks for that plan task.
+   */
+  @Override
+  public FetchScanTasksResponse fetchScanTasks(
+      IcebergRequestContext context,
+      TableIdentifier tableIdentifier,
+      FetchScanTasksRequest request) {
+    return dispatcher.fetchScanTasks(context, tableIdentifier, request);
+  }
+
   @Override
   public Optional<String> getTableMetadataLocation(
       IcebergRequestContext context, TableIdentifier tableIdentifier) {

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

@@ -24,9 +24,11 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -135,6 +137,20 @@ PlanTableScanResponse planTableScan(
       TableIdentifier tableIdentifier,
       PlanTableScanRequest scanRequest);
 
+  /**
+   * Fetch the scan tasks for a {@code plan-task} returned by a prior {@link #planTableScan} call,
+   * completing the second step of the Iceberg REST two-step scan planning protocol.
+   *
+   * @param context Iceberg REST request context information.
+   * @param tableIdentifier The Iceberg table identifier.
+   * @param request The request carrying the {@code plan-task}.
+   * @return A {@link FetchScanTasksResponse} containing the scan tasks for that plan task.
+   */
+  FetchScanTasksResponse fetchScanTasks(
+      IcebergRequestContext context,
+      TableIdentifier tableIdentifier,
+      FetchScanTasksRequest request);
+
   /**
    * Retrieves the metadata file location for a table without loading full table metadata. This is
    * an optional fast path for catalogs that support cheap metadata location retrieval.

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

@@ -41,9 +41,11 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -231,6 +233,16 @@ public PlanTableScanResponse planTableScan(
         .planTableScan(tableIdentifier, scanRequest, context.requestCredentialVending(), privilege);
   }
 
+  @Override
+  public FetchScanTasksResponse fetchScanTasks(
+      IcebergRequestContext context,
+      TableIdentifier tableIdentifier,
+      FetchScanTasksRequest request) {
+    return icebergCatalogWrapperManager
+        .getCatalogWrapper(context.catalogName())
+        .fetchScanTasks(tableIdentifier, request);
+  }
+
   @Override
   public Optional<String> getTableMetadataLocation(
       IcebergRequestContext context, TableIdentifier tableIdentifier) {

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

@@ -75,9 +75,11 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.RESTUtil;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.ReportMetricsRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -526,6 +528,9 @@ public Response planTableScan(
       @Encoded() @PathParam("table") @AuthorizationMetadata(type = EntityType.TABLE) String table,
       PlanTableScanRequest scanRequest,
       @HeaderParam(X_ICEBERG_ACCESS_DELEGATION) String accessDelegation) {
+    if (scanRequest == null) {
+      return missingRequestBody("plan table scan");
+    }
     boolean isCredentialVending = isCredentialVending(accessDelegation);
     String catalogName = IcebergRESTUtils.getCatalogName(prefix);
     Namespace icebergNS =
@@ -559,6 +564,68 @@ public Response planTableScan(
     }
   }
 
+  /**
+   * Fetch scan tasks endpoint. Completes the second step of the Iceberg REST two-step scan planning
+   * protocol: a client exchanges a {@code plan-task} handed out by {@code POST
+   * .../tables/{table}/plan} for the scan tasks it covers.
+   *
+   * @param prefix The catalog prefix
+   * @param namespace The namespace
+   * @param table The table name
+   * @param request The request containing the {@code plan-task}
+   * @return Response containing the scan tasks for the given plan task
+   */
+  @POST
+  @Path("{table}/tasks")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Timed(name = "fetch-scan-tasks." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "fetch-scan-tasks", absolute = true)
+  @AuthorizationExpression(
+      expression =
+          "ANY(OWNER, METALAKE, CATALOG) || "
+              + "SCHEMA_OWNER_WITH_USE_CATALOG || "
+              + "ANY_USE_CATALOG && ANY_USE_SCHEMA && (TABLE::OWNER || ANY_SELECT_TABLE || ANY_MODIFY_TABLE)",
+      accessMetadataType = MetadataObject.Type.TABLE)
+  public Response fetchScanTasks(
+      @PathParam("prefix") @AuthorizationMetadata(type = EntityType.CATALOG) String prefix,
+      @Encoded() @PathParam("namespace") @AuthorizationMetadata(type = EntityType.SCHEMA)
+          String namespace,
+      @Encoded() @PathParam("table") @AuthorizationMetadata(type = EntityType.TABLE) String table,
+      FetchScanTasksRequest request) {
+    if (request == null) {
+      return missingRequestBody("fetch scan tasks");
+    }
+    String catalogName = IcebergRESTUtils.getCatalogName(prefix);
+    Namespace icebergNS =
+        RESTUtil.decodeNamespace(namespace, IcebergRESTUtils.NAMESPACE_SEPARATOR_URLENCODED_UTF_8);
+    String tableName = RESTUtil.decodeString(table);
+    LOG.info(
+        "Fetch scan tasks, catalog: {}, namespace: {}, table: {}, planTask: {}",
+        catalogName,
+        icebergNS,
+        tableName,
+        request.planTask());
+
+    try {
+      return Utils.doAs(
+          httpRequest,
+          () -> {
+            TableIdentifier tableIdentifier = TableIdentifier.of(icebergNS, tableName);
+            IcebergRequestContext context =
+                new IcebergRequestContext(httpServletRequest(), catalogName);
+
+            FetchScanTasksResponse response =
+                tableOperationDispatcher.fetchScanTasks(context, tableIdentifier, request);
+
+            return IcebergRESTUtils.ok(response);
+          });
+    } catch (Exception e) {
+      LOG.error("Failed to fetch scan tasks: {}", e.getMessage(), e);
+      return IcebergExceptionMapper.toRESTResponse(e);
+    }
+  }
+
   /**
    * Filters the {@link LoadTableResponse} to include only snapshots that are directly referenced by
    * the table's refs (branches and tags). This implements the {@code snapshots=refs} query
@@ -586,6 +653,16 @@ static LoadTableResponse filterSnapshotsByRefs(LoadTableResponse loadTableRespon
         .build();
   }
 
+  /**
+   * Builds a 400 response for a request that arrived without a body. Jersey passes a {@code null}
+   * entity when the body is absent, which would otherwise surface as a 500 from a downstream NPE
+   * even though the Iceberg REST specification expects a 400 for a malformed request.
+   */
+  private static Response missingRequestBody(String operation) {
+    return IcebergExceptionMapper.toRESTResponse(
+        new IllegalArgumentException("Missing request body for " + operation));
+  }
+
   private static Response buildResponseWithETag(LoadTableResponse loadTableResponse) {
     return IcebergRESTUtils.buildResponseWithETag(loadTableResponse);
   }