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 bba86e983 fix(table): pin guarded refs in RollbackToSnapshot and
ExpireSnapshots (#2301)
bba86e983 is described below
commit bba86e9838efacfb9fce566d238ca215ae2d9301
Author: Harrison Crosse <[email protected]>
AuthorDate: Mon Oct 26 13:16:20 2026 -0400
fix(table): pin guarded refs in RollbackToSnapshot and ExpireSnapshots
(#2301)
* fix(table): pin guarded refs in RollbackToSnapshot and ExpireSnapshots
Since #2248 the SQL, Glue, Hive and Hadoop catalogs report a failed
commit requirement as ErrCommitFailed, so with commit.retry.num-retries
above zero doCommit retries it and rewrites the commit branch's
AssertRefSnapshotID to the refreshed head. RollbackToSnapshot and
ExpireSnapshots rely on that assertion to detect a concurrent change, so
a stale rollback could discard a peer's append and a stale expiry could
remove snapshots the moved branch still retains. REST was already
affected through 409 responses.
Pin the refs these operations guard, including branches the transaction
staged, using the same mechanism as Transaction.AssertRefSnapshotID.
Pins and the staged updates are now published under one lock, so a
failed apply leaves no pin. A pinned assertion that no longer holds
after refresh fails the commit without spending the rest of the retry
budget.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): skip guards for a no-op ExpireSnapshots
An ExpireSnapshots call that finds nothing to expire stages no update,
but it still pinned every ref. In a transaction that also appends, the
pin on the commit branch stopped the append from rebasing onto a peer's
commit, so the retry failed instead of succeeding.
Return without adding requirements or pins when there is nothing to
expire, while still rejecting a transaction that has already been
committed.
Signed-off-by: Harrison Crosse <[email protected]>
* test: trim pinned-ref test helpers and clarify retry docs
Signed-off-by: Harrison Crosse <[email protected]>
* docs: list every built-in catalog in the retry support note
Signed-off-by: Harrison Crosse <[email protected]>
* docs: limit the catalog support note to conflicting requirement failures
Signed-off-by: Harrison Crosse <[email protected]>
---------
Signed-off-by: Harrison Crosse <[email protected]>
Co-authored-by: Matt Topol <[email protected]>
---
catalog/sql/sql_test.go | 243 +++++++++++++++++++++++++++++++++++
table/table.go | 15 +--
table/transaction.go | 85 +++++++-----
table/transaction_assert_ref_test.go | 26 ++++
website/src/concurrent-writes.md | 12 +-
5 files changed, 335 insertions(+), 46 deletions(-)
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index c2250d7c7..7e74d8b39 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -18,6 +18,7 @@
package sql_test
import (
+ "cmp"
"context"
"database/sql"
"database/sql/driver"
@@ -1496,6 +1497,248 @@ func (s *SqliteCatalogTestSuite)
TestDropTableNotExist() {
}
}
+// stageStringRow stages one row for the string field "foo" on tx.
+func (s *SqliteCatalogTestSuite) stageStringRow(ctx context.Context, tx
*table.Transaction, value string) {
+ arrowSchema := arrow.NewSchema([]arrow.Field{{Name: "foo", Type:
arrow.BinaryTypes.String}}, nil)
+ bldr := array.NewStringBuilder(memory.DefaultAllocator)
+ defer bldr.Release()
+ bldr.Append(value)
+ arr := bldr.NewArray()
+ defer arr.Release()
+ rec := array.NewRecordBatch(arrowSchema, []arrow.Array{arr}, 1)
+ defer rec.Release()
+ arrTable := array.NewTableFromRecords(arrowSchema,
[]arrow.RecordBatch{rec})
+ defer arrTable.Release()
+
+ s.Require().NoError(tx.AppendTable(ctx, arrTable, 1024, nil))
+}
+
+// appendStringRow commits one row to branch, creating the branch from
+// main's head if it does not exist.
+func (s *SqliteCatalogTestSuite) appendStringRow(ctx context.Context, tbl
*table.Table, branch, value string) *table.Table {
+ tx, err := tbl.NewTransactionOnBranch(branch)
+ s.Require().NoError(err)
+ s.stageStringRow(ctx, tx, value)
+ out, err := tx.Commit(ctx)
+ s.Require().NoError(err)
+
+ return out
+}
+
+// waitPastSnapshots blocks until the wall clock in milliseconds is past
+// every snapshot timestamp of meta, so an expiry with WithOlderThan(0)
+// treats them all as old enough.
+func (s *SqliteCatalogTestSuite) waitPastSnapshots(meta table.Metadata) {
+ var newest int64
+ for _, snap := range meta.Snapshots() {
+ newest = max(newest, snap.TimestampMs)
+ }
+ s.Require().Eventually(func() bool { return time.Now().UnixMilli() >
newest },
+ time.Second, time.Millisecond)
+}
+
+// createRetryingStringTable creates a single-column string table that
+// retries conflicting commits numRetries times with millisecond waits,
+// then appends one row per value.
+func (s *SqliteCatalogTestSuite) createRetryingStringTable(ctx
context.Context, cat *sqlcat.Catalog, numRetries string, values ...string)
table.Identifier {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String,
Required: true,
+ })
+ tblID := s.randomTableIdentifier()
+ s.Require().NoError(cat.CreateNamespace(ctx,
catalog.NamespaceFromIdent(tblID), nil))
+
+ tbl, err := cat.CreateTable(ctx, tblID, schema,
catalog.WithProperties(iceberg.Properties{
+ table.CommitNumRetriesKey: numRetries,
+ table.CommitMinRetryWaitMsKey: "1",
+ table.CommitMaxRetryWaitMsKey: "2",
+ }))
+ s.Require().NoError(err)
+ for _, v := range values {
+ tbl = s.appendStringRow(ctx, tbl, table.MainBranch, v)
+ }
+
+ return tblID
+}
+
+// An ExpireSnapshots call with nothing to expire must not stop the
+// transaction's append from rebasing onto a peer's append.
+func (s *SqliteCatalogTestSuite)
TestStaleAppendWithNoOpExpireRetriesAfterPeerAppend() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "2", "seed")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ tx := stale.NewTransaction()
+ s.stageStringRow(ctx, tx, "stale")
+ s.Require().NoError(tx.ExpireSnapshots())
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, table.MainBranch, "peer")
+ peerHead := peer.Metadata().CurrentSnapshot().SnapshotID
+
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Len(current.Metadata().Snapshots(), 3)
+ parent := current.Metadata().CurrentSnapshot().ParentSnapshotID
+ s.Require().NotNil(parent)
+ s.Equal(peerHead, *parent)
+}
+
+// A rollback staged before a peer advanced the branch must not be
+// replayed against the new head, which would drop the peer's append.
+func (s *SqliteCatalogTestSuite) TestStaleRollbackFailsAfterPeerAppend() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "2", "seed", "second")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ first := stale.Metadata().CurrentSnapshot().ParentSnapshotID
+ s.Require().NotNil(first)
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, table.MainBranch, "peer")
+ peerHead := peer.Metadata().CurrentSnapshot().SnapshotID
+
+ tx := stale.NewTransaction()
+ s.Require().NoError(tx.RollbackToSnapshot(*first))
+ _, err = tx.Commit(ctx)
+ s.Require().Error(err)
+ s.ErrorIs(err, table.ErrCommitFailed)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(peerHead, current.Metadata().CurrentSnapshot().SnapshotID)
+ s.Len(current.Metadata().Snapshots(), 3)
+}
+
+// Snapshot expiry staged before a peer rolled the branch back must not
+// be replayed against the new head: its retention window was computed
+// from the old head, so it would expire snapshots the rolled-back
+// branch still retains.
+func (s *SqliteCatalogTestSuite)
TestStaleExpireSnapshotsFailsAfterPeerRollback() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "2", "s1", "s2", "s3",
"s4")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ snaps := slices.Clone(stale.Metadata().Snapshots())
+ s.Require().Len(snaps, 4)
+ slices.SortFunc(snaps, func(a, b table.Snapshot) int { return
cmp.Compare(a.SequenceNumber, b.SequenceNumber) })
+ ids := make([]int64, len(snaps))
+ for i, snap := range snaps {
+ ids[i] = snap.SnapshotID
+ }
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peerTx := peer.NewTransaction()
+ s.Require().NoError(peerTx.RollbackToSnapshot(ids[1]))
+ _, err = peerTx.Commit(ctx)
+ s.Require().NoError(err)
+
+ s.waitPastSnapshots(stale.Metadata())
+
+ tx := stale.NewTransaction()
+ s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(3),
table.WithOlderThan(0), table.WithPostCommit(false)))
+ _, err = tx.Commit(ctx)
+ s.Require().Error(err)
+ s.ErrorIs(err, table.ErrCommitFailed)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(ids[1], current.Metadata().CurrentSnapshot().SnapshotID)
+ var remaining []int64
+ for _, snap := range current.Metadata().Snapshots() {
+ remaining = append(remaining, snap.SnapshotID)
+ }
+ s.ElementsMatch(ids, remaining)
+}
+
+// A rollback of a branch the transaction staged must not be replayed
+// against a branch a peer created meanwhile, which would point the
+// peer's branch back at main and drop its append.
+func (s *SqliteCatalogTestSuite)
TestStaleRollbackOfStagedBranchFailsAfterPeerCreatesIt() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "2", "seed", "second")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ mainHead := stale.Metadata().CurrentSnapshot().SnapshotID
+
+ tx, err := stale.NewTransactionOnBranch("feature")
+ s.Require().NoError(err)
+ s.stageStringRow(ctx, tx, "stale")
+ s.Require().NoError(tx.RollbackToSnapshot(mainHead))
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, "feature", "peer")
+ peerHead := peer.Metadata().SnapshotByName("feature").SnapshotID
+
+ _, err = tx.Commit(ctx)
+ s.Require().Error(err)
+ s.ErrorIs(err, table.ErrCommitFailed)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(peerHead,
current.Metadata().SnapshotByName("feature").SnapshotID)
+ s.Equal(mainHead, current.Metadata().CurrentSnapshot().SnapshotID)
+ s.Len(current.Metadata().Snapshots(), 3)
+}
+
+// Snapshot expiry in a transaction that staged a new branch must not be
+// replayed against a branch a peer created meanwhile: its retention was
+// computed without the peer's branch.
+func (s *SqliteCatalogTestSuite)
TestStaleExpireWithStagedBranchFailsAfterPeerCreatesIt() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "2", "seed", "second")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, "feature", "peer")
+ peerHead := peer.Metadata().SnapshotByName("feature").SnapshotID
+ s.waitPastSnapshots(peer.Metadata())
+
+ tx, err := stale.NewTransactionOnBranch("feature")
+ s.Require().NoError(err)
+ s.stageStringRow(ctx, tx, "stale")
+ s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(1),
table.WithOlderThan(0), table.WithPostCommit(false)))
+ _, err = tx.Commit(ctx)
+ s.Require().Error(err)
+ s.ErrorIs(err, table.ErrCommitFailed)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(peerHead,
current.Metadata().SnapshotByName("feature").SnapshotID)
+ var remaining []int64
+ for _, snap := range current.Metadata().Snapshots() {
+ remaining = append(remaining, snap.SnapshotID)
+ }
+ var expected []int64
+ for _, snap := range peer.Metadata().Snapshots() {
+ expected = append(expected, snap.SnapshotID)
+ }
+ s.ElementsMatch(expected, remaining)
+}
+
func (s *SqliteCatalogTestSuite) TestPurgeTable() {
tests := []struct {
cat *sqlcat.Catalog
diff --git a/table/table.go b/table/table.go
index ad2c6792b..0ab929594 100644
--- a/table/table.go
+++ b/table/table.go
@@ -564,9 +564,8 @@ type commitOpts struct {
// in snapshotProducer.commitManifests for the rationale.
noReplay bool
- // pinnedRefs names branches with an explicit requirement from
- // Transaction.AssertRefSnapshotID, whose assertions must not be
- // rewritten to the fresh branch head between retries (see
+ // pinnedRefs names refs whose assertions guard the commit and must
+ // not be rewritten to the fresh head between retries (see
// Transaction.pinnedRefs).
pinnedRefs map[string]struct{}
}
@@ -712,7 +711,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// never succeed — fail now instead of burning the
remaining
// retries on it.
if err := validatePinnedRefRequirements(reqs,
co.pinnedRefs, current); err != nil {
- return nil, fmt.Errorf("%w: explicit ref
requirement failed: %w", ErrCommitFailed, err)
+ return nil, fmt.Errorf("%w: pinned ref
requirement failed: %w", ErrCommitFailed, err)
}
if err := validateBranchRequirement(reqs, co.branch,
current); err != nil {
return nil, err
@@ -878,10 +877,10 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// deleted underneath us), reqs is returned unchanged — newConflict-
// Context will surface the divergence on the next pre-flight pass.
//
-// Assertions on branches in pinned were registered explicitly by the
-// committer (Transaction.AssertRefSnapshotID) for compare-and-swap
-// semantics and are never rewritten: a branch that has changed must
-// fail the commit, not be replayed against the new head.
+// Assertions on refs in pinned guard the commit (an explicit
+// Transaction.AssertRefSnapshotID, a rollback, or snapshot expiry) and
+// are never rewritten: a ref that has changed must fail the commit, not
+// be replayed against the new head.
func rewriteRefSnapshotRequirements(reqs []Requirement, branch string, fresh
Metadata, pinned map[string]struct{}) []Requirement {
if branch == "" || fresh == nil {
return reqs
diff --git a/table/transaction.go b/table/transaction.go
index e8369e6f9..62d9e5366 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -94,11 +94,11 @@ type Transaction struct {
// removals; see the flag site in commitManifests.
noReplay bool
- // pinnedRefs names branches with an explicit AssertRefSnapshotID
- // requirement. doCommit's refresh-and-replay must not rewrite their
- // assertions to the fresh branch head between retries: the caller
- // opted into compare-and-swap semantics, so a branch that has
- // changed fails the commit.
+ // pinnedRefs names refs whose AssertRefSnapshotID requirement guards
+ // the commit: those from AssertRefSnapshotID, RollbackToSnapshot, and
+ // ExpireSnapshots. doCommit's refresh-and-replay must not rewrite
+ // their assertions to the fresh head between retries, so a ref that
+ // has changed fails the commit.
pinnedRefs map[string]struct{}
mx sync.Mutex
@@ -151,6 +151,11 @@ func (t *Transaction) apply(updates []Update, reqs
[]Requirement) error {
t.mx.Lock()
defer t.mx.Unlock()
+ return t.applyLocked(updates, reqs)
+}
+
+// The caller must hold t.mx.
+func (t *Transaction) applyLocked(updates []Update, reqs []Requirement) error {
meta, err := t.txnMeta()
if err != nil {
return err
@@ -449,28 +454,30 @@ func (t *Transaction) AssertRefSnapshotID(branch string)
error {
id := t.baseRefSnapshotID(branch)
- // Record the pin before the requirement can become visible to a
- // concurrent Commit, which would otherwise rewrite the requirement
- // to the fresh head on retry and silently void the
- // compare-and-swap. The transient pin is harmless on its own and
- // is rolled back if the requirement fails to apply.
- t.mx.Lock()
- if t.pinnedRefs == nil {
- t.pinnedRefs = make(map[string]struct{})
- }
- _, alreadyPinned := t.pinnedRefs[branch]
- t.pinnedRefs[branch] = struct{}{}
- t.mx.Unlock()
+ return t.applyPinned(nil, []Requirement{AssertRefSnapshotID(branch,
id)}, []string{branch})
+}
- if err := t.apply(nil, []Requirement{AssertRefSnapshotID(branch, id)});
err != nil {
- if !alreadyPinned {
- t.mx.Lock()
- delete(t.pinnedRefs, branch)
- t.mx.Unlock()
- }
+// applyPinned applies updates and reqs and pins refs, so doCommit fails
+// the commit when a pinned ref has moved instead of rewriting its
+// assertion to the fresh head. Updates and pins are published under one
+// hold of t.mx, so a concurrent Commit sees both or neither, and a failed
+// apply publishes no pins.
+func (t *Transaction) applyPinned(updates []Update, reqs []Requirement, refs
[]string) error {
+ if err := t.checkNotNil(); err != nil {
+ return err
+ }
+ t.mx.Lock()
+ defer t.mx.Unlock()
+ if err := t.applyLocked(updates, reqs); err != nil {
return err
}
+ if t.pinnedRefs == nil {
+ t.pinnedRefs = make(map[string]struct{}, len(refs))
+ }
+ for _, ref := range refs {
+ t.pinnedRefs[ref] = struct{}{}
+ }
return nil
}
@@ -581,16 +588,17 @@ func (t *Transaction) RollbackToSnapshot(snapshotID
int64) error {
update := meta.NewRetainingSnapshotRefUpdate(branch, snapshotID,
BranchRef)
- // Assert the base branch head so a concurrent head move fails the
- // commit instead of the rollback clobbering it. When the branch was
- // staged by this transaction (absent on the base), the update that
- // created it already carries its own base-state requirement.
+ // Assert and pin the base branch head so a concurrent head move fails
+ // the commit instead of the rollback clobbering it. A branch staged by
+ // this transaction (absent on the base) already carries an absence
+ // assertion, and the pin keeps a retry from rebasing it onto a branch a
+ // peer created.
var reqs []Requirement
if id := t.baseRefSnapshotID(branch); id != nil {
reqs = append(reqs, AssertRefSnapshotID(branch, id))
}
- return t.apply([]Update{update}, reqs)
+ return t.applyPinned([]Update{update}, reqs, []string{branch})
}
func (t *Transaction) UpdateSpec(caseSensitive bool) *UpdateSpec {
@@ -744,6 +752,7 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
cfg = expireSnapshotsCfg{postCommit: true}
updates []Update
reqs []Requirement
+ pinned []string
snapsToKeep = make(map[int64]struct{})
nowMs = time.Now().UnixMilli()
)
@@ -792,14 +801,15 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
retainedRefs := make(map[string]SnapshotRef, len(meta.refs))
for refName, ref := range meta.refs {
- // Assert the ref's base snapshot id so we don't accidentally
- // expire snapshots that are now referenced by concurrently
- // updated refs. Refs the transaction itself staged (absent on
- // the base) get no assertion here: the update that created
- // them carries its own base-state requirement.
+ // Assert and pin the ref's base snapshot id so we don't expire
+ // snapshots that concurrently updated refs now reference. Refs
+ // the transaction itself staged (absent on the base) already
+ // carry an absence assertion, which the pin keeps from being
+ // rebased.
if id := t.baseRefSnapshotID(refName); id != nil {
reqs = append(reqs, AssertRefSnapshotID(refName, id))
}
+ pinned = append(pinned, refName)
snap, err := meta.SnapshotByID(ref.SnapshotID)
if err != nil {
@@ -878,7 +888,14 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
updates = append(updates,
NewRemoveSnapshotsUpdate(snapsToDelete, cfg.postCommit))
}
- return t.apply(updates, reqs)
+ // A no-op expiry must not pin the commit branch, which would stop the
+ // transaction's other updates from rebasing on retry. apply still
+ // rejects a transaction that has already been committed.
+ if len(updates) == 0 {
+ return t.apply(nil, nil)
+ }
+
+ return t.applyPinned(updates, reqs, pinned)
}
func (t *Transaction) AppendTable(ctx context.Context, tbl arrow.Table,
batchSize int64, snapshotProps iceberg.Properties) error {
diff --git a/table/transaction_assert_ref_test.go
b/table/transaction_assert_ref_test.go
index 97b61bd98..de6935714 100644
--- a/table/transaction_assert_ref_test.go
+++ b/table/transaction_assert_ref_test.go
@@ -177,6 +177,32 @@ func TestTransactionAssertRefSnapshotID(t *testing.T) {
})
}
+func TestNoOpExpireSnapshotsDoesNotBlockRebase(t *testing.T) {
+ head := int64(100)
+ base := newConflictTestMetadataWithProps(t, &head, assertRefRetryProps)
+ tbl, cat := newAssertRefTestTable(t, base, graftSnapshotOnto(t, base,
MainBranch, 200))
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets":
"42"}))
+ require.NoError(t, tx.ExpireSnapshots())
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), cat.attempts.Load())
+ assert.Equal(t, "42", cat.metadata.Properties()["offsets"])
+}
+
+func TestNoOpExpireSnapshotsRejectsCommittedTransaction(t *testing.T) {
+ head := int64(100)
+ base := newConflictTestMetadataWithProps(t, &head, assertRefRetryProps)
+ tbl, _ := newAssertRefTestTable(t, base, base)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets":
"42"}))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.ErrorContains(t, tx.ExpireSnapshots(), "already been committed")
+}
+
// Why: ref assertions are deduplicated by (type, ref), so assertions
// for distinct branches must both be enforced, an explicit requirement
// and a producer assertion requiring the same snapshot must collapse to
diff --git a/website/src/concurrent-writes.md b/website/src/concurrent-writes.md
index c86a086d6..5dcf03847 100644
--- a/website/src/concurrent-writes.md
+++ b/website/src/concurrent-writes.md
@@ -72,7 +72,11 @@ rejects when concurrent deletes touch the same files.
iceberg-go can refresh the table and replay the operation against the latest
snapshot between attempts. On each retry the table metadata is reloaded, the
producer's validators re-run against the fresh snapshot, and the commit is
-re-submitted only if it is still valid.
+re-submitted only if it is still valid. `RollbackToSnapshot`, `ExpireSnapshots`
+and `AssertRefSnapshotID` pin the refs they guard, so a retry does not rebase
+those assertions. If a guarded ref has moved, the commit fails with
+`ErrCommitFailed` instead of retrying. An `ExpireSnapshots` that removes
nothing
+adds no guards.
Retry is **off by default** (`commit.retry.num-retries` is `0`). Opt in by
setting the retry properties — see [Manifest and
commit](./configuration.md#manifest-and-commit)
@@ -87,6 +91,6 @@ _, err := cat.CreateTable(ctx, ident, schema,
)
```
-> **Catalog support:** retry currently engages on the REST catalog, which wraps
-> commit conflicts as `ErrCommitFailed`. The Glue, SQL, and Hive catalogs do
-> not yet wrap their conflict errors, so the retry loop will not fire on them.
+> **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.
