You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "viirya (via GitHub)" <gi...@apache.org> on 2026/05/25 17:58:42 UTC
[PR] [SPARK-57056][SQL] Add SupportsBranching DSv2 interface and branching DDL [spark]
viirya opened a new pull request, #56102:
URL: https://github.com/apache/spark/pull/56102
### What changes were proposed in this pull request?
Add a standard DSv2 mix-in interface `SupportsBranching`, plus the SQL surface to manage branches:
```java
public interface SupportsBranching extends Table {
TableBranch createBranch(String name, OptionalLong sourceSnapshotId);
default TableBranch replaceBranch(String name, OptionalLong sourceSnapshotId);
boolean dropBranch(String name);
TableBranch fastForward(String branch, String targetBranch);
default TableBranch[] listBranches();
}
```
With companion value type `TableBranch(name, snapshotId, creationTimeMs)` and `SupportsBranching.BranchAlreadyExistsException` for the duplicate-create case.
New DDL:
```sql
ALTER TABLE t CREATE [OR REPLACE] BRANCH [IF NOT EXISTS] name [AS OF VERSION <integer>]
ALTER TABLE t DROP BRANCH [IF EXISTS] name
ALTER TABLE t FAST FORWARD branch TO target
SHOW BRANCHES (FROM | IN) t
```
Implementation:
- Define the interface and `TableBranch` value type under `sql/catalyst/.../connector/catalog/`.
- Extend the ANTLR grammar with the four DDL forms; register `BRANCH`, `BRANCHES`, `FAST`, `FORWARD` as non-reserved keywords; update `docs/sql-ref-ansi-compliance.md`.
- Add logical plans (`CreateBranch` / `DropBranch` / `FastForwardBranch` / `ShowBranches`) and exec nodes; dispatch through `ResolvedTable` in `DataSourceV2Strategy`.
- Add `DataSourceV2Implicits.asBranchable` and `QueryCompilationErrors.tableDoesNotSupportBranchingError` so non-branching tables fail with a clear message.
- Add error condition `CREATE_BRANCH_WITH_IF_NOT_EXISTS_AND_REPLACE` for the conflicting clauses.
- Implement `SupportsBranching` on `InMemoryTable` for testing.
Reads and writes against a specific branch (`SELECT ... FOR BRANCH 'x'`, `INSERT ... FOR BRANCH 'x'`) are out of scope here and are added in the follow-up SPARK-57057.
### Why are the changes needed?
Apache Iceberg and similar table formats support named branches as a first-class concept, but Spark today only exposes branching through connector-specific SQL extensions (e.g. `IcebergSparkSessionExtensions`). A standard DSv2 interface lets any data source expose branching through built-in Spark SQL, the same way `SupportsDelete` / `TruncatableTable` / `SupportsRowLevelOperations` standardize their respective capabilities.
### Does this PR introduce _any_ user-facing change?
Yes. New SQL DDL is recognized:
```sql
ALTER TABLE t CREATE [OR REPLACE] BRANCH [IF NOT EXISTS] name [AS OF VERSION <integer>]
ALTER TABLE t DROP BRANCH [IF EXISTS] name
ALTER TABLE t FAST FORWARD branch TO target
SHOW BRANCHES (FROM | IN) t
```
Data sources that do not implement `SupportsBranching` are unaffected — running the new DDL against them raises `AnalysisException` ("does not support branching"). Four new non-reserved keywords (`BRANCH`, `BRANCHES`, `FAST`, `FORWARD`) are added; they remain usable as identifiers in non-DDL contexts.
### How was this patch tested?
- `DDLParserSuite`: 5 new parser tests covering all four DDL forms and the `IF NOT EXISTS` / `OR REPLACE` conflict error.
- `SupportsBranchingSuite`: 12 new integration tests exercising the new DDL end-to-end against `InMemoryTable`, including positive cases for each operation and negative cases for duplicates, missing branches, fast-forward direction, and non-branching tables.
- All 152 existing tests in `DDLParserSuite` still pass; `TableIdentifierParserSuite` (7 tests) confirms the new keywords don't regress identifier handling.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.7)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314860972
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3747,6 +3718,12 @@
],
"sqlState" : "42000"
},
+ "INVALID_FAST_FORWARD" : {
Review Comment:
`INVALID_FAST_FORWARD` -> `INVALID_FASTFORWARD`
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
Review Comment:
Please remove this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335884639
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1697,6 +1697,59 @@ case class RecoverPartitions(child: LogicalPlan) extends UnaryCommand {
copy(child = newChild)
}
+/**
+ * The logical plan of the ALTER TABLE ... CREATE BRANCH command.
+ */
+case class CreateBranch(
+ child: LogicalPlan,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): CreateBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... DROP BRANCH command.
+ */
+case class DropBranch(
+ child: LogicalPlan,
+ branchName: String,
+ ifExists: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): DropBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... FASTFORWARD BRANCH command.
+ */
+case class FastForwardBranch(
+ child: LogicalPlan,
+ branchName: String,
+ targetBranchName: String) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): FastForwardBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the SHOW BRANCHES command.
+ */
+case class ShowBranches(
+ child: LogicalPlan,
+ override val output: Seq[Attribute] = ShowBranches.getOutputAttrs) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): ShowBranches =
+ copy(child = newChild)
+}
+
+object ShowBranches {
+ def getOutputAttrs: Seq[Attribute] = Seq(
+ AttributeReference("name", StringType, nullable = false)(),
+ AttributeReference("snapshot_id", org.apache.spark.sql.types.LongType, nullable = true)(),
+ AttributeReference("creation_time_ms",
+ org.apache.spark.sql.types.LongType, nullable = false)())
Review Comment:
In this context, `TimestampType` instead of `LongType`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335834314
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ dropBranch(name);
+ return createBranch(name, sourceSnapshotId);
Review Comment:
This looks a little too simple because `dropBranch` returns `false` instead of throwing exception. Specifically, how do we guarantee atomicity? For example, two Spark connect sessions invoke `replaceBranch` together? According to this code, a session S1 successfully drops the branch, `B`, but another session S2 can steal the branch `B` with S3's `sourceSnapshotId`.
If we ignore `Atomicity` in `replaceBranch`, we need to clarify that in the interface documentation because it's going to be used with Apache Iceberg eventually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314750938
##########
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
Review Comment:
Agreed — `verb BRANCH name` matches `CREATE BRANCH` / `DROP BRANCH` and reads better.
Changed the grammar to:
```
ALTER TABLE t FASTFORWARD BRANCH <branch> TO <target>
```
and replaced the previous `FAST` / `FORWARD` lexer tokens with a single `FASTFORWARD` token (they were only used by this rule).
One context note: Apache Iceberg's Spark extension uses `ALTER TABLE t FAST FORWARD <branch> TO <target>` (two tokens, no `BRANCH`), so this DDL form diverges from Iceberg's surface. I think the Spark-native form should follow Spark's own `CREATE BRANCH` / `DROP BRANCH` shape rather than Iceberg's, but flagging it in case you'd rather match Iceberg's syntax for ease of migration. Going with `FASTFORWARD BRANCH` for now — happy to switch if you prefer otherwise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314865144
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BranchingExec.scala:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.SupportsBranching
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Physical plan node for ALTER TABLE ... CREATE BRANCH.
+ */
+case class CreateBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val source = sourceSnapshotId match {
+ case Some(id) => OptionalLong.of(id)
+ case None => OptionalLong.empty()
+ }
+ if (replace) {
+ table.replaceBranch(branchName, source)
+ } else {
+ try {
+ table.createBranch(branchName, source)
+ } catch {
+ case _: BranchAlreadyExistsException if ifNotExists =>
+ // ignored: branch already exists and IF NOT EXISTS was requested
+ }
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... DROP BRANCH.
+ */
+case class DropBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ ifExists: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val removed = table.dropBranch(branchName)
+ if (!removed && !ifExists) {
+ throw new BranchNotFoundException(branchName, table.name())
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... FAST FORWARD.
Review Comment:
Please revise this description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314872612
##########
sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala:
##########
@@ -214,7 +214,7 @@ trait ThriftServerWithSparkContextSuite extends SharedThriftServer {
val sessionHandle = client.openSession(user, "")
val infoValue = client.getInfo(sessionHandle, GetInfoType.CLI_ODBC_KEYWORDS)
// scalastyle:off line.size.limit
- assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,BEGIN,BERNOULLI,BETWEEN,BIGINT,BINARY,BINDING,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,END,ENFORCED,ESCAPE,ES
CAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS
,OR,ORDER,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFO
RM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNION,UNIQUE,UNKNOWN,UNLOCK,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE")
+ assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,BEGIN,BERNOULLI,BETWEEN,BIGINT,BINARY,BINDING,BOOLEAN,BOTH,BRANCH,BRANCHES,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,END,EN
FORCED,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FAST,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FORWARD,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSE
T,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRAILING,TRA
NSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNION,UNIQUE,UNKNOWN,UNLOCK,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE")
Review Comment:
ditto. `,FAST,` still exists.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3326925700
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsBranchingSuite.scala:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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
+
+import java.util
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException, InvalidFastForwardException}
+import org.apache.spark.sql.connector.catalog.{Column, Identifier, InMemoryCatalog, SupportsBranching, Table, TableCapability}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+import org.apache.spark.sql.types.StructType
+
+class SupportsBranchingSuite extends QueryTest with DatasourceV2SQLBase {
+
+ private def branchingTable(tableName: String): SupportsBranching = {
+ val cat = catalog("testcat").asInstanceOf[InMemoryCatalog]
+ cat.loadTable(Identifier.of(Array("ns"), tableName)).asInstanceOf[SupportsBranching]
+ }
+
+ private def withBranchingTable(tableName: String)(f: String => Unit): Unit = {
+ val fqName = s"testcat.ns.$tableName"
+ sql(s"CREATE TABLE $fqName (id bigint, data string) USING foo")
+ try {
+ f(fqName)
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $fqName")
+ }
+ }
+
+ test("CREATE BRANCH on an InMemoryTable creates a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val t = branchingTable("t")
+ val branches = t.listBranches()
+ assert(branches.length == 1)
+ assert(branches(0).name == "dev")
+ }
+ }
+
+ test("CREATE BRANCH with explicit snapshot id") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH staging AS OF VERSION 7")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.name == "staging")
+ assert(branch.snapshotId.isPresent && branch.snapshotId.getAsLong == 7L)
+ }
+ }
+
+ test("CREATE BRANCH IF NOT EXISTS is idempotent") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq CREATE BRANCH IF NOT EXISTS dev")
+ assert(branchingTable("t").listBranches().length == 1)
+ }
+ }
+
+ test("CREATE BRANCH without IF NOT EXISTS fails on duplicate") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val ex = intercept[BranchAlreadyExistsException] {
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ }
+ assert(ex.getMessage.contains("dev"))
+ assert(ex.getCondition == "BRANCH_ALREADY_EXISTS")
+ }
+ }
+
+ test("CREATE OR REPLACE BRANCH overwrites an existing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE OR REPLACE BRANCH dev AS OF VERSION 9")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.snapshotId.getAsLong == 9L)
+ }
+ }
+
+ test("DROP BRANCH removes a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq DROP BRANCH dev")
+ assert(branchingTable("t").listBranches().isEmpty)
+ }
+ }
+
+ test("DROP BRANCH IF EXISTS does not fail on missing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq DROP BRANCH IF EXISTS missing")
+ }
+ }
+
+ test("DROP BRANCH fails on missing branch when IF EXISTS not specified") {
+ withBranchingTable("t") { fq =>
+ val ex = intercept[BranchNotFoundException] {
+ sql(s"ALTER TABLE $fq DROP BRANCH missing")
+ }
+ assert(ex.getMessage.contains("missing"))
+ assert(ex.getCondition == "BRANCH_NOT_FOUND")
+ }
+ }
+
+ test("FASTFORWARD BRANCH updates branch to target's snapshot") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 5")
+ sql(s"ALTER TABLE $fq FASTFORWARD BRANCH dev TO main")
+ val dev = branchingTable("t").listBranches().find(_.name == "dev").get
+ assert(dev.snapshotId.getAsLong == 5L)
+ }
+ }
+
+ test("FASTFORWARD BRANCH fails when target is behind") {
Review Comment:
Do we have a test coverage of `BranchNotFoundException` on `FASTFORWARD BRANCH` syntax?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3301226851
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -50,7 +53,50 @@ class InMemoryTable(
override val id: String = UUID.randomUUID().toString)
extends InMemoryBaseTable(name, columns, partitioning, properties, constraints, distribution,
ordering, numPartitions, advisoryPartitionSize, isDistributionStrictlyRequired,
- numRowsPerSplit) with SupportsDelete {
+ numRowsPerSplit) with SupportsDelete with SupportsBranching {
+
+ private val branches = new ConcurrentHashMap[String, TableBranch]()
+
+ override def createBranch(branchName: String,
+ sourceSnapshotId: OptionalLong): TableBranch = {
+ val snapshot = if (sourceSnapshotId.isPresent) sourceSnapshotId.getAsLong
+ else version().toLong
+ val branch = new TableBranch(branchName, snapshot, System.currentTimeMillis())
+ val existing = branches.putIfAbsent(branchName, branch)
+ if (existing != null) {
+ throw new SupportsBranching.BranchAlreadyExistsException(branchName)
+ }
+ branch
+ }
+
+ override def dropBranch(branchName: String): Boolean = {
+ branches.remove(branchName) != null
+ }
+
+ override def fastForward(branchName: String, targetBranchName: String): TableBranch = {
+ val current = branches.get(branchName)
+ if (current == null) {
+ throw new java.util.NoSuchElementException(s"Branch not found: $branchName")
Review Comment:
Agreed — switched to the typed `BranchNotFoundException` introduced in `SupportsBranching` (per your comment on the interface).
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -50,7 +53,50 @@ class InMemoryTable(
override val id: String = UUID.randomUUID().toString)
extends InMemoryBaseTable(name, columns, partitioning, properties, constraints, distribution,
ordering, numPartitions, advisoryPartitionSize, isDistributionStrictlyRequired,
- numRowsPerSplit) with SupportsDelete {
+ numRowsPerSplit) with SupportsDelete with SupportsBranching {
+
+ private val branches = new ConcurrentHashMap[String, TableBranch]()
+
+ override def createBranch(branchName: String,
+ sourceSnapshotId: OptionalLong): TableBranch = {
+ val snapshot = if (sourceSnapshotId.isPresent) sourceSnapshotId.getAsLong
+ else version().toLong
+ val branch = new TableBranch(branchName, snapshot, System.currentTimeMillis())
+ val existing = branches.putIfAbsent(branchName, branch)
+ if (existing != null) {
+ throw new SupportsBranching.BranchAlreadyExistsException(branchName)
+ }
+ branch
+ }
+
+ override def dropBranch(branchName: String): Boolean = {
+ branches.remove(branchName) != null
+ }
+
+ override def fastForward(branchName: String, targetBranchName: String): TableBranch = {
+ val current = branches.get(branchName)
+ if (current == null) {
+ throw new java.util.NoSuchElementException(s"Branch not found: $branchName")
+ }
+ val target = branches.get(targetBranchName)
+ if (target == null) {
+ throw new java.util.NoSuchElementException(s"Branch not found: $targetBranchName")
Review Comment:
Same fix — switching to `BranchNotFoundException`. The `IllegalArgumentException` for the fast-forward direction check is also now `InvalidFastForwardException` for the same reason.
On the DSv2 / Iceberg convention question:
- Spark's DSv2 surface uses typed `NoSuch*Exception` classes (`NoSuchTableException`, `NoSuchPartitionException`, `NoSuchNamespaceException`) for missing-entity errors; the plain JDK `NoSuchElementException` is not used.
- Iceberg doesn't have a standardized DSv2 interface for branching at all — internally it uses `ValidationException` / `IllegalArgumentException` in `ManageSnapshots` / branching procedures, but that's an Iceberg-internal choice rather than a DSv2 convention. Since this PR is proposing the DSv2 interface, I think we should follow Spark's DSv2 naming here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335834314
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ dropBranch(name);
+ return createBranch(name, sourceSnapshotId);
Review Comment:
This looks a little too simple because `dropBranch` returns `false` instead of throwing exception. Specifically, how do we guarantee atomicity? For example, two Spark connect sessions invoke `replaceBranch` together? According to this code, a session S1 successfully drops the branch, `B`, but another session S2 can steal the branch `B` with S3's `sourceSnapshotId`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314883720
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1697,6 +1697,59 @@ case class RecoverPartitions(child: LogicalPlan) extends UnaryCommand {
copy(child = newChild)
}
+/**
+ * The logical plan of the ALTER TABLE ... CREATE BRANCH command.
+ */
+case class CreateBranch(
+ child: LogicalPlan,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): CreateBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... DROP BRANCH command.
+ */
+case class DropBranch(
+ child: LogicalPlan,
+ branchName: String,
+ ifExists: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): DropBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... FASTFORWARD BRANCH command.
+ */
+case class FastForwardBranch(
+ child: LogicalPlan,
+ branchName: String,
+ targetBranchName: String) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): FastForwardBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the SHOW BRANCHES command.
+ */
+case class ShowBranches(
+ child: LogicalPlan,
+ override val output: Seq[Attribute] = ShowBranches.getOutputAttrs) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): ShowBranches =
+ copy(child = newChild)
+}
+
+object ShowBranches {
+ def getOutputAttrs: Seq[Attribute] = Seq(
+ AttributeReference("name", StringType, nullable = false)(),
+ AttributeReference("snapshot_id", org.apache.spark.sql.types.LongType, nullable = true)(),
Review Comment:
Just a question for my understanding: is `snapshot_id` nullable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4815682882
> Could you rebase this PR once more, @viirya ?
Rebased. Thanks @dongjoon-hyun.
This touches the public DSv2 API so it needs some more community consensus. I will leave it open for that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3326964756
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3747,6 +3718,12 @@
],
"sqlState" : "42000"
},
+ "INVALID_FASTFORWARD" : {
+ "message" : [
+ "Cannot fast-forward branch <branchName> on table <relationName> to <targetBranchName>: <reason>"
Review Comment:
nit. We need `.` at the end to be consistent with other messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314866419
##########
sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out:
##########
@@ -141,6 +143,7 @@ EXTENDED false
EXTERNAL false
EXTRACT false
FALSE false
+FAST false
Review Comment:
ditto.
##########
sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out:
##########
@@ -153,6 +156,7 @@ FOR false
FOREIGN false
FORMAT false
FORMATTED false
+FORWARD false
Review Comment:
ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3319599117
##########
sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out:
##########
@@ -141,6 +143,7 @@ EXTENDED false
EXTERNAL false
EXTRACT false
FALSE true
+FAST false
Review Comment:
Regenerated — `keywords-enforced.sql.out` now has `FASTFORWARD` in place of `FAST` and no `FORWARD` row.
##########
sql/core/src/test/resources/sql-tests/results/keywords.sql.out:
##########
@@ -141,6 +143,7 @@ EXTENDED false
EXTERNAL false
EXTRACT false
FALSE false
+FAST false
Review Comment:
Regenerated — `keywords.sql.out` updated identically.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4812402203
Could you rebase this PR once more, @viirya ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3300723524
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -50,7 +53,50 @@ class InMemoryTable(
override val id: String = UUID.randomUUID().toString)
extends InMemoryBaseTable(name, columns, partitioning, properties, constraints, distribution,
ordering, numPartitions, advisoryPartitionSize, isDistributionStrictlyRequired,
- numRowsPerSplit) with SupportsDelete {
+ numRowsPerSplit) with SupportsDelete with SupportsBranching {
+
+ private val branches = new ConcurrentHashMap[String, TableBranch]()
+
+ override def createBranch(branchName: String,
+ sourceSnapshotId: OptionalLong): TableBranch = {
+ val snapshot = if (sourceSnapshotId.isPresent) sourceSnapshotId.getAsLong
+ else version().toLong
+ val branch = new TableBranch(branchName, snapshot, System.currentTimeMillis())
+ val existing = branches.putIfAbsent(branchName, branch)
+ if (existing != null) {
+ throw new SupportsBranching.BranchAlreadyExistsException(branchName)
+ }
+ branch
+ }
+
+ override def dropBranch(branchName: String): Boolean = {
+ branches.remove(branchName) != null
+ }
+
+ override def fastForward(branchName: String, targetBranchName: String): TableBranch = {
+ val current = branches.get(branchName)
+ if (current == null) {
+ throw new java.util.NoSuchElementException(s"Branch not found: $branchName")
+ }
+ val target = branches.get(targetBranchName)
+ if (target == null) {
+ throw new java.util.NoSuchElementException(s"Branch not found: $targetBranchName")
Review Comment:
ditto.
Please correct me if this is the standard of DSv2 and Apache Iceberg.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4619944672
Thanks for another pass @dongjoon-hyun! Force-pushed \`35b6b6dc8ea\` addressing the comments:
- **Indentation** in `BranchingExec` — fixed.
- **\`creation_time_ms\` → \`creation_time\` (`TimestampType`)** — renamed column and switched type; `ShowBranchesExec` converts long ms to internal microseconds.
- **\`fastForward\` → \`fastForwardBranch\`** — renamed for consistency with the other `*Branch` methods.
- **\`replaceBranch\` atomicity** — made the method abstract (removed the racy drop+create default impl) and documented the atomicity contract. `InMemoryTable` now overrides it under a `synchronized` block. Real connectors like Iceberg already have an atomic `manageSnapshots().replaceBranch()` primitive.
- **`docs/sql-ref-syntax-*.md`** — will add in a follow-up PR.
- **\`dropBranch\` `ignoreIfNotExists` parameter** — pushed back gently in the inline thread; the current DSv2 catalog API (`TableCatalog.dropTable`, `SupportsNamespaces.dropNamespace`) uses `boolean` return rather than `ignoreIfNotExists`. Happy to revisit if you'd prefer the V1 `SessionCatalog` style.
`SupportsBranchingSuite` (15/15) and scalafmt all pass locally. PR #56103 rebased on top and tests green (23/23). PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314867154
##########
sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out:
##########
@@ -141,6 +143,7 @@ EXTENDED false
EXTERNAL false
EXTRACT false
FALSE true
+FAST false
Review Comment:
ditto.
##########
sql/core/src/test/resources/sql-tests/results/keywords.sql.out:
##########
@@ -141,6 +143,7 @@ EXTENDED false
EXTERNAL false
EXTRACT false
FALSE false
+FAST false
Review Comment:
ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314889148
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BranchingExec.scala:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.SupportsBranching
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Physical plan node for ALTER TABLE ... CREATE BRANCH.
+ */
+case class CreateBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val source = sourceSnapshotId match {
+ case Some(id) => OptionalLong.of(id)
+ case None => OptionalLong.empty()
+ }
+ if (replace) {
+ table.replaceBranch(branchName, source)
+ } else {
+ try {
+ table.createBranch(branchName, source)
+ } catch {
+ case _: BranchAlreadyExistsException if ifNotExists =>
+ // ignored: branch already exists and IF NOT EXISTS was requested
+ }
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... DROP BRANCH.
+ */
+case class DropBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ ifExists: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val removed = table.dropBranch(branchName)
+ if (!removed && !ifExists) {
+ throw new BranchNotFoundException(branchName, table.name())
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... FAST FORWARD.
+ */
+case class FastForwardBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ targetBranchName: String) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ table.fastForward(branchName, targetBranchName)
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for SHOW BRANCHES.
+ */
+case class ShowBranchesExec(
+ output: Seq[Attribute],
+ table: SupportsBranching) extends LeafV2CommandExec {
+
+ override protected def run(): Seq[InternalRow] = {
+ table.listBranches().toSeq.map { branch =>
+ val snapshot: Any = if (branch.snapshotId().isPresent) {
+ branch.snapshotId().getAsLong
+ } else {
+ null
Review Comment:
We need a test coverage for this code path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314864556
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala:
##########
@@ -3662,4 +3662,77 @@ class DDLParserSuite extends AnalysisTest {
)
}
}
+
+ test("alter table: create branch") {
+ comparePlans(
+ parsePlan("ALTER TABLE t CREATE BRANCH b"),
+ CreateBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... CREATE BRANCH"),
+ "b",
+ None,
+ ifNotExists = false,
+ replace = false))
+
+ comparePlans(
+ parsePlan("ALTER TABLE t CREATE BRANCH IF NOT EXISTS b"),
+ CreateBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... CREATE BRANCH"),
+ "b",
+ None,
+ ifNotExists = true,
+ replace = false))
+
+ comparePlans(
+ parsePlan("ALTER TABLE t CREATE OR REPLACE BRANCH b AS OF VERSION 42"),
+ CreateBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... CREATE BRANCH"),
+ "b",
+ Some(42L),
+ ifNotExists = false,
+ replace = true))
+ }
+
+ test("alter table: create branch with both REPLACE and IF NOT EXISTS is rejected") {
+ val sqlText = "ALTER TABLE t CREATE OR REPLACE BRANCH IF NOT EXISTS b"
+ checkError(
+ exception = parseException(sqlText),
+ condition = "CREATE_BRANCH_WITH_IF_NOT_EXISTS_AND_REPLACE",
+ parameters = Map("tableName" -> "t", "branchName" -> "b"),
+ context = ExpectedContext(fragment = sqlText, start = 0, stop = sqlText.length - 1))
+ }
+
+ test("alter table: drop branch") {
+ comparePlans(
+ parsePlan("ALTER TABLE t DROP BRANCH b"),
+ DropBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... DROP BRANCH"),
+ "b",
+ ifExists = false))
+
+ comparePlans(
+ parsePlan("ALTER TABLE t DROP BRANCH IF EXISTS b"),
+ DropBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... DROP BRANCH"),
+ "b",
+ ifExists = true))
+ }
+
+ test("alter table: fast forward") {
Review Comment:
Please revise the test case name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314861486
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
Review Comment:
Please remove this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4590295712
cc @cloud-fan Please let me know what do you think about this idea. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3326956837
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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]
Review Comment:
While I'm rechecking this part, I notice that we use `VERSION AS OF` instead of `AS OF VERSION` according to the existing Apache Spark Time Travel Syntax.
https://github.com/apache/spark/blob/06ffa273d0213eb14a96c923eba7901f05b23e08/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TimeTravelSpec.scala#L34-L36
Given that, it seems that we use like the following.
```
SELECT * FROM t VERSION AS OF v1 (O)
SELECT * FROM t AS OF VERSION v1 (X)
```
So, maybe the following is more consistent inside Apache Spark? WDYT, @viirya ?
```
- ALTER TABLE t CREATE BRANCH b AS OF VERSION v1
+ ALTER TABLE t CREATE BRANCH b VERSION AS OF v1
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3300720036
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.NoSuchElementException;
+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
+ * @throws NoSuchElementException if the requested {@code sourceSnapshotId} does not exist
+ */
+ 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 IllegalArgumentException} otherwise.
+ *
+ * @return the updated {@link TableBranch} after fast-forwarding
+ * @throws NoSuchElementException if either branch is missing
+ * @throws IllegalArgumentException 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];
+ }
+
+ /**
+ * Thrown when a branch with the requested name already exists.
+ */
+ class BranchAlreadyExistsException extends RuntimeException {
Review Comment:
What about something like `BranchNotFound` or `InvalidFastForward`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3300722271
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala:
##########
@@ -50,7 +53,50 @@ class InMemoryTable(
override val id: String = UUID.randomUUID().toString)
extends InMemoryBaseTable(name, columns, partitioning, properties, constraints, distribution,
ordering, numPartitions, advisoryPartitionSize, isDistributionStrictlyRequired,
- numRowsPerSplit) with SupportsDelete {
+ numRowsPerSplit) with SupportsDelete with SupportsBranching {
+
+ private val branches = new ConcurrentHashMap[String, TableBranch]()
+
+ override def createBranch(branchName: String,
+ sourceSnapshotId: OptionalLong): TableBranch = {
+ val snapshot = if (sourceSnapshotId.isPresent) sourceSnapshotId.getAsLong
+ else version().toLong
+ val branch = new TableBranch(branchName, snapshot, System.currentTimeMillis())
+ val existing = branches.putIfAbsent(branchName, branch)
+ if (existing != null) {
+ throw new SupportsBranching.BranchAlreadyExistsException(branchName)
+ }
+ branch
+ }
+
+ override def dropBranch(branchName: String): Boolean = {
+ branches.remove(branchName) != null
+ }
+
+ override def fastForward(branchName: String, targetBranchName: String): TableBranch = {
+ val current = branches.get(branchName)
+ if (current == null) {
+ throw new java.util.NoSuchElementException(s"Branch not found: $branchName")
Review Comment:
This looks wired to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335890752
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BranchingExec.scala:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.SupportsBranching
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Physical plan node for ALTER TABLE ... CREATE BRANCH.
+ */
+case class CreateBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val source = sourceSnapshotId match {
+ case Some(id) => OptionalLong.of(id)
+ case None => OptionalLong.empty()
+ }
+ if (replace) {
+ table.replaceBranch(branchName, source)
+ } else {
+ try {
+ table.createBranch(branchName, source)
+ } catch {
+ case _: BranchAlreadyExistsException if ifNotExists =>
+ // ignored: branch already exists and IF NOT EXISTS was requested
Review Comment:
nit. indentation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314871169
##########
sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala:
##########
@@ -210,7 +210,7 @@ class SparkConnectDatabaseMetaDataSuite extends ConnectFunSuite with RemoteSpark
val metadata = conn.getMetaData
// scalastyle:off line.size.limit
// CURRENT_PATH and SYSTEM are excluded: getSQLKeywords drops SQL:2003 reserved words (see companion).
- assert(metadata.getSQLKeywords === "ADD,AFTER,AGGREGATE,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPROX,ARCHIVE,ASC,BERNOULLI,BINDING,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,ENFORCED,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIMIT,LINES,LIST,LOAD,LOCATION,L
OCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,ROLE,ROLES,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNLOCK,UNPIVOT,UNSET,UNTIL,USE,
VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,X,YEARS,ZONE")
+ assert(metadata.getSQLKeywords === "ADD,AFTER,AGGREGATE,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPROX,ARCHIVE,ASC,BERNOULLI,BINDING,BRANCH,BRANCHES,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,ENFORCED,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FAST,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FORWARD,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIM
IT,LINES,LIST,LOAD,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,ROLE,ROLES,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UN
LOCK,UNPIVOT,UNSET,UNTIL,USE,VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,X,YEARS,ZONE")
Review Comment:
The test case seems to be outdated because it has `,FAST,` still
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3319601924
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BranchingExec.scala:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.SupportsBranching
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Physical plan node for ALTER TABLE ... CREATE BRANCH.
+ */
+case class CreateBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val source = sourceSnapshotId match {
+ case Some(id) => OptionalLong.of(id)
+ case None => OptionalLong.empty()
+ }
+ if (replace) {
+ table.replaceBranch(branchName, source)
+ } else {
+ try {
+ table.createBranch(branchName, source)
+ } catch {
+ case _: BranchAlreadyExistsException if ifNotExists =>
+ // ignored: branch already exists and IF NOT EXISTS was requested
+ }
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... DROP BRANCH.
+ */
+case class DropBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ ifExists: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val removed = table.dropBranch(branchName)
+ if (!removed && !ifExists) {
+ throw new BranchNotFoundException(branchName, table.name())
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... FAST FORWARD.
+ */
+case class FastForwardBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ targetBranchName: String) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ table.fastForward(branchName, targetBranchName)
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for SHOW BRANCHES.
+ */
+case class ShowBranchesExec(
+ output: Seq[Attribute],
+ table: SupportsBranching) extends LeafV2CommandExec {
+
+ override protected def run(): Seq[InternalRow] = {
+ table.listBranches().toSeq.map { branch =>
+ val snapshot: Any = if (branch.snapshotId().isPresent) {
+ branch.snapshotId().getAsLong
+ } else {
+ null
Review Comment:
The new `SHOW BRANCHES` test on the suite exercises this code path (`branch.snapshotId().isPresent` false → `null`). See the reply on the suite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4912277771
Close this for now. I will revisit this if necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya closed pull request #56102: [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL
URL: https://github.com/apache/spark/pull/56102
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314862607
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
Review Comment:
And, please add FASTFORWARD somewhere.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3354225892
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ dropBranch(name);
+ return createBranch(name, sourceSnapshotId);
Review Comment:
Good catch — the default impl can't guarantee atomicity by composition, so I made `replaceBranch` abstract (removed the default impl) and updated the javadoc to require atomicity:
> Implementations must perform the replacement atomically with respect to concurrent `createBranch`, `dropBranch`, and other `replaceBranch` calls on the same branch name. A naive drop-then-create implementation is not sufficient because it allows a concurrent caller to observe the branch as missing and create it with a different snapshot in the window between the two operations.
This forces every implementer to think about atomicity, and real connectors like Iceberg already have atomic primitives (`manageSnapshots().replaceBranch()`) to delegate to. The `InMemoryTable` test impl now overrides `replaceBranch` under a `branches.synchronized` block.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ 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);
Review Comment:
The existing DSv2 catalog interfaces follow a `boolean drop*(...)` convention rather than an `ignoreIfNotExists` parameter:
- [`TableCatalog.dropTable(Identifier): boolean`](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java#L360) — "true if a table was deleted, false if no table exists for the identifier"
- [`SupportsNamespaces.dropNamespace(String[], boolean cascade): boolean`](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsNamespaces.java#L159)
Following that pattern keeps `SupportsBranching` shape-consistent with its siblings — the caller (`DropBranchExec`) decides whether to throw `BranchNotFoundException` based on the boolean and the `IF EXISTS` clause, which mirrors how `DropTableExec` / `DropNamespaceExec` handle the same case.
The `ignoreIfNotExists`-style API exists in Spark's internal V1 `SessionCatalog`, but that's a different layer (the V1 `ExternalCatalog` contract, not the v2 connector interface). Happy to revisit if you'd prefer the V1 style here for some other reason, but I'd lean toward keeping the DSv2 convention.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3319600482
##########
sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala:
##########
@@ -210,7 +210,7 @@ class SparkConnectDatabaseMetaDataSuite extends ConnectFunSuite with RemoteSpark
val metadata = conn.getMetaData
// scalastyle:off line.size.limit
// CURRENT_PATH and SYSTEM are excluded: getSQLKeywords drops SQL:2003 reserved words (see companion).
- assert(metadata.getSQLKeywords === "ADD,AFTER,AGGREGATE,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPROX,ARCHIVE,ASC,BERNOULLI,BINDING,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,ENFORCED,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIMIT,LINES,LIST,LOAD,LOCATION,L
OCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,ROLE,ROLES,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNLOCK,UNPIVOT,UNSET,UNTIL,USE,
VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,X,YEARS,ZONE")
+ assert(metadata.getSQLKeywords === "ADD,AFTER,AGGREGATE,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPROX,ARCHIVE,ASC,BERNOULLI,BINDING,BRANCH,BRANCHES,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,ENFORCED,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FAST,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FORWARD,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIM
IT,LINES,LIST,LOAD,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,ROLE,ROLES,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UN
LOCK,UNPIVOT,UNSET,UNTIL,USE,VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,X,YEARS,ZONE")
Review Comment:
Updated — `FAST` and `FORWARD` removed, `FASTFORWARD` inserted between `EXTENDED` and `FIELDS`.
##########
sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala:
##########
@@ -214,7 +214,7 @@ trait ThriftServerWithSparkContextSuite extends SharedThriftServer {
val sessionHandle = client.openSession(user, "")
val infoValue = client.getInfo(sessionHandle, GetInfoType.CLI_ODBC_KEYWORDS)
// scalastyle:off line.size.limit
- assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,BEGIN,BERNOULLI,BETWEEN,BIGINT,BINARY,BINDING,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,END,ENFORCED,ESCAPE,ES
CAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS
,OR,ORDER,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFO
RM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNION,UNIQUE,UNKNOWN,UNLOCK,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE")
+ assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,BEGIN,BERNOULLI,BETWEEN,BIGINT,BINARY,BINDING,BOOLEAN,BOTH,BRANCH,BRANCHES,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,END,EN
FORCED,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FAST,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FORWARD,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSE
T,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRAILING,TRA
NSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNION,UNIQUE,UNKNOWN,UNLOCK,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE")
Review Comment:
Updated identically in the Thrift server suite — `FAST` and `FORWARD` replaced with `FASTFORWARD` between `FALSE` and `FETCH`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1697,6 +1697,59 @@ case class RecoverPartitions(child: LogicalPlan) extends UnaryCommand {
copy(child = newChild)
}
+/**
+ * The logical plan of the ALTER TABLE ... CREATE BRANCH command.
+ */
+case class CreateBranch(
+ child: LogicalPlan,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): CreateBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... DROP BRANCH command.
+ */
+case class DropBranch(
+ child: LogicalPlan,
+ branchName: String,
+ ifExists: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): DropBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... FASTFORWARD BRANCH command.
+ */
+case class FastForwardBranch(
+ child: LogicalPlan,
+ branchName: String,
+ targetBranchName: String) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): FastForwardBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the SHOW BRANCHES command.
+ */
+case class ShowBranches(
+ child: LogicalPlan,
+ override val output: Seq[Attribute] = ShowBranches.getOutputAttrs) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): ShowBranches =
+ copy(child = newChild)
+}
+
+object ShowBranches {
+ def getOutputAttrs: Seq[Attribute] = Seq(
+ AttributeReference("name", StringType, nullable = false)(),
+ AttributeReference("snapshot_id", org.apache.spark.sql.types.LongType, nullable = true)(),
Review Comment:
Yes — `snapshot_id` is nullable. `TableBranch.snapshotId()` is `OptionalLong`, so an implementation that records a branch with no snapshot reference (for example, a branch created on a table that has no commits yet) surfaces as `NULL` in `SHOW BRANCHES`.
Found a related bug in the latest push: `InMemoryTable.createBranch` was using `version().toLong` as a fallback "current snapshot" even when the table had never been written, which would pretend snapshot `0` exists for a fresh table. Fixed that so when `tableVersion == 0` the branch carries `OptionalLong.empty()`, and added test coverage for the new `NULL` code path — see the reply on the suite below.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsBranchingSuite.scala:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import java.util
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException, InvalidFastForwardException}
+import org.apache.spark.sql.connector.catalog.{Column, Identifier, InMemoryCatalog, SupportsBranching, Table, TableCapability}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+import org.apache.spark.sql.types.StructType
+
+class SupportsBranchingSuite extends QueryTest with DatasourceV2SQLBase {
+
+ private def branchingTable(tableName: String): SupportsBranching = {
+ val cat = catalog("testcat").asInstanceOf[InMemoryCatalog]
+ cat.loadTable(Identifier.of(Array("ns"), tableName)).asInstanceOf[SupportsBranching]
+ }
+
+ private def withBranchingTable(tableName: String)(f: String => Unit): Unit = {
+ val fqName = s"testcat.ns.$tableName"
+ sql(s"CREATE TABLE $fqName (id bigint, data string) USING foo")
+ try {
+ f(fqName)
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $fqName")
+ }
+ }
+
+ test("CREATE BRANCH on an InMemoryTable creates a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val t = branchingTable("t")
+ val branches = t.listBranches()
+ assert(branches.length == 1)
+ assert(branches(0).name == "dev")
+ }
+ }
+
+ test("CREATE BRANCH with explicit snapshot id") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH staging AS OF VERSION 7")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.name == "staging")
+ assert(branch.snapshotId.isPresent && branch.snapshotId.getAsLong == 7L)
+ }
+ }
+
+ test("CREATE BRANCH IF NOT EXISTS is idempotent") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq CREATE BRANCH IF NOT EXISTS dev")
+ assert(branchingTable("t").listBranches().length == 1)
+ }
+ }
+
+ test("CREATE BRANCH without IF NOT EXISTS fails on duplicate") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val ex = intercept[BranchAlreadyExistsException] {
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ }
+ assert(ex.getMessage.contains("dev"))
+ assert(ex.getCondition == "BRANCH_ALREADY_EXISTS")
+ }
+ }
+
+ test("CREATE OR REPLACE BRANCH overwrites an existing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE OR REPLACE BRANCH dev AS OF VERSION 9")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.snapshotId.getAsLong == 9L)
+ }
+ }
+
+ test("DROP BRANCH removes a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq DROP BRANCH dev")
+ assert(branchingTable("t").listBranches().isEmpty)
+ }
+ }
+
+ test("DROP BRANCH IF EXISTS does not fail on missing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq DROP BRANCH IF EXISTS missing")
+ }
+ }
+
+ test("DROP BRANCH fails on missing branch when IF EXISTS not specified") {
+ withBranchingTable("t") { fq =>
+ val ex = intercept[BranchNotFoundException] {
+ sql(s"ALTER TABLE $fq DROP BRANCH missing")
+ }
+ assert(ex.getMessage.contains("missing"))
+ assert(ex.getCondition == "BRANCH_NOT_FOUND")
+ }
+ }
+
+ test("FASTFORWARD BRANCH updates branch to target's snapshot") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 5")
+ sql(s"ALTER TABLE $fq FASTFORWARD BRANCH dev TO main")
+ val dev = branchingTable("t").listBranches().find(_.name == "dev").get
+ assert(dev.snapshotId.getAsLong == 5L)
+ }
+ }
+
+ test("FASTFORWARD BRANCH fails when target is behind") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 5")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 1")
+ val ex = intercept[InvalidFastForwardException] {
+ sql(s"ALTER TABLE $fq FASTFORWARD BRANCH dev TO main")
+ }
+ assert(ex.getMessage.contains("behind"))
+ assert(ex.getCondition == "INVALID_FAST_FORWARD")
+ }
+ }
+
+ test("SHOW BRANCHES lists branches with snapshot ids") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 2")
+ val rows = sql(s"SHOW BRANCHES IN $fq").collect()
+ val byName = rows.map(r => r.getString(0) -> r.getLong(1)).toMap
Review Comment:
Added a new test case `"SHOW BRANCHES surfaces NULL snapshot_id when the branch has no snapshot"` that creates a branch on a freshly created table (no commits) and verifies `rows(0).isNullAt(1) == true`. This required fixing `InMemoryTable.createBranch` so that when no explicit snapshot is supplied **and** the table has never been written, the resulting `TableBranch` carries `OptionalLong.empty()` instead of `OptionalLong.of(0L)` — the previous fallback was pretending snapshot `0` exists for an empty table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3319598145
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BranchingExec.scala:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.SupportsBranching
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Physical plan node for ALTER TABLE ... CREATE BRANCH.
+ */
+case class CreateBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val source = sourceSnapshotId match {
+ case Some(id) => OptionalLong.of(id)
+ case None => OptionalLong.empty()
+ }
+ if (replace) {
+ table.replaceBranch(branchName, source)
+ } else {
+ try {
+ table.createBranch(branchName, source)
+ } catch {
+ case _: BranchAlreadyExistsException if ifNotExists =>
+ // ignored: branch already exists and IF NOT EXISTS was requested
+ }
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... DROP BRANCH.
+ */
+case class DropBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ ifExists: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val removed = table.dropBranch(branchName)
+ if (!removed && !ifExists) {
+ throw new BranchNotFoundException(branchName, table.name())
+ }
+ Seq.empty
+ }
+}
+
+/**
+ * Physical plan node for ALTER TABLE ... FAST FORWARD.
Review Comment:
Expanded to spell out the operation: "Physical plan node for ALTER TABLE ... FASTFORWARD BRANCH, which advances `branchName` to the head snapshot of `targetBranchName`."
##########
sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out:
##########
@@ -141,6 +143,7 @@ EXTENDED false
EXTERNAL false
EXTRACT false
FALSE false
+FAST false
Review Comment:
Regenerated — `FAST` row replaced with `FASTFORWARD`.
##########
sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out:
##########
@@ -153,6 +156,7 @@ FOR false
FOREIGN false
FORMAT false
FORMATTED false
+FORWARD false
Review Comment:
Regenerated — `FORWARD` row removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3319596052
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3747,6 +3718,12 @@
],
"sqlState" : "42000"
},
+ "INVALID_FAST_FORWARD" : {
Review Comment:
Done — renamed to `INVALID_FASTFORWARD` for consistency with the `FASTFORWARD` token and the exception class name.
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
Review Comment:
Done — removed the `FAST` row.
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
Review Comment:
Done — removed the `FORWARD` row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335873689
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
+|FASTFORWARD|non-reserved|non-reserved|non-reserved|
Review Comment:
Maybe, you want to add the syntax documentation, `docs/sql-ref-syntax-*.md`, later?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335835225
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ 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)
Review Comment:
Like other APIs, `(create|drop|replace)Branch` and `listBranches`, can we have `fastForwardBranch` instead of `fastForward`?
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ dropBranch(name);
+ return createBranch(name, sourceSnapshotId);
Review Comment:
This looks a little too simple because `dropBranch` returns `false` instead of throwing exception. Specifically, how do we guarantee atomicity? For example, two Spark connect sessions invoke `replaceBranch` together?
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ 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);
Review Comment:
It might be better to have `ignoreIfNotExists` parameter together (true by default) in order to follow Spark's conventional interface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3335834314
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ dropBranch(name);
+ return createBranch(name, sourceSnapshotId);
Review Comment:
This looks a little too simple because `dropBranch` returns `false` instead of throwing exception. Specifically, how do we guarantee atomicity? For example, two Spark connect sessions invoke `replaceBranch` together? According to this code, a session S1 successfully drops the branch, `B`, but another session S2 can steal the branch `B` with S3's `sourceSnapshotId`.
If we ignore `Atomicity` in `replaceBranch`, we need to clarify that in the interface documentation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3327894163
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3747,6 +3718,12 @@
],
"sqlState" : "42000"
},
+ "INVALID_FASTFORWARD" : {
+ "message" : [
+ "Cannot fast-forward branch <branchName> on table <relationName> to <targetBranchName>: <reason>"
Review Comment:
Done — added trailing period.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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]
Review Comment:
Agreed — `VERSION AS OF` matches Spark's existing time-travel syntax. Changed the grammar to:
```
ALTER TABLE t CREATE [OR REPLACE] BRANCH [IF NOT EXISTS] name [VERSION AS OF <id>]
```
Updated all callers (parser tests, suite tests, javadoc on `SupportsBranching`, PR description). This diverges from Apache Iceberg's `AS OF VERSION` form for the same reason `FASTFORWARD BRANCH` did — the Spark-native surface should be internally consistent first.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsBranchingSuite.scala:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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
+
+import java.util
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException, InvalidFastForwardException}
+import org.apache.spark.sql.connector.catalog.{Column, Identifier, InMemoryCatalog, SupportsBranching, Table, TableCapability}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+import org.apache.spark.sql.types.StructType
+
+class SupportsBranchingSuite extends QueryTest with DatasourceV2SQLBase {
+
+ private def branchingTable(tableName: String): SupportsBranching = {
+ val cat = catalog("testcat").asInstanceOf[InMemoryCatalog]
+ cat.loadTable(Identifier.of(Array("ns"), tableName)).asInstanceOf[SupportsBranching]
+ }
+
+ private def withBranchingTable(tableName: String)(f: String => Unit): Unit = {
+ val fqName = s"testcat.ns.$tableName"
+ sql(s"CREATE TABLE $fqName (id bigint, data string) USING foo")
+ try {
+ f(fqName)
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $fqName")
+ }
+ }
+
+ test("CREATE BRANCH on an InMemoryTable creates a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val t = branchingTable("t")
+ val branches = t.listBranches()
+ assert(branches.length == 1)
+ assert(branches(0).name == "dev")
+ }
+ }
+
+ test("CREATE BRANCH with explicit snapshot id") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH staging AS OF VERSION 7")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.name == "staging")
+ assert(branch.snapshotId.isPresent && branch.snapshotId.getAsLong == 7L)
+ }
+ }
+
+ test("CREATE BRANCH IF NOT EXISTS is idempotent") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq CREATE BRANCH IF NOT EXISTS dev")
+ assert(branchingTable("t").listBranches().length == 1)
+ }
+ }
+
+ test("CREATE BRANCH without IF NOT EXISTS fails on duplicate") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val ex = intercept[BranchAlreadyExistsException] {
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ }
+ assert(ex.getMessage.contains("dev"))
+ assert(ex.getCondition == "BRANCH_ALREADY_EXISTS")
+ }
+ }
+
+ test("CREATE OR REPLACE BRANCH overwrites an existing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE OR REPLACE BRANCH dev AS OF VERSION 9")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.snapshotId.getAsLong == 9L)
+ }
+ }
+
+ test("DROP BRANCH removes a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq DROP BRANCH dev")
+ assert(branchingTable("t").listBranches().isEmpty)
+ }
+ }
+
+ test("DROP BRANCH IF EXISTS does not fail on missing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq DROP BRANCH IF EXISTS missing")
+ }
+ }
+
+ test("DROP BRANCH fails on missing branch when IF EXISTS not specified") {
+ withBranchingTable("t") { fq =>
+ val ex = intercept[BranchNotFoundException] {
+ sql(s"ALTER TABLE $fq DROP BRANCH missing")
+ }
+ assert(ex.getMessage.contains("missing"))
+ assert(ex.getCondition == "BRANCH_NOT_FOUND")
+ }
+ }
+
+ test("FASTFORWARD BRANCH updates branch to target's snapshot") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 5")
+ sql(s"ALTER TABLE $fq FASTFORWARD BRANCH dev TO main")
+ val dev = branchingTable("t").listBranches().find(_.name == "dev").get
+ assert(dev.snapshotId.getAsLong == 5L)
+ }
+ }
+
+ test("FASTFORWARD BRANCH fails when target is behind") {
Review Comment:
Added two cases:
- `FASTFORWARD BRANCH fails when the source branch is missing` — intercepts `BranchNotFoundException` with `getCondition == "BRANCH_NOT_FOUND"` for the source branch.
- `FASTFORWARD BRANCH fails when the target branch is missing` — same for the target.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4581367227
Thanks again @dongjoon-hyun! Force-pushed `f395f6d1cb3` addressing all four comments:
**Scalafmt CI failure** — the `mvn-scalafmt` plugin wanted the new `BranchAlreadyExistsException` / `BranchNotFoundException` constructors' `Map(...)` literals on a single line (they fit under 98 cols). Verified locally with `scalafmt 3.8.6` against `dev/.scalafmt.conf`.
**Grammar `AS OF VERSION` → `VERSION AS OF`** — agreed, matches Spark's existing time-travel syntax. Updated grammar, AST builder, javadoc, parser tests, suite tests, and PR description.
**`BranchNotFoundException` coverage on `FASTFORWARD BRANCH`** — added two cases: source branch missing and target branch missing.
**Trailing period on `INVALID_FASTFORWARD`** — done.
`SupportsBranchingSuite` (15/15), `DDLParserSuite` create/fastforward branch tests, and scalafmt all pass locally. PR #56103 rebased on top and also rebuilds/tests green (23/23). PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3354223767
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BranchingExec.scala:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.SupportsBranching
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Physical plan node for ALTER TABLE ... CREATE BRANCH.
+ */
+case class CreateBranchExec(
+ table: SupportsBranching,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends LeafV2CommandExec {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val source = sourceSnapshotId match {
+ case Some(id) => OptionalLong.of(id)
+ case None => OptionalLong.empty()
+ }
+ if (replace) {
+ table.replaceBranch(branchName, source)
+ } else {
+ try {
+ table.createBranch(branchName, source)
+ } catch {
+ case _: BranchAlreadyExistsException if ifNotExists =>
+ // ignored: branch already exists and IF NOT EXISTS was requested
Review Comment:
Fixed — added the missing 2-space indent so the comment sits inside the case body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1697,6 +1697,59 @@ case class RecoverPartitions(child: LogicalPlan) extends UnaryCommand {
copy(child = newChild)
}
+/**
+ * The logical plan of the ALTER TABLE ... CREATE BRANCH command.
+ */
+case class CreateBranch(
+ child: LogicalPlan,
+ branchName: String,
+ sourceSnapshotId: Option[Long],
+ ifNotExists: Boolean,
+ replace: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): CreateBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... DROP BRANCH command.
+ */
+case class DropBranch(
+ child: LogicalPlan,
+ branchName: String,
+ ifExists: Boolean) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): DropBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... FASTFORWARD BRANCH command.
+ */
+case class FastForwardBranch(
+ child: LogicalPlan,
+ branchName: String,
+ targetBranchName: String) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): FastForwardBranch =
+ copy(child = newChild)
+}
+
+/**
+ * The logical plan of the SHOW BRANCHES command.
+ */
+case class ShowBranches(
+ child: LogicalPlan,
+ override val output: Seq[Attribute] = ShowBranches.getOutputAttrs) extends UnaryCommand {
+ override protected def withNewChildInternal(newChild: LogicalPlan): ShowBranches =
+ copy(child = newChild)
+}
+
+object ShowBranches {
+ def getOutputAttrs: Seq[Attribute] = Seq(
+ AttributeReference("name", StringType, nullable = false)(),
+ AttributeReference("snapshot_id", org.apache.spark.sql.types.LongType, nullable = true)(),
+ AttributeReference("creation_time_ms",
+ org.apache.spark.sql.types.LongType, nullable = false)())
Review Comment:
Good point — changed the column to `TimestampType` and renamed it from `creation_time_ms` to `creation_time` (the `_ms` suffix would be misleading now that the type is timestamp). `ShowBranchesExec.run` converts `TableBranch.creationTimeMs()` (long ms) to microseconds for the internal row.
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
+|FASTFORWARD|non-reserved|non-reserved|non-reserved|
Review Comment:
Yes — I'll add the syntax documentation (`docs/sql-ref-syntax-ddl-{create,drop}-branch.md`, `sql-ref-syntax-aux-show-branches.md`, and the `ALTER TABLE ... FASTFORWARD BRANCH` section) in a follow-up PR.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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;
+import org.apache.spark.sql.catalyst.analysis.BranchAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.BranchNotFoundException;
+import org.apache.spark.sql.catalyst.analysis.InvalidFastForwardException;
+
+/**
+ * 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 [VERSION AS OF snapshotId]
+ * ALTER TABLE t DROP BRANCH [IF EXISTS] name
+ * ALTER TABLE t FASTFORWARD BRANCH 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)
+ throws BranchAlreadyExistsException;
+
+ /**
+ * 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)
+ throws BranchAlreadyExistsException {
+ 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)
Review Comment:
Renamed to `fastForwardBranch` for consistency with the other `*Branch` methods. `InMemoryTable` override, `FastForwardBranchExec` call site, and `@throws` javadoc updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4826813221
Sure! Thank you for keeping this up-to-date.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4559992400
Thanks for the follow-up review @dongjoon-hyun! Force-pushed `8395ee9d6ac` addressing both points:
**Exceptions now extend `AnalysisException` (following the `TableCatalog` pattern):**
- Removed the three nested classes from `SupportsBranching.java`.
- Added `BranchAlreadyExistsException` to `alreadyExistException.scala`, plus `BranchNotFoundException` and `InvalidFastForwardException` to `noSuchItemsExceptions.scala`, all extending `AnalysisException` and constructed via error class + message parameters.
- New error conditions in `error-conditions.json`: `BRANCH_ALREADY_EXISTS` (42710), `BRANCH_NOT_FOUND` (42704), `INVALID_FAST_FORWARD` (42000).
- Interface methods now declare these typed exceptions in their `throws` clauses; `InMemoryTable`, `DropBranchExec`, and `SupportsBranchingSuite` updated accordingly.
**SQL syntax changed to `FASTFORWARD BRANCH`** (single-word verb + `BRANCH` + name, matching `CREATE BRANCH` / `DROP BRANCH`):
- Lexer: replaced `FAST` / `FORWARD` with a single `FASTFORWARD` token.
- Parser: `ALTER TABLE t FASTFORWARD BRANCH <branch> TO <target>`.
- Tests and `keyword-context` updated.
`SupportsBranchingSuite` (12/12) and the parser/`DDLParserSuite` fast-forward test pass locally; scalastyle is clean. PTAL when you get a chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3314887336
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsBranchingSuite.scala:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import java.util
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.analysis.{BranchAlreadyExistsException, BranchNotFoundException, InvalidFastForwardException}
+import org.apache.spark.sql.connector.catalog.{Column, Identifier, InMemoryCatalog, SupportsBranching, Table, TableCapability}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+import org.apache.spark.sql.types.StructType
+
+class SupportsBranchingSuite extends QueryTest with DatasourceV2SQLBase {
+
+ private def branchingTable(tableName: String): SupportsBranching = {
+ val cat = catalog("testcat").asInstanceOf[InMemoryCatalog]
+ cat.loadTable(Identifier.of(Array("ns"), tableName)).asInstanceOf[SupportsBranching]
+ }
+
+ private def withBranchingTable(tableName: String)(f: String => Unit): Unit = {
+ val fqName = s"testcat.ns.$tableName"
+ sql(s"CREATE TABLE $fqName (id bigint, data string) USING foo")
+ try {
+ f(fqName)
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $fqName")
+ }
+ }
+
+ test("CREATE BRANCH on an InMemoryTable creates a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val t = branchingTable("t")
+ val branches = t.listBranches()
+ assert(branches.length == 1)
+ assert(branches(0).name == "dev")
+ }
+ }
+
+ test("CREATE BRANCH with explicit snapshot id") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH staging AS OF VERSION 7")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.name == "staging")
+ assert(branch.snapshotId.isPresent && branch.snapshotId.getAsLong == 7L)
+ }
+ }
+
+ test("CREATE BRANCH IF NOT EXISTS is idempotent") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq CREATE BRANCH IF NOT EXISTS dev")
+ assert(branchingTable("t").listBranches().length == 1)
+ }
+ }
+
+ test("CREATE BRANCH without IF NOT EXISTS fails on duplicate") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ val ex = intercept[BranchAlreadyExistsException] {
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ }
+ assert(ex.getMessage.contains("dev"))
+ assert(ex.getCondition == "BRANCH_ALREADY_EXISTS")
+ }
+ }
+
+ test("CREATE OR REPLACE BRANCH overwrites an existing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE OR REPLACE BRANCH dev AS OF VERSION 9")
+ val branch = branchingTable("t").listBranches().head
+ assert(branch.snapshotId.getAsLong == 9L)
+ }
+ }
+
+ test("DROP BRANCH removes a branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev")
+ sql(s"ALTER TABLE $fq DROP BRANCH dev")
+ assert(branchingTable("t").listBranches().isEmpty)
+ }
+ }
+
+ test("DROP BRANCH IF EXISTS does not fail on missing branch") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq DROP BRANCH IF EXISTS missing")
+ }
+ }
+
+ test("DROP BRANCH fails on missing branch when IF EXISTS not specified") {
+ withBranchingTable("t") { fq =>
+ val ex = intercept[BranchNotFoundException] {
+ sql(s"ALTER TABLE $fq DROP BRANCH missing")
+ }
+ assert(ex.getMessage.contains("missing"))
+ assert(ex.getCondition == "BRANCH_NOT_FOUND")
+ }
+ }
+
+ test("FASTFORWARD BRANCH updates branch to target's snapshot") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 5")
+ sql(s"ALTER TABLE $fq FASTFORWARD BRANCH dev TO main")
+ val dev = branchingTable("t").listBranches().find(_.name == "dev").get
+ assert(dev.snapshotId.getAsLong == 5L)
+ }
+ }
+
+ test("FASTFORWARD BRANCH fails when target is behind") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 5")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 1")
+ val ex = intercept[InvalidFastForwardException] {
+ sql(s"ALTER TABLE $fq FASTFORWARD BRANCH dev TO main")
+ }
+ assert(ex.getMessage.contains("behind"))
+ assert(ex.getCondition == "INVALID_FAST_FORWARD")
+ }
+ }
+
+ test("SHOW BRANCHES lists branches with snapshot ids") {
+ withBranchingTable("t") { fq =>
+ sql(s"ALTER TABLE $fq CREATE BRANCH dev AS OF VERSION 1")
+ sql(s"ALTER TABLE $fq CREATE BRANCH main AS OF VERSION 2")
+ val rows = sql(s"SHOW BRANCHES IN $fq").collect()
+ val byName = rows.map(r => r.getString(0) -> r.getLong(1)).toMap
Review Comment:
Can we have a test coverage for nullable snapshot IDs because we cannot use `r.getLong(1)` for null value?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4566641278
Thanks for the thorough review @dongjoon-hyun! Force-pushed `4580a96a2ec` addressing all 16 comments:
**Naming & docs**
- Renamed error class `INVALID_FAST_FORWARD` → `INVALID_FASTFORWARD`.
- Removed `FAST` / `FORWARD` rows and added `FASTFORWARD` in `docs/sql-ref-ansi-compliance.md`.
- Renamed `DDLParserSuite` test from `"alter table: fast forward"` to `"alter table: fastforward branch"`.
- Expanded `FastForwardBranchExec` doc: now spells out the operation.
- Updated the PR description to match the current grammar and exception model.
**Outdated golden / keyword strings**
- Regenerated `keywords.sql.out`, `keywords-enforced.sql.out`, `nonansi/keywords.sql.out`.
- Updated the hard-coded keyword strings in `SparkConnectDatabaseMetaDataSuite` and `ThriftServerWithSparkContextSuite`.
**Nullable snapshot_id coverage** (your question on `v2Commands.scala:1748` and the two requests for test coverage):
- `snapshot_id` is intentionally nullable. Found that `InMemoryTable.createBranch` was using `version().toLong` as a fallback "current snapshot" — which pretends snapshot `0` exists for an empty table. Fixed it so when `tableVersion == 0` the branch carries `OptionalLong.empty()`.
- Added `SHOW BRANCHES surfaces NULL snapshot_id when the branch has no snapshot` to `SupportsBranchingSuite`, asserting `rows(0).isNullAt(1)`. This now exercises the `null` branch in `ShowBranchesExec`.
`SupportsBranchingSuite` (13/13), `DDLParserSuite` fast-forward test, `SQLQueryTestSuite` keyword cases (6/6), and `SparkConnectDatabaseMetaDataSuite.getSQLKeywords` all pass locally; scalastyle is clean. The follow-up PR #56103 is rebased on top and also rebuilds/tests green. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3319597065
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -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|
Review Comment:
Added `FASTFORWARD` (non-reserved in all three contexts) in alphabetical position between `FALSE` and `FETCH`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala:
##########
@@ -3662,4 +3662,77 @@ class DDLParserSuite extends AnalysisTest {
)
}
}
+
+ test("alter table: create branch") {
+ comparePlans(
+ parsePlan("ALTER TABLE t CREATE BRANCH b"),
+ CreateBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... CREATE BRANCH"),
+ "b",
+ None,
+ ifNotExists = false,
+ replace = false))
+
+ comparePlans(
+ parsePlan("ALTER TABLE t CREATE BRANCH IF NOT EXISTS b"),
+ CreateBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... CREATE BRANCH"),
+ "b",
+ None,
+ ifNotExists = true,
+ replace = false))
+
+ comparePlans(
+ parsePlan("ALTER TABLE t CREATE OR REPLACE BRANCH b AS OF VERSION 42"),
+ CreateBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... CREATE BRANCH"),
+ "b",
+ Some(42L),
+ ifNotExists = false,
+ replace = true))
+ }
+
+ test("alter table: create branch with both REPLACE and IF NOT EXISTS is rejected") {
+ val sqlText = "ALTER TABLE t CREATE OR REPLACE BRANCH IF NOT EXISTS b"
+ checkError(
+ exception = parseException(sqlText),
+ condition = "CREATE_BRANCH_WITH_IF_NOT_EXISTS_AND_REPLACE",
+ parameters = Map("tableName" -> "t", "branchName" -> "b"),
+ context = ExpectedContext(fragment = sqlText, start = 0, stop = sqlText.length - 1))
+ }
+
+ test("alter table: drop branch") {
+ comparePlans(
+ parsePlan("ALTER TABLE t DROP BRANCH b"),
+ DropBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... DROP BRANCH"),
+ "b",
+ ifExists = false))
+
+ comparePlans(
+ parsePlan("ALTER TABLE t DROP BRANCH IF EXISTS b"),
+ DropBranch(
+ UnresolvedTable(Seq("t"), "ALTER TABLE ... DROP BRANCH"),
+ "b",
+ ifExists = true))
+ }
+
+ test("alter table: fast forward") {
Review Comment:
Renamed to `"alter table: fastforward branch"`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3300716075
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.NoSuchElementException;
Review Comment:
Does this interface need to rely on `NoSuchElementException`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3305441240
##########
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
Review Comment:
For SQL syntax:
1. When we use this as `verb`, this should be one-word, `fast-forward` or `fastforward`.
2. To be consistent with `CREATE BRANCH` and `DROP BRNCH`, we may need to define `FASTFORWARD BRANCH` like the following.
```
CREATE BRANCH X
DROP BRNCH X
FASTFORWARD BRANCH X
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #56102:
URL: https://github.com/apache/spark/pull/56102#discussion_r3301225769
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.NoSuchElementException;
Review Comment:
Good catch — relying on the generic JDK `NoSuchElementException` makes the interface semantically weaker and inconsistent with the rest of the DSv2 API (e.g. `NoSuchTableException`, `NoSuchPartitionException`).
Fixed in the latest push: dedicated typed exceptions `BranchNotFoundException` and `InvalidFastForwardException` are now defined alongside `BranchAlreadyExistsException`, and the `java.util.NoSuchElementException` import is gone.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsBranching.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.NoSuchElementException;
+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
+ * @throws NoSuchElementException if the requested {@code sourceSnapshotId} does not exist
+ */
+ 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 IllegalArgumentException} otherwise.
+ *
+ * @return the updated {@link TableBranch} after fast-forwarding
+ * @throws NoSuchElementException if either branch is missing
+ * @throws IllegalArgumentException 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];
+ }
+
+ /**
+ * Thrown when a branch with the requested name already exists.
+ */
+ class BranchAlreadyExistsException extends RuntimeException {
Review Comment:
Done. The latest push adds both `BranchNotFoundException` and `InvalidFastForwardException` as nested classes alongside `BranchAlreadyExistsException`, following the same naming pattern (`Exception` suffix). Each method's `@throws` javadoc now references the new typed exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57056][SQL] Add `SupportsBranching` DSv2 interface and branching DDL [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #56102:
URL: https://github.com/apache/spark/pull/56102#issuecomment-4540055016
Thanks for the review @dongjoon-hyun!
Force-pushed `a13783190d4` addressing all four review comments:
- Removed the dependency on `java.util.NoSuchElementException`; introduced typed `BranchNotFoundException` and `InvalidFastForwardException` alongside `BranchAlreadyExistsException`, following the existing DSv2 `NoSuch*Exception` naming pattern.
- Updated `InMemoryTable.fastForward` and `DropBranchExec` to throw the new typed exceptions.
- Fixed the two scalastyle errors (`toLowerCase` without `Locale.ROOT` and `util.Collections.emptySet()`) in `SupportsBranchingSuite`.
- Updated the suite to intercept the new typed exceptions.
`sql/Test/scalastyle` is clean locally and the full `SupportsBranchingSuite` still passes. PTAL when you get a chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org