Core: Move decision about remote scan planning from Spark to Core

Author: nastra

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

@@ -144,6 +144,11 @@ protected PlanningMode deletePlanningMode() {
 
   @Override
   protected CloseableIterable<ScanTask> doPlanFiles() {
+    if (table() instanceof SupportsDistributedScanPlanning
+        && !((SupportsDistributedScanPlanning) table()).allowDistributedPlanning()) {
+      return table().newBatchScan().planFiles();
+    }
+
     Snapshot snapshot = snapshot();
 
     List<ManifestFile> deleteManifests = findMatchingDeleteManifests(snapshot);

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

@@ -38,7 +38,8 @@
  * BaseTable using a {@link StaticTableOperations}. This way no Catalog related calls are needed
  * when reading the table data after deserialization.
  */
-public class BaseTable implements Table, HasTableOperations, Serializable {
+public class BaseTable
+    implements Table, HasTableOperations, Serializable, SupportsDistributedScanPlanning {
   private final TableOperations ops;
   private final String name;
   private final MetricsReporter reporter;

…/iceberg/RequiresRemoteScanPlanning.java …erg/SupportsDistributedScanPlanning.javacore/src/main/java/org/apache/iceberg/RequiresRemoteScanPlanning.java renamed to core/src/main/java/org/apache/iceberg/SupportsDistributedScanPlanning.java core/src/main/java/org/apache/iceberg/RequiresRemoteScanPlanning.java renamed to core/src/main/java/org/apache/iceberg/SupportsDistributedScanPlanning.java

@@ -18,5 +18,9 @@
  */
 package org.apache.iceberg;
 
-/** Marker interface to indicate whether a Table requires remote scan planning */
-public interface RequiresRemoteScanPlanning {}
+/** Marker interface to indicate whether a Table supports distributed scan planning */
+public interface SupportsDistributedScanPlanning {
+  default boolean allowDistributedPlanning() {
+    return true;
+  }
+}

core/src/main/java/org/apache/iceberg/rest/RESTTable.java

@@ -25,13 +25,13 @@
 import org.apache.iceberg.BatchScan;
 import org.apache.iceberg.BatchScanAdapter;
 import org.apache.iceberg.ImmutableTableScanContext;
-import org.apache.iceberg.RequiresRemoteScanPlanning;
+import org.apache.iceberg.SupportsDistributedScanPlanning;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.metrics.MetricsReporter;
 
-class RESTTable extends BaseTable implements RequiresRemoteScanPlanning {
+class RESTTable extends BaseTable implements SupportsDistributedScanPlanning {
   private final RESTClient client;
   private final Supplier<Map<String, String>> headers;
   private final MetricsReporter reporter;
@@ -84,4 +84,9 @@ public TableScan newScan() {
   public BatchScan newBatchScan() {
     return new BatchScanAdapter(newScan());
   }
+
+  @Override
+  public boolean allowDistributedPlanning() {
+    return false;
+  }
 }

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

@@ -34,7 +34,6 @@
 import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.MetricsModes;
 import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.RequiresRemoteScanPlanning;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.SparkDistributedDataScan;
@@ -761,9 +760,7 @@ public StructType readSchema() {
   }
 
   private BatchScan newBatchScan() {
-    if (table instanceof RequiresRemoteScanPlanning) {
-      return table.newBatchScan();
-    } else if (table instanceof BaseTable && readConf.distributedPlanningEnabled()) {
+    if (table instanceof BaseTable && readConf.distributedPlanningEnabled()) {
       return new SparkDistributedDataScan(spark, table, readConf);
     } else {
       return table.newBatchScan();