(iceberg-go) branch main updated: fix(table): reject tags as transaction targets (#1657)

[ze...@apache.org]

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 0bb9e83ee fix(table): reject tags as transaction targets (#1657)
0bb9e83ee is described below

commit 0bb9e83eec356c80e2883dbfc47aa618c050574b
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Fri Aug 7 19:28:10 2026 +0200

    fix(table): reject tags as transaction targets (#1657)
    
    * fix(table): reject tags as transaction targets
    
    * fix(table): preserve branch-only transaction targets
---
 table/requirements.go              | 18 ++++++++++--
 table/table.go                     | 35 ++++++++++++++++++++++-
 table/transaction.go               | 33 +++++++++++++++++++++-
 table/transaction_internal_test.go | 57 ++++++++++++++++++++++++++++++++++++++
 4 files changed, 139 insertions(+), 4 deletions(-)

diff --git a/table/requirements.go b/table/requirements.go
index d6004a656..f0daa2741 100644
--- a/table/requirements.go
+++ b/table/requirements.go
@@ -155,8 +155,9 @@ func (a *assertTableUuid) Validate(meta Metadata) error {
 
 type assertRefSnapshotID struct {
 	baseRequirement
-	Ref        string `json:"ref"`
-	SnapshotID *int64 `json:"snapshot-id"`
+	Ref           string `json:"ref"`
+	SnapshotID    *int64 `json:"snapshot-id"`
+	requireBranch bool
 }
 
 // AssertRefSnapshotID creates a requirement which ensures that the table branch
@@ -170,6 +171,15 @@ func AssertRefSnapshotID(ref string, id *int64) Requirement {
 	}
 }
 
+func assertBranchRefSnapshotID(ref string, id *int64) Requirement {
+	return &assertRefSnapshotID{
+		baseRequirement: baseRequirement{Type: reqAssertRefSnapshotID},
+		Ref:             ref,
+		SnapshotID:      id,
+		requireBranch:   true,
+	}
+}
+
 func (a *assertRefSnapshotID) Validate(meta Metadata) error {
 	if meta == nil {
 		return errors.New("Requirement failed: current table metadata does not exist")
@@ -185,6 +195,10 @@ func (a *assertRefSnapshotID) Validate(meta Metadata) error {
 	}
 
 	if r != nil {
+		if a.requireBranch && r.SnapshotRefType != BranchRef {
+			return fmt.Errorf("requirement failed: ref %q is a %s; tags cannot be transaction targets", a.Ref, r.SnapshotRefType)
+		}
+
 		if a.SnapshotID == nil {
 			return fmt.Errorf("requirement failed: %s %q was created concurrently", r.SnapshotRefType, a.Ref)
 		}
diff --git a/table/table.go b/table/table.go
index 430fb9dc4..ac0cbf0b9 100644
--- a/table/table.go
+++ b/table/table.go
@@ -189,6 +189,18 @@ func (t Table) newBrokenTransaction(branch string, err error) *Transaction {
 // callers to receive the precise initialization error instead of hitting
 // panic/undefined behavior later.
 func (t Table) NewTransactionOnBranchWithError(branch string) (*Transaction, error) {
+	for name, ref := range t.metadata.Refs() {
+		if name != branch {
+			continue
+		}
+		if ref.SnapshotRefType != BranchRef {
+			return nil, fmt.Errorf("%w: ref %q is a %s; tags cannot be transaction targets",
+				iceberg.ErrInvalidArgument, branch, ref.SnapshotRefType)
+		}
+
+		break
+	}
+
 	meta, err := MetadataBuilderFromBase(t.metadata, t.metadataLocation)
 	if err != nil {
 		return nil, err
@@ -592,6 +604,9 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen
 			}
 			current = fresh.metadata
 			reqs = rewriteRefSnapshotRequirements(reqs, co.branch, current)
+			if err := validateBranchRequirement(reqs, co.branch, current); err != nil {
+				return nil, err
+			}
 
 			// Rebuild snapshot manifest lists to inherit all files committed
 			// by concurrent writers since the snapshot was originally built.
@@ -606,6 +621,10 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen
 			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)
@@ -739,7 +758,11 @@ func rewriteRefSnapshotRequirements(reqs []Requirement, branch string, fresh Met
 	for i, r := range reqs {
 		if a, ok := r.(*assertRefSnapshotID); ok && a.Ref == branch {
 			newID := head.SnapshotID
-			out[i] = AssertRefSnapshotID(branch, &newID)
+			if a.requireBranch {
+				out[i] = assertBranchRefSnapshotID(branch, &newID)
+			} else {
+				out[i] = AssertRefSnapshotID(branch, &newID)
+			}
 
 			continue
 		}
@@ -749,6 +772,16 @@ func rewriteRefSnapshotRequirements(reqs []Requirement, branch string, fresh Met
 	return out
 }
 
+func validateBranchRequirement(reqs []Requirement, branch string, meta Metadata) error {
+	for _, req := range reqs {
+		if ref, ok := req.(*assertRefSnapshotID); ok && ref.requireBranch && ref.Ref == branch {
+			return ref.Validate(meta)
+		}
+	}
+
+	return nil
+}
+
 // rebuildSnapshotUpdates returns a new slice of updates where any
 // addSnapshotUpdate that carries a rebuildManifestList closure has its
 // snapshot regenerated to inherit all data files committed to the branch
diff --git a/table/transaction.go b/table/transaction.go
index b5648c909..2b8bc7fb4 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -240,6 +240,37 @@ func requirementSemanticKey(r Requirement) (string, error) {
 	return string(data), nil
 }
 
+func transactionRequirements(reqs []Requirement, branch string, base Metadata) []Requirement {
+	if branch == "" {
+		branch = MainBranch
+	}
+
+	out := slices.Clone(reqs)
+	for i, req := range out {
+		if ref, ok := req.(*assertRefSnapshotID); ok && ref.Ref == branch {
+			if ref.requireBranch {
+				return out
+			}
+
+			out[i] = assertBranchRefSnapshotID(branch, ref.SnapshotID)
+
+			return out
+		}
+	}
+
+	var snapshotID *int64
+	for name, ref := range base.Refs() {
+		if name == branch {
+			id := ref.SnapshotID
+			snapshotID = &id
+
+			break
+		}
+	}
+
+	return append(out, assertBranchRefSnapshotID(branch, snapshotID))
+}
+
 // addValidator appends a conflict validator under t.mx. Producers
 // that register validators from outside doCommit (RowDelta, RewriteFiles)
 // must use this helper rather than mutating t.validators directly —
@@ -2373,7 +2404,7 @@ func (t *Transaction) Commit(ctx context.Context) (*Table, error) {
 	}
 
 	if len(meta.updates) > 0 {
-		reqs := append(slices.Clone(t.reqs), AssertTableUUID(meta.uuid))
+		reqs := append(transactionRequirements(t.reqs, t.branch, t.tbl.metadata), AssertTableUUID(meta.uuid))
 		tbl, err := t.tbl.doCommit(ctx, meta.updates, reqs,
 			withCommitBranch(t.branch),
 			withCommitValidators(t.validators...),
diff --git a/table/transaction_internal_test.go b/table/transaction_internal_test.go
index 3f10e509d..a92a6d2a3 100644
--- a/table/transaction_internal_test.go
+++ b/table/transaction_internal_test.go
@@ -97,6 +97,63 @@ func TestNewTransactionOnBranchWithErrorReturnsTransactionInitError(t *testing.T
 	require.Nil(t, txn)
 }
 
+func TestNewTransactionOnBranchRejectsTags(t *testing.T) {
+	txn := newTransactionWithSnapshotRefs(t)
+	require.NoError(t, txn.meta.SetSnapshotRef("release", 10, TagRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	tbl := New(Identifier{"db", "table"}, meta, "metadata.json", nil, nil)
+
+	before := tbl.Metadata()
+	transaction, err := tbl.NewTransactionOnBranchWithError("release")
+	require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+	require.Nil(t, transaction)
+	require.True(t, before.Equals(tbl.Metadata()))
+
+	legacyTransaction := tbl.NewTransactionOnBranch("release")
+	require.ErrorIs(t, legacyTransaction.initErr, iceberg.ErrInvalidArgument)
+	require.ErrorContains(t, legacyTransaction.SetProperties(iceberg.Properties{"k": "v"}), "tags cannot be transaction targets")
+}
+
+func TestTransactionCommitRejectsSameIDTagRace(t *testing.T) {
+	seed := newTransactionWithSnapshotRefs(t)
+	baseMeta, err := seed.meta.Build()
+	require.NoError(t, err)
+
+	head := baseMeta.SnapshotByName(MainBranch)
+	require.NotNil(t, head)
+	tagBuilder, err := MetadataBuilderFromBase(baseMeta, "")
+	require.NoError(t, err)
+	require.NoError(t, tagBuilder.SetSnapshotRef(MainBranch, head.SnapshotID, TagRef))
+	tagMeta, err := tagBuilder.Build()
+	require.NoError(t, err)
+
+	cat := &headTrackingCatalog{metadata: tagMeta}
+	tbl := New(Identifier{"db", "tag-race"}, baseMeta, "metadata.json",
+		func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }, cat)
+	txn := tbl.NewTransaction()
+	require.NoError(t, txn.SetProperties(iceberg.Properties{"k": "v"}))
+
+	_, err = txn.Commit(context.Background())
+	require.Error(t, err)
+	require.ErrorContains(t, err, "tags cannot be transaction targets")
+	require.Equal(t, int32(1), cat.attempts.Load())
+}
+
+func TestNewTransactionOnBranchAllowsBranchesAndNewRefs(t *testing.T) {
+	txn := newTransactionWithSnapshotRefs(t)
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	tbl := New(Identifier{"db", "table"}, meta, "metadata.json", nil, nil)
+
+	for _, branch := range []string{MainBranch, "feature", "new-branch"} {
+		transaction, err := tbl.NewTransactionOnBranchWithError(branch)
+		require.NoError(t, err, branch)
+		require.NotNil(t, transaction, branch)
+	}
+}
+
 func TestNewTransactionOnBranchKeepsLegacySignatureAndFailsOnUse(t *testing.T) {
 	baseMeta, err := NewMetadata(simpleSchema(), iceberg.UnpartitionedSpec, UnsortedSortOrder, "table-location", nil)
 	require.NoError(t, err, "new metadata")