Skip to content

Spark: Backport the branch write options to Spark 3.5 and Spark 4.0 #17842

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nimesh1601 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Spark: Backport the branch write options to Spark 3.5 and Spark 4.0 #17842

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a1e289a
Spark 3.5, 4.0: Add branch write option to SparkWriteOptions
nimesh1601 Aug 27, 2026
9efd0c0
adding tests for precedence
nimesh1601 Aug 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down
3 changes: 3 additions & 0 deletions spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down
50 changes: 50 additions & 0 deletions spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
Expand Down
11 changes: 11 additions & 0 deletions spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down
3 changes: 3 additions & 0 deletions spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down
50 changes: 50 additions & 0 deletions spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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");
}

@uros-b uros-b Aug 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No test exercises the PR's headline behavioral guarantee: option branch takes precedence over a session WAP branch (i.e., optionBranch != null while wapEnabled() == true && wapBranch != null -> returns option branch, bypassing the WAP guards). This is the one new code path introduced by the PR; the companion docs PR #17827 explicitly documents this precedence, but it is unverified by any test in either module. Please add a case asserting the option wins when WAP is also configured.

@nimesh1601 nimesh1601 Aug 29, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the review @uros-b ! you're right that the option-wins-over-WAP path wasn't covered. I've added a test for it to  TestSparkWriteConf  in both the v3.5 and v4.0 modules. While adding this, I noticed the original v4.1 PR (#15288) doesn't unit-test the option ->branch resolution either - it only changed the SparkWriteConf  constructors mechanically and covered branch precedence via the SQL-identifier integration tests. So this actually adds coverage the source PR lacked. Happy to open a small follow-up PR adding the sameTestSparkWriteConf  case to v4.1 for parity if that's useful.

@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);
Expand Down
Loading