Support procedure fast_forward for iceberg

presto-docs/src/main/sphinx/connector/iceberg.rst

@@ -917,6 +917,36 @@ Examples:
 
     CALL iceberg.system.remove_orphan_files(schema => 'db', table_name => 'sample');
 
+Fast Forward Branch
+^^^^^^^^^^^^^^^^^^^
+
+This procedure advances the current snapshot of the specified branch to a more recent snapshot from another branch without replaying any intermediate snapshots.
+``branch`` can be fast-forwarded up to the ``to`` snapshot if ``branch`` is an ancestor of ``to``.
+
+The following arguments are available:
+
+===================== ========== =============== =======================================================================
+Argument Name         required   type            Description
+===================== ========== =============== =======================================================================
+``schema``            ✔️         string          Schema of the table to update
+
+``table_name``        ✔️         string          Name of the table to update
+
+``branch``            ✔️         string          The branch you want to fast-forward
+
+``to``                ✔️         string          The branch you want to fast-forward to
+===================== ========== =============== =======================================================================
+
+Examples:
+
+* Fast-forward the ``dev`` branch to the latest snapshot of the ``main`` branch ::
+
+    CALL iceberg.system.fast_forward('schema_name', 'table_name', 'dev', 'main');
+
+* Given the branch named ``branch1`` does not exist yet, create a new branch named ``branch1``  and set it's current snapshot equal to the latest snapshot of the ``main`` branch ::
+
+    CALL iceberg.system.fast_forward('schema_name', 'table_name', 'branch1', 'main');
+
 SQL Support
 -----------
 

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergCommonModule.java

@@ -40,6 +40,7 @@
 import com.facebook.presto.iceberg.nessie.IcebergNessieConfig;
 import com.facebook.presto.iceberg.optimizer.IcebergPlanOptimizerProvider;
 import com.facebook.presto.iceberg.procedure.ExpireSnapshotsProcedure;
+import com.facebook.presto.iceberg.procedure.FastForwardBranchProcedure;
 import com.facebook.presto.iceberg.procedure.RegisterTableProcedure;
 import com.facebook.presto.iceberg.procedure.RemoveOrphanFiles;
 import com.facebook.presto.iceberg.procedure.RollbackToSnapshotProcedure;
@@ -159,6 +160,7 @@ public void setup(Binder binder)
         procedures.addBinding().toProvider(UnregisterTableProcedure.class).in(Scopes.SINGLETON);
         procedures.addBinding().toProvider(ExpireSnapshotsProcedure.class).in(Scopes.SINGLETON);
         procedures.addBinding().toProvider(RemoveOrphanFiles.class).in(Scopes.SINGLETON);
+        procedures.addBinding().toProvider(FastForwardBranchProcedure.class).in(Scopes.SINGLETON);
         procedures.addBinding().toProvider(SetCurrentSnapshotProcedure.class).in(Scopes.SINGLETON);
 
         // for orc

presto-iceberg/src/main/java/com/facebook/presto/iceberg/procedure/FastForwardBranchProcedure.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed 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 com.facebook.presto.iceberg.procedure;
+
+import com.facebook.presto.iceberg.IcebergMetadataFactory;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.SchemaTableName;
+import com.facebook.presto.spi.connector.ConnectorMetadata;
+import com.facebook.presto.spi.procedure.Procedure;
+import com.google.common.collect.ImmutableList;
+import org.apache.iceberg.Table;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+
+import java.lang.invoke.MethodHandle;
+
+import static com.facebook.presto.common.block.MethodHandleUtil.methodHandle;
+import static com.facebook.presto.common.type.StandardTypes.VARCHAR;
+import static com.facebook.presto.iceberg.IcebergUtil.getIcebergTable;
+import static java.util.Objects.requireNonNull;
+
+public class FastForwardBranchProcedure
+        implements Provider<Procedure>
+{
+    private static final MethodHandle FAST_FORWARD = methodHandle(
+            FastForwardBranchProcedure.class,
+            "fastForwardToBranch",
+            ConnectorSession.class,
+            String.class,
+            String.class,
+            String.class,
+            String.class);
+
+    private final IcebergMetadataFactory metadataFactory;
+
+    @Inject
+    public FastForwardBranchProcedure(IcebergMetadataFactory metadataFactory)
+    {
+        this.metadataFactory = requireNonNull(metadataFactory, "metadataFactory is null");
+    }
+
+    @Override
+    public Procedure get()
+    {
+        return new Procedure(
+                "system",
+                "fast_forward",
+                ImmutableList.of(
+                        new Procedure.Argument("schema", VARCHAR),
+                        new Procedure.Argument("table_name", VARCHAR),
+                        new Procedure.Argument("branch", VARCHAR),
+                        new Procedure.Argument("to", VARCHAR)),
+                FAST_FORWARD.bindTo(this));
+    }
+
+    public void fastForwardToBranch(ConnectorSession clientSession, String schemaName, String tableName, String fromBranch, String targetBranch)
+    {
+        SchemaTableName schemaTableName = new SchemaTableName(schemaName, tableName);
+        ConnectorMetadata metadata = metadataFactory.create();
+        Table icebergTable = getIcebergTable(metadata, clientSession, schemaTableName);
+        icebergTable.manageSnapshots().fastForwardBranch(fromBranch, targetBranch).commit();
+    }
+}