diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java index 2f98aec4e705..34ac89fe6dcf 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java @@ -490,6 +490,17 @@ public boolean caseSensitive() { } public String branch() { + String optionBranch = confParser.stringConf().option(SparkWriteOptions.BRANCH).parseOptional(); + ValidationException.check( + branch == null || optionBranch == null || optionBranch.equals(branch), + "Must not specify different branches in both table identifier and write option, " + + "got [%s] in identifier and [%s] in options", + branch, + optionBranch); + if (optionBranch != null) { + return optionBranch; + } + if (wapEnabled()) { String wapId = wapId(); String wapBranch = diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java index 09bfd762b7e6..a4a02de0dc21 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java @@ -23,6 +23,9 @@ public class SparkWriteOptions { private SparkWriteOptions() {} + // Overrides the target branch for write operations + public static final String BRANCH = "branch"; + // Fileformat for write operations(default: Table write.format.default ) public static final String WRITE_FORMAT = "write-format"; diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java index 89daf195ca73..46f7fdc0567a 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java @@ -58,6 +58,7 @@ import org.apache.iceberg.TableProperties; import org.apache.iceberg.UpdateProperties; import org.apache.iceberg.deletes.DeleteGranularity; +import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.spark.sql.internal.SQLConf; @@ -84,6 +85,55 @@ public void after() { sql("DROP TABLE IF EXISTS %s", tableName); } + @TestTemplate + public void writeBranchOption() { + Table table = validationCatalog.loadTable(tableIdent); + + SparkWriteConf branchFromOption = + new SparkWriteConf( + spark, table, null, ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + assertThat(branchFromOption.branch()).isEqualTo("branchA"); + + SparkWriteConf matchingIdentifierAndOption = + new SparkWriteConf( + spark, table, "branchA", ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + assertThat(matchingIdentifierAndOption.branch()).isEqualTo("branchA"); + } + + @TestTemplate + public void writeBranchOptionConflictsWithIdentifier() { + Table table = validationCatalog.loadTable(tableIdent); + + SparkWriteConf conflicting = + new SparkWriteConf( + spark, table, "branchB", ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + + assertThatThrownBy(conflicting::branch) + .isInstanceOf(ValidationException.class) + .hasMessageContaining( + "Must not specify different branches in both table identifier and write option"); + } + + @TestTemplate + public void writeBranchOptionTakesPrecedenceOverWapBranch() { + Table table = validationCatalog.loadTable(tableIdent); + table.updateProperties().set(TableProperties.WRITE_AUDIT_PUBLISH_ENABLED, "true").commit(); + + withSQLConf( + ImmutableMap.of(SparkSQLProperties.WAP_BRANCH, "wapBranch"), + () -> { + // With WAP enabled and no write option, the session WAP branch is used. + SparkWriteConf wapConf = new SparkWriteConf(spark, table, null, ImmutableMap.of()); + assertThat(wapConf.branch()).isEqualTo("wapBranch"); + + // The branch write option takes precedence over the session WAP branch. + SparkWriteConf optionConf = + new SparkWriteConf( + spark, table, null, ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + assertThat(optionConf.branch()).isEqualTo("branchA"); + }); + } + @TestTemplate public void testOptionCaseInsensitive() { Table table = validationCatalog.loadTable(tableIdent); diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java index 3a34a1eb83a0..f8a21955e63f 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java @@ -496,6 +496,17 @@ public boolean caseSensitive() { } public String branch() { + String optionBranch = confParser.stringConf().option(SparkWriteOptions.BRANCH).parseOptional(); + ValidationException.check( + branch == null || optionBranch == null || optionBranch.equals(branch), + "Must not specify different branches in both table identifier and write option, " + + "got [%s] in identifier and [%s] in options", + branch, + optionBranch); + if (optionBranch != null) { + return optionBranch; + } + if (wapEnabled()) { String wapId = wapId(); String wapBranch = diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java index b7a5597aea14..f945d36fb5e7 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java @@ -23,6 +23,9 @@ public class SparkWriteOptions { private SparkWriteOptions() {} + // Overrides the target branch for write operations + public static final String BRANCH = "branch"; + // Fileformat for write operations(default: Table write.format.default ) public static final String WRITE_FORMAT = "write-format"; diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java index c5cfbe62b1be..f4e22091ce40 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java @@ -59,6 +59,7 @@ import org.apache.iceberg.TableProperties; import org.apache.iceberg.UpdateProperties; import org.apache.iceberg.deletes.DeleteGranularity; +import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.spark.sql.internal.SQLConf; @@ -86,6 +87,55 @@ public void after() { sql("DROP TABLE IF EXISTS %s", tableName); } + @TestTemplate + public void writeBranchOption() { + Table table = validationCatalog.loadTable(tableIdent); + + SparkWriteConf branchFromOption = + new SparkWriteConf( + spark, table, null, ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + assertThat(branchFromOption.branch()).isEqualTo("branchA"); + + SparkWriteConf matchingIdentifierAndOption = + new SparkWriteConf( + spark, table, "branchA", ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + assertThat(matchingIdentifierAndOption.branch()).isEqualTo("branchA"); + } + + @TestTemplate + public void writeBranchOptionConflictsWithIdentifier() { + Table table = validationCatalog.loadTable(tableIdent); + + SparkWriteConf conflicting = + new SparkWriteConf( + spark, table, "branchB", ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + + assertThatThrownBy(conflicting::branch) + .isInstanceOf(ValidationException.class) + .hasMessageContaining( + "Must not specify different branches in both table identifier and write option"); + } + + @TestTemplate + public void writeBranchOptionTakesPrecedenceOverWapBranch() { + Table table = validationCatalog.loadTable(tableIdent); + table.updateProperties().set(TableProperties.WRITE_AUDIT_PUBLISH_ENABLED, "true").commit(); + + withSQLConf( + ImmutableMap.of(SparkSQLProperties.WAP_BRANCH, "wapBranch"), + () -> { + // With WAP enabled and no write option, the session WAP branch is used. + SparkWriteConf wapConf = new SparkWriteConf(spark, table, null, ImmutableMap.of()); + assertThat(wapConf.branch()).isEqualTo("wapBranch"); + + // The branch write option takes precedence over the session WAP branch. + SparkWriteConf optionConf = + new SparkWriteConf( + spark, table, null, ImmutableMap.of(SparkWriteOptions.BRANCH, "branchA")); + assertThat(optionConf.branch()).isEqualTo("branchA"); + }); + } + @TestTemplate public void testOptionCaseInsensitive() { Table table = validationCatalog.loadTable(tableIdent);