[ #9047] feat(iceberg):Support submit table scan plan (#9050)

### What changes were proposed in this pull request?

Support submit table scan plan.

### Why are the changes needed?

Fix: #([9047](#9047))

### Does this PR introduce _any_ user-facing change?

N/A

### How was this patch tested?

Author: sunxiaojian

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

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

docs/iceberg-rest-service.md

@@ -22,7 +22,6 @@ There are some key difference between Gravitino Iceberg REST server and Gravitin
 - Supports the Apache Iceberg REST API defined in Iceberg 1.10, and supports all namespace and table interfaces. The following interfaces are not implemented yet:
   - multi table transaction
   - pagination
-  - scan planning
 - Works as a catalog proxy, supporting `Hive` and `JDBC` as catalog backend.
 - Supports credential vending for `S3`、`GCS`、`OSS` and `ADLS`.
 - Supports different storages like `S3`, `HDFS`, `OSS`, `GCS`, `ADLS` and provides the capability to support other storages.

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

@@ -22,8 +22,11 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Stream;
@@ -40,15 +43,27 @@
 import org.apache.gravitino.storage.GCSProperties;
 import org.apache.gravitino.utils.MapUtils;
 import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.IncrementalAppendScan;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Scan;
+import org.apache.iceberg.ScanTaskParser;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.ServiceUnavailableException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.rest.PlanStatus;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 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;
 
 /** Process Iceberg REST specific operations, like credential vending. */
 public class CatalogWrapperForREST extends IcebergCatalogWrapper {
@@ -207,6 +222,211 @@ private Credential getCredential(
     return credential;
   }
 
+  /**
+   * Plan table scan and return scan tasks.
+   *
+   * <p>This method performs server-side scan planning to optimize query performance by reducing
+   * client-side metadata loading and enabling parallel task execution.
+   *
+   * <p>Implementation uses synchronous scan planning (COMPLETED status) where tasks are returned
+   * immediately as serialized JSON strings. This is different from asynchronous mode (SUBMITTED
+   * status) where a plan ID is returned for later retrieval.
+   *
+   * <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.
+   * @return PlanTableScanResponse with status=COMPLETED and serialized planTasks.
+   * @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) {
+
+    LOG.debug(
+        "Planning scan for table: {}, snapshotId: {}, startSnapshotId: {}, endSnapshotId: {}, select: {}, caseSensitive: {}",
+        tableIdentifier,
+        scanRequest.snapshotId(),
+        scanRequest.startSnapshotId(),
+        scanRequest.endSnapshotId(),
+        scanRequest.select(),
+        scanRequest.caseSensitive());
+
+    try {
+      Table table = catalog.loadTable(tableIdentifier);
+      CloseableIterable<FileScanTask> fileScanTasks =
+          createFilePlanScanTasks(table, tableIdentifier, scanRequest);
+
+      List<String> planTasks = new ArrayList<>();
+      Map<Integer, PartitionSpec> specsById = new HashMap<>();
+      List<org.apache.iceberg.DeleteFile> deleteFiles = new ArrayList<>();
+
+      try (fileScanTasks) {
+        for (FileScanTask fileScanTask : fileScanTasks) {
+          try {
+            String taskString = ScanTaskParser.toJson(fileScanTask);
+            planTasks.add(taskString);
+
+            int specId = fileScanTask.spec().specId();
+            if (!specsById.containsKey(specId)) {
+              specsById.put(specId, fileScanTask.spec());
+            }
+
+            if (!fileScanTask.deletes().isEmpty()) {
+              deleteFiles.addAll(fileScanTask.deletes());
+            }
+          } catch (Exception e) {
+            throw new RuntimeException(
+                String.format(
+                    "Failed to serialize scan task 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);
+      }
+
+      List<DeleteFile> uniqueDeleteFiles =
+          deleteFiles.stream().distinct().collect(java.util.stream.Collectors.toList());
+
+      if (planTasks.isEmpty()) {
+        LOG.info(
+            "Scan planning returned no tasks for table: {}. Table may be empty or fully filtered.",
+            tableIdentifier);
+      }
+
+      PlanTableScanResponse.Builder responseBuilder =
+          PlanTableScanResponse.builder()
+              .withPlanStatus(PlanStatus.COMPLETED)
+              .withPlanTasks(planTasks)
+              .withSpecsById(specsById);
+
+      if (!uniqueDeleteFiles.isEmpty()) {
+        responseBuilder.withDeleteFiles(uniqueDeleteFiles);
+        LOG.debug(
+            "Included {} delete files in scan plan for table: {}",
+            uniqueDeleteFiles.size(),
+            tableIdentifier);
+      }
+
+      return responseBuilder.build();
+
+    } catch (IllegalArgumentException e) {
+      LOG.error("Invalid scan request for table {}: {}", tableIdentifier, e.getMessage());
+      throw new IllegalArgumentException("Invalid scan parameters: " + e.getMessage(), e);
+    } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+      LOG.error("Table not found during scan planning: {}", tableIdentifier);
+      throw e;
+    } catch (Exception e) {
+      LOG.error("Unexpected error during scan planning for table: {}", tableIdentifier, e);
+      throw new RuntimeException(
+          "Scan planning failed for table " + tableIdentifier + ": " + e.getMessage(), e);
+    }
+  }
+
+  /**
+   * Create and plan a scan based on the scan request.
+   *
+   * <p>If both start and end snapshot IDs are provided, uses IncrementalAppendScan. Otherwise, uses
+   * regular TableScan.
+   *
+   * @param table The table to scan
+   * @param tableIdentifier The table identifier for logging
+   * @param scanRequest The scan request parameters
+   * @return CloseableIterable of FileScanTask
+   */
+  private CloseableIterable<FileScanTask> createFilePlanScanTasks(
+      Table table, TableIdentifier tableIdentifier, PlanTableScanRequest scanRequest) {
+    Long startSnapshotId = scanRequest.startSnapshotId();
+    Long endSnapshotId = scanRequest.endSnapshotId();
+    // Use IncrementalAppendScan if both start and end snapshot IDs are provided
+    if (startSnapshotId != null && endSnapshotId != null) {
+      if (startSnapshotId >= endSnapshotId) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Invalid snapshot range: startSnapshotId (%d) must be less than endSnapshotId (%d)",
+                startSnapshotId, endSnapshotId));
+      }
+      LOG.debug(
+          "Using IncrementalAppendScan for table: {}, from snapshot: {} to snapshot: {}",
+          tableIdentifier,
+          startSnapshotId,
+          endSnapshotId);
+      IncrementalAppendScan incrementalScan =
+          table
+              .newIncrementalAppendScan()
+              .fromSnapshotInclusive(startSnapshotId)
+              .toSnapshot(endSnapshotId);
+      incrementalScan = applyScanRequest(incrementalScan, scanRequest);
+      return incrementalScan.planFiles();
+    } else {
+      TableScan tableScan = table.newScan();
+      if (scanRequest.snapshotId() != null && scanRequest.snapshotId() != 0L) {
+        tableScan = tableScan.useSnapshot(scanRequest.snapshotId());
+        LOG.debug("Applied snapshot filter: snapshot-id={}", scanRequest.snapshotId());
+      }
+      tableScan = applyScanRequest(tableScan, scanRequest);
+      return tableScan.planFiles();
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T extends Scan> T applyScanRequest(T scan, PlanTableScanRequest scanRequest) {
+    scan = (T) scan.caseSensitive(scanRequest.caseSensitive());
+    LOG.debug("Applied case-sensitive: {}", scanRequest.caseSensitive());
+    scan = applyScanFilter(scan, scanRequest);
+    scan = applyScanSelect(scan, scanRequest);
+    scan = applyScanStatsFields(scan, scanRequest);
+
+    return scan;
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T extends Scan> T applyScanFilter(T scan, PlanTableScanRequest scanRequest) {
+    if (scanRequest.filter() != null) {
+      try {
+        scan = (T) scan.filter(scanRequest.filter());
+        LOG.debug("Applied filter expression: {}", scanRequest.filter());
+      } catch (Exception e) {
+        LOG.error("Failed to apply filter expression: {}", e.getMessage(), e);
+        throw new IllegalArgumentException("Invalid filter expression: " + e.getMessage(), e);
+      }
+    }
+    return scan;
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T extends Scan> T applyScanSelect(T scan, PlanTableScanRequest scanRequest) {
+    if (scanRequest.select() != null && !scanRequest.select().isEmpty()) {
+      try {
+        scan = (T) scan.select(scanRequest.select());
+        LOG.debug("Applied column projection: {}", scanRequest.select());
+      } catch (Exception e) {
+        LOG.error("Failed to apply column projection: {}", e.getMessage(), e);
+        throw new IllegalArgumentException("Invalid column selection: " + e.getMessage(), e);
+      }
+    }
+    return scan;
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T extends Scan> T applyScanStatsFields(T scan, PlanTableScanRequest scanRequest) {
+    if (scanRequest.statsFields() != null && !scanRequest.statsFields().isEmpty()) {
+      try {
+        scan = (T) scan.includeColumnStats(scanRequest.statsFields());
+        LOG.debug("Applied statistics fields: {}", scanRequest.statsFields());
+      } catch (Exception e) {
+        LOG.error("Failed to apply statistics fields: {}", e.getMessage(), e);
+        throw new IllegalArgumentException("Invalid statistics fields: " + e.getMessage(), e);
+      }
+    }
+    return scan;
+  }
+
   @VisibleForTesting
   static Map<String, String> checkForCompatibility(
       Map<String, String> properties, Map<String, String> deprecatedProperties) {

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

@@ -39,6 +39,9 @@
 import org.apache.gravitino.listener.api.event.IcebergLoadTableEvent;
 import org.apache.gravitino.listener.api.event.IcebergLoadTableFailureEvent;
 import org.apache.gravitino.listener.api.event.IcebergLoadTablePreEvent;
+import org.apache.gravitino.listener.api.event.IcebergPlanTableScanEvent;
+import org.apache.gravitino.listener.api.event.IcebergPlanTableScanFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergPlanTableScanPreEvent;
 import org.apache.gravitino.listener.api.event.IcebergRenameTableEvent;
 import org.apache.gravitino.listener.api.event.IcebergRenameTableFailureEvent;
 import org.apache.gravitino.listener.api.event.IcebergRenameTablePreEvent;
@@ -52,11 +55,13 @@
 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.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.PlanTableScanResponse;
 
 /**
  * {@code IcebergTableEventDispatcher} is a decorator for {@link IcebergTableOperationExecutor} that
@@ -263,4 +268,36 @@ public LoadCredentialsResponse getTableCredentials(
     eventBus.dispatchEvent(new IcebergLoadTableCredentialEvent(context, gravitinoNameIdentifier));
     return loadCredentialsResponse;
   }
+
+  /**
+   * Plan table scan and return scan tasks.
+   *
+   * <p>Dispatches pre/post events for table scan planning operation.
+   *
+   * @param context Iceberg REST request context information.
+   * @param tableIdentifier The Iceberg table identifier.
+   * @param scanRequest The scan request parameters
+   * @return A PlanTableScanResponse containing the scan plan with plan-id and tasks
+   */
+  @Override
+  public PlanTableScanResponse planTableScan(
+      IcebergRequestContext context,
+      TableIdentifier tableIdentifier,
+      PlanTableScanRequest scanRequest) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRESTUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), tableIdentifier);
+    eventBus.dispatchEvent(new IcebergPlanTableScanPreEvent(context, gravitinoNameIdentifier));
+    PlanTableScanResponse planTableScanResponse;
+    try {
+      planTableScanResponse =
+          icebergTableOperationDispatcher.planTableScan(context, tableIdentifier, scanRequest);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new IcebergPlanTableScanFailureEvent(context, gravitinoNameIdentifier, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(new IcebergPlanTableScanEvent(context, gravitinoNameIdentifier));
+    return planTableScanResponse;
+  }
 }

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

@@ -35,11 +35,13 @@
 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.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.PlanTableScanResponse;
 
 public class IcebergTableHookDispatcher implements IcebergTableOperationDispatcher {
 
@@ -156,6 +158,27 @@ public LoadCredentialsResponse getTableCredentials(
     return dispatcher.getTableCredentials(context, tableIdentifier);
   }
 
+  /**
+   * Plan table scan and return scan tasks.
+   *
+   * <p>This method performs server-side scan planning to optimize query performance by reducing
+   * client-side metadata loading and enabling parallel task execution.
+   *
+   * @param context Iceberg REST request context information.
+   * @param tableIdentifier The Iceberg table identifier.
+   * @param scanRequest The scan request parameters including filters, projections, snapshot-id,
+   *     etc.
+   * @return A PlanTableScanResponse containing the scan plan with plan-id and tasks. The response
+   *     format follows Iceberg REST API specification.
+   */
+  @Override
+  public PlanTableScanResponse planTableScan(
+      IcebergRequestContext context,
+      TableIdentifier tableIdentifier,
+      PlanTableScanRequest scanRequest) {
+    return dispatcher.planTableScan(context, tableIdentifier, scanRequest);
+  }
+
   private void importTable(String catalogName, Namespace namespace, String tableName) {
     TableDispatcher tableDispatcher = GravitinoEnv.getInstance().tableDispatcher();
     if (tableDispatcher != null) {