sunxiaojian commented on code in PR #9050:
URL: https://github.com/apache/gravitino/pull/9050#discussion_r2564447105
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java:
##########
@@ -192,6 +206,186 @@ private Credential getCredential(LoadTableResponse
loadTableResponse) {
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: {}, select: {},
caseSensitive: {}",
+ tableIdentifier,
+ scanRequest.snapshotId(),
+ scanRequest.select(),
+ scanRequest.caseSensitive());
+
+ try {
+ Table table = catalog.loadTable(tableIdentifier);
+ if (table == null) {
+ throw new NoSuchTableException("Table not found: %s", tableIdentifier);
+ }
+
+ TableScan tableScan = table.newScan();
+ tableScan = applyScanRequest(tableScan, scanRequest);
+
+ List<String> planTasks = new ArrayList<>();
+ Map<Integer, PartitionSpec> specsById = new HashMap<>();
+ List<org.apache.iceberg.DeleteFile> deleteFiles = new ArrayList<>();
+
+ try (CloseableIterable<FileScanTask> fileScanTasks =
tableScan.planFiles()) {
+ 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) {
+ LOG.warn(
+ "Failed to serialize scan task for table: {}, skipping task.
Error: {}",
+ tableIdentifier,
+ e.getMessage());
+ }
+ }
+ } 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);
+ }
+
+ if (!planTasks.isEmpty() && specsById.isEmpty()) {
+ LOG.error(
+ "Internal error: planTasks is not empty ({} tasks) but specsById
is empty for table: {}",
+ planTasks.size(),
+ tableIdentifier);
+ throw new IllegalStateException("Scan planning produced tasks but no
partition specs");
+ }
+
+ 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);
+ }
+
+ PlanTableScanResponse response = responseBuilder.build();
+
+ String snapshotInfo =
+ scanRequest.snapshotId() != null ?
String.valueOf(scanRequest.snapshotId()) : "current";
+ LOG.info(
+ "Successfully planned {} scan tasks for table: {}, snapshot: {}",
+ planTasks.size(),
+ tableIdentifier,
+ snapshotInfo);
+
+ return response;
+
+ } 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);
+ }
+ }
+
+ private TableScan applyScanRequest(TableScan tableScan, PlanTableScanRequest
scanRequest) {
Review Comment:
@FANNG1 When startSnapshotId and endSnapshotId are present, use
IncrementalAppendScan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
