You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/08/27 11:13:00 UTC
(iceberg-go) branch main updated: fix(table): fence multi-table commits with the target branch requirement (#1844) (#1858)
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 181d0f617 fix(table): fence multi-table commits with the target branch requirement (#1844) (#1858)
181d0f617 is described below
commit 181d0f617af8ad7e80ec28ff7c20b4186dfef3b0
Author: Badal Prasad Singh <ba...@datazip.io>
AuthorDate: Thu Aug 27 16:42:55 2026 +0530
fix(table): fence multi-table commits with the target branch requirement (#1844) (#1858)
* fix: branch req
Signed-off-by: badalprasadsingh <ba...@datazip.io>
* fix: multi-table commit fencing
Signed-off-by: badalprasadsingh <ba...@datazip.io>
* fix: lint
Signed-off-by: badalprasadsingh <ba...@datazip.io>
---------
Signed-off-by: badalprasadsingh <ba...@datazip.io>
---
catalog/multi_table_transaction.go | 29 +++-
catalog/multi_table_transaction_test.go | 152 ++++++++++++++++++
table/commit.go | 12 +-
table/transaction.go | 3 +
table/transaction_internal_test.go | 262 ++++++++++++++++++++++++++++++++
5 files changed, 447 insertions(+), 11 deletions(-)
diff --git a/catalog/multi_table_transaction.go b/catalog/multi_table_transaction.go
index ca0defaf0..4c31934e1 100644
--- a/catalog/multi_table_transaction.go
+++ b/catalog/multi_table_transaction.go
@@ -72,6 +72,9 @@ func NewMultiTableTransaction(cat Catalog) (*MultiTableTransaction, error) {
// with all other transactions in this multi-table transaction.
// Returns an error if the transaction is nil, already committed, or
// targets a table that was already added.
+//
+// A transaction may still be empty here and gain updates before Commit,
+// so emptiness is only decided when the payloads are built.
func (m *MultiTableTransaction) AddTransaction(tx *table.Transaction) error {
if tx == nil {
return errors.New("transaction must not be nil")
@@ -102,8 +105,14 @@ func (m *MultiTableTransaction) AddTransaction(tx *table.Transaction) error {
// Commit extracts pending changes from all added transactions and
// commits them atomically. On success, all transactions are marked
-// as committed. On failure, no transactions are marked committed
-// and the caller may retry.
+// as committed. On failure, no transactions are marked committed.
+//
+// A retry must be rebuilt from freshly loaded tables:
+// each payload asserts the branch head its transaction read.
+//
+// Transactions that stage no updates are left out of the request, since
+// a catalog may reject a table change carrying none. They are still
+// marked committed. A request with nothing left to send is not made.
//
// PostCommit hooks are not executed. Because the multi-table commit
// endpoint returns 204 No Content, callers must LoadTable individually
@@ -118,18 +127,26 @@ func (m *MultiTableTransaction) Commit(ctx context.Context) error {
}
commits := make([]table.TableCommit, 0, len(m.txns))
- for _, tx := range m.txns {
+ names := make([]string, 0, len(m.txns))
+ for i, tx := range m.txns {
tc, err := tx.TableCommit()
if err != nil {
return err
}
+ if len(tc.Updates) == 0 {
+ continue
+ }
+
commits = append(commits, tc)
+ names = append(names, m.tableNames[i])
}
- if err := m.cat.CommitTransaction(ctx, commits); err != nil {
- return fmt.Errorf("commit transaction for tables [%s]: %w",
- strings.Join(m.tableNames, ", "), err)
+ if len(commits) > 0 {
+ if err := m.cat.CommitTransaction(ctx, commits); err != nil {
+ return fmt.Errorf("commit transaction for tables [%s]: %w",
+ strings.Join(names, ", "), err)
+ }
}
m.committed = true
diff --git a/catalog/multi_table_transaction_test.go b/catalog/multi_table_transaction_test.go
index 3008fd41b..87ee69e5a 100644
--- a/catalog/multi_table_transaction_test.go
+++ b/catalog/multi_table_transaction_test.go
@@ -21,6 +21,7 @@ import (
"context"
"errors"
"iter"
+ "strings"
"testing"
"github.com/apache/iceberg-go"
@@ -277,3 +278,154 @@ func TestCommitAndReloadPartialFailure(t *testing.T) {
// First table was loaded successfully before the second failed.
assert.Len(t, tables, 1)
}
+
+func mtxTableWithHead(t *testing.T, name string, headID, childID int64) (*table.Table, table.Metadata) {
+ t.Helper()
+
+ base := mtxTestTable(t, "db", name).Metadata()
+ withHead := mtxGraftSnapshot(t, base, headID, nil)
+ advanced := mtxGraftSnapshot(t, withHead, childID, &headID)
+
+ return table.New(table.Identifier{"db", name}, withHead, "", nil, nil), advanced
+}
+
+func mtxGraftSnapshot(t *testing.T, base table.Metadata, id int64, parent *int64) table.Metadata {
+ t.Helper()
+
+ builder, err := table.MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.AddSnapshot(&table.Snapshot{
+ SnapshotID: id,
+ ParentSnapshotID: parent,
+ SequenceNumber: base.LastSequenceNumber() + 1,
+ TimestampMs: base.LastUpdatedMillis() + 1,
+ Summary: &table.Summary{Operation: table.OpAppend},
+ }))
+ require.NoError(t, builder.SetSnapshotRef(table.MainBranch, id, table.BranchRef))
+ out, err := builder.Build()
+ require.NoError(t, err)
+
+ return out
+}
+
+// Why: with a distinct head per table, a shared or missing assertion would still let one table's concurrent writer through.
+func TestMultiTableTransactionFencesEachBranchHead(t *testing.T) {
+ tbl1, advanced1 := mtxTableWithHead(t, "t1", 100, 101)
+ tbl2, advanced2 := mtxTableWithHead(t, "t2", 200, 201)
+
+ stub := &stubCatalog{}
+ mtx := &MultiTableTransaction{cat: stub}
+
+ for _, tbl := range []*table.Table{tbl1, tbl2} {
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(map[string]string{"offsets": "42"}))
+ require.NoError(t, mtx.AddTransaction(tx))
+ }
+
+ require.NoError(t, mtx.Commit(t.Context()))
+ require.Len(t, stub.commits, 1)
+ require.Len(t, stub.commits[0], 2)
+
+ cases := []struct {
+ commit table.TableCommit
+ base table.Metadata
+ advanced table.Metadata
+ }{
+ {stub.commits[0][0], tbl1.Metadata(), advanced1},
+ {stub.commits[0][1], tbl2.Metadata(), advanced2},
+ }
+
+ for _, tc := range cases {
+ name := tc.commit.Identifier[len(tc.commit.Identifier)-1]
+ t.Run(name, func(t *testing.T) {
+ var refReqs int
+ for _, req := range tc.commit.Requirements {
+ if req.GetType() != "assert-ref-snapshot-id" {
+ continue
+ }
+ refReqs++
+ assert.NoError(t, req.Validate(tc.base),
+ "the assertion must hold against the state the writer read")
+ assert.Error(t, req.Validate(tc.advanced),
+ "a concurrent commit on main must fail the assertion")
+ }
+ assert.Equal(t, 1, refReqs, "expected exactly one snapshot-id assertion for main")
+ })
+ }
+}
+
+type validatingStubCatalog struct {
+ metadata map[string]table.Metadata
+ calls int
+}
+
+func (s *validatingStubCatalog) CommitTransaction(_ context.Context, commits []table.TableCommit) error {
+ s.calls++
+ for _, c := range commits {
+ meta := s.metadata[strings.Join(c.Identifier, ".")]
+ for _, req := range c.Requirements {
+ if err := req.Validate(meta); err != nil {
+ return err
+ }
+ }
+ }
+
+ return nil
+}
+
+// Why: the branch fence only pays off if the catalog rejects a payload built against a head
+// another writer has since moved.
+func TestMultiTableTransactionFailsAfterConcurrentAdvance(t *testing.T) {
+ tbl, advanced := mtxTableWithHead(t, "t1", 100, 101)
+
+ cat := &validatingStubCatalog{metadata: map[string]table.Metadata{"db.t1": advanced}}
+ mtx := &MultiTableTransaction{cat: cat}
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(map[string]string{"offsets": "42"}))
+ require.NoError(t, mtx.AddTransaction(tx))
+
+ err := mtx.Commit(t.Context())
+ require.Error(t, err, "a moved branch head must fail the atomic batch")
+ assert.ErrorContains(t, err, "has changed")
+ assert.Equal(t, 1, cat.calls)
+}
+
+// Why: a catalog may reject a table change with no updates, which would fail the whole batch
+// over an entry that changes nothing.
+func TestMultiTableTransactionSkipsTransactionsWithoutUpdates(t *testing.T) {
+ t.Run("empty transactions are left out of the request", func(t *testing.T) {
+ stub := &stubCatalog{}
+ mtx := &MultiTableTransaction{cat: stub}
+
+ empty := mtxTestTable(t, "db", "t1").NewTransaction()
+ require.NoError(t, mtx.AddTransaction(empty))
+
+ staged := mtxTestTable(t, "db", "t2").NewTransaction()
+ require.NoError(t, staged.SetProperties(map[string]string{"k": "v"}))
+ require.NoError(t, mtx.AddTransaction(staged))
+
+ require.NoError(t, mtx.Commit(t.Context()))
+ require.Len(t, stub.commits, 1)
+ require.Len(t, stub.commits[0], 1, "only the transaction with updates is submitted")
+ assert.Equal(t, table.Identifier{"db", "t2"}, stub.commits[0][0].Identifier)
+
+ _, err := empty.TableCommit()
+ assert.ErrorContains(t, err, "already been committed",
+ "a skipped transaction is still marked committed")
+ })
+
+ t.Run("a batch with nothing to send makes no request", func(t *testing.T) {
+ stub := &stubCatalog{}
+ mtx := &MultiTableTransaction{cat: stub}
+
+ tx := mtxTestTable(t, "db", "t1").NewTransaction()
+ require.NoError(t, mtx.AddTransaction(tx))
+
+ require.NoError(t, mtx.Commit(t.Context()))
+ assert.Empty(t, stub.commits, "no table change means no request")
+
+ _, err := tx.TableCommit()
+ assert.ErrorContains(t, err, "already been committed")
+ })
+}
diff --git a/table/commit.go b/table/commit.go
index c0684a5c2..22aa98606 100644
--- a/table/commit.go
+++ b/table/commit.go
@@ -41,8 +41,9 @@ type TableCommit struct {
// this method directly — it handles extraction, commit, and lifecycle
// management automatically.
//
-// The method automatically includes an AssertTableUUID requirement, matching
-// the behavior of [Transaction.Commit].
+// The returned requirements are the ones [Transaction.Commit] would send,
+// so the multi-table path is fenced by the target branch's base head.
+// A transaction with no updates yields an empty payload and enforces none.
//
// TableCommit does not mark the transaction as committed — the caller is
// responsible for either calling Commit (single-table) or submitting the
@@ -77,9 +78,10 @@ func (t *Transaction) TableCommit() (TableCommit, error) {
}, nil
}
- reqs := make([]Requirement, len(t.reqs), len(t.reqs)+1)
- copy(reqs, t.reqs)
- reqs = append(reqs, AssertTableUUID(meta.uuid))
+ // Same derivation as Commit. Copying t.reqs alone leaves metadata-only transactions unfenced:
+ // their updates stage no ref requirement. The helper returns a fresh
+ // slice, so appending here cannot touch t.reqs.
+ reqs := append(transactionRequirements(t.reqs, t.branch, t.tbl.metadata), AssertTableUUID(meta.uuid))
updates := make([]Update, len(meta.updates))
copy(updates, meta.updates)
diff --git a/table/transaction.go b/table/transaction.go
index baf32c921..769cf448f 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -323,6 +323,9 @@ func (t *Transaction) planningSnapshot(meta *MetadataBuilder) *Snapshot {
return meta.currentSnapshotForRef(t.branch)
}
+// transactionRequirements returns the requirements a commit on branch must carry.
+// The result is always a fresh allocation, never aliasing reqs, so callers may append to it
+// while holding the transaction lock.
func transactionRequirements(reqs []Requirement, branch string, base Metadata) []Requirement {
if branch == "" {
branch = MainBranch
diff --git a/table/transaction_internal_test.go b/table/transaction_internal_test.go
index 848ee90a5..0ff15efee 100644
--- a/table/transaction_internal_test.go
+++ b/table/transaction_internal_test.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/google/uuid"
+ "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -1836,3 +1837,264 @@ func TestRollbackToSnapshotCommitRejectsBranchTurnedTag(t *testing.T) {
require.ErrorContains(t, err, "tags cannot be transaction targets")
require.Equal(t, int32(1), cat.attempts.Load(), "a type conflict must not be retried")
}
+
+type reqCapturingCatalog struct {
+ metadata Metadata
+ reqs [][]Requirement
+}
+
+func (c *reqCapturingCatalog) LoadTable(_ context.Context, ident Identifier) (*Table, error) {
+ return New(ident, c.metadata, "",
+ func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }, c), nil
+}
+
+func (c *reqCapturingCatalog) CommitTable(_ context.Context, _ Identifier, reqs []Requirement, updates []Update) (Metadata, string, error) {
+ c.reqs = append(c.reqs, reqs)
+ meta, err := UpdateTableMetadata(c.metadata, updates, "")
+ if err != nil {
+ return nil, "", err
+ }
+ c.metadata = meta
+
+ return meta, "", nil
+}
+
+func refRequirements(t *testing.T, reqs []Requirement, ref string) []*assertRefSnapshotID {
+ t.Helper()
+
+ var out []*assertRefSnapshotID
+ for _, r := range reqs {
+ if r.GetType() != reqAssertRefSnapshotID {
+ continue
+ }
+ req, ok := r.(*assertRefSnapshotID)
+ require.True(t, ok, "requirement of type %s must be an *assertRefSnapshotID", r.GetType())
+ if req.Ref == ref {
+ out = append(out, req)
+ }
+ }
+
+ return out
+}
+
+func soleRefRequirement(t *testing.T, reqs []Requirement, ref string) *assertRefSnapshotID {
+ t.Helper()
+
+ found := refRequirements(t, reqs, ref)
+ require.Len(t, found, 1, "expected exactly one snapshot-id assertion for ref %q", ref)
+
+ return found[0]
+}
+
+func metadataWithRef(t *testing.T, base Metadata, name string, refType RefType) Metadata {
+ t.Helper()
+
+ head := base.SnapshotByName(MainBranch)
+ require.NotNil(t, head, "base must have a main branch head")
+
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.SetSnapshotRef(name, head.SnapshotID, refType))
+ out, err := builder.Build()
+ require.NoError(t, err)
+
+ return out
+}
+
+func multiTableTestTable(t *testing.T, meta Metadata) *Table {
+ t.Helper()
+
+ return New(Identifier{"db", "multi-table"}, meta, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil },
+ &headTrackingCatalog{metadata: meta})
+}
+
+// Why: TableCommit copied the staged requirements verbatim, and metadata-only transactions
+// stage no ref requirement of their own.
+func TestTableCommitFencesTargetBranch(t *testing.T) {
+ head := int64(100)
+
+ t.Run("metadata-only transaction pins the base branch head", func(t *testing.T) {
+ base := newConflictTestMetadata(t, &head)
+ tx := multiTableTestTable(t, base).NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets": "42"}))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ req := soleRefRequirement(t, tc.Requirements, MainBranch)
+ require.NotNil(t, req.SnapshotID)
+ assert.Equal(t, head, *req.SnapshotID)
+ assert.True(t, req.requireBranch, "the target ref must be asserted as a branch")
+
+ assert.NoError(t, req.Validate(base))
+ err = req.Validate(graftSnapshotOnto(t, base, MainBranch, 200))
+ require.Error(t, err, "a concurrent snapshot on main must fail the assertion")
+ assert.ErrorContains(t, err, "has changed")
+ })
+
+ t.Run("explicit ref assertion is kept without duplication", func(t *testing.T) {
+ base := newConflictTestMetadata(t, &head)
+ tx := multiTableTestTable(t, base).NewTransaction()
+ require.NoError(t, tx.AssertRefSnapshotID(MainBranch))
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets": "42"}))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ req := soleRefRequirement(t, tc.Requirements, MainBranch)
+ require.NotNil(t, req.SnapshotID)
+ assert.Equal(t, head, *req.SnapshotID)
+ })
+
+ t.Run("branch transaction pins only that branch", func(t *testing.T) {
+ base := metadataWithRef(t, newConflictTestMetadata(t, &head), "audit", BranchRef)
+ tx := multiTableTestTable(t, base).NewTransactionOnBranch("audit")
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets": "42"}))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ req := soleRefRequirement(t, tc.Requirements, "audit")
+ require.NotNil(t, req.SnapshotID)
+ assert.Equal(t, head, *req.SnapshotID)
+ assert.Empty(t, refRequirements(t, tc.Requirements, MainBranch),
+ "a branch transaction must not fence main")
+ })
+
+ t.Run("absent target branch requires absence", func(t *testing.T) {
+ base := newConflictTestMetadata(t, &head)
+ tx := multiTableTestTable(t, base).NewTransactionOnBranch("audit")
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets": "42"}))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ req := soleRefRequirement(t, tc.Requirements, "audit")
+ assert.Nil(t, req.SnapshotID, "a branch absent on the base must be required to stay absent")
+
+ assert.NoError(t, req.Validate(base))
+ err = req.Validate(metadataWithRef(t, base, "audit", BranchRef))
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "created concurrently")
+ })
+
+ // The branch-typed assertion is client-side only: requireBranch has no wire form, so a REST catalog
+ // enforces snapshot-id equality alone.
+ t.Run("target name created as a tag is rejected locally", func(t *testing.T) {
+ base := newConflictTestMetadata(t, &head)
+ tx := multiTableTestTable(t, base).NewTransactionOnBranch("audit")
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets": "42"}))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ req := soleRefRequirement(t, tc.Requirements, "audit")
+ err = req.Validate(metadataWithRef(t, base, "audit", TagRef))
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "tags cannot be transaction targets")
+ })
+
+ t.Run("repeated calls are stable and leave the transaction unchanged", func(t *testing.T) {
+ base := newConflictTestMetadata(t, &head)
+ tx := multiTableTestTable(t, base).NewTransaction()
+ require.NoError(t, tx.AssertRefSnapshotID(MainBranch))
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"offsets": "42"}))
+
+ staged := soleRefRequirement(t, tx.reqs, MainBranch)
+ require.False(t, staged.requireBranch,
+ "the staged requirement starts untyped; the payload upgrade must not happen in place")
+
+ first, err := tx.TableCommit()
+ require.NoError(t, err)
+ second, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ assert.Equal(t, first.Requirements, second.Requirements)
+ assert.Len(t, tx.reqs, 1)
+ assert.False(t, staged.requireBranch,
+ "building the payload must not mutate the transaction's own requirements")
+ })
+}
+
+// Why: whole-list equality, so any future divergence between the two paths fails here
+// rather than only when a ref requirement goes missing.
+func TestTableCommitRequirementsMatchCommit(t *testing.T) {
+ head := int64(100)
+
+ for _, branch := range []string{MainBranch, "audit"} {
+ t.Run(branch, func(t *testing.T) {
+ base := metadataWithRef(t, newConflictTestMetadata(t, &head), "audit", BranchRef)
+
+ cat := &reqCapturingCatalog{metadata: base}
+ committed := New(Identifier{"db", "multi-table"}, base, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }, cat)
+ payload := multiTableTestTable(t, base)
+
+ // The catalog advances its metadata on commit; both sides must still derive their requirements
+ // from the same base.
+ require.Equal(t, base, committed.Metadata())
+ require.Equal(t, base, payload.Metadata())
+
+ txCommit := committed.NewTransactionOnBranch(branch)
+ require.NoError(t, txCommit.SetProperties(iceberg.Properties{"offsets": "42"}))
+ _, err := txCommit.Commit(t.Context())
+ require.NoError(t, err)
+ require.Len(t, cat.reqs, 1)
+
+ txPayload := payload.NewTransactionOnBranch(branch)
+ require.NoError(t, txPayload.SetProperties(iceberg.Properties{"offsets": "42"}))
+ tc, err := txPayload.TableCommit()
+ require.NoError(t, err)
+
+ assert.Equal(t, cat.reqs[0], tc.Requirements,
+ "a multi-table payload must carry the same fencing as Commit")
+ })
+ }
+}
+
+// Why: a producer stages its own ref assertion, which must be reused rather than duplicated.
+func TestTableCommitWithSnapshotProducerPinsBaseHead(t *testing.T) {
+ tbl, _, _ := newProducerAssertRefTable(t)
+
+ seed := tbl.NewTransaction()
+ require.NoError(t, seed.AddFiles(t.Context(), nil, nil, false))
+ tbl, err := seed.Commit(t.Context())
+ require.NoError(t, err)
+
+ baseHead := tbl.CurrentSnapshot()
+ require.NotNil(t, baseHead)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), nil, nil, false))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ req := soleRefRequirement(t, tc.Requirements, MainBranch)
+ require.NotNil(t, req.SnapshotID)
+ assert.Equal(t, baseHead.SnapshotID, *req.SnapshotID)
+ assert.True(t, req.requireBranch, "the target ref must be asserted as a branch")
+
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+ stagedHead := staged.CurrentSnapshot()
+ require.NotNil(t, stagedHead)
+ assert.NotEqual(t, stagedHead.SnapshotID, *req.SnapshotID,
+ "the assertion must not name a snapshot the catalog has never seen")
+}
+
+// Why: a transaction with nothing to commit enforces nothing on the single-table path either.
+func TestTableCommitWithoutUpdatesStaysEmpty(t *testing.T) {
+ head := int64(100)
+ base := newConflictTestMetadata(t, &head)
+
+ tx := multiTableTestTable(t, base).NewTransaction()
+ require.NoError(t, tx.AssertRefSnapshotID(MainBranch))
+
+ tc, err := tx.TableCommit()
+ require.NoError(t, err)
+
+ assert.Equal(t, []Requirement{}, tc.Requirements, "an empty payload must serialize as [], not null")
+ assert.Equal(t, []Update{}, tc.Updates, "an empty payload must serialize as [], not null")
+}