Add support for DML with properties in engine + Add support for branching in Iceberg

core/trino-grammar/src/main/antlr4/io/trino/grammar/sql/SqlBase.g4

@@ -71,8 +71,10 @@ statement
          (COMMENT string)?
          (WITH properties)?                                            #createTable
     | DROP TABLE (IF EXISTS)? qualifiedName                            #dropTable
-    | INSERT INTO qualifiedName columnAliases? rootQuery               #insertInto
-    | DELETE FROM qualifiedName (WHERE booleanExpression)?             #delete
+    | INSERT INTO qualifiedName ('@' branch=string)?
+       columnAliases? rootQuery                                        #insertInto
+    | DELETE FROM qualifiedName ('@' branch=string)?
+         (WHERE booleanExpression)?                                    #delete
     | TRUNCATE TABLE qualifiedName                                     #truncateTable
     | COMMENT ON TABLE qualifiedName IS (string | NULL)                #commentTable
     | COMMENT ON VIEW qualifiedName IS (string | NULL)                 #commentView
@@ -192,10 +194,10 @@ statement
     | DESCRIBE OUTPUT identifier                                       #describeOutput
     | SET PATH pathSpecification                                       #setPath
     | SET TIME ZONE (LOCAL | expression)                               #setTimeZone
-    | UPDATE qualifiedName
+    | UPDATE qualifiedName ('@' branch=string)?
         SET updateAssignment (',' updateAssignment)*
         (WHERE where=booleanExpression)?                               #update
-    | MERGE INTO qualifiedName (AS? identifier)?
+    | MERGE INTO qualifiedName (AS? identifier)? ('@' branch=string)?
         USING relation ON expression mergeCase+                        #merge
     ;
 

core/trino-main/src/main/java/io/trino/security/AccessControl.java

@@ -37,6 +37,8 @@
 import java.util.Optional;
 import java.util.Set;
 
+import static io.trino.spi.security.AccessDeniedException.denyCreateBranchAndTag;
+import static io.trino.spi.security.AccessDeniedException.denyDropBranchAndTag;
 import static io.trino.spi.security.AccessDeniedException.denySetViewAuthorization;
 
 public interface AccessControl
@@ -620,4 +622,24 @@ default Map<ColumnSchema, ViewExpression> getColumnMasks(SecurityContext context
     {
         return ImmutableMap.of();
     }
+
+    /**
+     * Check if identity is allowed to create the specified branch or tag.
+     *
+     * @throws AccessDeniedException if not allowed
+     */
+    default void checkCanCreateBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        denyCreateBranchAndTag(tableName.toString(), name);
+    }
+
+    /**
+     * Check if identity is allowed to drop the specified branch or tag.
+     *
+     * @throws AccessDeniedException if not allowed
+     */
+    default void canCanDropBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        denyDropBranchAndTag(tableName.toString(), name);
+    }
 }

core/trino-main/src/main/java/io/trino/security/AccessControlManager.java

@@ -1456,6 +1456,32 @@ private ConnectorAccessControl getConnectorAccessControl(TransactionId transacti
         return connectorAccessControl;
     }
 
+    @Override
+    public void checkCanCreateBranchAndTag(SecurityContext securityContext, QualifiedObjectName tableName, String name)
+    {
+        requireNonNull(securityContext, "securityContext is null");
+        requireNonNull(tableName, "tableName is null");
+
+        checkCanAccessCatalog(securityContext, tableName.catalogName());
+
+        systemAuthorizationCheck(control -> control.checkCanCreateBranchAndTag(securityContext.toSystemSecurityContext(), tableName.asCatalogSchemaTableName(), name));
+
+        catalogAuthorizationCheck(tableName.catalogName(), securityContext, (control, context) -> control.checkCanCreateBranchAndTag(context, tableName.asSchemaTableName(), name));
+    }
+
+    @Override
+    public void canCanDropBranchAndTag(SecurityContext securityContext, QualifiedObjectName tableName, String name)
+    {
+        requireNonNull(securityContext, "securityContext is null");
+        requireNonNull(tableName, "tableName is null");
+
+        checkCanAccessCatalog(securityContext, tableName.catalogName());
+
+        systemAuthorizationCheck(control -> control.checkCanDropBranchAndTag(securityContext.toSystemSecurityContext(), tableName.asCatalogSchemaTableName(), name));
+
+        catalogAuthorizationCheck(tableName.catalogName(), securityContext, (control, context) -> control.checkCanDropBranchAndTag(context, tableName.asSchemaTableName(), name));
+    }
+
     @Managed
     @Nested
     public CounterStat getAuthorizationSuccess()

core/trino-main/src/main/java/io/trino/security/AllowAllAccessControl.java

@@ -291,4 +291,10 @@ public void checkCanDropFunction(SecurityContext context, QualifiedObjectName fu
 
     @Override
     public void checkCanShowCreateFunction(SecurityContext context, QualifiedObjectName functionName) {}
+
+    @Override
+    public void checkCanCreateBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name) {}
+
+    @Override
+    public void canCanDropBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name) {}
 }

core/trino-main/src/main/java/io/trino/security/DenyAllAccessControl.java

@@ -39,6 +39,7 @@
 import static io.trino.spi.security.AccessDeniedException.denyCommentColumn;
 import static io.trino.spi.security.AccessDeniedException.denyCommentTable;
 import static io.trino.spi.security.AccessDeniedException.denyCommentView;
+import static io.trino.spi.security.AccessDeniedException.denyCreateBranchAndTag;
 import static io.trino.spi.security.AccessDeniedException.denyCreateCatalog;
 import static io.trino.spi.security.AccessDeniedException.denyCreateFunction;
 import static io.trino.spi.security.AccessDeniedException.denyCreateMaterializedView;
@@ -51,6 +52,7 @@
 import static io.trino.spi.security.AccessDeniedException.denyDenyEntityPrivilege;
 import static io.trino.spi.security.AccessDeniedException.denyDenySchemaPrivilege;
 import static io.trino.spi.security.AccessDeniedException.denyDenyTablePrivilege;
+import static io.trino.spi.security.AccessDeniedException.denyDropBranchAndTag;
 import static io.trino.spi.security.AccessDeniedException.denyDropCatalog;
 import static io.trino.spi.security.AccessDeniedException.denyDropColumn;
 import static io.trino.spi.security.AccessDeniedException.denyDropFunction;
@@ -575,4 +577,16 @@ public void checkCanShowCreateFunction(SecurityContext context, QualifiedObjectN
     {
         denyShowCreateFunction(functionName.toString());
     }
+
+    @Override
+    public void checkCanCreateBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        denyCreateBranchAndTag(tableName.toString(), name);
+    }
+
+    @Override
+    public void canCanDropBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        denyDropBranchAndTag(tableName.toString(), name);
+    }
 }

core/trino-main/src/main/java/io/trino/security/ForwardingAccessControl.java

@@ -535,4 +535,16 @@ public Map<ColumnSchema, ViewExpression> getColumnMasks(SecurityContext context,
     {
         return delegate().getColumnMasks(context, tableName, columns);
     }
+
+    @Override
+    public void checkCanCreateBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        delegate().checkCanCreateBranchAndTag(context, tableName, name);
+    }
+
+    @Override
+    public void canCanDropBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        delegate().canCanDropBranchAndTag(context, tableName, name);
+    }
 }

core/trino-main/src/main/java/io/trino/security/InjectedConnectorAccessControl.java

@@ -534,6 +534,20 @@ public Map<ColumnSchema, ViewExpression> getColumnMasks(ConnectorSecurityContext
         throw new TrinoException(NOT_SUPPORTED, "Column masking not supported");
     }
 
+    @Override
+    public void checkCanCreateBranchAndTag(ConnectorSecurityContext context, SchemaTableName tableName, String name)
+    {
+        checkArgument(context == null, "context must be null");
+        accessControl.checkCanCreateBranchAndTag(securityContext, new QualifiedObjectName(catalogName, tableName.getSchemaName(), tableName.getTableName()), name);
+    }
+
+    @Override
+    public void checkCanDropBranchAndTag(ConnectorSecurityContext context, SchemaTableName tableName, String name)
+    {
+        checkArgument(context == null, "context must be null");
+        accessControl.canCanDropBranchAndTag(securityContext, new QualifiedObjectName(catalogName, tableName.getSchemaName(), tableName.getTableName()), name);
+    }
+
     private QualifiedObjectName getQualifiedObjectName(SchemaTableName schemaTableName)
     {
         return new QualifiedObjectName(catalogName, schemaTableName.getSchemaName(), schemaTableName.getTableName());

core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java

@@ -301,6 +301,7 @@
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.collect.Iterables.getLast;
 import static com.google.common.collect.Iterables.getOnlyElement;
+import static io.airlift.slice.Slices.utf8Slice;
 import static io.trino.SystemSessionProperties.getMaxGroupingSets;
 import static io.trino.metadata.FunctionResolver.toPath;
 import static io.trino.metadata.GlobalFunctionCatalog.isBuiltinFunctionName;
@@ -363,6 +364,7 @@
 import static io.trino.spi.StandardErrorCode.VIEW_IS_RECURSIVE;
 import static io.trino.spi.StandardErrorCode.VIEW_IS_STALE;
 import static io.trino.spi.connector.MaterializedViewFreshness.Freshness.FRESH;
+import static io.trino.spi.connector.PointerType.TARGET_ID;
 import static io.trino.spi.connector.StandardWarningCode.REDUNDANT_ORDER_BY;
 import static io.trino.spi.function.FunctionKind.AGGREGATE;
 import static io.trino.spi.function.FunctionKind.WINDOW;
@@ -580,7 +582,8 @@ protected Scope visitInsert(Insert insert, Optional<Scope> scope)
             Scope queryScope = analyze(insert.getQuery(), Optional.empty(), false);
 
             // verify the insert destination columns match the query
-            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, targetTable);
+            Optional<TableVersion> endVersion = insert.getTable().getBranch().map(branch -> new TableVersion(TARGET_ID, VARCHAR, utf8Slice(branch)));
+            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, targetTable, Optional.empty(), endVersion);
             Optional<TableHandle> targetTableHandle = redirection.tableHandle();
             targetTable = redirection.redirectedTableName().orElse(targetTable);
             if (targetTableHandle.isEmpty()) {
@@ -821,7 +824,8 @@ protected Scope visitDelete(Delete node, Optional<Scope> scope)
                 throw semanticException(NOT_SUPPORTED, node, "Deleting from views is not supported");
             }
 
-            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName);
+            Optional<TableVersion> endVersion = node.getTable().getBranch().map(branch -> new TableVersion(TARGET_ID, VARCHAR, utf8Slice(branch)));
+            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName, Optional.empty(), endVersion);
             QualifiedObjectName tableName = redirection.redirectedTableName().orElse(originalName);
             TableHandle handle = redirection.tableHandle()
                     .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, table, "Table '%s' does not exist", tableName));
@@ -3368,7 +3372,8 @@ protected Scope visitUpdate(Update update, Optional<Scope> scope)
 
             analysis.setUpdateType("UPDATE");
 
-            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName);
+            Optional<TableVersion> endVersion = update.getTable().getBranch().map(branch -> new TableVersion(TARGET_ID, VARCHAR, utf8Slice(branch)));
+            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName, Optional.empty(), endVersion);
             QualifiedObjectName tableName = redirection.redirectedTableName().orElse(originalName);
             TableHandle handle = redirection.tableHandle()
                     .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, table, "Table '%s' does not exist", tableName));
@@ -3498,7 +3503,8 @@ protected Scope visitMerge(Merge merge, Optional<Scope> scope)
 
             analysis.setUpdateType("MERGE");
 
-            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalTableName);
+            Optional<TableVersion> endVersion = merge.getTargetTable().getBranch().map(branch -> new TableVersion(TARGET_ID, VARCHAR, utf8Slice(branch)));
+            RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalTableName, Optional.empty(), endVersion);
             QualifiedObjectName tableName = redirection.redirectedTableName().orElse(originalTableName);
             TableHandle targetTableHandle = redirection.tableHandle()
                     .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, table, "Table '%s' does not exist", tableName));
@@ -5854,11 +5860,17 @@ private OutputColumn createOutputColumn(Field field)
         private RedirectionAwareTableHandle getTableHandle(Table table, QualifiedObjectName name, Optional<Scope> scope)
         {
             if (table.getQueryPeriod().isPresent()) {
+                verify(table.getBranch().isEmpty(), "branch must be empty");
                 Optional<TableVersion> startVersion = extractTableVersion(table, table.getQueryPeriod().get().getStart(), scope);
                 Optional<TableVersion> endVersion = extractTableVersion(table, table.getQueryPeriod().get().getEnd(), scope);
                 return metadata.getRedirectionAwareTableHandle(session, name, startVersion, endVersion);
             }
-            return metadata.getRedirectionAwareTableHandle(session, name);
+            if (table.getBranch().isPresent()) {
+                verify(table.getQueryPeriod().isEmpty(), "query period must be empty");
+                Optional<TableVersion> endVersion = table.getBranch().map(branch -> new TableVersion(TARGET_ID, VARCHAR, utf8Slice(branch)));
+                return metadata.getRedirectionAwareTableHandle(session, name, Optional.empty(), endVersion);
+            }
+            return metadata.getRedirectionAwareTableHandle(session, name, Optional.empty(), Optional.empty());
         }
 
         /**

core/trino-main/src/main/java/io/trino/tracing/TracingAccessControl.java

@@ -782,6 +782,24 @@ public Map<ColumnSchema, ViewExpression> getColumnMasks(SecurityContext context,
         }
     }
 
+    @Override
+    public void checkCanCreateBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        Span span = startSpan("checkCanCreateBranchAndTag");
+        try (var _ = scopedSpan(span)) {
+            delegate.checkCanCreateBranchAndTag(context, tableName, name);
+        }
+    }
+
+    @Override
+    public void canCanDropBranchAndTag(SecurityContext context, QualifiedObjectName tableName, String name)
+    {
+        Span span = startSpan("canCanDropBranchAndTag");
+        try (var _ = scopedSpan(span)) {
+            delegate.canCanDropBranchAndTag(context, tableName, name);
+        }
+    }
+
     private Span startSpan(String methodName)
     {
         return tracer.spanBuilder("AccessControl." + methodName)

core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java

@@ -1006,7 +1006,10 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
     @Override
     public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
     {
-        throw new UnsupportedOperationException();
+        if (startVersion.isPresent() || endVersion.isPresent()) {
+            throw new UnsupportedOperationException();
+        }
+        return noRedirection(getTableHandle(session, tableName));
     }
 
     @Override

core/trino-parser/src/main/java/io/trino/sql/SqlFormatter.java

@@ -1080,6 +1080,8 @@ protected Void visitMerge(Merge node, Integer indent)
             builder.append("MERGE INTO ")
                     .append(formatName(node.getTargetTable().getName()));
 
+            node.getTargetTable().getBranch().ifPresent(branch -> builder.append("@").append(branch));
+
             node.getTargetAlias().ifPresent(value -> builder
                     .append(' ')
                     .append(formatName(value)));
@@ -1464,6 +1466,8 @@ protected Void visitDelete(Delete node, Integer indent)
             builder.append("DELETE FROM ")
                     .append(formatName(node.getTable().getName()));
 
+            node.getTable().getBranch().ifPresent(branch -> builder.append("@").append(branch));
+
             node.getWhere().ifPresent(where -> builder
                     .append(" WHERE ")
                     .append(formatExpression(where)));
@@ -1898,6 +1902,8 @@ protected Void visitInsert(Insert node, Integer indent)
             builder.append("INSERT INTO ")
                     .append(formatName(node.getTarget()));
 
+            node.getTable().getBranch().ifPresent(branch -> builder.append("@").append(branch));
+
             node.getColumns().ifPresent(columns -> builder
                     .append(" (")
                     .append(Joiner.on(", ").join(columns))
@@ -1914,8 +1920,12 @@ protected Void visitInsert(Insert node, Integer indent)
         protected Void visitUpdate(Update node, Integer indent)
         {
             builder.append("UPDATE ")
-                    .append(formatName(node.getTable().getName()))
-                    .append(" SET");
+                    .append(formatName(node.getTable().getName()));
+
+            node.getTable().getBranch().ifPresent(branch -> builder.append("@").append(branch));
+
+            builder.append(" SET");
+
             int setCounter = node.getAssignments().size() - 1;
             for (UpdateAssignment assignment : node.getAssignments()) {
                 builder.append("\n")