This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 00acae94e fix(table): mark transaction committed only after successful
commit (#1603)
00acae94e is described below
commit 00acae94e06b8c387b1326c040d16ef51b2a99d8
Author: Badal Prasad Singh <[email protected]>
AuthorDate: Fri Jul 31 02:10:32 2026 +0530
fix(table): mark transaction committed only after successful commit (#1603)
## Description
Fixes #1602, issue of setting `t.committed = true` before invoking
`doCommit`.
Moved `t.committed = true` so it is set only after `doCommit` returns
successfully. The transaction is left usable for retry. Add a test for
it.
---------
Signed-off-by: badalprasadsingh <[email protected]>
---
table/commit_retry_test.go | 96 +++++++++++++++++++++++++++++++++++++++++++++-
table/transaction.go | 21 ++++++++--
2 files changed, 112 insertions(+), 5 deletions(-)
diff --git a/table/commit_retry_test.go b/table/commit_retry_test.go
index 87b740a07..e698ed08b 100644
--- a/table/commit_retry_test.go
+++ b/table/commit_retry_test.go
@@ -50,6 +50,7 @@ type sequentialCatalog struct {
loadMeta Metadata // optional: returned by LoadTable if non-nil
errs []error
attempts atomic.Int32
+ lastReqs []Requirement
}
func (c *sequentialCatalog) LoadTable(_ context.Context, ident Identifier)
(*Table, error) {
@@ -62,7 +63,8 @@ func (c *sequentialCatalog) LoadTable(_ context.Context,
ident Identifier) (*Tab
func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, c), nil
}
-func (c *sequentialCatalog) CommitTable(_ context.Context, _ Identifier, _
[]Requirement, updates []Update) (Metadata, string, error) {
+func (c *sequentialCatalog) CommitTable(_ context.Context, _ Identifier, reqs
[]Requirement, updates []Update) (Metadata, string, error) {
+ c.lastReqs = reqs
n := int(c.attempts.Add(1)) - 1 // 0-indexed
if n < len(c.errs) && c.errs[n] != nil {
return nil, "", c.errs[n]
@@ -944,3 +946,95 @@ func TestDoCommit_RetryProgressesFreshMeta(t *testing.T) {
require.NotContains(t, wfs.files, cat.observedManifestLists[1],
"attempt-1 rebuild manifest list must be cleaned as orphan
after success")
}
+
+// To verify a retried commit does not accumulate duplicates.
+func countAssertTableUUID(reqs []Requirement) int {
+ n := 0
+ for _, r := range reqs {
+ if r.GetType() == "assert-table-uuid" {
+ n++
+ }
+ }
+
+ return n
+}
+
+func TestTransactionCommit_RetriableAfterCleanConflict(t *testing.T) {
+ // Default retry config (numRetries == 0): each Commit is a single
+ // CommitTable attempt. The first fails with a clean conflict, the
second succeeds.
+ 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.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed, "first commit must surface the
clean conflict")
+ 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.
+ 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.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")
+
+ assert.Equal(t, 1, countAssertTableUUID(cat.lastReqs),
+ "retry must not append a duplicate AssertTableUUID")
+
+ _, err = tx.Commit(t.Context())
+ assert.ErrorContains(t, err, "already been committed")
+}
+
+// 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
+ // exhausts its internal retries and returns ErrCommitFailed.
+ cat := &sequentialCatalog{
+ errs: []error{ErrCommitFailed, ErrCommitFailed,
ErrCommitFailed},
+ }
+ tbl := newRetryTestTable(t, cat, iceberg.Properties{
+ CommitNumRetriesKey: "2",
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ })
+ cat.metadata = tbl.Metadata()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(map[string]string{"key": "value"}))
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ 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")
+}
+
+func TestTransactionCommit_TerminalOnUnknownState(t *testing.T) {
+ cat := &sequentialCatalog{
+ errs: []error{errors.New("simulated 5xx: internal server
error")},
+ }
+ tbl := newRetryTestTable(t, cat, nil)
+ cat.metadata = tbl.Metadata()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(map[string]string{"key": "value"}))
+
+ _, err := tx.Commit(t.Context())
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, ErrCommitFailed, "test must exercise the
unknown-state path")
+ assert.True(t, tx.committed, "unknown-state failure must mark the
transaction terminal")
+
+ // A retry must be rejected — retrying could double-apply if the first
+ // attempt actually landed at the catalog.
+ _, err = tx.Commit(t.Context())
+ assert.ErrorContains(t, err, "already been committed")
+ assert.Equal(t, int32(1), cat.attempts.Load(), "terminal failure must
not reach the catalog again")
+}
diff --git a/table/transaction.go b/table/transaction.go
index 6e91a35ca..b5648c909 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -2372,18 +2372,29 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
return nil, errors.New("transaction has already been committed")
}
- t.committed = true
-
if len(meta.updates) > 0 {
- t.reqs = append(t.reqs, AssertTableUUID(meta.uuid))
- tbl, err := t.tbl.doCommit(ctx, meta.updates, t.reqs,
+ reqs := append(slices.Clone(t.reqs), AssertTableUUID(meta.uuid))
+ tbl, err := t.tbl.doCommit(ctx, meta.updates, reqs,
withCommitBranch(t.branch),
withCommitValidators(t.validators...),
)
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) {
+ t.committed = true
+ }
+
return tbl, err
}
+ // Mark committed after the catalog accepts but before
PostCommit runs.
+ // A PostCommit failure must not leave the transaction
retriable;
+ // otherwise a retry would re-run the catalog commit.
+ t.committed = true
+
for _, u := range meta.updates {
if perr := u.PostCommit(ctx, t.tbl, tbl); perr != nil {
err = errors.Join(err, perr)
@@ -2393,6 +2404,8 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
return tbl, err
}
+ t.committed = true
+
return t.tbl, nil
}
