badalprasadsingh commented on code in PR #2384:
URL: https://github.com/apache/iceberg-go/pull/2384#discussion_r4526237240
##########
cmd/iceberg/branch_tag_test.go:
##########
@@ -784,3 +786,165 @@ func (l *loadTrackingCatalog) CommitTable(
) (table.Metadata, string, error) {
return nil, "", errors.New("unexpected CommitTable call")
}
+
+// CLI test catalog validates the same requirements as an optimistic catalog.
+type branchPromotionCatalog struct {
+ *branchTagCommitCatalog
+ meta table.Metadata
+ moveMainOnCommit bool
+}
+
+func (c *branchPromotionCatalog) LoadTable(_ context.Context, ident
table.Identifier) (*table.Table, error) {
+ return table.New(ident, c.meta, "",
+ func(context.Context) (icebergio.IO, error) { return
icebergio.LocalFS{}, nil }, c), nil
+}
+
+func (c *branchPromotionCatalog) CommitTable(_ context.Context, _
table.Identifier,
+ reqs []table.Requirement, updates []table.Update,
+) (table.Metadata, string, error) {
+ if c.moveMainOnCommit {
+ // A peer changes only main after the CLI loaded the table.
+ b, err := table.MetadataBuilderFromBase(c.meta, "")
+ if err != nil {
+ return nil, "", err
+ }
+ if err := b.SetSnapshotRef(table.MainBranch, 6000,
table.BranchRef); err != nil {
+ return nil, "", err
+ }
+ c.meta, err = b.Build()
+ if err != nil {
+ return nil, "", err
+ }
+ c.moveMainOnCommit = false
+ }
+ for _, r := range reqs {
+ if err := r.Validate(c.meta); err != nil {
+ return nil, "", err
+ }
+ }
+ changed, err := table.UpdateTableMetadata(c.meta, updates, "")
+ if err != nil {
+ return nil, "", err
+ }
+ c.meta = changed
+
+ return changed, "", nil
+}
+
+func newBranchPromotionTestCatalog(t *testing.T) *branchPromotionCatalog {
+ t.Helper()
+ original, err := table.ParseMetadataBytes([]byte(branchTagTestMetadata))
+ require.NoError(t, err)
+ b, err := table.MetadataBuilderFromBase(original, "")
+ require.NoError(t, err)
+ parent := int64(5000)
+ require.NoError(t, b.AddSnapshot(&table.Snapshot{
+ SnapshotID: 6000, ParentSnapshotID: &parent,
+ SequenceNumber: 2, TimestampMs:
original.SnapshotByID(5000).TimestampMs + 1000,
+ Summary: &table.Summary{Operation: table.OpAppend},
+ }))
+ require.NoError(t, b.SetSnapshotRef("audit", 6000, table.BranchRef))
+ meta, err := b.Build()
Review Comment:
Since #2462 turned retries on by default,
`TestBranchPromotionCLIAllowsUnrelatedMainMove` also passes with
`tbl.NewTransaction()`, because the retry rebases main. Could you set
`commit.retry.num-retries` to 0 here, so the test still checks that both
commands use NewTransactionOnBranch
Wrapping `tc.run` in `captureBranchTagExit` would also show the error
instead of exiting the test run.
##########
table/manage_snapshot_refs_test.go:
##########
@@ -0,0 +1,497 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "encoding/json"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func manageSnapshotRefsFixture(t *testing.T) Metadata {
+ t.Helper()
+ base := newConflictTestMetadataWithChain(t, []int64{100, 200, 300})
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.SetSnapshotRef(MainBranch, 100, BranchRef))
+ require.NoError(t, builder.SetSnapshotRef("qa", 100, BranchRef,
+ WithMaxRefAgeMs(1000), WithMaxSnapshotAgeMs(2000),
WithMinSnapshotsToKeep(3)))
+ require.NoError(t, builder.SetSnapshotRef("audit", 300, BranchRef))
+ require.NoError(t, builder.SetSnapshotRef("release", 300, TagRef,
WithMaxRefAgeMs(5000)))
+ require.NoError(t, builder.AddSnapshot(&Snapshot{
+ SnapshotID: 900, SequenceNumber: 4,
+ TimestampMs: base.LastUpdatedMillis() + 1000,
+ Summary: &Summary{Operation: OpAppend},
+ }))
+ require.NoError(t, builder.SetSnapshotRef("diverged", 900, BranchRef))
+ result, err := builder.Build()
+ require.NoError(t, err)
+
+ return result
+}
+
+func manageSnapshotRefsRetryFixture(t *testing.T) Metadata {
+ t.Helper()
+ base := manageSnapshotRefsFixture(t)
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.SetProperties(assertRefRetryProps))
+ result, err := builder.Build()
+ require.NoError(t, err)
+
+ return result
+}
+
+func TestFastForwardBranch(t *testing.T) {
+ t.Run("preserves retention and leaves source unchanged", func(t
*testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("qa", "audit"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("audit").SnapshotID)
+ assert.Equal(t, int64(100),
cat.metadata.CurrentSnapshot().SnapshotID)
+ for name, ref := range cat.metadata.Refs() {
+ if name == "qa" {
+ require.NotNil(t, ref.MaxRefAgeMs)
+ require.NotNil(t, ref.MaxSnapshotAgeMs)
+ require.NotNil(t, ref.MinSnapshotsToKeep)
+ assert.Equal(t, int64(1000), *ref.MaxRefAgeMs)
+ assert.Equal(t, int64(2000),
*ref.MaxSnapshotAgeMs)
+ assert.Equal(t, 3, *ref.MinSnapshotsToKeep)
+ }
+ }
+ })
+ t.Run("tag source and main target", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch(MainBranch, "release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(300),
cat.metadata.CurrentSnapshot().SnapshotID)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("release").SnapshotID)
+ })
+ t.Run("creates absent target", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("fresh", "audit"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ var found bool
+ for name, ref := range cat.metadata.Refs() {
+ if name == "fresh" {
+ found = true
+ assert.Equal(t, int64(300), ref.SnapshotID)
+ assert.Equal(t, BranchRef, ref.Type)
+ assert.Nil(t, ref.MinSnapshotsToKeep)
+ }
+ }
+ assert.True(t, found)
+ })
+ t.Run("uses transaction staged source", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranch("audit", 200))
+ require.NoError(t, tx.FastForwardBranch("qa", "audit"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(200),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ assert.Equal(t, int64(200),
cat.metadata.SnapshotByName("audit").SnapshotID)
+ })
+ t.Run("same heads are no-op", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("audit", "release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Zero(t, cat.attempts.Load())
+ assert.ErrorContains(t, tx.FastForwardBranch("audit",
"release"), "already been committed")
+ })
+ t.Run("reject invalid refs and divergent heads", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, _ := newAssertRefTestTable(t, base, base)
+ for _, tc := range []struct{ target, source string }{
+ {"qa", "missing"},
+ {"release", "audit"},
+ {"audit", "qa"},
+ {"qa", "diverged"},
+ {"", "audit"},
+ {"qa", ""},
+ } {
+ tx := tbl.NewTransaction()
+ require.ErrorIs(t, tx.FastForwardBranch(tc.target,
tc.source), iceberg.ErrInvalidArgument)
+ assert.Empty(t, tx.meta.updates)
+ }
+ })
+}
+
+func TestBranchPromotionPinsBothRefs(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ stage func(*Transaction) error
+ move func(*MetadataBuilder) error
+ wantErr error
+ }{
+ {
+ name: "target moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("qa", 200, BranchRef) },
+ },
+ {
+ name: "source moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("audit", 200, BranchRef) },
+ },
+ {
+ name: "replace-from-ref target moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("qa", 200, BranchRef) },
+ },
+ {
+ name: "replace-from-ref source moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef("qa", "release") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("release", 200, TagRef) },
+ },
+ {
+ name: "replace-from-ref main target moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef(MainBranch, "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "replace-from-ref main source moved on no-op",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef("qa", MainBranch) },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "target became tag with same id",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("qa", 100, TagRef) },
+ wantErr: ErrTransactionUnusable,
+ },
+ {
+ name: "main target moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch(MainBranch, "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "main source moved",
Review Comment:
the real move with main as source has no test: this case and line185 are
no-ops, since qa and main are both at 100. A no-op ReplaceBranch on main has no
test either.
better: start main at 200 and add both cases
##########
table/manage_snapshot_refs_test.go:
##########
@@ -0,0 +1,497 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "encoding/json"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func manageSnapshotRefsFixture(t *testing.T) Metadata {
+ t.Helper()
+ base := newConflictTestMetadataWithChain(t, []int64{100, 200, 300})
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.SetSnapshotRef(MainBranch, 100, BranchRef))
+ require.NoError(t, builder.SetSnapshotRef("qa", 100, BranchRef,
+ WithMaxRefAgeMs(1000), WithMaxSnapshotAgeMs(2000),
WithMinSnapshotsToKeep(3)))
+ require.NoError(t, builder.SetSnapshotRef("audit", 300, BranchRef))
+ require.NoError(t, builder.SetSnapshotRef("release", 300, TagRef,
WithMaxRefAgeMs(5000)))
+ require.NoError(t, builder.AddSnapshot(&Snapshot{
+ SnapshotID: 900, SequenceNumber: 4,
+ TimestampMs: base.LastUpdatedMillis() + 1000,
+ Summary: &Summary{Operation: OpAppend},
+ }))
+ require.NoError(t, builder.SetSnapshotRef("diverged", 900, BranchRef))
+ result, err := builder.Build()
+ require.NoError(t, err)
+
+ return result
+}
+
+func manageSnapshotRefsRetryFixture(t *testing.T) Metadata {
+ t.Helper()
+ base := manageSnapshotRefsFixture(t)
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.SetProperties(assertRefRetryProps))
+ result, err := builder.Build()
+ require.NoError(t, err)
+
+ return result
+}
+
+func TestFastForwardBranch(t *testing.T) {
+ t.Run("preserves retention and leaves source unchanged", func(t
*testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("qa", "audit"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("audit").SnapshotID)
+ assert.Equal(t, int64(100),
cat.metadata.CurrentSnapshot().SnapshotID)
+ for name, ref := range cat.metadata.Refs() {
+ if name == "qa" {
+ require.NotNil(t, ref.MaxRefAgeMs)
+ require.NotNil(t, ref.MaxSnapshotAgeMs)
+ require.NotNil(t, ref.MinSnapshotsToKeep)
+ assert.Equal(t, int64(1000), *ref.MaxRefAgeMs)
+ assert.Equal(t, int64(2000),
*ref.MaxSnapshotAgeMs)
+ assert.Equal(t, 3, *ref.MinSnapshotsToKeep)
+ }
+ }
+ })
+ t.Run("tag source and main target", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch(MainBranch, "release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(300),
cat.metadata.CurrentSnapshot().SnapshotID)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("release").SnapshotID)
+ })
+ t.Run("creates absent target", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("fresh", "audit"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ var found bool
+ for name, ref := range cat.metadata.Refs() {
+ if name == "fresh" {
+ found = true
+ assert.Equal(t, int64(300), ref.SnapshotID)
+ assert.Equal(t, BranchRef, ref.Type)
+ assert.Nil(t, ref.MinSnapshotsToKeep)
+ }
+ }
+ assert.True(t, found)
+ })
+ t.Run("uses transaction staged source", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranch("audit", 200))
+ require.NoError(t, tx.FastForwardBranch("qa", "audit"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(200),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ assert.Equal(t, int64(200),
cat.metadata.SnapshotByName("audit").SnapshotID)
+ })
+ t.Run("same heads are no-op", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("audit", "release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Zero(t, cat.attempts.Load())
+ assert.ErrorContains(t, tx.FastForwardBranch("audit",
"release"), "already been committed")
+ })
+ t.Run("reject invalid refs and divergent heads", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, _ := newAssertRefTestTable(t, base, base)
+ for _, tc := range []struct{ target, source string }{
+ {"qa", "missing"},
+ {"release", "audit"},
+ {"audit", "qa"},
+ {"qa", "diverged"},
+ {"", "audit"},
+ {"qa", ""},
+ } {
+ tx := tbl.NewTransaction()
+ require.ErrorIs(t, tx.FastForwardBranch(tc.target,
tc.source), iceberg.ErrInvalidArgument)
+ assert.Empty(t, tx.meta.updates)
+ }
+ })
+}
+
+func TestBranchPromotionPinsBothRefs(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ stage func(*Transaction) error
+ move func(*MetadataBuilder) error
+ wantErr error
+ }{
+ {
+ name: "target moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("qa", 200, BranchRef) },
+ },
+ {
+ name: "source moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("audit", 200, BranchRef) },
+ },
+ {
+ name: "replace-from-ref target moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("qa", 200, BranchRef) },
+ },
+ {
+ name: "replace-from-ref source moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef("qa", "release") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("release", 200, TagRef) },
+ },
+ {
+ name: "replace-from-ref main target moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef(MainBranch, "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "replace-from-ref main source moved on no-op",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranchFromRef("qa", MainBranch) },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "target became tag with same id",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef("qa", 100, TagRef) },
+ wantErr: ErrTransactionUnusable,
+ },
+ {
+ name: "main target moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch(MainBranch, "audit") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "main source moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch("qa", MainBranch) },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "no-op main target moved",
+ stage: func(tx *Transaction) error { return
tx.FastForwardBranch(MainBranch, "qa") },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ {
+ name: "replace main moved",
+ stage: func(tx *Transaction) error { return
tx.ReplaceBranch(MainBranch, 900) },
+ move: func(b *MetadataBuilder) error { return
b.SetSnapshotRef(MainBranch, 200, BranchRef) },
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ base := manageSnapshotRefsRetryFixture(t)
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, tc.move(builder))
+ current, err := builder.Build()
+ require.NoError(t, err)
+
+ tbl, cat := newAssertRefTestTable(t, base, current)
+ tx := tbl.NewTransaction()
+ // Force a catalog commit even when the promotion is a
no-op.
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{"promotion-marker": "staged"}))
+ require.NoError(t, tc.stage(tx))
+ _, err = tx.Commit(t.Context())
+ wantErr := ErrCommitFailed
+ if tc.wantErr != nil {
+ wantErr = tc.wantErr
+ }
+ require.ErrorIs(t, err, wantErr)
+ if tc.wantErr != nil {
+ require.NotErrorIs(t, err, ErrCommitFailed)
+ }
+ assert.Equal(t, int32(1), cat.attempts.Load(),
+ "a pinned ref must abort on refresh without a
second commit attempt")
+ assert.NotContains(t, cat.metadata.Properties(),
"promotion-marker")
+ })
+ }
+}
+
+func TestFastForwardBranchRejectsConcurrentTargetCreation(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ // Another writer creates the target after our transaction loaded
metadata.
+ require.NoError(t, builder.SetSnapshotRef("fresh", 200, BranchRef))
+ current, err := builder.Build()
+ require.NoError(t, err)
+
+ tbl, cat := newAssertRefTestTable(t, base, current)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.FastForwardBranch("fresh", "audit"))
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.Equal(t, int64(200),
cat.metadata.SnapshotByName("fresh").SnapshotID,
+ "promotion must not overwrite a concurrently created branch")
+}
+
+func TestReplaceBranchFromRef(t *testing.T) {
+ t.Run("moves branch to source tag preserving retention", func(t
*testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranchFromRef("qa", "release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("release").SnapshotID)
+ for name, ref := range cat.metadata.Refs() {
+ if name == "qa" {
+ require.NotNil(t, ref.MaxRefAgeMs)
+ require.NotNil(t, ref.MaxSnapshotAgeMs)
+ require.NotNil(t, ref.MinSnapshotsToKeep)
+ assert.Equal(t, int64(1000), *ref.MaxRefAgeMs)
+ assert.Equal(t, int64(2000),
*ref.MaxSnapshotAgeMs)
+ assert.Equal(t, 3, *ref.MinSnapshotsToKeep)
+ }
+ }
+ })
+
+ t.Run("moves a divergent branch", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranchFromRef("diverged",
"release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("diverged").SnapshotID)
+ assert.Equal(t, int64(300),
cat.metadata.SnapshotByName("release").SnapshotID)
+ })
+
+ t.Run("creates absent target", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranchFromRef("published",
"release"))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, int64(300),
cat.metadata.SnapshotByName("published").SnapshotID)
+ for name, ref := range cat.metadata.Refs() {
+ if name == "published" {
+ assert.Equal(t, BranchRef, ref.Type)
+ assert.Nil(t, ref.MaxRefAgeMs)
+ }
+ }
+ })
+
+ t.Run("rejects source absent or tag target", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, _ := newAssertRefTestTable(t, base, base)
+ for _, tc := range []struct{ target, source string }{
+ {"qa", "missing"},
+ {"release", "qa"},
+ {"qa", ""},
+ {"", "release"},
+ } {
+ tx := tbl.NewTransaction()
+ require.ErrorIs(t, tx.ReplaceBranchFromRef(tc.target,
tc.source), iceberg.ErrInvalidArgument)
+ assert.Empty(t, tx.meta.updates)
+ }
+ })
+}
+
+func TestReplaceBranch(t *testing.T) {
+ t.Run("divergent snapshot and retention", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, cat := newAssertRefTestTable(t, base, base)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranch("qa", 900))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, int64(900),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ for name, ref := range cat.metadata.Refs() {
+ if name == "qa" {
+ assert.Equal(t, int64(1000), *ref.MaxRefAgeMs)
+ assert.Equal(t, int64(2000),
*ref.MaxSnapshotAgeMs)
+ assert.Equal(t, 3, *ref.MinSnapshotsToKeep)
+ }
+ }
+ })
+ t.Run("reject invalid inputs", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, _ := newAssertRefTestTable(t, base, base)
+ for _, tc := range []struct {
+ name string
+ id int64
+ }{{"release", 100}, {"qa", 99999}, {"", 200}, {"fresh", 200}} {
+ tx := tbl.NewTransaction()
+ require.ErrorIs(t, tx.ReplaceBranch(tc.name, tc.id),
iceberg.ErrInvalidArgument)
+ assert.Empty(t, tx.meta.updates)
+ }
+ })
+ t.Run("unknown snapshot keeps error identity and explains refs mode",
func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ tbl, _ := newAssertRefTestTable(t, base, base)
+ err := tbl.NewTransaction().ReplaceBranch("qa", 99999)
+ require.ErrorIs(t, err, ErrSnapshotNotFound)
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ assert.ErrorContains(t, err, "snapshot-loading-mode=refs")
+ })
+ t.Run("concurrent target move", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ peer, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, peer.SetSnapshotRef("qa", 200, BranchRef))
+ current, err := peer.Build()
+ require.NoError(t, err)
+ tbl, cat := newAssertRefTestTable(t, base, current)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranch("qa", 900))
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.Equal(t, int64(200),
cat.metadata.SnapshotByName("qa").SnapshotID)
+ })
+ t.Run("concurrent tag conversion", func(t *testing.T) {
+ base := manageSnapshotRefsFixture(t)
+ peer, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, peer.SetSnapshotRef("qa", 100, TagRef))
+ current, err := peer.Build()
+ require.NoError(t, err)
+ tbl, _ := newAssertRefTestTable(t, base, current)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceBranch("qa", 200))
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
Review Comment:
P1: ci fails here: since #2462, commits retry 4 times by default, so this
now ends with `ErrTransactionUnusable`, not `ErrCommitFailed`.
better: expect that and NotErrorIs(ErrCommitFailed, like "target became tag
with same id" does, and check that qa is still a tag?
##########
table/manage_snapshot_refs.go:
##########
@@ -0,0 +1,155 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+// FastForwardBranch advances target to the snapshot referenced by source.
+// Source can be a branch or tag and is never changed. If target does not
+// exist, it is created with default retention settings. If target exists,
+// it must be a branch and its snapshot must be an ancestor of source's
+// snapshot. The target keeps its retention settings.
+//
+// Both refs are pinned to their snapshot IDs in the transaction's base
+// metadata. A concurrent change to either ID fails Commit. This also stops
+// the normal retry of a co-staged append if a pinned ref moves. No catalog's
+// snapshot-ID requirement detects retention-only changes, and REST catalogs
+// also cannot detect a ref type change with the same snapshot ID.
+// Fast-forward needs the full ancestor chain. With a REST catalog configured
+// with snapshot-loading-mode=refs, a valid fast-forward can be rejected;
+// reload with snapshot-loading-mode=all before calling this method.
+// For non-main targets, use Table.NewTransactionOnBranch(target) so an
+// unrelated main head move does not conflict with the transaction's implicit
+// commit-branch requirement.
+// Calls after other ref changes in this transaction use the staged ref heads.
+// A transaction without updates does not contact the catalog.
+func (t *Transaction) FastForwardBranch(target, source string) error {
+ return t.moveBranchToRef(target, source, true)
+}
+
+// ReplaceBranchFromRef moves target to the snapshot named by source without
+// checking ancestry, matching Java's ManageSnapshots.replaceBranch(from, to).
+// Source may be a branch or tag and is left unchanged. If target is absent,
+// it is created as a branch with default retention; otherwise the target
+// branch retains its existing retention settings.
+// For non-main targets, use Table.NewTransactionOnBranch(target) to avoid
+// requiring main to remain unchanged during the commit.
+//
+// Both ref heads are pinned to their base snapshot IDs, so concurrent head
+// moves fail the commit and prevent the normal retry of a co-staged append.
+// No catalog's snapshot-ID requirement detects retention-only changes, and
+// REST catalogs also cannot detect ref type changes with the same snapshot ID.
+func (t *Transaction) ReplaceBranchFromRef(target, source string) error {
+ return t.moveBranchToRef(target, source, false)
+}
+
+func (t *Transaction) moveBranchToRef(target, source string, fastForward bool)
error {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return err
+ }
+ if target == "" || source == "" {
+ return fmt.Errorf("%w: target and source ref names must not be
empty", iceberg.ErrInvalidArgument)
+ }
+
+ sourceRef, ok := meta.refs[source]
+ if !ok {
+ return fmt.Errorf("%w: source ref %q does not exist",
iceberg.ErrInvalidArgument, source)
+ }
+
+ targetRef, exists := meta.refs[target]
+ if exists && targetRef.Type != BranchRef {
+ return fmt.Errorf("%w: target ref %q is a %s, not a branch",
iceberg.ErrInvalidArgument, target, targetRef.Type)
+ }
+ requirements := []Requirement{assertBranchRefSnapshotID(target,
t.baseRefSnapshotID(target))}
+ if source != target {
+ requirements = append(requirements, AssertRefSnapshotID(source,
t.baseRefSnapshotID(source)))
+ }
+ if exists {
+ if targetRef.SnapshotID == sourceRef.SnapshotID {
+ // No metadata update is needed, but pin both refs so
that a
+ // staged metadata-only commit cannot miss a concurrent
move.
+ return t.applyPinned(nil, requirements,
[]string{target, source})
+ }
+ lookup := func(id int64) *Snapshot {
+ snap, _ := meta.SnapshotByID(id)
+
+ return snap
+ }
+ if fastForward && !IsAncestorOf(sourceRef.SnapshotID,
targetRef.SnapshotID, lookup) {
+ return fmt.Errorf("%w: branch %q snapshot %d is not an
ancestor of %q snapshot %d (REST snapshot-loading-mode=refs may omit ancestors;
reload with snapshot-loading-mode=all)",
+ iceberg.ErrInvalidArgument, target,
targetRef.SnapshotID, source, sourceRef.SnapshotID)
+ }
+ }
+
+ update := meta.NewRetainingSnapshotRefUpdate(target,
sourceRef.SnapshotID, BranchRef)
+
+ return t.applyPinned([]Update{update}, requirements, []string{target,
source})
Review Comment:
P1: the ancestor check only runs when staging: if source points at a
snapshot staged here, a retry can rebuild it onto a peer's rollback of main,
and the target silently loses rows: iceberg-java fails this commit.
could you also pin `t.branch` for a staged source, and add a test?
--
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]
