Skip to content

Spark 4.1 | 4.0 | 3.5 | 3.4: Fail publish_changes procedure if there's more than one matching snapshot #14955

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
SamWheating wants to merge 6 commits into
base: main
Choose a base branch
from
Open

Spark 4.1 | 4.0 | 3.5 | 3.4: Fail publish_changes procedure if there's more than one matching snapshot #14955

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
df314e9
Fail publish_changes procedure if there's multiple matching snapshots
SamWheating Jan 2, 2026
8acad33
rewrite publish_changes procedure to early-exit on duplicated wap.id
SamWheating Jan 19, 2026
98ecb30
Update docs for publish_changes procedure
SamWheating Jan 19, 2026
115b302
run spotlessApply
SamWheating Jan 19, 2026
3b6421e
Update docs/docs/spark-procedures.md
SamWheating Jan 24, 2026
c6502ef
backport fix to spark 3.4, 3.5, 4.0
SamWheating Jan 27, 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
2 changes: 2 additions & 0 deletions docs/docs/spark-procedures.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,8 @@ publish_changes creates a new snapshot from an existing snapshot without alterin

Only append and dynamic overwrite snapshots can be successfully published.

The `publish_changes` procedure will fail if there are multiple snapshots in the table with the provided `wap_id`.

!!! info
This procedure invalidates all cached Spark plans that reference the affected table.

Expand Down
20 changes: 20 additions & 0 deletions ...nsions/src/test/java/org/apache/iceberg/spark/extensions/TestPublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,26 @@ public void testApplyInvalidWapId() {
.hasMessage("Cannot apply unknown WAP ID 'not_valid'");
}

@TestTemplate
public void testApplyDuplicateWapId() {

String wapId = "wap_id_1";

sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'true')", tableName, WRITE_AUDIT_PUBLISH_ENABLED);

spark.conf().set("spark.wap.id", wapId);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b')", tableName);

assertThatThrownBy(
() -> sql("CALL %s.system.publish_changes('%s', '%s')", catalogName, tableIdent, wapId))
.isInstanceOf(ValidationException.class)
.hasMessage(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID 'wap_id_1'");
}

@TestTemplate
public void testInvalidApplyWapChangesCases() {
assertThatThrownBy(
Expand Down
27 changes: 15 additions & 12 deletions ...v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,8 @@
*/
package org.apache.iceberg.spark.procedures;

import java.util.Optional;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.spark.procedures.SparkProcedures.ProcedureBuilder;
import org.apache.iceberg.util.WapUtil;
import org.apache.spark.sql.catalyst.InternalRow;
Expand Down Expand Up @@ -92,21 +90,26 @@ public InternalRow[] call(InternalRow args) {
return modifyIcebergTable(
tableIdent,
table -> {
Optional<Snapshot> wapSnapshot =
Optional.ofNullable(
Iterables.find(
table.snapshots(),
snapshot -> wapId.equals(WapUtil.stagedWapId(snapshot)),
null));
if (!wapSnapshot.isPresent()) {
Snapshot matchingSnap = null;
for (Snapshot snap : table.snapshots()) {
if (wapId.equals(WapUtil.stagedWapId(snap))) {
if (matchingSnap != null) {
throw new ValidationException(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID '%s'",
wapId);
} else {
matchingSnap = snap;
}
}
}

if (matchingSnap == null) {
throw new ValidationException("Cannot apply unknown WAP ID '%s'", wapId);
}

long wapSnapshotId = wapSnapshot.get().snapshotId();
long wapSnapshotId = matchingSnap.snapshotId();
table.manageSnapshots().cherrypick(wapSnapshotId).commit();

Snapshot currentSnapshot = table.currentSnapshot();

InternalRow outputRow = newInternalRow(wapSnapshotId, currentSnapshot.snapshotId());
return new InternalRow[] {outputRow};
});
Expand Down
20 changes: 20 additions & 0 deletions ...nsions/src/test/java/org/apache/iceberg/spark/extensions/TestPublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,26 @@ public void testApplyInvalidWapId() {
.hasMessage("Cannot apply unknown WAP ID 'not_valid'");
}

@TestTemplate
public void testApplyDuplicateWapId() {

String wapId = "wap_id_1";

sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'true')", tableName, WRITE_AUDIT_PUBLISH_ENABLED);

spark.conf().set("spark.wap.id", wapId);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b')", tableName);

assertThatThrownBy(
() -> sql("CALL %s.system.publish_changes('%s', '%s')", catalogName, tableIdent, wapId))
.isInstanceOf(ValidationException.class)
.hasMessage(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID 'wap_id_1'");
}

@TestTemplate
public void testInvalidApplyWapChangesCases() {
assertThatThrownBy(
Expand Down
27 changes: 15 additions & 12 deletions ...v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,8 @@
*/
package org.apache.iceberg.spark.procedures;

import java.util.Optional;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.spark.procedures.SparkProcedures.ProcedureBuilder;
import org.apache.iceberg.util.WapUtil;
import org.apache.spark.sql.catalyst.InternalRow;
Expand Down Expand Up @@ -92,21 +90,26 @@ public InternalRow[] call(InternalRow args) {
return modifyIcebergTable(
tableIdent,
table -> {
Optional<Snapshot> wapSnapshot =
Optional.ofNullable(
Iterables.find(
table.snapshots(),
snapshot -> wapId.equals(WapUtil.stagedWapId(snapshot)),
null));
if (!wapSnapshot.isPresent()) {
Snapshot matchingSnap = null;
for (Snapshot snap : table.snapshots()) {
if (wapId.equals(WapUtil.stagedWapId(snap))) {
if (matchingSnap != null) {
throw new ValidationException(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID '%s'",
wapId);
} else {
matchingSnap = snap;
}
}
}

if (matchingSnap == null) {
throw new ValidationException("Cannot apply unknown WAP ID '%s'", wapId);
}

long wapSnapshotId = wapSnapshot.get().snapshotId();
long wapSnapshotId = matchingSnap.snapshotId();
table.manageSnapshots().cherrypick(wapSnapshotId).commit();

Snapshot currentSnapshot = table.currentSnapshot();

InternalRow outputRow = newInternalRow(wapSnapshotId, currentSnapshot.snapshotId());
return new InternalRow[] {outputRow};
});
Expand Down
20 changes: 20 additions & 0 deletions ...nsions/src/test/java/org/apache/iceberg/spark/extensions/TestPublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,26 @@ public void testApplyInvalidWapId() {
.hasMessage("Cannot apply unknown WAP ID 'not_valid'");
}

@TestTemplate
public void testApplyDuplicateWapId() {

String wapId = "wap_id_1";

sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'true')", tableName, WRITE_AUDIT_PUBLISH_ENABLED);

spark.conf().set("spark.wap.id", wapId);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b')", tableName);

assertThatThrownBy(
() -> sql("CALL %s.system.publish_changes('%s', '%s')", catalogName, tableIdent, wapId))
.isInstanceOf(ValidationException.class)
.hasMessage(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID 'wap_id_1'");
}

@TestTemplate
public void testInvalidApplyWapChangesCases() {
assertThatThrownBy(
Expand Down
27 changes: 15 additions & 12 deletions ...v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,8 @@
package org.apache.iceberg.spark.procedures;

import java.util.Iterator;
import java.util.Optional;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.spark.procedures.SparkProcedures.ProcedureBuilder;
import org.apache.iceberg.util.WapUtil;
import org.apache.spark.sql.catalyst.InternalRow;
Expand Down Expand Up @@ -97,21 +95,26 @@ public Iterator<Scan> call(InternalRow args) {
return modifyIcebergTable(
tableIdent,
table -> {
Optional<Snapshot> wapSnapshot =
Optional.ofNullable(
Iterables.find(
table.snapshots(),
snapshot -> wapId.equals(WapUtil.stagedWapId(snapshot)),
null));
if (!wapSnapshot.isPresent()) {
Snapshot matchingSnap = null;
for (Snapshot snap : table.snapshots()) {
if (wapId.equals(WapUtil.stagedWapId(snap))) {
if (matchingSnap != null) {
throw new ValidationException(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID '%s'",
wapId);
} else {
matchingSnap = snap;
}
}
}

if (matchingSnap == null) {
throw new ValidationException("Cannot apply unknown WAP ID '%s'", wapId);
}

long wapSnapshotId = wapSnapshot.get().snapshotId();
long wapSnapshotId = matchingSnap.snapshotId();
table.manageSnapshots().cherrypick(wapSnapshotId).commit();

Snapshot currentSnapshot = table.currentSnapshot();

InternalRow outputRow = newInternalRow(wapSnapshotId, currentSnapshot.snapshotId());
return asScanIterator(OUTPUT_TYPE, outputRow);
});
Expand Down
20 changes: 20 additions & 0 deletions ...nsions/src/test/java/org/apache/iceberg/spark/extensions/TestPublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,26 @@ public void testApplyInvalidWapId() {
.hasMessage("Cannot apply unknown WAP ID 'not_valid'");
}

@TestTemplate
public void testApplyDuplicateWapId() {

String wapId = "wap_id_1";

sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'true')", tableName, WRITE_AUDIT_PUBLISH_ENABLED);

spark.conf().set("spark.wap.id", wapId);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b')", tableName);

assertThatThrownBy(
() -> sql("CALL %s.system.publish_changes('%s', '%s')", catalogName, tableIdent, wapId))
.isInstanceOf(ValidationException.class)
.hasMessage(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID 'wap_id_1'");
}

@TestTemplate
public void testInvalidApplyWapChangesCases() {
assertThatThrownBy(
Expand Down
27 changes: 15 additions & 12 deletions ...v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,8 @@
package org.apache.iceberg.spark.procedures;

import java.util.Iterator;
import java.util.Optional;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.spark.procedures.SparkProcedures.ProcedureBuilder;
import org.apache.iceberg.util.WapUtil;
import org.apache.spark.sql.catalyst.InternalRow;
Expand Down Expand Up @@ -97,21 +95,26 @@ public Iterator<Scan> call(InternalRow args) {
return modifyIcebergTable(
tableIdent,
table -> {
Optional<Snapshot> wapSnapshot =
Optional.ofNullable(
Iterables.find(
table.snapshots(),
snapshot -> wapId.equals(WapUtil.stagedWapId(snapshot)),
null));
if (!wapSnapshot.isPresent()) {
Snapshot matchingSnap = null;
for (Snapshot snap : table.snapshots()) {
if (wapId.equals(WapUtil.stagedWapId(snap))) {
if (matchingSnap != null) {
throw new ValidationException(
"Cannot apply non-unique WAP ID. Found multiple snapshots with WAP ID '%s'",
wapId);
} else {
matchingSnap = snap;
}
}
}

if (matchingSnap == null) {
throw new ValidationException("Cannot apply unknown WAP ID '%s'", wapId);
}

long wapSnapshotId = wapSnapshot.get().snapshotId();
long wapSnapshotId = matchingSnap.snapshotId();
table.manageSnapshots().cherrypick(wapSnapshotId).commit();

Snapshot currentSnapshot = table.currentSnapshot();

InternalRow outputRow = newInternalRow(wapSnapshotId, currentSnapshot.snapshotId());
return asScanIterator(OUTPUT_TYPE, outputRow);
});
Expand Down