laskoviymishka commented on code in PR #1858:
URL: https://github.com/apache/iceberg-go/pull/1858#discussion_r3850546348
##########
catalog/multi_table_transaction.go:
##########
@@ -102,8 +102,10 @@ 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:
Review Comment:
A metadata-only transaction with no staged updates returns `{Requirements:
[], Updates: []}` from `TableCommit`, and `Commit` here forwards every
collected transaction unconditionally. A conformant REST catalog can reject a
`TableChange` with zero updates (400/422), which would fail the whole atomic
batch over one no-op entry.
I'd either filter zero-update transactions out in `Commit`, or document on
`AddTransaction`/`Commit` that a no-update transaction shouldn't be added. wdyt?
##########
catalog/multi_table_transaction_test.go:
##########
@@ -277,3 +277,78 @@ 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(context.Background()))
Review Comment:
Small consistency thing: the new tests in `transaction_internal_test.go` all
use `t.Context()`, but this one reaches for `context.Background()`. I'd match
`t.Context()` here.
##########
table/commit.go:
##########
@@ -77,9 +78,9 @@ 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.
+ reqs := append(transactionRequirements(t.reqs, t.branch,
t.tbl.metadata), AssertTableUUID(meta.uuid))
Review Comment:
This is safe today only because `transactionRequirements` always starts with
`slices.Clone(reqs)`, so the returned slice never aliases `t.reqs`. That's a
non-local invariant now, and we're appending into the result while holding
`t.mx`; the old `make + copy` spelled the ownership out.
I'd add a one-liner here noting the returned slice is a fresh allocation, or
state that guarantee in `transactionRequirements`'s doc, so a future fast-path
return can't silently reintroduce aliasing.
##########
table/transaction_internal_test.go:
##########
@@ -1785,3 +1786,258 @@ 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")
+ })
+
+ t.Run("target name created as a tag is rejected", 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))
Review Comment:
This sub-case proves the tag guard fires through a local `Validate` call,
but that isn't what runs on the multi-table path this PR is fencing.
`MultiTableTransaction.Commit` hands the payload to `CommitTransaction` with no
local retry or `Validate` loop, and `requireBranch` has no wire representation,
so the server only sees
`{"type":"assert-ref-snapshot-id","ref":"audit","snapshot-id":...}` and
enforces snapshot-id equality, not ref type.
So the "created as a tag is rejected" protection is real for single-table
`Commit` but silently absent once the payload goes out over REST. I'd scope
this with a comment to client-side validation, so it isn't read as multi-table
tag-race coverage. wdyt?
##########
table/transaction_internal_test.go:
##########
@@ -1785,3 +1786,258 @@ 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")
+ })
+
+ t.Run("target name created as a tag is rejected", 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)
+
+ 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 := multiTableTestTable(t,
base).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.Empty(t, tc.Requirements)
Review Comment:
The `NotNil` pair already catches a nil return, but the intent, a non-nil
zero-length slice that serializes as `[]` and not `null`, reads more directly
as `assert.Equal(t, []Requirement{}, tc.Requirements)` (same for `Updates`).
One assertion, exact intent.
##########
table/transaction_internal_test.go:
##########
@@ -1785,3 +1786,258 @@ 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")
+ })
+
+ t.Run("target name created as a tag is rejected", 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)
+
+ 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 := multiTableTestTable(t,
base).NewTransactionOnBranch(branch)
Review Comment:
This passes only because `txPayload` is rebuilt from `base`, the same
starting metadata the `Commit` side used. `reqCapturingCatalog.CommitTable`
advances `c.metadata` in place, so if someone later seeds this from the
post-commit metadata, the two sides would start from different heads and the
equality could pass for the wrong reason.
I'd assert the two transactions start from identical metadata, or at least
leave a comment pinning that assumption.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
