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 e904fb445 fix(table): treat requirement failures as commit conflicts
(#2100)
e904fb445 is described below
commit e904fb445780226acf43812afe78c1bd6ea6c80c
Author: Harrison Crosse <[email protected]>
AuthorDate: Tue Oct 27 16:50:21 2026 -0400
fix(table): treat requirement failures as commit conflicts (#2100)
* fix(table): treat requirement failures as commit conflicts
Catalogs that validate commit requirements client-side (Glue, SQL,
Hadoop) returned a plain error when a requirement failed, such as a
branch that moved since the table was loaded. The REST catalog returns
the same failure as a 409, which already maps to ErrCommitFailed, so the
same race was retryable through REST and terminal elsewhere. Java,
PyIceberg, and iceberg-rust all report these failures as retryable
commit conflicts.
Requirement failures now wrap ErrCommitFailed, keeping their messages.
Two failures stay terminal because a refresh cannot fix them: targeting
a tag as a transaction branch, and metadata with no last assigned
partition id.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): shorten requirementFailed comment
Signed-off-by: Harrison Crosse <[email protected]>
* chore: restore go.sum
Signed-off-by: Harrison Crosse <[email protected]>
* test(catalog/sql): drop redundant test comment
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): pin guarded refs in RollbackToSnapshot and ExpireSnapshots
RollbackToSnapshot asserts its branch head and ExpireSnapshots asserts
every ref so that a concurrent ref move fails the commit. On retry,
doCommit rewrites unpinned assertions on the commit branch to the
refreshed head, so a stale rollback could drop a peer's append and a
stale expiry could remove snapshots that are back inside the retention
window after a peer rollback.
Pin the guarded refs the same way Transaction.AssertRefSnapshotID
does, so the retry fails fast with ErrCommitFailed instead of replaying
against the new head. This includes a branch the transaction itself
staged: its absence assertion would otherwise be rebased onto a branch
a peer created meanwhile. The REST catalog already maps HTTP 409 to
ErrCommitFailed, so REST tables with retries enabled were exposed
before requirement failures started wrapping it; pinning fixes this
for every catalog.
Pins are now published under the same lock hold as the apply that
registers their requirements, so a failed registration publishes no pin
and cannot withdraw one that an overlapping registration relies on.
Snapshot producers still assert unpinned so they can rebase, and the
CLI branch and tag commands call CommitTable directly without retry.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): fail fast on non-rebased requirement failures
On retry, doCommit rebases only unpinned AssertRefSnapshotID
requirements on the commit branch. Every other requirement (schema,
partition spec, sort order, table UUID, last assigned ids, other refs,
pinned refs) is resubmitted unchanged, so one the refreshed metadata
violates fails identically on every attempt, spending the backoff and a
LoadTable per retry.
After the refresh, validate those requirements against the fresh
metadata and return immediately on failure, as Java does with
ValidationFailureException. This generalizes the existing pinned-ref
check. The error still wraps ErrCommitFailed, so callers can reload
and rebuild the commit.
Collect the manifests a rewrite producer wrote in the deferred cleanup,
so this and every other safe early exit removes them; previously only
an exhausted retry budget or a successful commit collected them.
Cleanup is still skipped when CommitTable fails in an unknown state.
A failed commit that removed files the staged updates reference (a
rewrite's merged manifests, or the attempt-0 manifest list a rebuild
replaced) now also wraps the new ErrTransactionUnusable sentinel, and
the transaction latches unusable. Later Commit, TableCommit (and so
catalog.MultiTableTransaction) and staging calls return an error
wrapping ErrTransactionUnusable instead of resubmitting references to
deleted files. A failure that removed nothing stays retriable.
Signed-off-by: Harrison Crosse <[email protected]>
* docs: update catalog retry support
The concurrent-writes guide said only the REST catalog engages commit
retry. REST maps HTTP 409 to ErrCommitFailed, and Glue, SQL, Hive and
Hadoop wrap their conflicts and failed requirements. Retry stays off
unless commit.retry.num-retries is set, and requirements other than the
commit branch head assertion fail without retrying.
Also correct the API guide, which said Commit retries automatically,
and note on ErrCommitFailed how the retry loop treats requirements and
that an error also matching ErrTransactionUnusable needs a new
transaction.
Signed-off-by: Harrison Crosse <[email protected]>
* chore: trim comments and docs
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): skip guards for a no-op ExpireSnapshots
ExpireSnapshots pins every ref it asserts, including the commit branch.
When it stages nothing, the requirements and pins still applied, so a
transaction that wrote and ran a no-op expiry could no longer rebase
onto a peer's commit and failed after one attempt. Return before
applying when there is nothing to remove.
Also correct the comment on the ErrTransactionUnusable latch in
Transaction.Commit: a clean conflict is not retriable once cleanup has
removed staged files, which includes an ordinary append whose manifest
list was rebuilt. Add a transaction-level test for that append path.
Signed-off-by: Harrison Crosse <[email protected]>
* test(catalog/sql): cover an append with a no-op ExpireSnapshots
A transaction that appends and runs an ExpireSnapshots that removes
nothing must still rebase onto a peer's append on retry. Cover that
path on the SQL catalog: the commit succeeds on retry and both appends
are present.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): make a non-rebased requirement failure terminal
When a requirement the retry loop does not rebase fails after a
refresh, committing the same transaction again can only reproduce the
failure. The error matched only ErrCommitFailed, so a caller retrying
on errors.Is(err, ErrCommitFailed) would loop on it.
Wrap ErrTransactionUnusable as well, so Transaction.Commit latches the
transaction and callers can tell this case from a retry-safe conflict.
Requirement failures at staging time are unchanged. Update the sentinel
docs and the concurrent-writes guide.
Signed-off-by: Harrison Crosse <[email protected]>
* test(table): pin cleanup after a transient failure following a rebuild
A refresh error or a cancelled context after a rebuilt attempt still
runs the deferred cleanup, which removes the replaced manifest list the
staged updates reference. Cover both paths: the error wraps
ErrTransactionUnusable, the file is gone, and a second Commit on the
same transaction is rejected with ErrTransactionUnusable.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): make a conflict with retries disabled terminal
Transaction.Commit rebuilds its requirements from the table it was
created from, so a re-commit resends the same requirements. With
num-retries=0 nothing rebases them, so every ErrCommitFailed, including
a stale branch head, repeats on each re-commit and a caller retrying on
errors.Is(err, ErrCommitFailed) spins. A no-replay commit, which carries
snapshot-relative delete-file removals, aborts on its first conflict
for the same reason at any retry setting.
In both cases end the commit with ErrTransactionUnusable so
Transaction.Commit latches the transaction. With retries enabled, the
next Commit of a replayable transaction refreshes once and a
requirement that no longer holds already makes it unusable.
The terminal errors now lead with "transaction cannot be committed
again" instead of "refresh and try again". They still match the cause,
and ErrCommitFailed when the cause is a conflict. The cleanup wrap is
keyed on whether a removal was attempted and skips errors that already
match ErrTransactionUnusable.
Signed-off-by: Harrison Crosse <[email protected]>
* docs(table): describe when a transaction becomes unusable
Say that ErrCommitFailed is retryable unless the error also matches
ErrTransactionUnusable, and spell out when a transaction becomes
unusable: its commit conflicts with retries disabled or while carrying
delete-file removals that cannot be replayed, a requirement the retry
loop does not rebase fails against the refreshed table, or files the
staged updates reference were cleaned up, which follows any failure
once an attempt was rebuilt or a rewrite staged files, including a
refresh error or a cancelled context.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): resume a re-commit after a conflict from a refresh
Transaction.Commit rebuilt its requirements from the transaction's
original base, so a re-commit after a failed commit resent the same
stale requirements and updates. To keep a caller retrying on
ErrCommitFailed from spinning, the previous change made every conflict
with retries disabled, and every conflict of a non-replayable commit,
leave the transaction unusable, which also burned transactions a plain
re-commit would land.
Treat a re-commit after a failed commit as the next retry attempt, as
Java does: it refreshes the table, rebases the commit-branch
assertion, checks the requirements that are not rebased, rebuilds the
staged snapshots and runs the validators before its first submission.
The retry budget per Commit is unchanged. A resume always rebuilds, so
the submitted snapshot and the producers' state come from the same
pass, and the rewrite producer now publishes a pass's state only after
its file-count guard passes. A non-replayable commit pins its branch on
resume, so it resubmits only if the branch head is unchanged. Rebuilt
manifest lists are numbered after earlier calls' attempts so their
names stay unique. Drop the blanket latch.
doCommit now deletes only files it wrote that were not committed. On a
failure it deletes the manifest lists its rebuilds wrote, including a
partial rebuild's. The staged updates' own files, which a later Commit
resends, are deleted only once a commit lands, so a failed commit no
longer makes the transaction unusable. Only a failure of unknown commit
state is terminal.
After a refresh, a rejection that fails every later attempt the same
way makes the transaction unusable: a non-rebased requirement, a
validator conflict, a diverged base, a commit branch that became a tag
or was deleted, or a replaced table. A validator that fails to read
stays resumable. TableCommit refuses a transaction whose commit failed,
since only Commit refreshes it. A no-op ExpireSnapshots now checks that
the transaction is usable.
Signed-off-by: Harrison Crosse <[email protected]>
* docs(table): describe resuming a transaction after a conflict
Explain that calling Commit again after a failed commit refreshes and
replays like the next retry attempt, that staged files are kept until a
commit lands so an abandoned transaction leaves them, and for a rewrite
the manifests later attempts merged, for orphan file removal, that
TableCommit refuses a transaction whose commit failed, which rejections
after a refresh make the transaction unusable, and that a real
ExpireSnapshots deliberately pins the commit branch.
Signed-off-by: Harrison Crosse <[email protected]>
* fix(catalog): pass requirement errors through unchanged
ValidateRequirements wrapped every failed requirement in
ErrCommitFailed. Requirement errors already match ErrCommitFailed when
they report a concurrent change, so the wrap only added the sentinel to
failures that are not conflicts, such as a tag used as a transaction
target, and prefixed every message with the sentinel's text. Return the
requirement's error unchanged apart from the missing-table case.
Signed-off-by: Harrison Crosse <[email protected]>
* docs: limit the catalog support note to conflicting requirement failures
Signed-off-by: Harrison Crosse <[email protected]>
* fix(table): wrap validator conflicts and pin the dual-match contract
Wrap a validator conflict after a refresh with fmt.Errorf as the other
terminal rejections are, instead of a dedicated error type that
rewrote the validator's message. The message now keeps the validator
error as is, including ErrCommitFailed's text, which holds for a
conflict: a new transaction built from a reloaded table may land.
Every row of the terminal-rejection matrix now asserts whether its
error matches ErrCommitFailed, and the matrix gains a stale sort order
and a moved pinned ref. A SQLite test runs the loop a caller writes
around ErrCommitFailed: it lands on the next pass after a validator
conflict or a stale requirement, and stops after one pass when the
base diverged.
Signed-off-by: Harrison Crosse <[email protected]>
* docs(table): describe what ErrCommitFailed tells a caller
Open the ErrCommitFailed godoc with what it means for a caller: a
concurrent change caused the failure, so a new transaction built from a
reloaded table may succeed. Spell out which terminal rejections match it
and which do not, and say on ErrTransactionUnusable that this
transaction can't be committed again and that one error can match both.
Show the dual match in the concurrent-writes guide.
Signed-off-by: Harrison Crosse <[email protected]>
---------
Signed-off-by: Harrison Crosse <[email protected]>
Co-authored-by: Matt Topol <[email protected]>
---
catalog/internal/update_stage_test.go | 21 ++
catalog/internal/utils.go | 10 +-
catalog/sql/sql_test.go | 281 +++++++++++++++++++++
table/commit.go | 9 +-
table/commit_refresh_replay_test.go | 327 ++++++++++++++++++++++++
table/commit_retry_test.go | 410 +++++++++++++++++++++++++++++--
table/no_replay_removals_test.go | 146 ++++++++++-
table/rebuild_manifest_test.go | 6 +-
table/requirement_test.go | 33 +++
table/requirements.go | 44 ++--
table/requirements_internal_test.go | 12 +
table/rewrite_data_files.go | 11 +-
table/rewrite_files_test.go | 7 +
table/rewrite_manifests.go | 48 ++--
table/rewrite_manifests_internal_test.go | 53 ++++
table/rewrite_manifests_test.go | 174 ++++++++++++-
table/snapshot_producers.go | 7 +-
table/table.go | 235 +++++++++++++-----
table/transaction.go | 58 ++++-
table/transaction_assert_ref_test.go | 13 +-
website/src/api.md | 2 +-
website/src/concurrent-writes.md | 31 ++-
22 files changed, 1746 insertions(+), 192 deletions(-)
diff --git a/catalog/internal/update_stage_test.go
b/catalog/internal/update_stage_test.go
index 3095d943f..0c81b2205 100644
--- a/catalog/internal/update_stage_test.go
+++ b/catalog/internal/update_stage_test.go
@@ -19,6 +19,7 @@ package internal
import (
"context"
+ "errors"
"testing"
"github.com/apache/iceberg-go"
@@ -110,6 +111,15 @@ func TestUpdateAndStageTableUsesVersionZeroForCreate(t
*testing.T) {
require.Equal(t, 0, ParseMetadataVersion(staged.MetadataLocation()))
}
+// failingRequirement fails with err, standing in for a requirement whose
+// failure is not a concurrent change, such as a tag used as a transaction
+// target.
+type failingRequirement struct{ err error }
+
+func (r failingRequirement) Validate(table.Metadata) error { return r.err }
+
+func (r failingRequirement) GetType() string { return "failing" }
+
func TestValidateRequirementsClassifiesFailures(t *testing.T) {
metadata, err := table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "file:///tmp/table", nil)
require.NoError(t, err)
@@ -139,3 +149,14 @@ func TestValidateRequirementsClassifiesFailures(t
*testing.T) {
})
}
}
+
+// A requirement failure that is not a concurrent change must not gain
+// ErrCommitFailed, or a retry loop would treat it as one.
+func TestValidateRequirementsReturnsRequirementErrorUnchanged(t *testing.T) {
+ metadata, err := table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "file:///tmp/table", nil)
+ require.NoError(t, err)
+ errTag := errors.New("requirement failed: ref \"t\" is a tag")
+
+ err = ValidateRequirements(table.Identifier{"db", "tbl"}, metadata,
[]table.Requirement{failingRequirement{errTag}})
+ require.Same(t, errTag, err)
+}
diff --git a/catalog/internal/utils.go b/catalog/internal/utils.go
index 23713d6af..3ca437076 100644
--- a/catalog/internal/utils.go
+++ b/catalog/internal/utils.go
@@ -263,10 +263,10 @@ func ParseMetadataVersion(location string) int {
// ValidateRequirements checks commit requirements against the table's current
// metadata, which is nil when the table does not exist. A failed requirement
-// on a missing table is reported as catalog.ErrNoSuchTable; any other failure
-// means a concurrent writer changed the table, and is reported as
-// table.ErrCommitFailed so the commit retry loop can refresh and replay, as it
-// does for a REST 409.
+// on a missing table is reported as catalog.ErrNoSuchTable. Any other failure
+// is returned as the requirement reported it, which matches
+// table.ErrCommitFailed when a concurrent writer changed the table, so the
+// commit retry loop can refresh and replay, as it does for a REST 409.
func ValidateRequirements(ident table.Identifier, current table.Metadata, reqs
[]table.Requirement) error {
for _, r := range reqs {
if err := r.Validate(current); err != nil {
@@ -274,7 +274,7 @@ func ValidateRequirements(ident table.Identifier, current
table.Metadata, reqs [
return fmt.Errorf("%w: %s: %w",
catalog.ErrNoSuchTable, strings.Join(ident, "."), err)
}
- return fmt.Errorf("%w: %w", table.ErrCommitFailed, err)
+ return err
}
}
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index e5005314f..b320359d7 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -1561,6 +1561,287 @@ func (s *SqliteCatalogTestSuite)
createRetryingStringTable(ctx context.Context,
return tblID
}
+// storedStringValues returns the "foo" values in the scan opts select.
+func (s *SqliteCatalogTestSuite) storedStringValues(ctx context.Context, tbl
*table.Table, opts ...table.ScanOption) []string {
+ rows, err := tbl.Scan(opts...).ToArrowTable(ctx)
+ s.Require().NoError(err)
+ defer rows.Release()
+
+ var values []string
+ for i := range int(rows.NumCols()) {
+ if rows.Schema().Field(i).Name != "foo" {
+ continue
+ }
+ for _, chunk := range rows.Column(i).Data().Chunks() {
+ for j := range chunk.Len() {
+ values = append(values, chunk.ValueStr(j))
+ }
+ }
+ }
+
+ return values
+}
+
+func (s *SqliteCatalogTestSuite) TestStaleAppendRetriesRequirementFailure() {
+ ctx := context.Background()
+ tests := []struct {
+ name string
+ numRetries string
+ }{
+ {"retries enabled", "2"},
+ {"retries disabled", "0"},
+ }
+
+ for _, tt := range tests {
+ s.Run(tt.name, func() {
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat,
tt.numRetries, "seed")
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+
+ s.appendStringRow(ctx, peer, table.MainBranch, "peer")
+
+ tx := stale.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(), "has changed")
+
+ // The re-commit refreshes and rebases like the
next retry.
+ _, err = tx.Commit(ctx)
+ }
+ s.Require().NoError(err)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Len(current.Metadata().Snapshots(), 3)
+ s.ElementsMatch([]string{"seed", "peer", "stale"},
s.storedStringValues(ctx, current))
+ })
+ }
+}
+
+func (s *SqliteCatalogTestSuite) addStringColumn(ctx context.Context, tbl
*table.Table, name string) *table.Transaction {
+ tx := tbl.NewTransaction()
+ s.Require().NoError(tx.UpdateSchema(true, false).
+ AddColumn([]string{name}, iceberg.PrimitiveTypes.String, "",
false, nil).
+ Commit())
+
+ return tx
+}
+
+func (s *SqliteCatalogTestSuite) TestStaleSchemaUpdateIsRefusedOnRecommit() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "0", "seed")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ tx := s.addStringColumn(ctx, stale, "stale")
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ _, err = s.addStringColumn(ctx, peer, "peer").Commit(ctx)
+ s.Require().NoError(err)
+ peerMeta, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrCommitFailed)
+ s.Require().NotErrorIs(err, table.ErrTransactionUnusable)
+
+ // The re-commit's refresh shows the schema requirement can never hold,
+ // so it stops before the catalog.
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrTransactionUnusable)
+ s.Require().ErrorIs(err, table.ErrCommitFailed)
+ s.EqualError(err, "transaction cannot be committed again: requirement
no longer holds after refresh: "+
+ "Requirement failed: current schema id has changed: expected 0,
found 1")
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(peerMeta.MetadataLocation(), current.MetadataLocation(),
"nothing may be committed")
+}
+
+func (s *SqliteCatalogTestSuite) TestConflictClearedBeforeRecommitLands() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "0", "seed")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ seedHead := stale.Metadata().CurrentSnapshot().SnapshotID
+ tx := stale.NewTransaction()
+ s.stageStringRow(ctx, tx, "writer")
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, table.MainBranch, "peer")
+
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrCommitFailed)
+ s.Require().NotErrorIs(err, table.ErrTransactionUnusable)
+
+ // The peer rolls main back, so the branch the writer read is current
again.
+ rollback := peer.NewTransaction()
+ s.Require().NoError(rollback.RollbackToSnapshot(seedHead))
+ _, err = rollback.Commit(ctx)
+ s.Require().NoError(err)
+
+ committed, err := tx.Commit(ctx)
+ s.Require().NoError(err)
+ s.Equal(seedHead, *committed.CurrentSnapshot().ParentSnapshotID)
+ s.ElementsMatch([]string{"seed", "writer"}, s.storedStringValues(ctx,
committed))
+}
+
+// A delete staged before a peer appended a matching row is rejected by
+// the serializable validator on the re-commit, and every later attempt
+// would be rejected the same way.
+func (s *SqliteCatalogTestSuite)
TestStaleDeleteRejectedByValidatorIsUnusable() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "0", "seed", "peer")
+
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ tx := stale.NewTransaction()
+ s.Require().NoError(tx.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("foo"), "peer"), nil))
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, table.MainBranch, "peer")
+
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrCommitFailed)
+ s.Require().NotErrorIs(err, table.ErrTransactionUnusable)
+
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrTransactionUnusable)
+ s.Require().ErrorIs(err, table.ErrConflictingDataFiles)
+ s.Require().ErrorIs(err, table.ErrCommitFailed)
+ s.True(strings.HasPrefix(err.Error(), fmt.Sprintf("transaction cannot
be committed again: validation conflict: "+
+ "commit failed, refresh and try again: concurrent data files
added: snapshot %d added data file ",
+ peer.CurrentSnapshot().SnapshotID)), err.Error())
+ s.True(strings.HasSuffix(err.Error(), " matching filter
Equal(term=Reference(name='foo'), literal=peer)"), err.Error())
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(peer.MetadataLocation(), current.MetadataLocation(), "nothing
may be committed")
+}
+
+func (s *SqliteCatalogTestSuite) retryWithNewTransactions(ctx context.Context,
cat *sqlcat.Catalog, tblID table.Identifier,
+ maxPasses int, stage func(pass int, tbl *table.Table)
*table.Transaction,
+) []error {
+ var errs []error
+ for pass := 1; pass <= maxPasses; pass++ {
+ tbl, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ _, err = stage(pass, tbl).Commit(ctx)
+ errs = append(errs, err)
+ if !errors.Is(err, table.ErrCommitFailed) {
+ break
+ }
+ }
+
+ return errs
+}
+
+// A failure that matches ErrCommitFailed is one a new transaction built from
+// a reloaded table may get past, and one that does not match ends the loop.
+// Tables retry once, so the first pass refreshes and fails its terminal check.
+func (s *SqliteCatalogTestSuite)
TestRetryLoopOnCommitFailedWithNewTransactions() {
+ ctx := context.Background()
+ const maxPasses = 3
+
+ s.Run("validator conflict lands on the next pass", func() {
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "1", "seed",
"peer")
+
+ errs := s.retryWithNewTransactions(ctx, cat, tblID, maxPasses,
func(pass int, tbl *table.Table) *table.Transaction {
+ tx := tbl.NewTransaction()
+ s.Require().NoError(tx.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("foo"), "peer"), nil))
+ if pass == 1 {
+ s.appendStringRow(ctx, tbl, table.MainBranch,
"peer")
+ }
+
+ return tx
+ })
+ s.Require().Len(errs, 2)
+ s.ErrorIs(errs[0], table.ErrTransactionUnusable)
+ s.ErrorIs(errs[0], table.ErrConflictingDataFiles)
+ s.NoError(errs[1])
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.ElementsMatch([]string{"seed"}, s.storedStringValues(ctx,
current))
+ })
+
+ s.Run("stale requirement lands on the next pass", func() {
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "1", "seed")
+
+ errs := s.retryWithNewTransactions(ctx, cat, tblID, maxPasses,
func(pass int, tbl *table.Table) *table.Transaction {
+ tx := s.addStringColumn(ctx, tbl, "writer")
+ if pass == 1 {
+ _, err := s.addStringColumn(ctx, tbl,
"peer").Commit(ctx)
+ s.Require().NoError(err)
+ }
+
+ return tx
+ })
+ s.Require().Len(errs, 2)
+ s.ErrorIs(errs[0], table.ErrTransactionUnusable)
+ s.Contains(errs[0].Error(), "current schema id has changed")
+ s.NoError(errs[1])
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ for _, name := range []string{"peer", "writer"} {
+ _, ok := current.Schema().FindFieldByName(name)
+ s.Truef(ok, "column %s", name)
+ }
+ })
+
+ s.Run("diverged base stops after one pass", func() {
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createRetryingStringTable(ctx, cat, "1", "seed",
"second")
+
+ errs := s.retryWithNewTransactions(ctx, cat, tblID, maxPasses,
func(pass int, tbl *table.Table) *table.Transaction {
+ tx := tbl.NewTransaction()
+ s.Require().NoError(tx.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("foo"), "seed"), nil))
+ if pass == 1 {
+ parent :=
tbl.Metadata().CurrentSnapshot().ParentSnapshotID
+ s.Require().NotNil(parent)
+ rollback := tbl.NewTransaction()
+
s.Require().NoError(rollback.RollbackToSnapshot(*parent))
+ _, err := rollback.Commit(ctx)
+ s.Require().NoError(err)
+ }
+
+ return tx
+ })
+ s.Require().Len(errs, 1, "a failure that does not match
ErrCommitFailed must end the loop")
+ s.ErrorIs(errs[0], table.ErrTransactionUnusable)
+ s.ErrorIs(errs[0], table.ErrCommitDiverged)
+ s.NotErrorIs(errs[0], table.ErrCommitFailed)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.ElementsMatch([]string{"seed"}, s.storedStringValues(ctx,
current))
+ })
+}
+
// 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() {
diff --git a/table/commit.go b/table/commit.go
index 22aa98606..516c6e218 100644
--- a/table/commit.go
+++ b/table/commit.go
@@ -66,8 +66,13 @@ func (t *Transaction) TableCommit() (TableCommit, error) {
return TableCommit{}, err
}
- if t.committed {
- return TableCommit{}, errors.New("transaction has already been
committed")
+ if err := t.checkUsable(); err != nil {
+ return TableCommit{}, err
+ }
+ // TableCommit sends the transaction's original requirements and
updates,
+ // which may be stale after a failed commit. Only Commit refreshes them.
+ if t.failedCommits > 0 {
+ return TableCommit{}, errors.New("a previous commit of this
transaction failed, call Commit to resume it")
}
if len(meta.updates) == 0 {
diff --git a/table/commit_refresh_replay_test.go
b/table/commit_refresh_replay_test.go
index 304ad8d05..25116cd36 100644
--- a/table/commit_refresh_replay_test.go
+++ b/table/commit_refresh_replay_test.go
@@ -30,7 +30,9 @@ package table
import (
"context"
+ "errors"
"fmt"
+ "strings"
"sync/atomic"
"testing"
@@ -51,6 +53,7 @@ import (
type headTrackingCatalog struct {
metadata Metadata
attempts atomic.Int32
+ loads atomic.Int32
}
type identifierCapturingCatalog struct {
@@ -61,6 +64,8 @@ type identifierCapturingCatalog struct {
}
func (c *headTrackingCatalog) LoadTable(_ context.Context, ident Identifier)
(*Table, error) {
+ c.loads.Add(1)
+
return New(ident, c.metadata, "",
func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, c), nil
}
@@ -215,6 +220,328 @@ func TestDoCommit_ValidatorRejectsOnRefresh(t *testing.T)
{
"validator rejection on retry must abort before re-issuing
CommitTable")
}
+// Why: after a refresh, a rejection that does not depend on timing fails
+// every later attempt and every re-commit the same way, so the transaction
+// must become unusable rather than leave an errors.Is(ErrCommitFailed) loop
+// spinning, and its message must not tell the caller to try again. A
+// validator that fails to read must stay resumable.
+func TestResumeDeterministicRejectionsMakeTransactionUnusable(t *testing.T) {
+ head := int64(100)
+ noRetries := iceberg.Properties{}
+ // Rejections formatted like the validators in conflict_validation.go.
+ rejectConcurrent := func(sentinel error) conflictValidatorFunc {
+ return func(cc *conflictContext) error {
+ if len(cc.concurrent) > 0 {
+ return fmt.Errorf("%w: snapshot %d added
f.parquet", sentinel, cc.concurrent[0].SnapshotID)
+ }
+
+ return nil
+ }
+ }
+ withFeature := func(t *testing.T, base Metadata) Metadata {
+ b, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, b.SetSnapshotRef("feature", head, BranchRef))
+ out, err := b.Build()
+ require.NoError(t, err)
+
+ return out
+ }
+ rebuild := func(t *testing.T, base Metadata, edit
func(*MetadataBuilder)) Metadata {
+ b, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ edit(b)
+ out, err := b.Build()
+ require.NoError(t, err)
+
+ return out
+ }
+
+ tests := []struct {
+ name string
+ branch string
+ stage func(t *testing.T, tx *Transaction)
+ validator conflictValidatorFunc
+ peer func(t *testing.T, base Metadata) Metadata
+ want error
+ message string
+ matchesCommitFailed bool
+ // mentionsRetry marks validator causes, whose text includes
+ // ErrCommitFailed's.
+ mentionsRetry bool
+ }{
+ {
+ name: "diverged base", validator:
func(*conflictContext) error { return nil },
+ peer: func(t *testing.T, base Metadata) Metadata {
+ return rebuild(t, base, func(b
*MetadataBuilder) {
+ require.NoError(t,
b.AddSnapshot(&Snapshot{
+ SnapshotID: 300,
SequenceNumber: 2, TimestampMs: base.LastUpdatedMillis() + 1,
+ Summary: &Summary{Operation:
OpAppend},
+ }))
+ require.NoError(t,
b.SetSnapshotRef(MainBranch, 300, BranchRef))
+ })
+ },
+ want: ErrCommitDiverged,
+ message: "transaction cannot be committed again: base
diverged: commit diverged: base snapshot is no longer on the branch: ",
+ },
+ {
+ name: "conflicting data files", validator:
rejectConcurrent(ErrConflictingDataFiles),
+ peer: func(t *testing.T, base Metadata)
Metadata { return graftSnapshotOnto(t, base, MainBranch, 200) },
+ want: ErrConflictingDataFiles,
+ message: "transaction cannot be committed
again: validation conflict: commit failed, refresh and try again: concurrent
data files added: snapshot 200 added f.parquet",
+ matchesCommitFailed: true, mentionsRetry: true,
+ },
+ {
+ name: "conflicting delete files", validator:
rejectConcurrent(ErrConflictingDeleteFiles),
+ peer: func(t *testing.T, base Metadata)
Metadata { return graftSnapshotOnto(t, base, MainBranch, 200) },
+ want: ErrConflictingDeleteFiles,
+ message: "transaction cannot be committed
again: validation conflict: commit failed, refresh and try again: concurrent
delete files added: snapshot 200 added f.parquet",
+ matchesCommitFailed: true, mentionsRetry: true,
+ },
+ {
+ name: "data files missing", validator:
rejectConcurrent(ErrDataFilesMissing),
+ peer: func(t *testing.T, base Metadata)
Metadata { return graftSnapshotOnto(t, base, MainBranch, 200) },
+ want: ErrDataFilesMissing,
+ message: "transaction cannot be committed
again: validation conflict: commit failed, refresh and try again: referenced
data files missing: snapshot 200 added f.parquet",
+ matchesCommitFailed: true, mentionsRetry: true,
+ },
+ {
+ name: "wrapped conflict from any validator",
+ validator: func(cc *conflictContext) error {
+ if len(cc.concurrent) > 0 {
+ return fmt.Errorf("custom validator:
%w", ErrCommitFailed)
+ }
+
+ return nil
+ },
+ peer: func(t *testing.T, base Metadata)
Metadata { return graftSnapshotOnto(t, base, MainBranch, 200) },
+ message: "transaction cannot be committed
again: validation conflict: custom validator: commit failed, refresh and try
again",
+ matchesCommitFailed: true, mentionsRetry: true,
+ },
+ {
+ name: "commit branch became a tag", branch: "feature",
+ peer: func(t *testing.T, base Metadata) Metadata {
+ return rebuild(t, base, func(b
*MetadataBuilder) {
+ require.NoError(t,
b.RemoveSnapshotRef("feature"))
+ require.NoError(t,
b.SetSnapshotRef("feature", head, TagRef))
+ })
+ },
+ message: `transaction cannot be committed again: commit
branch is no longer a branch: requirement failed: ref "feature" is a tag; tags
cannot be transaction targets`,
+ },
+ {
+ name: "commit branch was deleted", branch: "feature",
+ peer: func(t *testing.T, base Metadata) Metadata {
+ return rebuild(t, base, func(b
*MetadataBuilder) { require.NoError(t, b.RemoveSnapshotRef("feature")) })
+ },
+ message: `transaction cannot be committed
again: commit branch is no longer a branch: requirement failed: branch or tag
"feature" is missing, expected 100`,
+ matchesCommitFailed: true,
+ },
+ {
+ name: "stale schema",
+ stage: func(t *testing.T, tx *Transaction) {
+ require.NoError(t, tx.UpdateSchema(true, false).
+ AddColumn([]string{"writer"},
iceberg.PrimitiveTypes.String, "", false, nil).Commit())
+ },
+ peer: func(t *testing.T, base Metadata) Metadata {
+ return rebuild(t, base, func(b
*MetadataBuilder) {
+ require.NoError(t,
b.AddSchema(iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1,
Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2,
Name: "peer", Type: iceberg.PrimitiveTypes.String},
+ )))
+ require.NoError(t,
b.SetCurrentSchemaID(-1))
+ })
+ },
+ message: "transaction cannot be committed again:
requirement no longer holds after refresh: " +
+ "Requirement failed: current schema id has
changed: expected 0, found 1",
+ matchesCommitFailed: true,
+ },
+ {
+ name: "stale default sort order",
+ stage: func(t *testing.T, tx *Transaction) {
require.NoError(t, tx.AssertDefaultShape()) },
+ peer: func(t *testing.T, base Metadata) Metadata {
+ return rebuild(t, base, func(b
*MetadataBuilder) {
+ order, err := NewSortOrder(1,
[]SortField{{
+ SourceIDs: []int{1}, Transform:
iceberg.IdentityTransform{}, Direction: SortASC, NullOrder: NullsFirst,
+ }})
+ require.NoError(t, err)
+ require.NoError(t,
b.AddSortOrder(&order))
+ require.NoError(t,
b.SetDefaultSortOrderID(-1))
+ })
+ },
+ message: "transaction cannot be committed again:
requirement no longer holds after refresh: " +
+ "requirement failed: default sort order id has
changed: expected 0, found 1",
+ matchesCommitFailed: true,
+ },
+ {
+ name: "pinned ref moved",
+ stage: func(t *testing.T, tx *Transaction) {
require.NoError(t, tx.AssertRefSnapshotID(MainBranch)) },
+ peer: func(t *testing.T, base Metadata) Metadata {
return graftSnapshotOnto(t, base, MainBranch, 200) },
+ message: "transaction cannot be committed again:
requirement no longer holds after refresh: " +
+ `requirement failed: branch "main" has changed:
expected id 100, found 200`,
+ matchesCommitFailed: true,
+ },
+ {
+ name: "non-replayable commit whose branch moved",
+ stage: func(_ *testing.T, tx *Transaction) {
tx.noReplay = true },
+ peer: func(t *testing.T, base Metadata) Metadata {
return graftSnapshotOnto(t, base, MainBranch, 200) },
+ message: "transaction cannot be committed again:
requirement no longer holds after refresh (" + noReplayHint + "): " +
+ `requirement failed: branch "main" has changed:
expected id 100, found 200`,
+ matchesCommitFailed: true,
+ },
+ {
+ name: "table was replaced",
+ peer: func(t *testing.T, base Metadata) Metadata {
+ return rebuild(t, base, func(b
*MetadataBuilder) { require.NoError(t, b.SetUUID(uuid.New())) })
+ },
+ want: ErrInvalidMetadata,
+ message: "transaction cannot be committed again: table
was replaced: ",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ base := withFeature(t,
newConflictTestMetadataWithProps(t, &head, noRetries))
+ tbl, cat := newAssertRefTestTable(t, base, tt.peer(t,
base))
+ tx := tbl.NewTransaction()
+ if tt.branch != "" {
+ branchTx, err :=
tbl.NewTransactionOnBranch(tt.branch)
+ require.NoError(t, err)
+ tx = branchTx
+ }
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{"k": "v"}))
+ if tt.stage != nil {
+ tt.stage(t, tx)
+ }
+ if tt.validator != nil {
+ tx.addValidator(tt.validator)
+ }
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ require.Equal(t, int32(1), cat.attempts.Load())
+ require.Equal(t, int32(0), cat.loads.Load(), "retries
are disabled")
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ assert.Equal(t, int32(1), cat.loads.Load(), "the
re-commit refreshes once")
+ if tt.want != nil {
+ require.ErrorIs(t, err, tt.want)
+ }
+ if tt.matchesCommitFailed {
+ require.ErrorIs(t, err, ErrCommitFailed)
+ } else {
+ require.NotErrorIs(t, err, ErrCommitFailed, "a
caller retrying on ErrCommitFailed must stop")
+ }
+ assert.True(t, strings.HasPrefix(err.Error(),
tt.message), err.Error())
+ if !tt.mentionsRetry {
+ assert.NotContains(t, err.Error(), "refresh and
try again")
+ }
+ assert.Equal(t, int32(1), cat.attempts.Load(), "the
resume must stop before the catalog")
+
+ _, err = tx.Commit(t.Context())
+ assert.EqualError(t, err, "transaction cannot be
committed again: a previous commit failed, build a new transaction")
+ assert.Equal(t, int32(1), cat.attempts.Load())
+ assert.Equal(t, int32(1), cat.loads.Load(), "an
unusable transaction must not refresh again")
+ })
+ }
+
+ t.Run("validator read failure stays resumable", func(t *testing.T) {
+ base := newConflictTestMetadataWithProps(t, &head, noRetries)
+ tbl, cat := newAssertRefTestTable(t, base, graftSnapshotOnto(t,
base, MainBranch, 200))
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"k":
"v"}))
+ readFailures := 1
+ tx.addValidator(func(cc *conflictContext) error {
+ if len(cc.concurrent) > 0 && readFailures > 0 {
+ readFailures--
+
+ return errors.New("simulated manifest read
failure")
+ }
+
+ return nil
+ })
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+
+ _, err = tx.Commit(t.Context())
+ require.EqualError(t, err, "simulated manifest read failure")
+
+ _, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), cat.attempts.Load())
+ })
+}
+
+// A stale schema requirement cannot be rebased, but a valid one must not
+// prevent retrying against a changed branch head.
+func TestDoCommit_NonRebasedRequirementFailsAfterOneRefresh(t *testing.T) {
+ writerHead := int64(100)
+ props := iceberg.Properties{
+ CommitNumRetriesKey: "3",
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ }
+
+ stalePeerSchema := func(t *testing.T, props iceberg.Properties)
(*headTrackingCatalog, *Transaction) {
+ writerBase := newConflictTestMetadataWithProps(t, &writerHead,
props)
+
+ builder, err := MetadataBuilderFromBase(writerBase, "")
+ require.NoError(t, err)
+ peerSchema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "peer", Type:
iceberg.PrimitiveTypes.String},
+ )
+ require.NoError(t, builder.AddSchema(peerSchema))
+ require.NoError(t, builder.SetCurrentSchemaID(-1))
+ peerChanged, err := builder.Build()
+ require.NoError(t, err)
+
+ tbl, cat := newAssertRefTestTable(t, writerBase, peerChanged)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.UpdateSchema(true, false).
+ AddColumn([]string{"writer"},
iceberg.PrimitiveTypes.String, "", false, nil).
+ Commit())
+
+ return cat, tx
+ }
+
+ const staleSchema = "transaction cannot be committed again: requirement
no longer holds after refresh: " +
+ "Requirement failed: current schema id has changed: expected 0,
found 1"
+
+ t.Run("stale schema update fails after one refresh", func(t *testing.T)
{
+ cat, tx := stalePeerSchema(t, props)
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ assert.EqualError(t, err, staleSchema)
+ assert.Equal(t, int32(1), cat.attempts.Load(), "the stale
requirement must not be resubmitted")
+ assert.Equal(t, int32(1), cat.loads.Load(), "the commit must
stop after the first refresh")
+ assert.Equal(t, 1, cat.metadata.CurrentSchema().ID, "the peer's
schema must survive")
+ })
+
+ t.Run("requirements that still hold do not block the rebase", func(t
*testing.T) {
+ writerBase := newConflictTestMetadataWithProps(t, &writerHead,
props)
+ advanced := graftSnapshotOnto(t, writerBase, MainBranch, 200)
+ tbl, cat := newAssertRefTestTable(t, writerBase, advanced)
+
+ reqs := []Requirement{
+ AssertRefSnapshotID(MainBranch, &writerHead),
+ AssertCurrentSchemaID(writerBase.CurrentSchema().ID),
+ AssertTableUUID(writerBase.TableUUID()),
+ }
+ _, err := tbl.doCommit(t.Context(),
+ []Update{NewSetPropertiesUpdate(iceberg.Properties{"k":
"v"})}, reqs,
+ withCommitBranch(MainBranch))
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), cat.attempts.Load())
+ assert.Equal(t, int32(1), cat.loads.Load())
+ assert.Equal(t, "v", cat.metadata.Properties()["k"])
+ })
+}
+
func TestRefreshPassesIdentifierCopy(t *testing.T) {
branchHead := int64(1)
meta := newConflictTestMetadataWithProps(t, &branchHead, nil)
diff --git a/table/commit_retry_test.go b/table/commit_retry_test.go
index a404cd852..b0c478be2 100644
--- a/table/commit_retry_test.go
+++ b/table/commit_retry_test.go
@@ -21,8 +21,10 @@ import (
"context"
"errors"
"fmt"
+ "maps"
"math"
"path/filepath"
+ "slices"
"strconv"
"strings"
"sync/atomic"
@@ -630,11 +632,9 @@ func TestDoCommit_OrphanNotCleanedOnUnknownError(t
*testing.T) {
"on unknown 5xx every rebuild's manifest list must be preserved
(one may be the live commit)")
}
-// TestDoCommit_OrphanCleanedOnCommitDiverged verifies that manifest-list files
-// orphaned by rebuild attempts are removed when ErrCommitDiverged is returned
-// by a conflict validator. Diverged commits are terminal (no retry), and since
-// neither of the orphaned files was ever accepted by the catalog, they are
safe
-// to delete. The defer cleanup runs with cleanupOrphans=true on this path.
+// TestDoCommit_OrphanCleanedOnCommitDiverged verifies that when a conflict
+// validator returns ErrCommitDiverged after a rebuild, the manifest list the
+// rebuild wrote is removed while the staged list the caller owns is kept.
func TestDoCommit_OrphanCleanedOnCommitDiverged(t *testing.T) {
spec := iceberg.NewPartitionSpec()
wfs, meta := newMemIOWithRetryMeta(t, spec)
@@ -684,18 +684,26 @@ func TestDoCommit_OrphanCleanedOnCommitDiverged(t
*testing.T) {
)
require.ErrorIs(t, err, ErrCommitDiverged)
- // The defer must have fired with cleanupOrphans=true and removed the
orphan.
- _, stillExists := wfs.files[originalManifestList]
- assert.False(t, stillExists,
- "orphaned manifest list must be removed even on
ErrCommitDiverged: "+
- "the file was never accepted by the catalog so it is
safe to delete")
+ assertOnlyManifestList(t, wfs, originalManifestList)
+}
+
+// assertOnlyManifestList checks that want is the only manifest list left in
+// wfs: a failed commit keeps the staged list and removes every rebuilt one.
+func assertOnlyManifestList(t *testing.T, wfs *memIO, want string) {
+ t.Helper()
+ var lists []string
+ for path := range wfs.files {
+ if strings.HasPrefix(filepath.Base(path), "snap-") {
+ lists = append(lists, path)
+ }
+ }
+ assert.Equal(t, []string{want}, lists)
}
// TestDoCommit_OrphanCleanedOnRetriesExhausted verifies that when every retry
-// attempt fails with ErrCommitFailed and the loop exits with the budget
-// exhausted, the defer still fires with cleanupOrphans=true. None of the
-// orphaned manifest-list files were ever accepted by the catalog, so they are
-// safe to delete on this terminal exit.
+// attempt fails with ErrCommitFailed, the manifest lists the rebuilds wrote,
+// including the final attempt's, are removed while the staged list the caller
+// owns is kept for a later Commit.
func TestDoCommit_OrphanCleanedOnRetriesExhausted(t *testing.T) {
spec := iceberg.NewPartitionSpec()
wfs, meta := newMemIOWithRetryMeta(t, spec)
@@ -724,10 +732,7 @@ func TestDoCommit_OrphanCleanedOnRetriesExhausted(t
*testing.T) {
require.ErrorIs(t, err, ErrCommitFailed,
"retries exhausted: terminal error must be ErrCommitFailed")
- _, stillExists := wfs.files[originalManifestList]
- assert.False(t, stillExists,
- "orphaned manifest list must be removed when retries are
exhausted with ErrCommitFailed: "+
- "none of the retry attempts were accepted, so all
orphans are safe to delete")
+ assertOnlyManifestList(t, wfs, originalManifestList)
}
// cancelAwareRemoveIO wraps memIO and, like blobfs, rejects Remove on a
@@ -763,7 +768,9 @@ func (c *cancelAfterCommitCatalog) CommitTable(ctx
context.Context, ident Identi
// TestDoCommit_OrphanCleanedOnCancellation verifies that orphan cleanup does
// not inherit the caller's cancellation: when doCommit exits because ctx was
-// canceled, an IO that honors ctx on Remove must still remove the orphans.
+// canceled, an IO that honors ctx on Remove must still remove the manifest
+// lists its rebuilds wrote, while the caller's staged list is kept for a
+// re-commit.
func TestDoCommit_OrphanCleanedOnCancellation(t *testing.T) {
spec := iceberg.NewPartitionSpec()
mem, meta := newMemIOWithRetryMeta(t, spec)
@@ -778,10 +785,11 @@ func TestDoCommit_OrphanCleanedOnCancellation(t
*testing.T) {
require.NoError(t, err)
originalManifestList :=
updates[0].(*addSnapshotUpdate).Snapshot.ManifestList
require.Contains(t, mem.files, originalManifestList)
+ filesBefore := slices.Sorted(maps.Keys(mem.files))
- // The first failure triggers a rebuild that orphans the original
manifest
- // list; the context is canceled during the second attempt, so doCommit
- // exits on the cancellation with orphan cleanup still enabled.
+ // The first failure triggers a rebuild that writes a new manifest list;
+ // the context is canceled during the second attempt, so doCommit exits
+ // on the cancellation with orphan cleanup still enabled.
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
cat := &cancelAfterCommitCatalog{
@@ -797,8 +805,8 @@ func TestDoCommit_OrphanCleanedOnCancellation(t *testing.T)
{
_, err = tbl.doCommit(ctx, updates, reqs, withCommitBranch(MainBranch))
require.ErrorIs(t, err, context.Canceled)
- assert.NotContains(t, mem.files, originalManifestList,
- "orphaned manifest list must be removed even when the commit
ends on cancellation")
+ assert.Equal(t, filesBefore, slices.Sorted(maps.Keys(mem.files)),
+ "rebuilt manifest lists must be removed even when the commit
ends on cancellation, and the staged list kept")
}
// ---------------------------------------------------------------------------
@@ -836,9 +844,20 @@ type progressingRebuildCatalog struct {
committedSnapshot *Snapshot // populated when CommitTable accepts
peerCount int // running peer-commit counter (=
total-data-files contributed by peers)
+
+ // failLoad makes that LoadTable call (1-based) fail. cancel runs once
+ // CommitTable has been called cancelAt times.
+ loads int
+ failLoad int
+ cancelAt int32
+ cancel context.CancelFunc
}
func (c *progressingRebuildCatalog) LoadTable(_ context.Context, ident
Identifier) (*Table, error) {
+ c.loads++
+ if c.loads == c.failLoad {
+ return nil, errors.New("simulated refresh failure")
+ }
c.loadedLastSeqNums = append(c.loadedLastSeqNums,
c.metadata.LastSequenceNumber())
return New(ident, c.metadata, "",
@@ -847,6 +866,9 @@ func (c *progressingRebuildCatalog) LoadTable(_
context.Context, ident Identifie
func (c *progressingRebuildCatalog) CommitTable(_ context.Context, _
Identifier, _ []Requirement, updates []Update) (Metadata, string, error) {
n := c.commitTableCalls.Add(1)
+ if c.cancel != nil && n == c.cancelAt {
+ defer c.cancel()
+ }
// Capture the manifest-list path and sequence number being submitted
// this attempt — the test asserts each retry produces a unique path
@@ -1101,17 +1123,19 @@ func
TestTransactionCommit_RetriableAfterCleanConflict(t *testing.T) {
_, err := tx.Commit(t.Context())
require.ErrorIs(t, err, ErrCommitFailed, "first commit must surface the
clean conflict")
+ assert.NotErrorIs(t, err, ErrTransactionUnusable)
assert.False(t, tx.committed, "clean-conflict failure must leave
committed == false")
assert.Equal(t, int32(1), cat.attempts.Load(), "first commit must reach
the catalog once")
- // The transaction stays usable: applying further changes and retrying
the
- // commit must both be allowed.
+ // Changes staged after the conflict are part of the re-commit.
require.NoError(t, tx.SetProperties(map[string]string{"key2":
"value2"}),
"apply must be allowed after a failed commit")
committed, err := tx.Commit(t.Context())
require.NoError(t, err, "commit retry must be allowed after a clean
conflict")
require.NotNil(t, committed)
+ assert.Equal(t, "value", committed.Properties()["key"])
+ assert.Equal(t, "value2", committed.Properties()["key2"])
assert.True(t, tx.committed, "committed must be set only after a
successful commit")
assert.Equal(t, int32(2), cat.attempts.Load(), "the retry must reach
the catalog")
@@ -1122,6 +1146,130 @@ func
TestTransactionCommit_RetriableAfterCleanConflict(t *testing.T) {
assert.ErrorContains(t, err, "already been committed")
}
+func TestTransactionCommit_ResumedConflictStaysResumable(t *testing.T) {
+ cat := &sequentialCatalog{errs: []error{ErrCommitFailed,
ErrCommitFailed}}
+ tbl := newRetryTestTable(t, cat, nil)
+ cat.metadata = tbl.Metadata()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(map[string]string{"key": "value"}))
+
+ for range 2 {
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ }
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, "value", committed.Properties()["key"])
+ assert.Equal(t, int32(3), cat.attempts.Load(), "each Commit submits
once with retries disabled")
+}
+
+func TestTransactionCommit_ResumeRebuildsAfterPeerSnapshotOnAnotherBranch(t
*testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ wfs := newMemIO(1<<20, nil)
+ meta, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder,
"mem://default/table-location",
+ iceberg.Properties{PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+ cat := &sequentialCatalog{metadata: meta}
+ tbl := newOCCTable(t, meta, wfs, cat)
+
+ seed := tbl.NewTransaction()
+ require.NoError(t, seed.AddDataFiles(t.Context(), []iceberg.DataFile{
+ newTestDataFile(t, spec,
"mem://default/table-location/data/f1.parquet", nil),
+ }, nil))
+ tbl, err = seed.Commit(t.Context())
+ require.NoError(t, err)
+ mainHead := tbl.CurrentSnapshot().SnapshotID
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{
+ newTestDataFile(t, spec,
"mem://default/table-location/data/f2.parquet", nil),
+ }, nil))
+
+ // A peer adds a snapshot on another branch. main is unchanged, but the
+ // staged snapshot's sequence number is now taken, so the catalog
+ // rejects the commit as a conflict.
+ builder, err := MetadataBuilderFromBase(cat.metadata, "")
+ require.NoError(t, err)
+ peerSeq := cat.metadata.LastSequenceNumber() + 1
+ require.NoError(t, builder.AddSnapshot(&Snapshot{
+ SnapshotID: mainHead + 1,
+ ParentSnapshotID: &mainHead,
+ SequenceNumber: peerSeq,
+ TimestampMs: cat.metadata.LastUpdatedMillis() + 1,
+ Summary: &Summary{Operation: OpAppend},
+ }))
+ require.NoError(t, builder.SetSnapshotRef("other", mainHead+1,
BranchRef))
+ cat.metadata, err = builder.Build()
+ require.NoError(t, err)
+ cat.errs = []error{nil, ErrCommitFailed}
+ var stagedList string
+ for _, u := range tx.meta.updates {
+ if su, ok := u.(*addSnapshotUpdate); ok {
+ stagedList = su.Snapshot.ManifestList
+ }
+ }
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ head := committed.CurrentSnapshot()
+ require.NotNil(t, head.ParentSnapshotID)
+ assert.Equal(t, mainHead, *head.ParentSnapshotID)
+ assert.Equal(t, peerSeq+1, head.SequenceNumber, "the resume must
renumber the staged snapshot")
+ assert.NotEqual(t, stagedList, head.ManifestList, "the rebuilt manifest
list must not reuse the first call's name")
+}
+
+func TestTransactionCommit_ResumeOfNonReplayableCommit(t *testing.T) {
+ const hint = "commit carries snapshot-relative delete-file removals and
cannot be replayed; reload the table and rebuild the removals"
+ head := int64(100)
+
+ t.Run("branch unchanged resubmits", func(t *testing.T) {
+ base := newConflictTestMetadataWithProps(t, &head, nil)
+ cat := &sequentialCatalog{metadata: base, errs:
[]error{ErrCommitFailed}}
+ tbl := New(Identifier{"db", "no-replay"}, base, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, cat)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"k":
"v"}))
+ tx.noReplay = true
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ assert.EqualError(t, err, "commit failed, refresh and try again
("+hint+")")
+
+ _, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), cat.attempts.Load())
+ })
+
+ t.Run("branch moved is unusable", func(t *testing.T) {
+ base := newConflictTestMetadataWithProps(t, &head, nil)
+ cat := &sequentialCatalog{metadata: base, loadMeta:
graftSnapshotOnto(t, base, MainBranch, 200), errs: []error{ErrCommitFailed}}
+ tbl := New(Identifier{"db", "no-replay"}, base, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, cat)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"k":
"v"}))
+ tx.noReplay = true
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.EqualError(t, err, "transaction cannot be committed
again: requirement no longer holds after refresh ("+hint+"): "+
+ `requirement failed: branch "main" has changed:
expected id 100, found 200`)
+ assert.Equal(t, int32(1), cat.attempts.Load(), "the resume must
stop before the catalog")
+ })
+}
+
// doCommit exhausts its own retry loop on ErrCommitFailed, the transaction
must still be left retriable.
func TestTransactionCommit_RetriableAfterExhaustedInternalRetries(t
*testing.T) {
// numRetries == 2 → doCommit makes 3 attempts, all clean conflicts, so
it
@@ -1141,8 +1289,218 @@ func
TestTransactionCommit_RetriableAfterExhaustedInternalRetries(t *testing.T)
_, err := tx.Commit(t.Context())
require.ErrorIs(t, err, ErrCommitFailed)
+ assert.NotErrorIs(t, err, ErrTransactionUnusable)
assert.Equal(t, int32(3), cat.attempts.Load(), "doCommit must exhaust
all internal attempts")
assert.False(t, tx.committed, "exhausted clean-conflict retries must
leave committed == false")
+
+ _, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int32(4), cat.attempts.Load())
+ assert.Equal(t, 1, countAssertTableUUID(cat.lastReqs),
+ "retry must not append a duplicate AssertTableUUID")
+}
+
+func TestTransactionCommit_CommittableAfterTransientFailureFollowingRebuild(t
*testing.T) {
+ tests := []struct {
+ name string
+ setup func(*progressingRebuildCatalog, context.CancelFunc)
+ wantErr error
+ message string
+ }{
+ {
+ name: "refresh error",
+ setup: func(c *progressingRebuildCatalog, _
context.CancelFunc) { c.failLoad = 2 },
+ message: "refresh table for retry: simulated refresh
failure",
+ },
+ {
+ name: "context cancelled during retry wait",
+ setup: func(c *progressingRebuildCatalog, cancel
context.CancelFunc) { c.cancelAt, c.cancel = 2, cancel },
+ wantErr: context.Canceled,
+ message: "context canceled",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ wfs, meta := newMemIOWithRetryMeta(t, spec)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ // The first two submissions conflict and the third
lands.
+ cat := &progressingRebuildCatalog{
+ metadata: meta, wfs: wfs, location:
"mem://default/table-location", branch: MainBranch, failTimes: 2,
+ }
+ tt.setup(cat, cancel)
+ tbl := newOCCTable(t, meta, wfs, cat)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(),
[]iceberg.DataFile{
+ newTestDataFile(t, spec,
"mem://default/table-location/data/f.parquet", nil),
+ }, nil))
+ var staged string
+ for _, u := range tx.meta.updates {
+ if su, ok := u.(*addSnapshotUpdate); ok {
+ staged = su.Snapshot.ManifestList
+ }
+ }
+ require.NotEmpty(t, staged)
+
+ _, err := tx.Commit(ctx)
+ if tt.wantErr != nil {
+ require.ErrorIs(t, err, tt.wantErr)
+ }
+ require.EqualError(t, err, tt.message)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ require.Equal(t, int32(2), cat.commitTableCalls.Load(),
"the failure must follow a rebuilt attempt")
+ assertOnlyManifestList(t, wfs, staged)
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int32(3), cat.commitTableCalls.Load())
+ require.NotNil(t,
committed.CurrentSnapshot().ParentSnapshotID)
+ assert.Equal(t, int64(9_002),
*committed.CurrentSnapshot().ParentSnapshotID, "the resume must rebase onto the
peers")
+ _, stillExists := wfs.files[staged]
+ assert.False(t, stillExists, "the staged list is
deleted once the commit lands")
+ _, live :=
wfs.files[committed.CurrentSnapshot().ManifestList]
+ assert.True(t, live)
+ })
+ }
+}
+
+func TestTransactionCommit_BusyTableWithRetriesDisabledResumesUntilItLands(t
*testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ wfs := newMemIO(1<<20, nil)
+ meta, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder,
"mem://default/table-location", nil)
+ require.NoError(t, err)
+ // Each failed submission grafts a peer commit, so every resume must
+ // rebuild onto a head that moves again before the next submission.
+ cat := &progressingRebuildCatalog{
+ metadata: meta, wfs: wfs, location:
"mem://default/table-location", branch: MainBranch, failTimes: 2,
+ }
+ tbl := newOCCTable(t, meta, wfs, cat)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{
+ newTestDataFile(t, spec,
"mem://default/table-location/data/f.parquet", nil),
+ }, nil))
+ var staged string
+ for _, u := range tx.meta.updates {
+ if su, ok := u.(*addSnapshotUpdate); ok {
+ staged = su.Snapshot.ManifestList
+ }
+ }
+
+ for range 2 {
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ assertOnlyManifestList(t, wfs, staged)
+ }
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int32(3), cat.commitTableCalls.Load(), "each Commit
submits once with retries disabled")
+ head := committed.CurrentSnapshot()
+ require.NotNil(t, head.ParentSnapshotID)
+ assert.Equal(t, int64(9_002), *head.ParentSnapshotID, "the last resume
must rebase onto both peers")
+ assert.Equal(t, "3", head.Summary.Properties["total-data-files"], "both
peers' files and ours")
+ assertOnlyManifestList(t, wfs, head.ManifestList)
+}
+
+func TestTransactionCommit_AbandonedAfterFailureLeavesOnlyStagedFiles(t
*testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ wfs, meta := newMemIOWithRetryMeta(t, spec)
+ cat := &progressingRebuildCatalog{
+ metadata: meta, wfs: wfs, location:
"mem://default/table-location", branch: MainBranch, failTimes: 99,
+ }
+ tbl := newOCCTable(t, meta, wfs, cat)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{
+ newTestDataFile(t, spec,
"mem://default/table-location/data/f.parquet", nil),
+ }, nil))
+ before := slices.Sorted(maps.Keys(wfs.files))
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.Equal(t, int32(4), cat.commitTableCalls.Load())
+
+ var after []string
+ for path := range wfs.files {
+ if !strings.HasPrefix(filepath.Base(path), "peer-") {
+ after = append(after, path)
+ }
+ }
+ slices.Sort(after)
+ assert.Equal(t, before, after, "only the staged files may remain, and
all of them")
+}
+
+func TestTransactionTableCommitRefusedAfterFailedCommit(t *testing.T) {
+ cat := &sequentialCatalog{errs: []error{ErrCommitFailed}}
+ tbl := newRetryTestTable(t, cat, nil)
+ cat.metadata = tbl.Metadata()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(map[string]string{"key": "value"}))
+ _, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+
+ _, err = tx.TableCommit()
+ require.EqualError(t, err, "a previous commit of this transaction
failed, call Commit to resume it")
+ assert.NotErrorIs(t, err, ErrTransactionUnusable)
+
+ _, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+}
+
+// A rebuild that writes the first staged snapshot's list and then fails on
+// the second still owns the list it wrote. The staged lists stay.
+func TestDoCommit_PartialRebuildFailureRemovesWrittenList(t *testing.T) {
+ head := int64(100)
+ base := newConflictTestMetadataWithProps(t, &head, iceberg.Properties{
+ CommitNumRetriesKey: "1",
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ })
+ wfs := newMemIO(1<<20, nil)
+ const loc = "mem://default/table-location/metadata/"
+ for _, name := range []string{"staged-1.avro", "staged-2.avro"} {
+ require.NoError(t, wfs.WriteFile(loc+name, []byte("list")))
+ }
+
+ rebuildErr := errors.New("simulated rebuild failure")
+ first := &addSnapshotUpdate{
+ baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+ Snapshot: &Snapshot{SnapshotID: 101, ParentSnapshotID: &head,
SequenceNumber: 2, ManifestList: loc + "staged-1.avro"},
+ rebuildManifestList: func(_ context.Context, _ Metadata, parent
*Snapshot, fio iceio.WriteFileIO, _ int) (*Snapshot, error) {
+ if err := fio.WriteFile(loc+"rebuilt-1.avro",
[]byte("list")); err != nil {
+ return nil, err
+ }
+ parentID := parent.SnapshotID
+
+ return &Snapshot{SnapshotID: 101, ParentSnapshotID:
&parentID, SequenceNumber: 3, ManifestList: loc + "rebuilt-1.avro"}, nil
+ },
+ }
+ firstID := int64(101)
+ second := &addSnapshotUpdate{
+ baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+ Snapshot: &Snapshot{SnapshotID: 102, ParentSnapshotID:
&firstID, SequenceNumber: 3, ManifestList: loc + "staged-2.avro"},
+ rebuildManifestList: func(context.Context, Metadata, *Snapshot,
iceio.WriteFileIO, int) (*Snapshot, error) {
+ return nil, rebuildErr
+ },
+ }
+
+ cat := &sequentialCatalog{metadata: base, loadMeta:
graftSnapshotOnto(t, base, MainBranch, 200), errs: []error{ErrCommitFailed}}
+ tbl := newOCCTable(t, base, wfs, cat)
+ _, err := tbl.doCommit(t.Context(), []Update{first, second}, nil,
withCommitBranch(MainBranch))
+ require.ErrorIs(t, err, rebuildErr)
+
+ assert.Contains(t, wfs.files, loc+"staged-1.avro")
+ assert.Contains(t, wfs.files, loc+"staged-2.avro")
+ assert.NotContains(t, wfs.files, loc+"rebuilt-1.avro", "the list the
failed rebuild wrote must be removed")
}
func TestTransactionCommit_TerminalOnUnknownState(t *testing.T) {
diff --git a/table/no_replay_removals_test.go b/table/no_replay_removals_test.go
index 36dbf2fd6..0cf55062d 100644
--- a/table/no_replay_removals_test.go
+++ b/table/no_replay_removals_test.go
@@ -26,6 +26,7 @@ package table_test
import (
"context"
+ "fmt"
"path/filepath"
"testing"
@@ -76,7 +77,8 @@ func newNoReplayV3Table(t *testing.T) (*table.Table,
*concurrentTestCatalog) {
// own merged DV; a stale writer (still seeing DV1 live) then commits
// its own supersession with retries enabled.
// Assertion: the stale commit fails wrapping ErrCommitFailed after
-// exactly one CommitTable attempt (no replay), and the table carries
+// exactly one CommitTable attempt (no replay), the re-commit finds the
+// branch moved and is refused before the catalog, and the table carries
// exactly one live DV — the peer's — so only the peer's deletes apply.
func TestMoRDeleteSupersedingDVFailsInsteadOfReplaying(t *testing.T) {
ctx := context.Background()
@@ -101,7 +103,8 @@ func TestMoRDeleteSupersedingDVFailsInsteadOfReplaying(t
*testing.T) {
require.NoError(t, staleTxn.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(3)), nil))
// Peer supersedes DV1 from the same view and wins the race.
- _, err = tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(2)), nil)
+ staleHead := tbl.CurrentSnapshot().SnapshotID
+ peer, err := tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(2)), nil)
require.NoError(t, err)
attemptsBefore := cat.attempts.Load()
@@ -111,13 +114,16 @@ func TestMoRDeleteSupersedingDVFailsInsteadOfReplaying(t
*testing.T) {
assert.Equal(t, attemptsBefore+1, cat.attempts.Load(),
"a commit carrying DV removals must fail on the first CAS
conflict, not refresh-and-replay")
- // The clean conflict leaves the transaction retriable, not latched
- // as committed: a naive same-transaction retry surfaces a fresh
- // conflict (its requirement still targets the stale base) rather
- // than "transaction has already been committed".
+ assert.NotErrorIs(t, err, table.ErrTransactionUnusable)
+ assert.Contains(t, err.Error(), "reload the table and rebuild the
removals")
+
+ attemptsBefore = cat.attempts.Load()
_, err = staleTxn.Commit(ctx)
- require.Error(t, err)
- assert.ErrorIs(t, err, table.ErrCommitFailed)
+ require.ErrorIs(t, err, table.ErrTransactionUnusable)
+ assert.EqualError(t, err, fmt.Sprintf("transaction cannot be committed
again: requirement no longer holds after refresh "+
+ "(commit carries snapshot-relative delete-file removals and
cannot be replayed; reload the table and rebuild the removals): "+
+ "requirement failed: branch \"main\" has changed: expected id
%d, found %d", staleHead, peer.CurrentSnapshot().SnapshotID))
+ assert.Equal(t, attemptsBefore, cat.attempts.Load(), "the re-commit
must stop before the catalog")
// The table is uncorrupted: exactly one live DV (the peer's merged
// one) references the data file, and only the peer's deletes apply.
@@ -231,6 +237,130 @@ func
TestReplaceFilesWithRemovalsFailsInsteadOfReplaying(t *testing.T) {
"the stale compaction must not have landed; ids 1 and 2 stay
deleted by the live position deletes")
}
+// A peer snapshot on another branch leaves main and every requirement
+// 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
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ dataPath := tbl.Location() + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc,
+ `[{"id": 1, "data": "alpha"}, {"id": 2, "data": "beta"}, {"id":
3, "data": "gamma"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(t.Context(),
[]string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ posDelPath := tbl.Location() + "/data/pos-del-001.parquet"
+ writeParquetFile(t, posDelPath, table.PositionalDeleteArrowSchema,
+ `[{"file_path": "`+dataPath+`", "pos": 1}]`)
+ tx2 := tbl.NewTransaction()
+ rd := tx2.NewRowDelta(nil)
+ rd.AddDeletes(newPosDeleteFile(t, posDelPath))
+ require.NoError(t, rd.Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+ mainHead := tbl.CurrentSnapshot().SnapshotID
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ compactedPath := tbl.Location() + "/data/data-compacted.parquet"
+ writeParquetFile(t, compactedPath, arrowSc, `[{"id": 1, "data":
"alpha"}, {"id": 3, "data": "gamma"}]`)
+ compacted, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ compactedPath, iceberg.ParquetFile, nil, nil, nil, 2, 512)
+ require.NoError(t, err)
+ staleTxn := tbl.NewTransaction()
+ require.NoError(t, staleTxn.ReplaceFiles(t.Context(),
+ []iceberg.DataFile{tasks[0].File},
[]iceberg.DataFile{compacted.Build()}, tasks[0].DeleteFiles, nil))
+
+ peerPath := tbl.Location() + "/data/data-peer.parquet"
+ writeParquetFile(t, peerPath, arrowSc, `[{"id": 9, "data": "peer"}]`)
+ peerTxn, err := tbl.NewTransactionOnBranch("other")
+ require.NoError(t, err)
+ require.NoError(t, peerTxn.AddDataFilesByPath(t.Context(),
[]string{peerPath}, nil, false))
+ _, err = peerTxn.Commit(t.Context())
+ require.NoError(t, err)
+
+ _, err = staleTxn.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ require.NotErrorIs(t, err, table.ErrTransactionUnusable)
+ assert.Contains(t, err.Error(), "reload the table and rebuild the
removals")
+
+ committed, err := staleTxn.Commit(t.Context())
+ require.NoError(t, err)
+ head := committed.CurrentSnapshot()
+ require.NotNil(t, head.ParentSnapshotID)
+ assert.Equal(t, mainHead, *head.ParentSnapshotID, "the resume must keep
the pinned parent")
+ assert.Equal(t, []int64{1, 3}, idsInTable(t, committed))
+ assert.Empty(t, deleteEntriesReferencing(t, committed,
map[string]struct{}{dataPath: {}, compactedPath: {}}),
+ "the compaction removed the only live delete")
+ other := committed.Metadata().SnapshotByName("other")
+ require.NotNil(t, other.ParentSnapshotID)
+ assert.Equal(t, mainHead, *other.ParentSnapshotID, "the peer branch is
untouched")
+}
+
+// A peer that removed a file this compaction replaces makes the resume's
+// rebuild fail the same way on every later attempt, so the transaction
+// must become unusable.
+func TestReplaceFilesResumeAfterPeerRemovedInputIsUnusable(t *testing.T) {
+ tbl, cat := newConcurrentRewriteTestTable(t)
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ dataPath := tbl.Location() + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[{"id": 1, "data": "alpha"},
{"id": 2, "data": "beta"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(t.Context(),
[]string{dataPath}, nil, false))
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{table.CommitNumRetriesKey: "0"}))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ replacement := func(name string) iceberg.DataFile {
+ path := tbl.Location() + "/data/" + name + ".parquet"
+ writeParquetFile(t, path, arrowSc, `[{"id": 1, "data":
"alpha"}, {"id": 2, "data": "beta"}]`)
+ b, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ path, iceberg.ParquetFile, nil, nil, nil, 2, 512)
+ require.NoError(t, err)
+
+ return b.Build()
+ }
+
+ staleTxn := tbl.NewTransaction()
+ require.NoError(t, staleTxn.ReplaceFiles(t.Context(),
+ []iceberg.DataFile{tasks[0].File},
[]iceberg.DataFile{replacement("stale")}, nil, nil))
+
+ peerTxn := tbl.NewTransaction()
+ require.NoError(t, peerTxn.ReplaceFiles(t.Context(),
+ []iceberg.DataFile{tasks[0].File},
[]iceberg.DataFile{replacement("peer")}, nil, nil))
+ _, err = peerTxn.Commit(t.Context())
+ require.NoError(t, err)
+
+ _, err = staleTxn.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ require.NotErrorIs(t, err, table.ErrTransactionUnusable)
+
+ attempts := cat.attempts.Load()
+ _, err = staleTxn.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrTransactionUnusable)
+ require.ErrorIs(t, err, table.ErrCommitDiverged)
+ require.NotErrorIs(t, err, table.ErrCommitFailed, "a caller retrying on
ErrCommitFailed must stop")
+ assert.EqualError(t, err, fmt.Sprintf("transaction cannot be committed
again: files to rewrite were removed: "+
+ "commit diverged: base snapshot is no longer on the branch: 1
data file(s) to rewrite are no longer on the branch head, e.g. [%q]", dataPath))
+ assert.NotContains(t, err.Error(), "refresh and try again")
+ assert.Equal(t, attempts, cat.attempts.Load(), "the resume must stop
before the catalog")
+
+ _, err = staleTxn.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrTransactionUnusable)
+ assert.Equal(t, attempts, cat.attempts.Load())
+}
+
// Regression guard: the noReplay flag must not leak to commits that
// remove only data files. Those removals are path-keyed and the retry
// rebuild fails terminally when the path is gone from the fresh base,
diff --git a/table/rebuild_manifest_test.go b/table/rebuild_manifest_test.go
index d6b619503..411324f6a 100644
--- a/table/rebuild_manifest_test.go
+++ b/table/rebuild_manifest_test.go
@@ -137,6 +137,7 @@ func
TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
MainBranch,
iceio.LocalFS{},
1,
+ false,
)
require.NoError(t, err)
require.Len(t, rebuilt, 1)
@@ -225,6 +226,7 @@ func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t
*testing.T) {
MainBranch,
iceio.LocalFS{},
1,
+ false,
)
require.NoError(t, err)
assert.False(t, called, "rebuild closure must not be called when parent
is already up-to-date")
@@ -259,6 +261,7 @@ func
TestRebuildSnapshotUpdates_RebuildsWhenTableStateMoved(t *testing.T) {
MainBranch,
iceio.LocalFS{},
1,
+ false,
)
require.NoError(t, err)
require.NotNil(t, receivedParent, "the rebuild must run even though the
parent is unchanged")
@@ -287,6 +290,7 @@ func
TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
MainBranch,
iceio.LocalFS{},
0,
+ false,
)
require.NoError(t, err)
assert.Empty(t, orphaned)
@@ -315,7 +319,7 @@ func TestRebuildSnapshotUpdates_PropagatesClosureError(t
*testing.T) {
},
}
- _, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd},
freshMeta, MainBranch, iceio.LocalFS{}, 1)
+ _, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd},
freshMeta, MainBranch, iceio.LocalFS{}, 1, false)
assert.ErrorIs(t, err, wantErr)
}
diff --git a/table/requirement_test.go b/table/requirement_test.go
index 19b41f7ac..550c44d26 100644
--- a/table/requirement_test.go
+++ b/table/requirement_test.go
@@ -20,6 +20,7 @@ package table_test
import (
"encoding/json"
"fmt"
+ "strings"
"testing"
"github.com/apache/iceberg-go/table"
@@ -310,3 +311,35 @@ func TestAssertRefSnapshotIDValidate(t *testing.T) {
assert.NoError(t, req.Validate(meta))
})
}
+
+func TestRequirementFailuresAreCommitConflicts(t *testing.T) {
+ meta, err :=
table.ParseMetadataBytes([]byte(table.ExampleTableMetadataV2))
+ require.NoError(t, err)
+
+ tests := []struct {
+ name string
+ req table.Requirement
+ meta table.Metadata
+ }{
+ {"table already exists", table.AssertCreate(), meta},
+ {"table uuid mismatch", table.AssertTableUUID(uuid.New()),
meta},
+ {"table metadata missing",
table.AssertTableUUID(meta.TableUUID()), nil},
+ {"ref has changed", table.AssertRefSnapshotID("test",
new(int64(1))), meta},
+ {"ref created concurrently", table.AssertRefSnapshotID("test",
nil), meta},
+ {"ref missing", table.AssertRefSnapshotID("nonexistent",
new(int64(1))), meta},
+ {"last assigned field id changed",
table.AssertLastAssignedFieldID(meta.LastColumnID() + 1), meta},
+ {"current schema id changed",
table.AssertCurrentSchemaID(meta.CurrentSchema().ID + 1), meta},
+ {"last assigned partition id changed",
table.AssertLastAssignedPartitionID(*meta.LastPartitionSpecID() + 1), meta},
+ {"default spec id changed",
table.AssertDefaultSpecID(meta.DefaultPartitionSpec() + 1), meta},
+ {"default sort order id changed",
table.AssertDefaultSortOrderID(meta.DefaultSortOrder() + 1), meta},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ err := tt.req.Validate(tt.meta)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrCommitFailed)
+ assert.Contains(t, strings.ToLower(err.Error()),
"requirement failed")
+ })
+ }
+}
diff --git a/table/requirements.go b/table/requirements.go
index a1abceed6..adcd0fc9f 100644
--- a/table/requirements.go
+++ b/table/requirements.go
@@ -227,6 +227,16 @@ func (b baseRequirement) GetType() string {
return b.Type
}
+func requirementFailed(format string, args ...any) error {
+ return &requirementError{msg: fmt.Sprintf(format, args...)}
+}
+
+type requirementError struct{ msg string }
+
+func (e *requirementError) Error() string { return e.msg }
+
+func (e *requirementError) Unwrap() error { return ErrCommitFailed }
+
type assertCreate struct {
baseRequirement
}
@@ -240,7 +250,7 @@ func AssertCreate() Requirement {
func (a *assertCreate) Validate(meta Metadata) error {
if meta != nil {
- return errors.New("Requirement failed: table already exists")
+ return requirementFailed("Requirement failed: table already
exists")
}
return nil
@@ -261,11 +271,11 @@ func AssertTableUUID(uuid uuid.UUID) Requirement {
func (a *assertTableUuid) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("Requirement failed: current table metadata
does not exist")
+ return requirementFailed("Requirement failed: current table
metadata does not exist")
}
if meta.TableUUID() != a.UUID {
- return fmt.Errorf("Requirement failed: UUID mismatch: %s !=
%s", meta.TableUUID(), a.UUID)
+ return requirementFailed("Requirement failed: UUID mismatch: %s
!= %s", meta.TableUUID(), a.UUID)
}
return nil
@@ -300,7 +310,7 @@ func assertBranchRefSnapshotID(ref string, id *int64)
Requirement {
func (a *assertRefSnapshotID) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("Requirement failed: current table metadata
does not exist")
+ return requirementFailed("Requirement failed: current table
metadata does not exist")
}
var r *SnapshotRef
@@ -318,14 +328,14 @@ func (a *assertRefSnapshotID) Validate(meta Metadata)
error {
}
if a.SnapshotID == nil {
- return fmt.Errorf("requirement failed: %s %q was
created concurrently", r.Type, a.Ref)
+ return requirementFailed("requirement failed: %s %q was
created concurrently", r.Type, a.Ref)
}
if r.SnapshotID != *a.SnapshotID {
- return fmt.Errorf("requirement failed: %s %q has
changed: expected id %d, found %d", r.Type, a.Ref, *a.SnapshotID, r.SnapshotID)
+ return requirementFailed("requirement failed: %s %q has
changed: expected id %d, found %d", r.Type, a.Ref, *a.SnapshotID, r.SnapshotID)
}
} else if a.SnapshotID != nil {
- return fmt.Errorf("requirement failed: branch or tag %q is
missing, expected %d", a.Ref, *a.SnapshotID)
+ return requirementFailed("requirement failed: branch or tag %q
is missing, expected %d", a.Ref, *a.SnapshotID)
}
return nil
@@ -347,11 +357,11 @@ func AssertLastAssignedFieldID(id int) Requirement {
func (a *assertLastAssignedFieldId) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("Requirement failed: current table metadata
does not exist")
+ return requirementFailed("Requirement failed: current table
metadata does not exist")
}
if meta.LastColumnID() != a.LastAssignedFieldID {
- return fmt.Errorf("Requirement failed: last assigned field id
has changed: expected %d, found %d", a.LastAssignedFieldID, meta.LastColumnID())
+ return requirementFailed("Requirement failed: last assigned
field id has changed: expected %d, found %d", a.LastAssignedFieldID,
meta.LastColumnID())
}
return nil
@@ -373,11 +383,11 @@ func AssertCurrentSchemaID(id int) Requirement {
func (a *assertCurrentSchemaId) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("Requirement failed: current table metadata
does not exist")
+ return requirementFailed("Requirement failed: current table
metadata does not exist")
}
if meta.CurrentSchema().ID != a.CurrentSchemaID {
- return fmt.Errorf("Requirement failed: current schema id has
changed: expected %d, found %d", a.CurrentSchemaID, meta.CurrentSchema().ID)
+ return requirementFailed("Requirement failed: current schema id
has changed: expected %d, found %d", a.CurrentSchemaID, meta.CurrentSchema().ID)
}
return nil
@@ -399,7 +409,7 @@ func AssertLastAssignedPartitionID(id int) Requirement {
func (a *assertLastAssignedPartitionId) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("Requirement failed: current table metadata
does not exist")
+ return requirementFailed("Requirement failed: current table
metadata does not exist")
}
lastPartitionSpecID := meta.LastPartitionSpecID()
@@ -408,7 +418,7 @@ func (a *assertLastAssignedPartitionId) Validate(meta
Metadata) error {
}
if *lastPartitionSpecID != a.LastAssignedPartitionID {
- return fmt.Errorf("requirement failed: last assigned partition
id has changed: expected %d, found %d", a.LastAssignedPartitionID,
*lastPartitionSpecID)
+ return requirementFailed("requirement failed: last assigned
partition id has changed: expected %d, found %d", a.LastAssignedPartitionID,
*lastPartitionSpecID)
}
return nil
@@ -430,11 +440,11 @@ func AssertDefaultSpecID(id int) Requirement {
func (a *assertDefaultSpecId) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("Requirement failed: current table metadata
does not exist")
+ return requirementFailed("Requirement failed: current table
metadata does not exist")
}
if meta.DefaultPartitionSpec() != a.DefaultSpecID {
- return fmt.Errorf("Requirement failed: default spec id has
changed: expected %d, found %d", a.DefaultSpecID, meta.DefaultPartitionSpec())
+ return requirementFailed("Requirement failed: default spec id
has changed: expected %d, found %d", a.DefaultSpecID,
meta.DefaultPartitionSpec())
}
return nil
@@ -456,11 +466,11 @@ func AssertDefaultSortOrderID(id int) Requirement {
func (a *assertDefaultSortOrderId) Validate(meta Metadata) error {
if meta == nil {
- return errors.New("requirement failed: current table metadata
does not exist")
+ return requirementFailed("requirement failed: current table
metadata does not exist")
}
if meta.DefaultSortOrder() != a.DefaultSortOrderID {
- return fmt.Errorf("requirement failed: default sort order id
has changed: expected %d, found %d", a.DefaultSortOrderID,
meta.DefaultSortOrder())
+ return requirementFailed("requirement failed: default sort
order id has changed: expected %d, found %d", a.DefaultSortOrderID,
meta.DefaultSortOrder())
}
return nil
diff --git a/table/requirements_internal_test.go
b/table/requirements_internal_test.go
index 5aa772d92..c75531d5f 100644
--- a/table/requirements_internal_test.go
+++ b/table/requirements_internal_test.go
@@ -34,6 +34,7 @@ func TestAssertLastAssignedPartitionIDValidate(t *testing.T) {
err := req.Validate(meta)
require.Error(t, err)
assert.EqualError(t, err, "requirement failed: last assigned
partition id is missing")
+ assert.NotErrorIs(t, err, ErrCommitFailed, "malformed metadata
is not a commit conflict")
})
t.Run("matches metadata last partition id", func(t *testing.T) {
@@ -52,3 +53,14 @@ func TestAssertLastAssignedPartitionIDValidate(t *testing.T)
{
assert.Contains(t, err.Error(), "last assigned partition id has
changed")
})
}
+
+func TestAssertBranchRefSnapshotIDRejectsTagWithoutRetry(t *testing.T) {
+ meta, err := ParseMetadataBytes([]byte(ExampleTableMetadataV2))
+ require.NoError(t, err)
+
+ // "test" is a tag. Refresh cannot make it a valid transaction target.
+ err = assertBranchRefSnapshotID("test",
new(int64(3051729675574597004))).Validate(meta)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "tags cannot be transaction targets")
+ assert.NotErrorIs(t, err, ErrCommitFailed)
+}
diff --git a/table/rewrite_data_files.go b/table/rewrite_data_files.go
index a9ca91016..cda61f6c7 100644
--- a/table/rewrite_data_files.go
+++ b/table/rewrite_data_files.go
@@ -598,8 +598,8 @@ func (t *Transaction) rewriteDataFilesPartial(ctx
context.Context, groups []Comp
if err != nil {
return nil, err
}
- if t.committed {
- return nil, errors.New("transaction has already been committed")
+ if err := t.checkUsable(); err != nil {
+ return nil, err
}
result := &RewriteResult{Table: t.tbl}
if len(meta.updates) > 0 || len(t.reqs) > 0 || len(t.validators) > 0 {
@@ -744,9 +744,10 @@ func (t *Transaction) rewriteDataFilesPartial(ctx
context.Context, groups []Comp
return result, err
}
- // ErrCommitFailed is the only error that proves the
catalog did not
- // commit. Every other error leaves commit state
unknown, so continuing
- // with the old table could apply later batches on
stale state.
+ // Only a conflict lets the loop move on to the next
batch, since each
+ // batch commits through its own transaction built on
the current
+ // table. Any other failure stops the rewrite. It may
be of unknown
+ // commit state, so continuing could apply later
batches on stale state.
if !errors.Is(err, ErrCommitFailed) {
t.committed = true
diff --git a/table/rewrite_files_test.go b/table/rewrite_files_test.go
index df2c512e8..9ddb30683 100644
--- a/table/rewrite_files_test.go
+++ b/table/rewrite_files_test.go
@@ -53,6 +53,9 @@ 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) {
@@ -68,6 +71,10 @@ 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/rewrite_manifests.go b/table/rewrite_manifests.go
index 66a8ef3ba..e2e7dade0 100644
--- a/table/rewrite_manifests.go
+++ b/table/rewrite_manifests.go
@@ -154,21 +154,11 @@ func newRewriteManifestsProducer(txn *Transaction, fs
iceio.WriteFileIO, props i
return prod
}
-// supersededManifests returns the merged manifests orphaned across OCC
retries,
-// safe to delete once the commit resolves. When the commit never landed
-// (committed is false) the final attempt's manifests are orphaned too — a
-// snapshot that never committed references nothing — so they are appended.
-func (r *rewriteManifests) supersededManifests(committed bool) []string {
- if committed {
- return r.superseded
- }
-
- paths := slices.Clone(r.superseded)
- for _, m := range r.added {
- paths = append(paths, m.FilePath())
- }
-
- return paths
+// supersededManifests returns the merged manifests earlier passes wrote and a
+// later pass replaced. The committed snapshot references only the last pass's
+// manifests, so these are safe to delete once the commit lands.
+func (r *rewriteManifests) supersededManifests() []string {
+ return r.superseded
}
func (r *rewriteManifests) parentDependentManifests(_ context.Context, parent
*Snapshot) ([]iceberg.ManifestFile, error) {
@@ -322,15 +312,6 @@ func (r *rewriteManifests) deleteWritten(ctx
context.Context, input, merged []ic
}
func (r *rewriteManifests) record(toRewrite, merged, kept
[]iceberg.ManifestFile) error {
- // A prior pass (a superseded OCC attempt) wrote its own merged
manifests
- // that the catalog never referenced. Carry their paths into the
- // producer's superseded set so commit() can delete them on success;
- // otherwise every retry leaks a full set of merged .avro files.
- for _, m := range r.added {
- r.superseded = append(r.superseded, m.FilePath())
- }
- r.added, r.rewritten = nil, nil
-
inPaths := make(map[string]struct{}, len(toRewrite))
for _, m := range toRewrite {
inPaths[m.FilePath()] = struct{}{}
@@ -343,14 +324,15 @@ func (r *rewriteManifests) record(toRewrite, merged, kept
[]iceberg.ManifestFile
// Bins of a single manifest pass through unchanged; only the manifests
// that actually appear on one side and not the other are
added/replaced.
// This drives the returned RewriteManifestsResult.
+ var added, rewritten []iceberg.ManifestFile
for _, m := range merged {
if _, ok := inPaths[m.FilePath()]; !ok {
- r.added = append(r.added, m)
+ added = append(added, m)
}
}
for _, m := range toRewrite {
if _, ok := outPaths[m.FilePath()]; !ok {
- r.rewritten = append(r.rewritten, m)
+ rewritten = append(rewritten, m)
}
}
@@ -360,11 +342,11 @@ func (r *rewriteManifests) record(toRewrite, merged, kept
[]iceberg.ManifestFile
// rewritten/added diff is equivalent to comparing the full
input/output and
// avoids re-walking the pass-throughs. The processed count doubles as
the
// entries-processed summary value.
- processed, err := manifestActiveFiles(r.base.io, r.rewritten)
+ processed, err := manifestActiveFiles(r.base.io, rewritten)
if err != nil {
return err
}
- written, err := manifestActiveFiles(r.base.io, r.added)
+ written, err := manifestActiveFiles(r.base.io, added)
if err != nil {
return err
}
@@ -372,6 +354,16 @@ func (r *rewriteManifests) record(toRewrite, merged, kept
[]iceberg.ManifestFile
return fmt.Errorf("rewrite manifests changed active file count:
%d before, %d after", processed, written)
}
+ // Publish only after the guard passes. On failure the caller deletes
this
+ // pass's merged files, so the producer must keep the previous pass's
state.
+ // The previous pass's merged manifests were never committed. Carry them
+ // into the superseded set so a successful commit deletes them, or every
+ // retry leaks a full set of merged files.
+ for _, m := range r.added {
+ r.superseded = append(r.superseded, m.FilePath())
+ }
+ r.added, r.rewritten = added, rewritten
+
// Publish into the shared result the caller holds so a retry's counts
win.
r.result.RewrittenManifests = r.rewritten
r.result.AddedManifests = r.added
diff --git a/table/rewrite_manifests_internal_test.go
b/table/rewrite_manifests_internal_test.go
index 642dac9a1..64eaeb029 100644
--- a/table/rewrite_manifests_internal_test.go
+++ b/table/rewrite_manifests_internal_test.go
@@ -22,6 +22,7 @@ import (
"context"
"crypto/sha256"
"encoding/binary"
+ "slices"
"strconv"
"strings"
"sync"
@@ -383,3 +384,55 @@ func TestRewriteManifestsCleansOrphansOnMergeFailure(t
*testing.T) {
require.Truef(t, preexisting, "merged manifest %s from the bin
that succeeded must be cleaned up", path)
}
}
+
+// A pass that fails the file-count guard deletes the files it merged. The
+// producer must keep the previous pass's state, which a resumed commit can
+// reuse, rather than point at the deleted files.
+func TestRewriteManifestsFailedPassKeepsPreviousState(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ schema := simpleSchema()
+ mem := newMemIO(1<<20, errLimitedWrite)
+ txn := createTestTransaction(t, mem, spec)
+
+ prod := newRewriteManifestsProducer(txn, mem, iceberg.Properties{},
rewriteManifestsCfg{targetSizeBytes: 8 << 20})
+ r := prod.producerImpl.(*rewriteManifests)
+
+ inputs := func(name string, addedFiles int32) []iceberg.ManifestFile {
+ out := make([]iceberg.ManifestFile, 2)
+ sequenceNumber := int64(1)
+ for i := range out {
+ df := newTestDataFile(t, spec,
"file://"+name+"-"+string(rune('a'+i))+".parquet", nil)
+ entries := []iceberg.ManifestEntry{
+
iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &prod.snapshotID,
&sequenceNumber, nil, df),
+ }
+ path := "table-location/metadata/" + name + "-" +
string(rune('a'+i)) + ".avro"
+ var buf bytes.Buffer
+ _, err := iceberg.WriteManifest(path, &buf, 2, spec,
schema, prod.snapshotID, entries)
+ require.NoError(t, err)
+ require.NoError(t, mem.WriteFile(path, buf.Bytes()))
+ out[i] = iceberg.NewManifestBuilder(2, path,
int64(buf.Len()), int32(spec.ID()), prod.snapshotID).
+
AddedFiles(addedFiles).Content(iceberg.ManifestContentData).Build()
+ }
+
+ return out
+ }
+
+ _, err := r.processManifests(inputs("good", 1))
+ require.NoError(t, err)
+ added, rewritten, superseded := slices.Clone(r.added),
slices.Clone(r.rewritten), slices.Clone(r.superseded)
+ require.NotEmpty(t, added)
+
+ // A header count that lies trips the guard after the merged file is
written.
+ _, err = r.processManifests(inputs("lying", 99))
+ require.Error(t, err)
+
+ assert.Equal(t, added, r.added)
+ assert.Equal(t, rewritten, r.rewritten)
+ assert.Equal(t, superseded, r.superseded)
+ assert.Equal(t, added, r.result.AddedManifests)
+ assert.Equal(t, rewritten, r.result.RewrittenManifests)
+ files := memKeys(mem)
+ for _, m := range r.added {
+ assert.Containsf(t, files, m.FilePath(), "manifest %s the
producer would commit must exist", m.FilePath())
+ }
+}
diff --git a/table/rewrite_manifests_test.go b/table/rewrite_manifests_test.go
index d89abce91..87de65987 100644
--- a/table/rewrite_manifests_test.go
+++ b/table/rewrite_manifests_test.go
@@ -1000,10 +1000,12 @@ func
TestRewriteManifestsCleansEverySupersededGeneration(t *testing.T) {
assert.Equal(t, 2, removed, "both superseded generations must be
cleaned up")
}
-// TestRewriteManifestsCleansSupersededOnExhaustedRetries asserts that when
every
-// commit attempt fails with ErrCommitFailed, every merged manifest written —
-// the superseded attempts and the final one — is cleaned up rather than
leaked.
-func TestRewriteManifestsCleansSupersededOnExhaustedRetries(t *testing.T) {
+// TestRewriteManifestsKeepsMergedManifestsUntilResumeLands asserts that a
+// rewrite whose retries are exhausted deletes none of its merged manifests,
+// since the transaction can still commit, and that the resume that lands
+// deletes every merged manifest the committed snapshot does not reference,
+// from both calls.
+func TestRewriteManifestsKeepsMergedManifestsUntilResumeLands(t *testing.T) {
ctx := context.Background()
dir := t.TempDir()
track := &trackingFS{}
@@ -1011,30 +1013,178 @@ func
TestRewriteManifestsCleansSupersededOnExhaustedRetries(t *testing.T) {
h0, h1, h2 := stagedRewriteHeads(t, ctx, dir, "2", fsF, track) // 3
attempts
- // Never succeed: every attempt conflicts, exhausting the retry budget.
- cat := &stagedHeadCatalog{current: h0, heads: []table.Metadata{h1, h2},
failures: 99, location: dir, fsF: fsF}
+ cat := &stagedHeadCatalog{current: h0, heads: []table.Metadata{h1, h2},
failures: 3, location: dir, fsF: fsF}
tbl := table.New(table.Identifier{"default", "staged"}, h0,
dir+"/metadata/00000.json", fsF, cat)
preExisting := track.snapshotCreated()
txn := tbl.NewTransaction()
- _, err := txn.RewriteManifests(ctx)
+ res, err := txn.RewriteManifests(ctx)
require.NoError(t, err)
_, err = txn.Commit(ctx)
- require.Error(t, err, "commit must fail once retries are exhausted")
require.ErrorIs(t, err, table.ErrCommitFailed)
+ require.NotErrorIs(t, err, table.ErrTransactionUnusable)
- // Three attempts wrote three merged manifests. None is referenced (no
commit
- // succeeded), so all three — the superseded ones and the final
attempt's —
- // must be cleaned on the exhausted-failure path.
merged := mergedManifestsCreated(track, preExisting)
require.Len(t, merged, 3, "three attempts each write one merged
manifest")
+ for _, p := range merged {
+ assert.Falsef(t, track.wasRemoved(p), "merged manifest %s must
be kept while the transaction can commit", p)
+ }
+
+ committed, err := txn.Commit(ctx)
+ require.NoError(t, err)
+ assertCommittedManifestsExist(t, committed, track, res)
+ merged = mergedManifestsCreated(track, preExisting)
+ require.Len(t, merged, 4, "the resume merges once more")
+ referenced := committedManifestPaths(t, committed, track)
for _, p := range merged {
- assert.Truef(t, track.wasRemoved(p), "orphaned manifest %s must
be cleaned on the failure path", p)
+ if _, ok := referenced[p]; ok {
+ continue
+ }
+ assert.Truef(t, track.wasRemoved(p), "superseded manifest %s
must be deleted once the commit lands", p)
}
}
+func committedManifestPaths(t *testing.T, committed *table.Table, fs iceio.IO)
map[string]struct{} {
+ t.Helper()
+ manifests, err := committed.CurrentSnapshot().Manifests(fs)
+ require.NoError(t, err)
+ paths := make(map[string]struct{}, len(manifests))
+ for _, m := range manifests {
+ paths[m.FilePath()] = struct{}{}
+ }
+
+ return paths
+}
+
+// assertCommittedManifestsExist checks that the committed snapshot's
+// manifest list and manifests exist and that res reports the manifests the
+// committed snapshot added.
+func assertCommittedManifestsExist(t *testing.T, committed *table.Table, track
*trackingFS, res *table.RewriteManifestsResult) {
+ t.Helper()
+ snap := committed.CurrentSnapshot()
+ require.NotNil(t, snap)
+ assert.FileExists(t, snap.ManifestList)
+ referenced := committedManifestPaths(t, committed, track)
+ for p := range referenced {
+ assert.FileExistsf(t, p, "committed manifest %s must exist", p)
+ }
+ require.NotEmpty(t, res.AddedManifests)
+ for _, m := range res.AddedManifests {
+ assert.Containsf(t, referenced, m.FilePath(), "result manifest
%s must be in the committed snapshot", m.FilePath())
+ }
+}
+
+func resortedMetadata(t *testing.T, base table.Metadata) table.Metadata {
+ t.Helper()
+ builder, err := table.MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ order, err := table.NewSortOrder(1, []table.SortField{{
+ SourceIDs: []int{1},
+ Transform: iceberg.IdentityTransform{},
+ Direction: table.SortASC,
+ NullOrder: table.NullsFirst,
+ }})
+ require.NoError(t, err)
+ require.NoError(t, builder.AddSortOrder(&order))
+ require.NoError(t, builder.SetDefaultSortOrderID(-1))
+ out, err := builder.Build()
+ require.NoError(t, err)
+
+ return out
+}
+
+func TestRewriteManifestsNonRebasedRequirementFailureKeepsStagedFiles(t
*testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+ track := &trackingFS{}
+ fsF := func(_ context.Context) (iceio.IO, error) { return track, nil }
+
+ h0, _, _ := stagedRewriteHeads(t, ctx, dir, "2", fsF, track)
+
+ cat := &stagedHeadCatalog{current: h0, heads:
[]table.Metadata{resortedMetadata(t, h0)}, failures: 1, location: dir, fsF: fsF}
+ tbl := table.New(table.Identifier{"default", "staged"}, h0,
dir+"/metadata/00000.json", fsF, cat)
+
+ preExisting := track.snapshotCreated()
+
+ txn := tbl.NewTransaction()
+ require.NoError(t, txn.AssertDefaultShape())
+ _, err := txn.RewriteManifests(ctx)
+ require.NoError(t, err)
+
+ _, err = txn.Commit(ctx)
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ require.ErrorIs(t, err, table.ErrTransactionUnusable)
+ assert.EqualError(t, err, "transaction cannot be committed again:
requirement no longer holds after refresh: "+
+ "requirement failed: default sort order id has changed:
expected 0, found 1")
+ require.EqualValues(t, 1, cat.commitTableCalls.Load(), "the violated
fence must stop the retry loop")
+
+ created := track.snapshotCreated()
+ for p := range created {
+ if _, ok := preExisting[p]; !ok {
+ assert.Falsef(t, track.wasRemoved(p), "staged file %s
must be kept", p)
+ }
+ }
+ require.Len(t, mergedManifestsCreated(track, preExisting), 1)
+
+ cat.current = h0
+ _, err = txn.Commit(ctx)
+ assert.ErrorIs(t, err, table.ErrTransactionUnusable)
+ assert.EqualValues(t, 1, cat.commitTableCalls.Load(), "the unusable
transaction must not reach the catalog")
+ assert.Same(t, h0, cat.current, "nothing may be committed")
+}
+
+// A peer that rolls main back to the transaction's base head between two
+// Commit calls must not let the resume submit the staged snapshot: the
+// rewrite producer has already moved past it and would delete its merged
+// manifests as superseded once the commit lands. On a v1 table the
+// sequence-number check that would otherwise force a rebuild is skipped.
+func TestRewriteManifestsResumeAfterPeerRollsBackToBaseHead(t *testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+ track := &trackingFS{}
+ fsF := func(_ context.Context) (iceio.IO, error) { return track, nil }
+
+ meta, err := table.NewMetadata(rewriteSchema(),
iceberg.UnpartitionedSpec, table.UnsortedSortOrder, dir,
+ iceberg.Properties{
+ table.PropertyFormatVersion: "1",
+ table.ManifestMergeEnabledKey: "false",
+ table.CommitMinRetryWaitMsKey: "0",
+ table.CommitMaxRetryWaitMsKey: "0",
+ table.CommitNumRetriesKey: "1",
+ })
+ require.NoError(t, err)
+ setup := table.New(table.Identifier{"default", "staged"}, meta,
dir+"/metadata/00000.json", fsF, &mergeCatalog{meta: meta})
+ setup = appendSeparateManifests(t, ctx, setup, track, dir, "stale", 3)
+ h0 := setup.Metadata()
+ baseHead := h0.CurrentSnapshot().SnapshotID
+ h1 := appendOne(t, ctx, setup, track, dir, "peer").Metadata()
+
+ // The first call conflicts twice, rebuilding onto the peer's append in
+ // between. Every later submission lands.
+ cat := &stagedHeadCatalog{current: h0, heads: []table.Metadata{h1},
failures: 2, location: dir, fsF: fsF}
+ tbl := table.New(table.Identifier{"default", "staged"}, h0,
dir+"/metadata/00000.json", fsF, cat)
+
+ txn := tbl.NewTransaction()
+ res, err := txn.RewriteManifests(ctx)
+ require.NoError(t, err)
+ _, err = txn.Commit(ctx)
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+
+ peer, err := cat.LoadTable(ctx, table.Identifier{"default", "staged"})
+ require.NoError(t, err)
+ rollback := peer.NewTransaction()
+ require.NoError(t, rollback.RollbackToSnapshot(baseHead))
+ _, err = rollback.Commit(ctx)
+ require.NoError(t, err)
+ require.Equal(t, baseHead, cat.current.CurrentSnapshot().SnapshotID)
+
+ committed, err := txn.Commit(ctx)
+ require.NoError(t, err)
+ assertCommittedManifestsExist(t, committed, track, res)
+}
+
// TestRewriteManifestsClusterByReusesOutputOnOCCRetry verifies that a
// concurrent append does not force an otherwise-valid clustered rewrite to
// read and rewrite every original entry again. The fresh parent retains all
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index 52024b537..d856a6958 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -67,10 +67,9 @@ type producerImpl interface {
// supersededAccumulator marks a producer that writes merged manifests OCC
// retries orphan. Only rewrite implements it.
type supersededAccumulator interface {
- // supersededManifests returns paths of merged manifests to delete once
the
- // commit resolves. When committed is false the commit never landed, so
the
- // final attempt's manifests are orphaned too and are included.
- supersededManifests(committed bool) []string
+ // supersededManifests returns paths of merged manifests that earlier
+ // passes wrote and a later pass replaced, to delete once the commit
lands.
+ supersededManifests() []string
}
// retryManifestRebuilder lets a producer reuse parent-independent manifest
diff --git a/table/table.go b/table/table.go
index 75d7d36cd..d0690372b 100644
--- a/table/table.go
+++ b/table/table.go
@@ -46,14 +46,35 @@ import (
"golang.org/x/sync/errgroup"
)
-// ErrCommitFailed is the sentinel error returned by catalogs when a
-// commit fails due to a concurrent modification (e.g. HTTP 409 Conflict
-// from the REST catalog, or a commit requirement that no longer holds).
-// Catalog implementations should wrap this error so that callers using
-// errors.Is(err, table.ErrCommitFailed) can detect retryable commit
-// conflicts.
+// ErrCommitFailed means a concurrent change caused the failure. A new
+// transaction built from a reloaded table may succeed. Catalogs return it
+// when a commit loses to a concurrent modification (e.g. HTTP 409 Conflict
+// from the REST catalog, or a commit requirement that no longer holds), and
+// catalog implementations should wrap it so callers can detect conflicts
+// with errors.Is(err, table.ErrCommitFailed). Calling Transaction.Commit
+// again retries the conflict unless the error also matches
+// ErrTransactionUnusable.
+//
+// After a refresh, a stale requirement, a validator conflict and a deleted
+// commit branch match ErrCommitFailed. A commit branch that became a tag, a
+// replaced table, a diverged base snapshot and files to rewrite that a
+// concurrent commit removed do not, so a retry loop keyed on ErrCommitFailed
+// stops.
var ErrCommitFailed = errors.New("commit failed, refresh and try again")
+// ErrTransactionUnusable means this transaction can't be committed again,
+// because after a refresh the commit fails a check that every later attempt
+// would fail too. An error can match both ErrTransactionUnusable and
+// ErrCommitFailed, in which case a new transaction built from a reloaded
+// table may still succeed. The checks are a requirement the retry loop does
+// not rebase (a pinned ref, the schema, partition spec or sort order, or a
+// non-replayable commit whose branch has moved), a validator conflict with a
+// concurrent commit, a diverged base snapshot, files to rewrite that a
+// concurrent commit removed, a commit branch that became a tag or was
+// deleted, and a replaced table. Only the commit branch's head assertion is
+// rebased.
+var ErrTransactionUnusable = errors.New("transaction cannot be committed
again")
+
// ErrWriteIORequired is returned by write paths when the table's file system
// does not implement io.WriteFileIO. Commit retries also fail fast on this
// condition because manifest-list rebuilds need write access; skipping that
@@ -568,6 +589,11 @@ type commitOpts struct {
// not be rewritten to the fresh head between retries (see
// Transaction.pinnedRefs).
pinnedRefs map[string]struct{}
+
+ // resumes counts earlier Commit calls of the same transaction that
+ // failed without committing. A resumed commit refreshes and replays
+ // before its first submission, as the next retry attempt would.
+ resumes int
}
type commitOption func(*commitOpts)
@@ -596,6 +622,10 @@ func withCommitPinnedRefs(refs map[string]struct{})
commitOption {
return func(o *commitOpts) { o.pinnedRefs = refs }
}
+func withCommitResume(resumes int) commitOption {
+ return func(o *commitOpts) { o.resumes = resumes }
+}
+
func (t Table) doCommit(ctx context.Context, updates []Update, reqs
[]Requirement, opts ...commitOption) (*Table, error) {
var co commitOpts
for _, apply := range opts {
@@ -638,20 +668,32 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// 0-based counter risks emitting 0 if a second success exit is ever
added.
var attemptsUsed int64 = 1
- // cleanupOrphans controls whether the defer below removes orphaned
manifest-list
- // files on exit. It defaults to true (clean on all safe exits) and is
set to
- // false only for the one unsafe case: a non-ErrCommitFailed error from
- // CommitTable, where the catalog may have silently accepted the commit
and one
- // of the "orphaned" files may actually be the live snapshot.
+ // doCommit deletes only files it wrote that were not committed. The
+ // staged updates' own files belong to the caller, which can commit them
+ // again after a failure, so they are deleted only once the commit
lands.
+ // cleanupOrphans is false when CommitTable fails in an unknown state,
+ // since the catalog may have accepted a snapshot that references them.
cleanupOrphans := true
+ committed := false
+ staged := stagedManifestLists(updates)
+ rebuiltLists := make(map[string]struct{}) // manifest lists this call's
rebuilds wrote
defer func() {
- if !cleanupOrphans || len(orphanedManifests) == 0 {
+ if !cleanupOrphans {
return
}
+ remove := slices.Collect(maps.Keys(rebuiltLists))
+ if committed {
+ remove = orphanedManifests
+ for _, u := range updates {
+ if su, ok := u.(*addSnapshotUpdate); ok &&
su.supersededSource != nil {
+ remove = append(remove,
su.supersededSource.supersededManifests()...)
+ }
+ }
+ }
// The commit may have failed because ctx was canceled; remove
the
// orphans without that cancellation so they are not left
behind.
cleanupCtx := context.WithoutCancel(ctx)
- for _, path := range orphanedManifests {
+ for _, path := range remove {
if removeErr := wfs.Remove(cleanupCtx, path); removeErr
!= nil {
log.Printf("Warning: failed to delete orphaned
manifest list %s: %v", path, removeErr)
}
@@ -667,12 +709,32 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// numRetries counts retries; total attempts = 1 initial + numRetries.
totalAttempts := cfg.numRetries + 1
+ // A rebuilt manifest list is named after its attempt, so number this
+ // call's attempts after every earlier call's. Every call reads the
retry
+ // config from the same base metadata, so totalAttempts is the same for
+ // each of them.
+ attemptBase := co.resumes * int(totalAttempts)
+
+ // A non-replayable commit never rebases its branch, so a resume
+ // fails unless the branch head is still the one it was built on.
+ pinned := co.pinnedRefs
+ if co.noReplay && co.resumes > 0 {
+ pinned = maps.Clone(pinned)
+ if pinned == nil {
+ pinned = make(map[string]struct{}, 1)
+ }
+ pinned[co.branch] = struct{}{}
+ }
+
// commitStart brackets the commit loop itself (not FS resolution above
or
// metadata cleanup below) so TotalDuration times only the CommitTable
// submission loop, matching Java's CommitReport.
commitStart := time.Now()
for attempt := range totalAttempts {
+ // After a refresh, a rejection that does not depend on timing
fails
+ // the same way on every later attempt and every re-commit.
+ refreshed := attempt != 0 || co.resumes > 0
if attempt != 0 {
wait := backoffDuration(attempt-1, cfg.minWaitMs,
cfg.maxWaitMs)
if timer == nil {
@@ -687,7 +749,8 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
return nil, context.Cause(retryCtx)
case <-timer.C:
}
-
+ }
+ if refreshed {
// Refresh-and-replay: reload the catalog's current
state,
// run the producers' validators against the fresh
// (base=t.metadata, current=fresh) conflict context,
and
@@ -701,39 +764,62 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
}
if t.metadata != nil && fresh.metadata != nil {
if err := validateTableUUID(t.identifier,
t.metadata.TableUUID(), fresh.metadata.TableUUID()); err != nil {
- return nil, err
+ return nil, fmt.Errorf("%w: table was
replaced: %w", ErrTransactionUnusable, err)
}
}
current = fresh.metadata
- reqs = rewriteRefSnapshotRequirements(reqs, co.branch,
current, co.pinnedRefs)
+ reqs = rewriteRefSnapshotRequirements(reqs, co.branch,
current, pinned)
+
+ if err := validateNonRebasedRequirements(reqs,
co.branch, pinned, current); err != nil {
+ if co.noReplay {
+ return nil, fmt.Errorf("%w: requirement
no longer holds after refresh (%s): %w",
+ ErrTransactionUnusable,
noReplayHint, err)
+ }
- // A pinned assertion the fresh catalog state violates
can
- // 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: pinned ref
requirement failed: %w", ErrCommitFailed, err)
+ return nil, fmt.Errorf("%w: requirement no
longer holds after refresh: %w", ErrTransactionUnusable, err)
}
- if err := validateBranchRequirement(reqs, co.branch,
current); err != nil {
- return nil, err
+ }
+
+ if err := validateBranchRequirement(reqs, co.branch, current);
err != nil {
+ if refreshed {
+ return nil, fmt.Errorf("%w: commit branch is no
longer a branch: %w", ErrTransactionUnusable, err)
}
+ return nil, err
+ }
+
+ if refreshed {
// Rebuild snapshot manifest lists to inherit all files
committed
// by concurrent writers since the snapshot was
originally built.
// Without this, the new snapshot's manifest list would
only
// contain its own files and callers scanning the
current snapshot
// would miss every concurrent writer's data.
- rebuiltUpdates, orphaned, rebuildErr :=
rebuildSnapshotUpdates(retryCtx, updates, current, co.branch, wfs, int(attempt))
+ // A resume always rebuilds on its first attempt, so
the submitted
+ // snapshot and the producers' state come from the same
pass. A
+ // producer that kept state from an earlier call could
otherwise
+ // submit an older snapshot whose manifests it now
counts as
+ // superseded and deletes on success.
+ forceRebuild := co.resumes > 0 && attempt == 0 &&
!co.noReplay
+ rebuiltUpdates, orphaned, rebuildErr :=
rebuildSnapshotUpdates(retryCtx, updates, current, co.branch, wfs,
attemptBase+int(attempt), forceRebuild)
+ // A failed rebuild still owns the lists it wrote
before failing.
+ for list := range stagedManifestLists(rebuiltUpdates) {
+ if _, ok := staged[list]; !ok {
+ rebuiltLists[list] = struct{}{}
+ }
+ }
if rebuildErr != nil {
+ // Replaced files missing from the branch head
stay missing, so
+ // every later attempt fails the same way.
+ if errors.Is(rebuildErr, ErrCommitDiverged) {
+ return nil, fmt.Errorf("%w: files to
rewrite were removed: %w", ErrTransactionUnusable, rebuildErr)
+ }
+
return nil, fmt.Errorf("rebuild manifest list
for retry attempt %d: %w", attempt, rebuildErr)
}
orphanedManifests = append(orphanedManifests,
orphaned...)
updates = rebuiltUpdates
}
- if err := validateBranchRequirement(reqs, co.branch, current);
err != nil {
- return nil, err
- }
-
// Pre-flight client-side conflict validation. Producers can
// reject commits whose semantics are violated by concurrent
// peers (partition-filter overlap, referenced-file removal)
@@ -757,10 +843,21 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
if ccErr != nil {
// ErrCommitDiverged — terminal, do not retry.
The
// sentinel deliberately does not wrap
ErrCommitFailed.
+ if errors.Is(ccErr, ErrCommitDiverged) {
+ return nil, fmt.Errorf("%w: base
diverged: %w", ErrTransactionUnusable, ccErr)
+ }
+
return nil, ccErr
}
for _, v := range co.validators {
if vErr := v(cc); vErr != nil {
+ // Conflicts with a concurrent commit
end the transaction.
+ // Other validator errors, such as read
failures, stay
+ // resumable.
+ if errors.Is(vErr, ErrCommitFailed) {
+ return nil, fmt.Errorf("%w:
validation conflict: %w", ErrTransactionUnusable, vErr)
+ }
+
return nil, vErr
}
}
@@ -772,6 +869,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
newMeta, newLoc, err = t.cat.CommitTable(retryCtx,
slices.Clone(t.identifier), reqs, updates)
if err == nil {
+ committed = true
attemptsUsed = int64(attempt) + 1
// Capture elapsed time at the commit boundary, before
orphan
// cleanup and deleteOldMetadata below (both can do
I/O).
@@ -788,7 +886,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
if !errors.Is(err, ErrCommitFailed) {
cleanupOrphans = false
- return nil, err
+ return nil, &commitStateUnknownError{err: err}
}
// Non-replayable commits fail on the first CAS conflict instead
@@ -796,25 +894,10 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// distinguishes this abort from an exhausted retry budget while
// preserving errors.Is(err, ErrCommitFailed).
if co.noReplay {
- return nil, fmt.Errorf("%w (commit carries
snapshot-relative delete-file removals and cannot be replayed; reload the table
and rebuild the removals)", err)
+ return nil, fmt.Errorf("%w (%s)", err, noReplayHint)
}
}
- // Inner data manifests written by superseded retry attempts (a rewrite
- // re-merges everything on each retry) are orphaned objects. On a safe
- // exhausted-ErrCommitFailed failure (err != nil here) nothing
committed, so
- // the accumulator also folds in the final attempt's manifests. On
success
- // the committed snapshot references those, so they are excluded. The
defer
- // skips cleanup only on the unsafe non-ErrCommitFailed path, which
returned
- // above with cleanupOrphans = false.
- for _, u := range updates {
- su, ok := u.(*addSnapshotUpdate)
- if !ok || su.supersededSource == nil {
- continue
- }
- orphanedManifests = append(orphanedManifests,
su.supersededSource.supersededManifests(err == nil)...)
- }
-
if err != nil {
return nil, err
}
@@ -874,8 +957,8 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
//
// Java's SnapshotProducer rewrites the same way between retries. If
// the branch is empty or the new head cannot be resolved (branch
-// deleted underneath us), reqs is returned unchanged — newConflict-
-// Context will surface the divergence on the next pre-flight pass.
+// deleted underneath us), reqs is returned unchanged, and the commit
+// branch check that follows rejects the commit.
//
// Assertions on refs in pinned guard the commit (an explicit
// Transaction.AssertRefSnapshotID, a rollback, or snapshot expiry) and
@@ -910,24 +993,17 @@ func rewriteRefSnapshotRequirements(reqs []Requirement,
branch string, fresh Met
return out
}
-// validatePinnedRefRequirements validates every assert-ref-snapshot-id
-// requirement on a pinned branch against the freshly refreshed catalog
-// metadata. A failure means the branch has changed from the required
-// snapshot: the assertion is never rewritten, so it can never hold and
-// the commit must fail instead of retrying.
-func validatePinnedRefRequirements(reqs []Requirement, pinned
map[string]struct{}, fresh Metadata) error {
- if len(pinned) == 0 {
- return nil
- }
+// validateNonRebasedRequirements checks the requirements a retry does not
+// rebase. Only unpinned commit-branch assertions are rebased, so any other
+// requirement that fails against refreshed metadata cannot succeed on retry.
+func validateNonRebasedRequirements(reqs []Requirement, branch string, pinned
map[string]struct{}, fresh Metadata) error {
for _, r := range reqs {
- a, ok := r.(*assertRefSnapshotID)
- if !ok {
- continue
- }
- if _, isPinned := pinned[a.Ref]; !isPinned {
- continue
+ if a, ok := r.(*assertRefSnapshotID); ok && a.Ref == branch {
+ if _, isPinned := pinned[a.Ref]; !isPinned {
+ continue
+ }
}
- if err := a.Validate(fresh); err != nil {
+ if err := r.Validate(fresh); err != nil {
return err
}
}
@@ -935,6 +1011,29 @@ func validatePinnedRefRequirements(reqs []Requirement,
pinned map[string]struct{
return nil
}
+const noReplayHint = "commit carries snapshot-relative delete-file removals
and cannot be replayed; reload the table and rebuild the removals"
+
+// commitStateUnknownError marks a CommitTable failure after which the catalog
+// may have accepted the commit. Every other doCommit failure committed
+// nothing.
+type commitStateUnknownError struct{ err error }
+
+func (e *commitStateUnknownError) Error() string { return e.err.Error() }
+
+func (e *commitStateUnknownError) Unwrap() error { return e.err }
+
+// stagedManifestLists returns the manifest list of every snapshot updates add.
+func stagedManifestLists(updates []Update) map[string]struct{} {
+ lists := make(map[string]struct{})
+ for _, u := range updates {
+ if su, ok := u.(*addSnapshotUpdate); ok && su.Snapshot != nil {
+ lists[su.Snapshot.ManifestList] = struct{}{}
+ }
+ }
+
+ return lists
+}
+
func validateBranchRequirement(reqs []Requirement, branch string, meta
Metadata) error {
for _, req := range reqs {
if ref, ok := req.(*assertRefSnapshotID); ok &&
ref.requireBranch && ref.Ref == branch {
@@ -993,13 +1092,14 @@ func stagedSnapshotStillValid(snap *Snapshot, fresh
Metadata) bool {
// It also returns the manifest-list file paths that were superseded by
// the rebuild (i.e., the paths from the input updates that were replaced).
// These become orphaned objects in object storage and should be removed
-// by the caller after a successful commit.
+// by the caller after a successful commit. On error the updates rebuilt so
+// far are still returned, so the caller can delete the lists they wrote.
//
// This is the manifest-layer "refresh-and-replay" step: the data files
// (already written to object storage) are reused as-is; only the manifest
// list is rewritten to include the fresh parent's manifests so that the
// rebuilt snapshot contains every committed file.
-func rebuildSnapshotUpdates(ctx context.Context, updates []Update, freshMeta
Metadata, branch string, fs icebergio.WriteFileIO, attempt int) (rebuilt
[]Update, orphanedPaths []string, err error) {
+func rebuildSnapshotUpdates(ctx context.Context, updates []Update, freshMeta
Metadata, branch string, fs icebergio.WriteFileIO, attempt int, force bool)
(rebuilt []Update, orphanedPaths []string, err error) {
// Must mirror createSnapshotProducer's attempt-0 fallback, or a
retried new
// branch rebuilds with a nil parent and silently drops main's data.
var freshHead *Snapshot
@@ -1019,7 +1119,8 @@ func rebuildSnapshotUpdates(ctx context.Context, updates
[]Update, freshMeta Met
// dropping the earlier snapshots' data. chainRebuilt forces every
snapshot
// after the first rebuild to rebuild too: its recorded parent is an
earlier
// staged snapshot that has itself just been reparented onto the fresh
head.
- chainRebuilt := false
+ // force rebuilds every snapshot from the start.
+ chainRebuilt := force
for i, u := range result {
su, ok := u.(*addSnapshotUpdate)
if !ok || su.rebuildManifestList == nil {
@@ -1044,7 +1145,7 @@ func rebuildSnapshotUpdates(ctx context.Context, updates
[]Update, freshMeta Met
newSnap, rebuildErr := su.rebuildManifestList(ctx, freshMeta,
freshHead, fs, attempt)
if rebuildErr != nil {
- return nil, nil, rebuildErr
+ return result, orphanedPaths, rebuildErr
}
result[i] = &addSnapshotUpdate{
diff --git a/table/transaction.go b/table/transaction.go
index 58833f665..2de3114b3 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -103,6 +103,25 @@ type Transaction struct {
mx sync.Mutex
committed bool
+
+ unusable bool
+
+ // failedCommits counts Commit calls that failed without committing. The
+ // next Commit resumes from a refresh, as the next retry attempt would.
+ failedCommits int
+}
+
+// The caller must hold t.mx.
+func (t *Transaction) checkUsable() error {
+ if t.unusable {
+ return fmt.Errorf("%w: a previous commit failed, build a new
transaction",
+ ErrTransactionUnusable)
+ }
+ if t.committed {
+ return errors.New("transaction has already been committed")
+ }
+
+ return nil
}
func (t *Transaction) ensureInitialized() error {
@@ -161,8 +180,8 @@ func (t *Transaction) applyLocked(updates []Update, reqs
[]Requirement) error {
return err
}
- if t.committed {
- return errors.New("transaction has already been committed")
+ if err := t.checkUsable(); err != nil {
+ return err
}
var stagedMeta *MetadataBuilder
@@ -746,6 +765,11 @@ func getRetentionInt64(props iceberg.Properties,
standardKey, legacyKey string,
// }
// newTbl, err := txn.Commit(ctx)
//
+// An expiry that removes something pins every ref it read, including the
+// commit branch. A transaction that also appends then fails rather than
+// rebasing onto a peer's append, since the expiry was computed from the old
+// head.
+//
// The "iceberg expire-snapshots" CLI command wraps the same operation.
func (t *Transaction) ExpireSnapshots(opts ...ExpireSnapshotsOpt) error {
var (
@@ -890,7 +914,7 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
// 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.
+ // rejects a committed or unusable transaction.
if len(updates) == 0 {
return t.apply(nil, nil)
}
@@ -2583,6 +2607,13 @@ func (t *Transaction) StagedTable() (*StagedTable,
error) {
// deferred when applying an operation adds no new requirement to the
transaction.
// The staged metadata is validated when a later transaction operation needs an
// immutable metadata view, or while the catalog applies the commit.
+//
+// After a failed commit, calling Commit again refreshes the table and replays
+// the staged changes like the next retry attempt, with a fresh retry budget.
+// The staged files are kept until a commit lands, so an abandoned transaction
+// leaves them, and for a rewrite the manifests later attempts merged, for
+// orphan file removal. TableCommit refuses a transaction whose commit failed,
+// since it does not refresh.
func (t *Transaction) Commit(ctx context.Context) (*Table, error) {
if err := t.checkNotNil(); err != nil {
return nil, err
@@ -2595,8 +2626,8 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
return nil, err
}
- if t.committed {
- return nil, errors.New("transaction has already been committed")
+ if err := t.checkUsable(); err != nil {
+ return nil, err
}
if len(meta.updates) > 0 {
@@ -2606,14 +2637,21 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
withCommitValidators(t.validators...),
withCommitNoReplay(t.noReplay),
withCommitPinnedRefs(t.pinnedRefs),
+ withCommitResume(t.failedCommits),
)
if err != nil {
- // A clean conflict (ErrCommitFailed) committed nothing
and stays
- // retriable. Any other failure leaves the commit state
unknown
- // (the catalog may have accepted it), so mark it
terminal to
- // avoid a double-apply on retry.
- if !errors.Is(err, ErrCommitFailed) {
+ // A failure that committed nothing stays committable,
and the next
+ // Commit resumes from a refresh. See
ErrTransactionUnusable for
+ // when it is not. A failure of unknown commit state is
terminal to
+ // avoid a double-apply.
+ var unknown *commitStateUnknownError
+ switch {
+ case errors.Is(err, ErrTransactionUnusable):
+ t.unusable = true
+ case errors.As(err, &unknown):
t.committed = true
+ default:
+ t.failedCommits++
}
return tbl, err
diff --git a/table/transaction_assert_ref_test.go
b/table/transaction_assert_ref_test.go
index de6935714..826e20220 100644
--- a/table/transaction_assert_ref_test.go
+++ b/table/transaction_assert_ref_test.go
@@ -191,16 +191,19 @@ func TestNoOpExpireSnapshotsDoesNotBlockRebase(t
*testing.T) {
assert.Equal(t, "42", cat.metadata.Properties()["offsets"])
}
-func TestNoOpExpireSnapshotsRejectsCommittedTransaction(t *testing.T) {
+func TestNoOpExpireSnapshotsRejectsFinishedTransaction(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())
+ committed := tbl.NewTransaction()
+ _, err := committed.Commit(t.Context())
require.NoError(t, err)
- assert.ErrorContains(t, tx.ExpireSnapshots(), "already been committed")
+ assert.ErrorContains(t, committed.ExpireSnapshots(), "already been
committed")
+
+ unusable := tbl.NewTransaction()
+ unusable.unusable = true
+ assert.ErrorIs(t, unusable.ExpireSnapshots(), ErrTransactionUnusable)
}
// Why: ref assertions are deduplicated by (type, ref), so assertions
diff --git a/website/src/api.md b/website/src/api.md
index 03edf9bef..162d7a726 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -393,7 +393,7 @@ txn, err := tbl.NewTransactionOnBranch("staging")
if err != nil { /* ... */ }
```
-`Commit` retries automatically on conflict (`ErrCommitFailed`) - tune via the
`commit.retry.*` table properties.
+`Commit` can retry on conflict (`ErrCommitFailed`) when
`commit.retry.num-retries` is set. See [Automatic
retry](./concurrent-writes.md#automatic-retry) for the `commit.retry.*` table
properties.
## Schema and partition evolution
diff --git a/website/src/concurrent-writes.md b/website/src/concurrent-writes.md
index 5dcf03847..6e830ba62 100644
--- a/website/src/concurrent-writes.md
+++ b/website/src/concurrent-writes.md
@@ -35,6 +35,9 @@ state. A failed validation surfaces as one of:
retried after refreshing (see below).
- `table.ErrCommitDiverged` — terminal. The base snapshot is no longer on the
branch at all, so a retry cannot reconcile the change.
+- `table.ErrTransactionUnusable` — this transaction can't be committed again.
+ If the error also matches `ErrCommitFailed`, retry in a new transaction from
a
+ reloaded table.
```go
import (
@@ -46,7 +49,9 @@ import (
_, err := txn.Commit(context.Background())
switch {
-case errors.Is(err, table.ErrCommitDiverged):
+case errors.Is(err, table.ErrTransactionUnusable) && errors.Is(err,
table.ErrCommitFailed):
+ // a concurrent change: retry in a new transaction from a reloaded table
+case errors.Is(err, table.ErrCommitDiverged), errors.Is(err,
table.ErrTransactionUnusable):
// unrecoverable: rebuild the operation from the latest table
case errors.Is(err, table.ErrCommitFailed):
// retriable: refresh and try again (the retry loop does this for you)
@@ -94,3 +99,27 @@ _, 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.
+
+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
+refs, must still hold. Otherwise, the commit fails without another attempt and
+the transaction is unusable.
+
+After a conflict, calling `Commit` again on the same transaction refreshes the
+table and replays the staged changes like the next retry attempt, so it can
+rebase onto a peer's commit even with retries disabled. Each call gets the
+configured retry budget. A commit carrying delete-file removals that cannot be
+replayed resubmits only if its branch has not moved. The rejections after a
+refresh that make the transaction unusable are listed on
+`ErrTransactionUnusable`.
+
+A failed commit keeps the files it staged until a later commit lands. A
+transaction you abandon leaves them for orphan file removal, as one with
+retries disabled always has. An abandoned manifest rewrite also leaves the
+manifests that later attempts merged. `TableCommit` refuses a transaction whose
+commit failed, since the multi-table path does not refresh. Call `Commit` to
+resume it, or build a new transaction.
+
+Because an `ExpireSnapshots` that removes something pins the commit branch, a
+transaction that also appends fails instead of rebasing onto a peer's append.
+This is deliberate, since the expiry was computed from the old head.
