This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 8fe62fb42 fix(table): treat stale snapshot values as commit conflicts
(#2342)
8fe62fb42 is described below
commit 8fe62fb42b778e72472f3571cc42d94c18f176bb
Author: Harrison Crosse <[email protected]>
AuthorDate: Wed Oct 28 15:42:45 2026 -0400
fix(table): treat stale snapshot values as commit conflicts (#2342)
When a peer commits on another branch, the commit branch requirement
still holds, but the metadata builder rejects the staged snapshot
because its sequence number is no longer above last-sequence-number, or
on v3 its first-row-id is behind next-row-id. The SQL, Glue, Hive and
Hadoop catalogs returned that rejection as a plain error, so doCommit
treated it as an unknown commit state: it did not retry, skipped
cleanup, and Transaction.Commit refused a re-commit, although nothing
was written and a refresh and rebuild would land it.
Wrap ErrCommitFailed for these two checks only, as Java does with
RetryableValidationException, which its REST handlers report as a 409.
The row-id rejection still matches ErrInvalidRowLineage. The timestamp
checks and every other builder error stay unchanged, since a rebuild
keeps the snapshot timestamp and would fail the same way.
Signed-off-by: Harrison Crosse <[email protected]>
---
catalog/sql/sql_test.go | 63 +++++++++++++++++++++++++++++++++
table/metadata.go | 9 ++---
table/metadata_builder_internal_test.go | 51 ++++++++++++++++++++++++++
table/no_replay_removals_test.go | 3 +-
table/rewrite_files_test.go | 7 ----
table/table.go | 4 +--
website/src/concurrent-writes.md | 4 ++-
7 files changed, 125 insertions(+), 16 deletions(-)
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index b320359d7..b8a39360e 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -1626,6 +1626,69 @@ func (s *SqliteCatalogTestSuite)
TestStaleAppendRetriesRequirementFailure() {
}
}
+// A peer append on another branch takes the staged snapshot's sequence number
+// or first-row-id.
+func (s *SqliteCatalogTestSuite)
TestStaleAppendRetriesAfterPeerOnAnotherBranch() {
+ ctx := context.Background()
+ tests := []struct {
+ name string
+ formatVersion int
+ numRetries string
+ // The v3 cases seed nothing, so the staged snapshot has no
parent and
+ // only the first-row-id check rejects it.
+ seed []string
+ conflict string
+ }{
+ {"sequence number, retries enabled", 2, "2", []string{"seed"},
""},
+ {"sequence number, retries disabled", 2, "0", []string{"seed"},
"must be > than last sequence number"},
+ {"first-row-id, retries enabled", 3, "2", nil, ""},
+ {"first-row-id, retries disabled", 3, "0", nil, "first-row-id 0
is behind table next-row-id 1"},
+ }
+
+ for _, tt := range tests {
+ s.Run(tt.name, func() {
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat,
tt.numRetries)
+ tbl, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ upgrade := tbl.NewTransaction()
+
s.Require().NoError(upgrade.UpgradeFormatVersion(tt.formatVersion))
+ tbl, err = upgrade.Commit(ctx)
+ s.Require().NoError(err)
+ for _, v := range tt.seed {
+ tbl = s.appendStringRow(ctx, tbl,
table.MainBranch, v)
+ }
+ mainHead := tbl.Metadata().CurrentSnapshot()
+
+ peer := s.appendStringRow(ctx, tbl, "other", "peer")
+ peerHead :=
peer.Metadata().SnapshotByName("other").SnapshotID
+
+ tx := tbl.NewTransaction()
+ s.stageStringRow(ctx, tx, "stale")
+ _, err = tx.Commit(ctx)
+ if tt.numRetries == "0" {
+ s.Require().ErrorIs(err, table.ErrCommitFailed)
+ s.Require().NotErrorIs(err,
table.ErrTransactionUnusable)
+ s.Contains(err.Error(), tt.conflict)
+
+ _, err = tx.Commit(ctx)
+ }
+ s.Require().NoError(err)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.ElementsMatch(append(slices.Clone(tt.seed), "stale"),
s.storedStringValues(ctx, current))
+ if mainHead == nil {
+
s.Nil(current.Metadata().CurrentSnapshot().ParentSnapshotID)
+ } else {
+ s.Equal(mainHead.SnapshotID,
*current.Metadata().CurrentSnapshot().ParentSnapshotID)
+ }
+ s.Equal(peerHead,
current.Metadata().SnapshotByName("other").SnapshotID)
+ })
+ }
+}
+
func (s *SqliteCatalogTestSuite) addStringColumn(ctx context.Context, tbl
*table.Table, name string) *table.Transaction {
tx := tbl.NewTransaction()
s.Require().NoError(tx.UpdateSchema(true, false).
diff --git a/table/metadata.go b/table/metadata.go
index 7ef3c66f7..6c15b229c 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -953,10 +953,11 @@ func (b *MetadataBuilder) addSnapshotInternal(snapshot
*Snapshot, preserveUpdate
} else if b.formatVersion >= 2 &&
snapshot.ParentSnapshotID != nil &&
snapshot.SequenceNumber <= lastSeqNum {
- return fmt.Errorf("can't add snapshot with sequence number %d,
must be > than last sequence number %d",
- snapshot.SequenceNumber, lastSeqNum)
+ return fmt.Errorf("%w: can't add snapshot with sequence number
%d, must be > than last sequence number %d",
+ ErrCommitFailed, snapshot.SequenceNumber, lastSeqNum)
}
+ // A rebuild keeps the timestamp, so these checks are not retryable.
if len(b.snapshotLog) > 0 {
last := b.snapshotLog[len(b.snapshotLog)-1]
if (snapshot.TimestampMs - last.TimestampMs) < -oneMinuteInMs {
@@ -1012,8 +1013,8 @@ func (b *MetadataBuilder)
validateAndUpdateRowLineage(snapshot *Snapshot) error
nextRowID := b.currentNextRowID()
if *snapshot.FirstRowID < nextRowID {
- return fmt.Errorf("%w: first-row-id %d is behind table
next-row-id %d",
- ErrInvalidRowLineage, *snapshot.FirstRowID, nextRowID)
+ return fmt.Errorf("%w: %w: first-row-id %d is behind table
next-row-id %d",
+ ErrCommitFailed, ErrInvalidRowLineage,
*snapshot.FirstRowID, nextRowID)
}
// next-row-id is non-decreasing across snapshots, not strictly
increasing:
diff --git a/table/metadata_builder_internal_test.go
b/table/metadata_builder_internal_test.go
index a740c6f44..e8dbfa308 100644
--- a/table/metadata_builder_internal_test.go
+++ b/table/metadata_builder_internal_test.go
@@ -1928,6 +1928,57 @@ func TestAddSnapshotOnBuilderWithoutBaseV3(t *testing.T)
{
require.Equal(t, addedRows, meta.NextRowID())
}
+func TestAddSnapshotStaleValuesAreCommitConflicts(t *testing.T) {
+ now := time.Now().UnixMilli()
+ parentID := int64(1)
+
+ t.Run("sequence number taken by a peer", func(t *testing.T) {
+ builder := freshMetadataBuilder(t, 2)
+ peer := freshBuilderSnapshot(1, nil, 1, now)
+ require.NoError(t, builder.AddSnapshot(&peer))
+ stale := freshBuilderSnapshot(2, &parentID, 1, now)
+
+ err := builder.AddSnapshot(&stale)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.ErrorContains(t, err, "can't add snapshot with sequence
number 1, must be > than last sequence number 1")
+ })
+
+ t.Run("first-row-id behind next-row-id", func(t *testing.T) {
+ builder := freshMetadataBuilder(t, 3)
+ firstRowID, addedRows := int64(0), int64(10)
+ peer := freshBuilderSnapshot(1, nil, 1, now)
+ peer.FirstRowID, peer.AddedRows = &firstRowID, &addedRows
+ require.NoError(t, builder.AddSnapshot(&peer))
+ stale := freshBuilderSnapshot(2, &parentID, 2, now)
+ stale.FirstRowID, stale.AddedRows = &firstRowID, &addedRows
+
+ err := builder.AddSnapshot(&stale)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.ErrorIs(t, err, ErrInvalidRowLineage)
+ require.ErrorContains(t, err, "first-row-id 0 is behind table
next-row-id 10")
+ })
+
+ t.Run("timestamp before last update", func(t *testing.T) {
+ builder := freshMetadataBuilder(t, 2)
+ first := freshBuilderSnapshot(1, nil, 1, now)
+ require.NoError(t, builder.AddSnapshot(&first))
+ stale := freshBuilderSnapshot(2, &parentID, 2,
now-2*oneMinuteInMs)
+
+ err := builder.AddSnapshot(&stale)
+ require.ErrorContains(t, err, "before last updated timestamp")
+ require.NotErrorIs(t, err, ErrCommitFailed)
+ })
+
+ t.Run("missing first-row-id", func(t *testing.T) {
+ builder := freshMetadataBuilder(t, 3)
+ snap := freshBuilderSnapshot(1, nil, 1, now)
+
+ err := builder.AddSnapshot(&snap)
+ require.ErrorIs(t, err, ErrInvalidRowLineage)
+ require.NotErrorIs(t, err, ErrCommitFailed)
+ })
+}
+
func TestAddSnapshotUpdateOnBuilderWithoutBase(t *testing.T) {
builder := freshMetadataBuilder(t, 2)
parentSnapshotID := int64(1)
diff --git a/table/no_replay_removals_test.go b/table/no_replay_removals_test.go
index 0cf55062d..2d1dfd6fb 100644
--- a/table/no_replay_removals_test.go
+++ b/table/no_replay_removals_test.go
@@ -241,8 +241,7 @@ func TestReplaceFilesWithRemovalsFailsInsteadOfReplaying(t
*testing.T) {
// unchanged but takes the staged snapshot's sequence number. The resume of a
// non-replayable compaction rebuilds onto the same pinned parent and lands.
func TestReplaceFilesWithRemovalsResumesAfterPeerOnAnotherBranch(t *testing.T)
{
- tbl, cat := newConcurrentRewriteTestTable(t)
- cat.restConflicts = true
+ tbl, _ := newConcurrentRewriteTestTable(t)
arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
require.NoError(t, err)
diff --git a/table/rewrite_files_test.go b/table/rewrite_files_test.go
index 9ddb30683..df2c512e8 100644
--- a/table/rewrite_files_test.go
+++ b/table/rewrite_files_test.go
@@ -53,9 +53,6 @@ type concurrentTestCatalog struct {
location string
fsF table.FSysF
attempts atomic.Int32
- // restConflicts reports a metadata builder rejection as a conflict, as
- // a REST catalog's 409 does.
- restConflicts bool
}
func (c *concurrentTestCatalog) LoadTable(_ context.Context, ident
table.Identifier) (*table.Table, error) {
@@ -71,10 +68,6 @@ func (c *concurrentTestCatalog) CommitTable(_
context.Context, _ table.Identifie
}
meta, err := table.UpdateTableMetadata(c.metadata, updates, "")
if err != nil {
- if c.restConflicts {
- return nil, "", fmt.Errorf("%w: %w",
table.ErrCommitFailed, err)
- }
-
return nil, "", err
}
c.metadata = meta
diff --git a/table/table.go b/table/table.go
index d0690372b..db12ced2f 100644
--- a/table/table.go
+++ b/table/table.go
@@ -1065,8 +1065,8 @@ func latestSnapshotForBranch(meta Metadata, branch
string) *Snapshot {
// stagedSnapshotStillValid reports whether a snapshot built on an earlier
// attempt would still be accepted by AddSnapshot against fresh. Both values it
// checks are table-wide, so a peer committing on ANOTHER branch invalidates
them
-// without moving this branch's head — the case the caller's parent check
misses,
-// and one AddSnapshot rejects terminally rather than retryably.
+// without moving this branch's head, which the caller's parent check misses.
+// Resubmitting such a snapshot unchanged would fail every retry.
func stagedSnapshotStillValid(snap *Snapshot, fresh Metadata) bool {
if snap == nil || fresh == nil {
return true
diff --git a/website/src/concurrent-writes.md b/website/src/concurrent-writes.md
index 6e830ba62..f3b0756cc 100644
--- a/website/src/concurrent-writes.md
+++ b/website/src/concurrent-writes.md
@@ -98,7 +98,9 @@ _, err := cat.CreateTable(ctx, ident, schema,
> **Catalog support:** the REST, SQL, Glue, Hive, and Hadoop catalogs report
> commit conflicts, including requirements that fail because of a concurrent
-> change, as `ErrCommitFailed`, so the retry loop engages on all of them.
+> change, as `ErrCommitFailed`, so the retry loop engages on all of them. A
+> commit on another branch that takes the staged snapshot's sequence number or
+> first-row-id is a conflict too, and the retry renumbers the snapshot.
Only the implicit assertion on the commit branch's head is rebased after a
refresh. Other requirements, including the schema, spec, sort order, and pinned
