Address review feedback

Author: sfc-gh-prsingh

core/src/main/java/org/apache/iceberg/RestTableScan.java

@@ -36,7 +36,7 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.ParallelIterable;
 
-public class RestTableScan extends DataTableScan {
+class RestTableScan extends DataTableScan {
   private final RESTClient client;
   private final String path;
   private final Supplier<Map<String, String>> headers;
@@ -48,7 +48,7 @@ public class RestTableScan extends DataTableScan {
   // TODO revisit if this property should be configurable
   private static final int FETCH_PLANNING_SLEEP_DURATION_MS = 1000;
 
-  public RestTableScan(
+  RestTableScan(
       Table table,
       Schema schema,
       TableScanContext context,
@@ -85,6 +85,9 @@ protected TableScan newRefinedScan(
 
   @Override
   public CloseableIterable<FileScanTask> planFiles() {
+    Long startSnapshotId = context().fromSnapshotId();
+    Long endSnapshotId = context().toSnapshotId();
+    Long snapshotId = snapshotId();
     List<String> selectedColumns =
         schema().columns().stream().map(Types.NestedField::name).collect(Collectors.toList());
 
@@ -96,10 +99,6 @@ public CloseableIterable<FileScanTask> planFiles() {
               .collect(Collectors.toList());
     }
 
-    Long startSnapshotId = context().fromSnapshotId();
-    Long endSnapshotId = context().toSnapshotId();
-    Long snapshotId = snapshotId();
-
     PlanTableScanRequest.Builder planTableScanRequestBuilder =
         new PlanTableScanRequest.Builder()
             .withSelect(selectedColumns)
@@ -127,12 +126,12 @@ public CloseableIterable<FileScanTask> planFiles() {
   }
 
   private CloseableIterable<FileScanTask> planTableScan(PlanTableScanRequest planTableScanRequest) {
-    // we need to inject specById map here and also the caseSensitive
     ParserContext context =
         ParserContext.builder()
             .add("specsById", table.specs())
             .add("caseSensitive", context().caseSensitive())
             .build();
+
     PlanTableScanResponse response =
         client.post(
             resourcePaths.planTableScan(tableIdentifier),
@@ -146,13 +145,15 @@ private CloseableIterable<FileScanTask> planTableScan(PlanTableScanRequest planT
     PlanStatus planStatus = response.planStatus();
     switch (planStatus) {
       case COMPLETED:
-        // List<FileScanTask> fileScanTasks = bindFileScanTasksWithSpec(response.fileScanTasks());
         return getScanTasksIterable(response.planTasks(), response.fileScanTasks());
       case SUBMITTED:
         return fetchPlanningResult(response.planId());
       case FAILED:
-        throw new RuntimeException(
+        throw new IllegalStateException(
             "Received \"failed\" status from service when planning a table scan");
+      case CANCELLED:
+        throw new IllegalStateException(
+            "Received \"cancelled\" status from service when planning a table scan");
       default:
         throw new RuntimeException(
             String.format("Invalid planStatus during planTableScan: %s", planStatus));
@@ -181,9 +182,10 @@ private CloseableIterable<FileScanTask> fetchPlanningResult(String planId) {
       PlanStatus planStatus = response.planStatus();
       switch (planStatus) {
         case COMPLETED:
-          // List<FileScanTask> fileScanTasks = bindFileScanTasksWithSpec(response.fileScanTasks());
           return getScanTasksIterable(response.planTasks(), response.fileScanTasks());
         case SUBMITTED:
+          // TODO: Think more about whether we should use a backoff strategy here.
+          // For now, we will just sleep for a fixed duration before checking the status again.
           try {
             Thread.sleep(FETCH_PLANNING_SLEEP_DURATION_MS);
           } catch (InterruptedException e) {

core/src/main/java/org/apache/iceberg/ScanTasksIterable.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg;
 
 import java.io.IOException;
+import java.util.ArrayDeque;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -43,7 +44,7 @@ public class ScanTasksIterable implements CloseableIterable<FileScanTask> {
   // parallelizing on this where a planTask produces a list of file scan tasks, as
   //  well more planTasks.
   private final String planTask;
-  private final List<FileScanTask> fileScanTasks;
+  private final ArrayDeque<FileScanTask> fileScanTasks;
   private final ExecutorService executorService;
   private final Map<Integer, PartitionSpec> specsById;
   private final boolean caseSensitive;
@@ -78,7 +79,7 @@ public ScanTasksIterable(
       Map<Integer, PartitionSpec> specsById,
       boolean caseSensitive) {
     this.planTask = null;
-    this.fileScanTasks = fileScanTasks;
+    this.fileScanTasks = new ArrayDeque<>(fileScanTasks);
     this.client = client;
     this.resourcePaths = resourcePaths;
     this.tableIdentifier = tableIdentifier;
@@ -111,14 +112,14 @@ private static class ScanTasksIterator implements CloseableIterator<FileScanTask
     private final TableIdentifier tableIdentifier;
     private final Supplier<Map<String, String>> headers;
     private String planTask;
-    private final List<FileScanTask> fileScanTasks;
+    private final ArrayDeque<FileScanTask> fileScanTasks;
     private final ExecutorService executorService;
     private final Map<Integer, PartitionSpec> specsById;
     private final boolean caseSensitive;
 
     ScanTasksIterator(
         String planTask,
-        List<FileScanTask> fileScanTasks,
+        ArrayDeque<FileScanTask> fileScanTasks,
         RESTClient client,
         ResourcePaths resourcePaths,
         TableIdentifier tableIdentifier,
@@ -131,7 +132,7 @@ private static class ScanTasksIterator implements CloseableIterator<FileScanTask
       this.tableIdentifier = tableIdentifier;
       this.headers = headers;
       this.planTask = planTask;
-      this.fileScanTasks = fileScanTasks != null ? fileScanTasks : Lists.newArrayList();
+      this.fileScanTasks = fileScanTasks != null ? fileScanTasks : new ArrayDeque<>();
       this.executorService = executorService;
       this.specsById = specsById;
       this.caseSensitive = caseSensitive;
@@ -158,7 +159,7 @@ public boolean hasNext() {
 
     @Override
     public FileScanTask next() {
-      return fileScanTasks.remove(0);
+      return fileScanTasks.removeFirst();
     }
 
     private void fetchScanTasks(String withPlanTask) {