You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by am...@apache.org on 2026/03/10 16:00:18 UTC
(iceberg) branch main updated: Core: Make sequence number conflicts retryable when there are concurrent commits (#15126)
This is an automated email from the ASF dual-hosted git repository.
amoghj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 1d6c685c13 Core: Make sequence number conflicts retryable when there are concurrent commits (#15126)
1d6c685c13 is described below
commit 1d6c685c1391df604d59f3335a1fba7a1af2c25f
Author: Xinyi Lu <12...@users.noreply.github.com>
AuthorDate: Tue Mar 10 09:00:11 2026 -0700
Core: Make sequence number conflicts retryable when there are concurrent commits (#15126)
---
.../iceberg/RetryableValidationException.java | 48 +++++++++++++
.../java/org/apache/iceberg/TableMetadata.java | 4 +-
.../org/apache/iceberg/rest/CatalogHandlers.java | 13 +++-
.../java/org/apache/iceberg/TestTableMetadata.java | 78 ++++++++++++++++++++++
.../org/apache/iceberg/rest/TestRESTCatalog.java | 57 ++++++++++++++++
5 files changed, 197 insertions(+), 3 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/RetryableValidationException.java b/core/src/main/java/org/apache/iceberg/RetryableValidationException.java
new file mode 100644
index 0000000000..0dbff48745
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/RetryableValidationException.java
@@ -0,0 +1,48 @@
+/*
+ * 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.iceberg;
+
+import com.google.errorprone.annotations.FormatMethod;
+import org.apache.iceberg.exceptions.ValidationException;
+
+/**
+ * A {@link ValidationException} that indicates a validation failure that can be fixed and retried.
+ *
+ * <p>This is specifically not a conflict. This is used when a validation failed because the commit
+ * includes stale values, such as a sequence number or first-row-id that is behind the current table
+ * state. Retrying the commit with refreshed metadata can resolve the failure.
+ */
+public class RetryableValidationException extends ValidationException {
+ @FormatMethod
+ public RetryableValidationException(String message, Object... args) {
+ super(message, args);
+ }
+
+ @FormatMethod
+ public RetryableValidationException(Throwable cause, String message, Object... args) {
+ super(cause, message, args);
+ }
+
+ @FormatMethod
+ public static void check(boolean test, String message, Object... args) {
+ if (!test) {
+ throw new RetryableValidationException(message, args);
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 5aa76d5680..43a67dd2be 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -1250,7 +1250,7 @@ public class TableMetadata implements Serializable {
"Snapshot already exists for id: %s",
snapshot.snapshotId());
- ValidationException.check(
+ RetryableValidationException.check(
formatVersion == 1
|| snapshot.sequenceNumber() > lastSequenceNumber
|| snapshot.parentId() == null,
@@ -1266,7 +1266,7 @@ public class TableMetadata implements Serializable {
if (formatVersion >= MIN_FORMAT_VERSION_ROW_LINEAGE) {
ValidationException.check(
snapshot.firstRowId() != null, "Cannot add a snapshot: first-row-id is null");
- ValidationException.check(
+ RetryableValidationException.check(
snapshot.firstRowId() != null && snapshot.firstRowId() >= nextRowId,
"Cannot add a snapshot, first-row-id is behind table next-row-id: %s < %s",
snapshot.firstRowId(),
diff --git a/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java b/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
index 2e88d52649..d75909b437 100644
--- a/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
+++ b/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
@@ -53,6 +53,7 @@ import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.IncrementalAppendScan;
import org.apache.iceberg.MetadataUpdate.UpgradeFormatVersion;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RetryableValidationException;
import org.apache.iceberg.Scan;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
@@ -634,7 +635,17 @@ public class CatalogHandlers {
// apply changes
TableMetadata.Builder metadataBuilder = TableMetadata.buildFrom(base);
- request.updates().forEach(update -> update.applyTo(metadataBuilder));
+ try {
+ request.updates().forEach(update -> update.applyTo(metadataBuilder));
+ } catch (RetryableValidationException e) {
+ // Validation failed because the commit includes stale values (e.g. sequence
+ // number or first-row-id behind the current table state). This is not a conflict.
+ // Server-side retry won't help since the stale values are in the request itself.
+ // Wrap as CommitFailedException so the client can retry with refreshed metadata.
+ throw new ValidationFailureException(
+ new CommitFailedException(
+ e, "Validation failed, please retry: %s", e.getMessage()));
+ }
TableMetadata updated = metadataBuilder.build();
if (updated.changes().isEmpty()) {
diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
index bddb878593..cb1decd2d8 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
@@ -1990,4 +1990,82 @@ public class TestTableMetadata {
assertThat(meta.changes()).anyMatch(u -> u instanceof MetadataUpdate.RemoveSchemas);
}
+
+ @Test
+ public void testAddSnapshotWithStaleSequenceNumberIsRetryable() {
+ TableMetadata base =
+ TableMetadata.newTableMetadata(
+ TEST_SCHEMA, PartitionSpec.unpartitioned(), "location", ImmutableMap.of());
+
+ // Advance lastSequenceNumber to 1 by adding a root snapshot
+ Snapshot s1 =
+ new BaseSnapshot(
+ 1,
+ 1L,
+ null,
+ System.currentTimeMillis(),
+ null,
+ null,
+ null,
+ "file:/s1.avro",
+ null,
+ null,
+ null);
+ TableMetadata withS1 = TableMetadata.buildFrom(base).addSnapshot(s1).build();
+
+ // A snapshot with seqNum=1 and non-null parentId is stale (1 is not > lastSequenceNumber=1)
+ Snapshot staleSnapshot =
+ new BaseSnapshot(
+ 1,
+ 2L,
+ 1L,
+ System.currentTimeMillis(),
+ null,
+ null,
+ null,
+ "file:/s2.avro",
+ null,
+ null,
+ null);
+
+ assertThatThrownBy(() -> TableMetadata.buildFrom(withS1).addSnapshot(staleSnapshot))
+ .isInstanceOf(RetryableValidationException.class)
+ .hasMessageContaining("Cannot add snapshot with sequence number");
+ }
+
+ @Test
+ public void testAddSnapshotWithStaleFirstRowIdIsRetryable() {
+ TableMetadata base =
+ TableMetadata.newTableMetadata(
+ TEST_SCHEMA,
+ PartitionSpec.unpartitioned(),
+ "location",
+ ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
+
+ // Advance nextRowId to 5 by adding a snapshot that allocates 5 rows
+ Snapshot s1 =
+ new BaseSnapshot(
+ 1,
+ 1L,
+ null,
+ System.currentTimeMillis(),
+ null,
+ null,
+ null,
+ "file:/s1.avro",
+ 0L,
+ 5L,
+ null);
+ TableMetadata withS1 = TableMetadata.buildFrom(base).addSnapshot(s1).build();
+ // nextRowId is now 5
+
+ // A snapshot with firstRowId=0 is stale (0 < nextRowId=5)
+ Snapshot staleSnapshot =
+ new BaseSnapshot(
+ 2, 2L, 1L, System.currentTimeMillis(), null, null, null, "file:/s2.avro", 0L, 1L, null);
+
+ assertThatThrownBy(() -> TableMetadata.buildFrom(withS1).addSnapshot(staleSnapshot))
+ .isInstanceOf(RetryableValidationException.class)
+ .hasMessageContaining("Cannot add a snapshot, first-row-id is behind table next-row-id");
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
index d03c5f9b88..56ab8f97f8 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
@@ -3683,6 +3683,63 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
any());
}
+ @Test
+ public void testSequenceNumberConflictThrowsCommitFailed() {
+ RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
+ RESTCatalog catalog = catalog(adapter);
+
+ catalog.createNamespace(TABLE.namespace());
+ catalog.buildTable(TABLE, SCHEMA).withPartitionSpec(SPEC).create();
+
+ DataFile fileOnMain =
+ DataFiles.builder(SPEC)
+ .withPath("/path/commit-test-file1.parquet")
+ .withFileSizeInBytes(10)
+ .withPartitionPath("id_bucket=0")
+ .withRecordCount(1)
+ .build();
+
+ catalog.loadTable(TABLE).newFastAppend().appendFile(fileOnMain).commit();
+
+ DataFile fileOnAnotherBranch =
+ DataFiles.builder(SPEC)
+ .withPath("/path/commit-test-conflicting.parquet")
+ .withFileSizeInBytes(10)
+ .withPartitionPath("id_bucket=0")
+ .withRecordCount(1)
+ .build();
+
+ // Before the next commit is processed by the server, advance the server's lastSequenceNumber
+ // by committing to a different branch. This simulates a concurrent request to a different
+ // branch
+ // that "beats" the commit to main.
+ Mockito.doAnswer(
+ invocation -> {
+ backendCatalog
+ .loadTable(TABLE)
+ .newFastAppend()
+ .appendFile(fileOnAnotherBranch)
+ .toBranch("other")
+ .commit();
+ return invocation.callRealMethod();
+ })
+ .when(adapter)
+ .execute(matches(HTTPMethod.POST, RESOURCE_PATHS.table(TABLE)), any(), any(), any());
+
+ DataFile anotherFileOnMain =
+ DataFiles.builder(SPEC)
+ .withPath("/path/commit-test-file2.parquet")
+ .withFileSizeInBytes(10)
+ .withPartitionPath("id_bucket=0")
+ .withRecordCount(1)
+ .build();
+
+ assertThatThrownBy(
+ () -> catalog.loadTable(TABLE).newFastAppend().appendFile(anotherFileOnMain).commit())
+ .isInstanceOf(CommitFailedException.class)
+ .hasMessageContaining("Validation failed, please retry");
+ }
+
private RESTCatalog catalog(RESTCatalogAdapter adapter) {
RESTCatalog catalog =
new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);