[SPARK-57057][SQL] Target a specific branch on SELECT / INSERT via temporal clause and session config

common/utils/src/main/resources/error/error-conditions.json

@@ -1445,6 +1445,12 @@
     ],
     "sqlState" : "22023"
   },
+  "CREATE_BRANCH_WITH_IF_NOT_EXISTS_AND_REPLACE" : {
+    "message" : [
+      "ALTER TABLE <tableName> CREATE BRANCH <branchName> with both IF NOT EXISTS and OR REPLACE is not allowed."
+    ],
+    "sqlState" : "42601"
+  },
   "CREATE_PERMANENT_VIEW_WITHOUT_ALIAS" : {
     "message" : [
       "Not allowed to create the permanent view <name> without explicitly assigning an alias for the expression <attr>."

docs/sql-ref-ansi-compliance.md

@@ -437,6 +437,8 @@ Below is a list of all the keywords in Spark SQL.
 |BINDING|non-reserved|non-reserved|non-reserved|
 |BOOLEAN|non-reserved|non-reserved|reserved|
 |BOTH|reserved|non-reserved|reserved|
+|BRANCH|non-reserved|non-reserved|non-reserved|
+|BRANCHES|non-reserved|non-reserved|non-reserved|
 |BUCKET|non-reserved|non-reserved|non-reserved|
 |BUCKETS|non-reserved|non-reserved|non-reserved|
 |BY|non-reserved|non-reserved|reserved|
@@ -546,6 +548,7 @@ Below is a list of all the keywords in Spark SQL.
 |EXTERNAL|non-reserved|non-reserved|reserved|
 |EXTRACT|non-reserved|non-reserved|reserved|
 |FALSE|reserved|non-reserved|reserved|
+|FAST|non-reserved|non-reserved|non-reserved|
 |FETCH|reserved|non-reserved|reserved|
 |FIELDS|non-reserved|non-reserved|non-reserved|
 |FILTER|reserved|non-reserved|reserved|
@@ -558,6 +561,7 @@ Below is a list of all the keywords in Spark SQL.
 |FOREIGN|reserved|non-reserved|reserved|
 |FORMAT|non-reserved|non-reserved|non-reserved|
 |FORMATTED|non-reserved|non-reserved|non-reserved|
+|FORWARD|non-reserved|non-reserved|non-reserved|
 |FOUND|non-reserved|non-reserved|non-reserved|
 |FROM|reserved|non-reserved|reserved|
 |FULL|reserved|strict-non-reserved|reserved|

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4

@@ -156,6 +156,8 @@ BINARY: 'BINARY';
 BINDING: 'BINDING';
 BOOLEAN: 'BOOLEAN';
 BOTH: 'BOTH';
+BRANCH: 'BRANCH';
+BRANCHES: 'BRANCHES';
 BUCKET: 'BUCKET';
 BUCKETS: 'BUCKETS';
 BY: 'BY';
@@ -264,6 +266,7 @@ EXTENDED: 'EXTENDED';
 EXTERNAL: 'EXTERNAL';
 EXTRACT: 'EXTRACT';
 FALSE: 'FALSE';
+FAST: 'FAST';
 FETCH: 'FETCH';
 FIELDS: 'FIELDS';
 FILTER: 'FILTER';
@@ -276,6 +279,7 @@ FOR: 'FOR';
 FOREIGN: 'FOREIGN';
 FORMAT: 'FORMAT';
 FORMATTED: 'FORMATTED';
+FORWARD: 'FORWARD';
 FOUND: 'FOUND';
 FROM: 'FROM';
 FULL: 'FULL';

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -316,6 +316,14 @@ statement
     | ALTER TABLE identifierReference
         DROP CONSTRAINT (IF EXISTS)? name=identifier
         (RESTRICT | CASCADE)?                                     #dropTableConstraint
+    | ALTER TABLE identifierReference
+        CREATE (OR REPLACE)? BRANCH (IF errorCapturingNot EXISTS)?
+        branch=identifier
+        (AS OF VERSION snapshotId=INTEGER_VALUE)?                      #createBranch
+    | ALTER TABLE identifierReference
+        DROP BRANCH (IF EXISTS)? branch=identifier                     #dropBranch
+    | ALTER TABLE identifierReference
+        FAST FORWARD branch=identifier TO target=identifier            #fastForwardBranch
     | DROP TABLE (IF EXISTS)? identifierReference PURGE?               #dropTable
     | DROP VIEW (IF EXISTS)? identifierReference                       #dropView
     | CREATE (OR REPLACE)? (GLOBAL? TEMPORARY)?
@@ -374,6 +382,7 @@ statement
     | SHOW VIEWS ((FROM | IN) identifierReference)?
         (LIKE? pattern=stringLit)?                                        #showViews
     | SHOW PARTITIONS identifierReference partitionSpec?               #showPartitions
+    | SHOW BRANCHES (FROM | IN) identifierReference                    #showBranches
     | SHOW functionScope=simpleIdentifier? FUNCTIONS ((FROM | IN) ns=identifierReference)?
         (LIKE? (legacy=multipartIdentifier | pattern=stringLit))?      #showFunctions
     | SHOW PROCEDURES ((FROM | IN) identifierReference)?               #showProcedures
@@ -587,11 +596,11 @@ query
     ;
 
 insertInto
-    : INSERT (WITH SCHEMA EVOLUTION)? OVERWRITE TABLE? identifierReference optionsClause? (partitionSpec (IF errorCapturingNot EXISTS)?)?  ((BY NAME) | identifierList)? #insertOverwriteTable
-    | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference optionsClause? partitionSpec? (IF errorCapturingNot EXISTS)? ((BY NAME) | identifierList)?   #insertIntoTable
-    | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference tableAlias optionsClause? (BY NAME)?
+    : INSERT (WITH SCHEMA EVOLUTION)? OVERWRITE TABLE? identifierReference temporalClause? optionsClause? (partitionSpec (IF errorCapturingNot EXISTS)?)?  ((BY NAME) | identifierList)? #insertOverwriteTable
+    | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference temporalClause? optionsClause? partitionSpec? (IF errorCapturingNot EXISTS)? ((BY NAME) | identifierList)?   #insertIntoTable
+    | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference temporalClause? tableAlias optionsClause? (BY NAME)?
         REPLACE (WHERE | ON) replaceCondition=booleanExpression        #insertIntoReplaceBooleanCond
-    | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference tableAlias optionsClause? (BY NAME)?
+    | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference temporalClause? tableAlias optionsClause? (BY NAME)?
         REPLACE USING identifierList                                   #insertIntoReplaceUsing
     | INSERT OVERWRITE LOCAL? DIRECTORY path=stringLit rowFormat? createFileFormat?                     #insertOverwriteHiveDir
     | INSERT OVERWRITE LOCAL? DIRECTORY (path=stringLit)? tableProvider (OPTIONS options=propertyList)? #insertOverwriteDir
@@ -921,8 +930,10 @@ fromClause
     ;
 
 temporalClause
-    : FOR? (SYSTEM_VERSION | VERSION) AS OF version
+    : FOR? (SYSTEM_VERSION | VERSION) AS OF BRANCH branch=stringLit
+    | FOR? (SYSTEM_VERSION | VERSION) AS OF version
     | FOR? (SYSTEM_TIME | TIMESTAMP) AS OF timestamp=valueExpression
+    | FOR BRANCH branch=stringLit
     ;
 
 changesClause
@@ -1957,6 +1968,8 @@ ansiNonReserved
     | BINARY_HEX
     | BINDING
     | BOOLEAN
+    | BRANCH
+    | BRANCHES
     | BUCKET
     | BUCKETS
     | BY
@@ -2042,6 +2055,7 @@ ansiNonReserved
     | EXTENDED
     | EXTERNAL
     | EXTRACT
+    | FAST
     | FIELDS
     | FILEFORMAT
     | FIRST
@@ -2050,6 +2064,7 @@ ansiNonReserved
     | FOLLOWING
     | FORMAT
     | FORMATTED
+    | FORWARD
     | FOUND
     | FUNCTION
     | FUNCTIONS
@@ -2340,6 +2355,8 @@ nonReserved
     | BINDING
     | BOOLEAN
     | BOTH
+    | BRANCH
+    | BRANCHES
     | BUCKET
     | BUCKETS
     | BY
@@ -2447,6 +2464,7 @@ nonReserved
     | EXTERNAL
     | EXTRACT
     | FALSE
+    | FAST
     | FETCH
     | FILTER
     | FIELDS
@@ -2459,6 +2477,7 @@ nonReserved
     | FOREIGN
     | FORMAT
     | FORMATTED
+    | FORWARD
     | FROM
     | FOUND
     | FUNCTION

sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala

@@ -712,6 +712,16 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
       ctx)
   }
 
+  def createBranchWithBothIfNotExistsAndReplaceError(
+      tableName: String,
+      branchName: String,
+      ctx: ParserRuleContext): Throwable = {
+    new ParseException(
+      errorClass = "CREATE_BRANCH_WITH_IF_NOT_EXISTS_AND_REPLACE",
+      messageParameters = Map("tableName" -> tableName, "branchName" -> branchName),
+      ctx)
+  }
+
   def temporaryViewWithSchemaBindingMode(ctx: StatementContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE.TEMPORARY_VIEW_WITH_SCHEMA_BINDING_MODE",

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java

@@ -0,0 +1,132 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import java.util.OptionalLong;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A mix-in interface for {@link Table} branching support. Data sources can implement this
+ * interface to expose multi-branch capabilities (such as Iceberg branches) through standard
+ * Spark SQL DDL:
+ *
+ * <pre>
+ *   ALTER TABLE t CREATE [OR REPLACE] BRANCH [IF NOT EXISTS] name [AS OF VERSION snapshotId]
+ *   ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ *   ALTER TABLE t FAST FORWARD branch TO target
+ *   SHOW BRANCHES IN t
+ * </pre>
+ *
+ * <p>The meaning of {@code snapshotId} is left to the data source. When a snapshot id is not
+ * supplied the implementation should branch from the current snapshot of the table.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface SupportsBranching extends Table {
+
+  /**
+   * Create a new branch on this table.
+   *
+   * @param name the branch name; must not be {@code null}
+   * @param sourceSnapshotId an optional snapshot id to branch from. If empty, the branch is
+   *                         created from the table's current snapshot.
+   * @return a {@link TableBranch} describing the new branch
+   * @throws BranchAlreadyExistsException if a branch with the given name already exists
+   */
+  TableBranch createBranch(String name, OptionalLong sourceSnapshotId);
+
+  /**
+   * Replace an existing branch (or create it if missing).
+   *
+   * <p>The default implementation drops the branch (if present) and then creates it again.
+   * Implementations may override this to perform the replacement atomically.
+   */
+  default TableBranch replaceBranch(String name, OptionalLong sourceSnapshotId) {
+    dropBranch(name);
+    return createBranch(name, sourceSnapshotId);
+  }
+
+  /**
+   * Drop the named branch.
+   *
+   * @return {@code true} if the branch was removed; {@code false} if it did not exist.
+   */
+  boolean dropBranch(String name);
+
+  /**
+   * Fast-forward {@code branch} to the head of {@code targetBranch}.
+   *
+   * <p>The operation succeeds only when the branch's current snapshot is an ancestor of the
+   * target branch's current snapshot. Implementations should throw
+   * {@link InvalidFastForwardException} otherwise.
+   *
+   * @return the updated {@link TableBranch} after fast-forwarding
+   * @throws BranchNotFoundException if either branch is missing
+   * @throws InvalidFastForwardException if the operation would not be a fast-forward
+   */
+  TableBranch fastForward(String branch, String targetBranch);
+
+  /**
+   * List all branches of this table. The default implementation returns an empty array.
+   */
+  default TableBranch[] listBranches() {
+    return new TableBranch[0];
+  }
+
+  /**
+   * Return a {@link Table} view that targets the named branch. Reads issued against the returned
+   * table see the rows currently on that branch; writes commit into that branch.
+   *
+   * <p>The returned table shares schema with the parent table. Implementations are free to return
+   * the same instance and remember the branch via internal state, or to return a wrapper.
+   *
+   * @throws BranchNotFoundException if the branch does not exist
+   */
+  Table loadBranch(String branchName);
+
+  /**
+   * Thrown when a branch with the requested name already exists.
+   */
+  class BranchAlreadyExistsException extends RuntimeException {
+    public BranchAlreadyExistsException(String branchName) {
+      super("Branch already exists: " + branchName);
+    }
+  }
+
+  /**
+   * Thrown when an operation references a branch that does not exist on the table.
+   */
+  class BranchNotFoundException extends RuntimeException {
+    public BranchNotFoundException(String branchName) {
+      super("Branch not found: " + branchName);
+    }
+  }
+
+  /**
+   * Thrown by {@link #fastForward(String, String)} when the requested update would not be a
+   * fast-forward, i.e. the branch's current snapshot is not an ancestor of the target
+   * branch's current snapshot.
+   */
+  class InvalidFastForwardException extends RuntimeException {
+    public InvalidFastForwardException(String message) {
+      super(message);
+    }
+  }
+}