feat(plugin-iceberg): Add DDL statements for CREATE BRANCH

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/utils/StatementUtils.java

@@ -21,6 +21,7 @@
 import com.facebook.presto.sql.tree.Analyze;
 import com.facebook.presto.sql.tree.Call;
 import com.facebook.presto.sql.tree.Commit;
+import com.facebook.presto.sql.tree.CreateBranch;
 import com.facebook.presto.sql.tree.CreateFunction;
 import com.facebook.presto.sql.tree.CreateMaterializedView;
 import com.facebook.presto.sql.tree.CreateRole;
@@ -131,6 +132,7 @@ private StatementUtils() {}
         builder.put(CreateType.class, QueryType.DATA_DEFINITION);
         builder.put(AddColumn.class, QueryType.DATA_DEFINITION);
         builder.put(CreateTable.class, QueryType.DATA_DEFINITION);
+        builder.put(CreateBranch.class, QueryType.DATA_DEFINITION);
         builder.put(RenameTable.class, QueryType.DATA_DEFINITION);
         builder.put(RenameColumn.class, QueryType.DATA_DEFINITION);
         builder.put(DropColumn.class, QueryType.DATA_DEFINITION);

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

@@ -1580,6 +1580,22 @@ Alter table operations are supported in the Iceberg connector::
 
      ALTER TABLE iceberg.web.page_views DROP TAG 'tag1';
 
+     ALTER TABLE iceberg.default.mytable CREATE BRANCH 'audit-branch';
+
+     ALTER TABLE iceberg.default.mytable CREATE BRANCH IF NOT EXISTS 'audit-branch';
+
+     ALTER TABLE iceberg.default.mytable CREATE OR REPLACE BRANCH 'audit-branch';
+
+     ALTER TABLE iceberg.default.mytable CREATE BRANCH 'audit-branch-system' FOR SYSTEM_VERSION AS OF 4176642711908913940;
+
+     ALTER TABLE iceberg.default.mytable CREATE BRANCH IF NOT EXISTS 'audit-branch-system' FOR SYSTEM_VERSION AS OF 4176642711908913940;
+
+     ALTER TABLE iceberg.default.mytable CREATE BRANCH 'audit-branch-retain' FOR SYSTEM_VERSION AS OF 4176642711908913940 RETAIN 7 DAYS;
+
+     ALTER TABLE iceberg.default.mytable CREATE BRANCH 'audit-branch-snap-retain' FOR SYSTEM_VERSION AS OF 4176642711908913940 RETAIN 7 DAYS WITH SNAPSHOT RETENTION 2 SNAPSHOTS 2 DAYS;
+
+     ALTER TABLE iceberg.default.mytable CREATE OR REPLACE BRANCH 'audit-branch-time' FOR SYSTEM_TIME AS OF TIMESTAMP '2026-01-02 17:30:35.247 Asia/Kolkata';
+
 To add a new column as a partition column, identify the transform functions for the column.
 The table is partitioned by the transformed value of the column::
 

presto-docs/src/main/sphinx/sql/alter-table.rst

@@ -17,6 +17,11 @@ Synopsis
     ALTER TABLE [ IF EXISTS ] name SET PROPERTIES (property_name=value, [, ...])
     ALTER TABLE [ IF EXISTS ] name DROP BRANCH [ IF EXISTS ] branch_name
     ALTER TABLE [ IF EXISTS ] name DROP TAG [ IF EXISTS ] tag_name
+    ALTER TABLE [ IF EXISTS ] name CREATE [ OR REPLACE ] BRANCH [ IF NOT EXISTS ] branch_name
+    ALTER TABLE [ IF EXISTS ] name CREATE [ OR REPLACE ] BRANCH [ IF NOT EXISTS ] branch_name FOR SYSTEM_VERSION AS OF version
+    ALTER TABLE [ IF EXISTS ] name CREATE [ OR REPLACE ] BRANCH [ IF NOT EXISTS ] branch_name FOR SYSTEM_TIME AS OF timestamp
+    ALTER TABLE [ IF EXISTS ] name CREATE [ OR REPLACE ] BRANCH [ IF NOT EXISTS ] branch_name FOR SYSTEM_VERSION AS OF version RETAIN retention_period
+    ALTER TABLE [ IF EXISTS ] name CREATE [ OR REPLACE ] BRANCH [ IF NOT EXISTS ] branch_name FOR SYSTEM_VERSION AS OF version RETAIN retention_period WITH SNAPSHOT RETENTION min_snapshots SNAPSHOTS min_retention_period
 
 Description
 -----------
@@ -29,6 +34,12 @@ The optional ``IF EXISTS`` (when used before the column name) clause causes the
 
 The optional ``IF NOT EXISTS`` clause causes the error to be suppressed if the column already exists.
 
+For ``CREATE BRANCH`` statements:
+
+* The optional ``OR REPLACE`` clause causes the branch to be replaced if it already exists.
+* The optional ``IF NOT EXISTS`` clause causes the error to be suppressed if the branch already exists.
+* ``OR REPLACE`` and ``IF NOT EXISTS`` cannot be specified together.
+
 Examples
 --------
 
@@ -104,6 +115,38 @@ Drop tag ``tag1`` from the ``users`` table::
 
     ALTER TABLE users DROP TAG 'tag1';
 
+Create branch ``branch1`` from the ``users`` table::
+
+    ALTER TABLE users CREATE BRANCH 'branch1';
+
+Create branch ``branch1`` from the ``users`` table only if it doesn't already exist::
+
+    ALTER TABLE users CREATE BRANCH IF NOT EXISTS 'branch1';
+
+Create or replace branch ``branch1`` from the ``users`` table::
+
+    ALTER TABLE users CREATE OR REPLACE BRANCH 'branch1';
+
+Create branch ``branch1`` from the ``users`` table for system version as of version 5::
+
+    ALTER TABLE users CREATE BRANCH 'branch1' FOR SYSTEM_VERSION AS OF 5;
+
+Create branch ``branch1`` from the ``users`` table for system version as of version 5, only if it doesn't exist::
+
+    ALTER TABLE users CREATE BRANCH IF NOT EXISTS 'branch1' FOR SYSTEM_VERSION AS OF 5;
+
+Create or replace branch ``branch1`` from the ``users`` table for system time as of timestamp '2026-01-02 17:30:35.247 Asia/Kolkata'::
+
+    ALTER TABLE users CREATE OR REPLACE BRANCH 'branch1' FOR SYSTEM_TIME AS OF TIMESTAMP '2026-01-02 17:30:35.247 Asia/Kolkata';
+
+Create branch ``branch1`` from the ``users`` table for system version as of version 5 with retention period of 30 days::
+
+    ALTER TABLE users CREATE BRANCH 'branch1' FOR SYSTEM_VERSION AS OF 5 RETAIN INTERVAL 30 DAY;
+
+Create branch ``branch1`` from the ``users`` table for system version as of version 5 with snapshot retention of minimum 3 snapshots and minimum retention period of 7 days::
+
+    ALTER TABLE users CREATE BRANCH 'branch1' FOR SYSTEM_VERSION AS OF 5 RETAIN INTERVAL 7 DAY WITH SNAPSHOT RETENTION 3 SNAPSHOTS INTERVAL 7 DAYS;
+
 See Also
 --------
 

presto-hive/src/main/java/com/facebook/presto/hive/security/LegacyAccessControl.java

@@ -301,6 +301,11 @@ public void checkCanDropBranch(ConnectorTransactionHandle transactionHandle, Con
     {
     }
 
+    @Override
+    public void checkCanCreateBranch(ConnectorTransactionHandle transactionHandle, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName)
+    {
+    }
+
     @Override
     public void checkCanDropTag(ConnectorTransactionHandle transactionHandle, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName)
     {

presto-hive/src/main/java/com/facebook/presto/hive/security/SqlStandardAccessControl.java

@@ -57,6 +57,7 @@
 import static com.facebook.presto.spi.security.AccessDeniedException.denyAddColumn;
 import static com.facebook.presto.spi.security.AccessDeniedException.denyAddConstraint;
 import static com.facebook.presto.spi.security.AccessDeniedException.denyCallProcedure;
+import static com.facebook.presto.spi.security.AccessDeniedException.denyCreateBranch;
 import static com.facebook.presto.spi.security.AccessDeniedException.denyCreateRole;
 import static com.facebook.presto.spi.security.AccessDeniedException.denyCreateSchema;
 import static com.facebook.presto.spi.security.AccessDeniedException.denyCreateTable;
@@ -269,6 +270,24 @@ public void checkCanDropBranch(ConnectorTransactionHandle transaction, Connector
         }
     }
 
+    @Override
+    public void checkCanCreateBranch(ConnectorTransactionHandle transaction, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName)
+    {
+        MetastoreContext metastoreContext = new MetastoreContext(
+                identity, context.getQueryId().getId(),
+                context.getClientInfo(),
+                context.getClientTags(),
+                context.getSource(),
+                Optional.empty(),
+                false,
+                HiveColumnConverterProvider.DEFAULT_COLUMN_CONVERTER_PROVIDER,
+                context.getWarningCollector(),
+                context.getRuntimeStats());
+        if (!isTableOwner(transaction, identity, metastoreContext, tableName)) {
+            denyCreateBranch(tableName.toString());
+        }
+    }
+
     @Override
     public void checkCanDropTag(ConnectorTransactionHandle transaction, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName)
     {

presto-hive/src/main/java/com/facebook/presto/hive/security/SystemTableAwareAccessControl.java

@@ -300,6 +300,12 @@ public void checkCanDropBranch(ConnectorTransactionHandle transactionHandle, Con
         delegate.checkCanDropBranch(transactionHandle, identity, context, tableName);
     }
 
+    @Override
+    public void checkCanCreateBranch(ConnectorTransactionHandle transactionHandle, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName)
+    {
+        delegate.checkCanCreateBranch(transactionHandle, identity, context, tableName);
+    }
+
     @Override
     public void checkCanDropTag(ConnectorTransactionHandle transactionHandle, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName)
     {

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

@@ -100,6 +100,7 @@
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileMetadata;
 import org.apache.iceberg.IsolationLevel;
+import org.apache.iceberg.ManageSnapshots;
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.MetricsModes.None;
@@ -243,6 +244,7 @@
 import static com.google.common.collect.Maps.transformValues;
 import static java.lang.Long.parseLong;
 import static java.lang.String.format;
+import static java.time.Duration.ofDays;
 import static java.util.Collections.singletonList;
 import static java.util.Objects.requireNonNull;
 import static org.apache.iceberg.MetadataColumns.ROW_POSITION;
@@ -1046,6 +1048,50 @@ public void dropBranch(ConnectorSession session, ConnectorTableHandle tableHandl
         }
     }
 
+    @Override
+    public void createBranch(
+            ConnectorSession session,
+            ConnectorTableHandle tableHandle,
+            String branchName,
+            boolean replace,
+            boolean ifNotExists,
+            Optional<ConnectorTableVersion> tableVersion,
+            Optional<Long> retainDays,
+            Optional<Integer> minSnapshotsToKeep,
+            Optional<Long> maxSnapshotAgeDays)
+    {
+        IcebergTableHandle icebergTableHandle = (IcebergTableHandle) tableHandle;
+        verify(icebergTableHandle.getIcebergTableName().getTableType() == DATA, "only the data table can have branch created");
+        Table icebergTable = getIcebergTable(session, icebergTableHandle.getSchemaTableName());
+
+        boolean branchExists = icebergTable.refs().containsKey(branchName);
+        if (ifNotExists && branchExists) {
+            return;
+        }
+        long targetSnapshotId = tableVersion.map(version -> getSnapshotIdForTableVersion(icebergTable, version))
+                .orElseGet(() -> {
+                    if (icebergTable.currentSnapshot() == null) {
+                        throw new PrestoException(NOT_FOUND, format("Table %s has no current snapshot", icebergTableHandle.getSchemaTableName().getTableName()));
+                    }
+                    return icebergTable.currentSnapshot().snapshotId();
+                });
+        ManageSnapshots manageSnapshots = icebergTable.manageSnapshots();
+        if (replace && branchExists) {
+            manageSnapshots.replaceBranch(branchName, targetSnapshotId);
+        }
+        else if (!branchExists) {
+            manageSnapshots.createBranch(branchName, targetSnapshotId);
+        }
+        else {
+            throw new PrestoException(ALREADY_EXISTS, format("Branch %s already exists in table %s", branchName, icebergTableHandle.getSchemaTableName().getTableName()));
+        }
+        // Apply retention policies if specified
+        retainDays.ifPresent(retainDs -> manageSnapshots.setMaxRefAgeMs(branchName, ofDays(retainDs).toMillis()));
+        minSnapshotsToKeep.ifPresent(minSnapshots -> manageSnapshots.setMinSnapshotsToKeep(branchName, minSnapshots));
+        maxSnapshotAgeDays.ifPresent(maxAgeDays -> manageSnapshots.setMaxSnapshotAgeMs(branchName, ofDays(maxAgeDays).toMillis()));
+        manageSnapshots.commit();
+    }
+
     @Override
     public void dropTag(ConnectorSession session, ConnectorTableHandle tableHandle, String tagName, boolean tagExists)
     {