Core: REST Scan Planning Task Implementation

build.gradle

@@ -377,6 +377,7 @@ project(':iceberg-core') {
     implementation libs.jackson.databind
     implementation libs.caffeine
     implementation libs.roaringbitmap
+    implementation libs.failsafe
     compileOnly(libs.hadoop3.client) {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.slf4j', module: 'slf4j-log4j12'

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

@@ -35,7 +35,7 @@
 
 /** Context object with optional arguments for a TableScan. */
 @Value.Immutable
-abstract class TableScanContext {
+public abstract class TableScanContext {
 
   @Nullable
   public abstract Long snapshotId();

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

@@ -26,6 +26,8 @@
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchPlanIdException;
+import org.apache.iceberg.exceptions.NoSuchPlanTaskException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
@@ -73,6 +75,14 @@ public static Consumer<ErrorResponse> tableCommitHandler() {
     return CommitErrorHandler.INSTANCE;
   }
 
+  public static Consumer<ErrorResponse> planErrorHandler() {
+    return PlanErrorHandler.INSTANCE;
+  }
+
+  public static Consumer<ErrorResponse> planTaskHandler() {
+    return PlanTaskErrorHandler.INSTANCE;
+  }
+
   public static Consumer<ErrorResponse> defaultErrorHandler() {
     return DefaultErrorHandler.INSTANCE;
   }
@@ -125,6 +135,46 @@ public void accept(ErrorResponse error) {
     }
   }
 
+  /** Plan level error handler. */
+  private static class PlanErrorHandler extends DefaultErrorHandler {
+    private static final ErrorHandler INSTANCE = new PlanErrorHandler();
+
+    @Override
+    public void accept(ErrorResponse error) {
+      if (error.code() == 404) {
+        if (NoSuchNamespaceException.class.getSimpleName().equals(error.type())) {
+          throw new NoSuchNamespaceException("%s", error.message());
+        } else if (NoSuchTableException.class.getSimpleName().equals(error.type())) {
+          throw new NoSuchTableException("%s", error.message());
+        } else {
+          throw new NoSuchPlanIdException("%s", error.message());
+        }
+      }
+
+      super.accept(error);
+    }
+  }
+
+  /** PlanTask level error handler. */
+  private static class PlanTaskErrorHandler extends DefaultErrorHandler {
+    private static final ErrorHandler INSTANCE = new PlanTaskErrorHandler();
+
+    @Override
+    public void accept(ErrorResponse error) {
+      if (error.code() == 404) {
+        if (NoSuchNamespaceException.class.getSimpleName().equals(error.type())) {
+          throw new NoSuchNamespaceException("%s", error.message());
+        } else if (NoSuchTableException.class.getSimpleName().equals(error.type())) {
+          throw new NoSuchTableException("%s", error.message());
+        } else {
+          throw new NoSuchPlanTaskException("%s", error.message());
+        }
+      }
+
+      super.accept(error);
+    }
+  }
+
   /** View commit error handler. */
   private static class ViewCommitErrorHandler extends DefaultErrorHandler {
     private static final ErrorHandler INSTANCE = new ViewCommitErrorHandler();

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

@@ -37,6 +37,10 @@ private RESTCatalogProperties() {}
 
   public static final String NAMESPACE_SEPARATOR = "namespace-separator";
 
+  // Enable planning on the REST server side
+  public static final String REST_SCAN_PLANNING_ENABLED = "rest-scan-planning-enabled";
+  public static final boolean REST_SCAN_PLANNING_ENABLED_DEFAULT = false;
+
   public enum SnapshotMode {
     ALL,
     REFS

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

@@ -41,6 +41,7 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.Transactions;
 import org.apache.iceberg.catalog.BaseViewSessionCatalog;
@@ -158,6 +159,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
   private MetricsReporter reporter = null;
   private boolean reportingViaRestEnabled;
   private Integer pageSize = null;
+  private boolean restScanPlanningEnabled;
   private CloseableGroup closeables = null;
   private Set<Endpoint> endpoints;
   private Supplier<Map<String, String>> mutationHeaders = Map::of;
@@ -270,6 +272,11 @@ public void initialize(String name, Map<String, String> unresolved) {
             RESTCatalogProperties.NAMESPACE_SEPARATOR,
             RESTUtil.NAMESPACE_SEPARATOR_URLENCODED_UTF_8);
 
+    this.restScanPlanningEnabled =
+        PropertyUtil.propertyAsBoolean(
+            mergedProps,
+            RESTCatalogProperties.REST_SCAN_PLANNING_ENABLED,
+            RESTCatalogProperties.REST_SCAN_PLANNING_ENABLED_DEFAULT);
     super.initialize(name, mergedProps);
   }
 
@@ -476,6 +483,13 @@ public Table loadTable(SessionContext context, TableIdentifier identifier) {
 
     trackFileIO(ops);
 
+    RESTTable restTable = restTableForScanPlanning(ops, finalIdentifier, tableClient);
+    // RestTable should be only be returned for non-metadata tables, because client would
+    // not have access to metadata files for example manifests, since all it needs is catalog.
+    if (restTable != null) {
+      return restTable;
+    }
+
     BaseTable table =
         new BaseTable(
             ops,
@@ -488,6 +502,23 @@ public Table loadTable(SessionContext context, TableIdentifier identifier) {
     return table;
   }
 
+  private RESTTable restTableForScanPlanning(
+      TableOperations ops, TableIdentifier finalIdentifier, RESTClient restClient) {
+    // server supports remote planning endpoint and server / client wants to do server side planning
+    if (endpoints.contains(Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN) && restScanPlanningEnabled) {
+      return new RESTTable(
+          ops,
+          fullTableName(finalIdentifier),
+          metricsReporter(paths.metrics(finalIdentifier), restClient),
+          restClient,
+          Map::of,
+          finalIdentifier,
+          paths,
+          endpoints);
+    }
+    return null;
+  }
+
   private void trackFileIO(RESTTableOperations ops) {
     if (io != ops.io()) {
       fileIOTracker.track(ops);
@@ -556,6 +587,11 @@ public Table registerTable(
 
     trackFileIO(ops);
 
+    RESTTable restTable = restTableForScanPlanning(ops, ident, tableClient);
+    if (restTable != null) {
+      return restTable;
+    }
+
     return new BaseTable(
         ops, fullTableName(ident), metricsReporter(paths.metrics(ident), tableClient));
   }
@@ -820,6 +856,11 @@ public Table create() {
 
       trackFileIO(ops);
 
+      RESTTable restTable = restTableForScanPlanning(ops, ident, tableClient);
+      if (restTable != null) {
+        return restTable;
+      }
+
       return new BaseTable(
           ops, fullTableName(ident), metricsReporter(paths.metrics(ident), tableClient));
     }

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

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.rest;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.ImmutableTableScanContext;
+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 {
+  private final RESTClient client;
+  private final Supplier<Map<String, String>> headers;
+  private final MetricsReporter reporter;
+  private final ResourcePaths resourcePaths;
+  private final TableIdentifier tableIdentifier;
+  private final Set<Endpoint> supportedEndpoints;
+
+  RESTTable(
+      TableOperations ops,
+      String name,
+      MetricsReporter reporter,
+      RESTClient client,
+      Supplier<Map<String, String>> headers,
+      TableIdentifier tableIdentifier,
+      ResourcePaths resourcePaths,
+      Set<Endpoint> supportedEndpoints) {
+    super(ops, name, reporter);
+    this.reporter = reporter;
+    this.client = client;
+    this.headers = headers;
+    this.tableIdentifier = tableIdentifier;
+    this.resourcePaths = resourcePaths;
+    this.supportedEndpoints = supportedEndpoints;
+  }
+
+  @Override
+  public TableScan newScan() {
+    return new RESTTableScan(
+        this,
+        schema(),
+        ImmutableTableScanContext.builder().metricsReporter(reporter).build(),
+        client,
+        headers.get(),
+        operations(),
+        tableIdentifier,
+        resourcePaths,
+        supportedEndpoints);
+  }
+}