badalprasadsingh commented on code in PR #2384: URL: https://github.com/apache/iceberg-go/pull/2384#discussion_r4455660426
########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,144 @@ +// 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 ( + "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 snapshot ID fails Commit. REST +// catalogs cannot check ref type or retention changes that leave the ID +// unchanged. Avoid changing the target ref concurrently with this operation. +// 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. +// +// Both ref heads are pinned to their base snapshot IDs, so concurrent head +// moves fail the commit. As with FastForwardBranch, REST requirements cannot +// detect changes to ref type or retention when its snapshot ID is unchanged. +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) + } + if sourceRef.Type != BranchRef && sourceRef.Type != TagRef { + return fmt.Errorf("%w: source ref %q has invalid type %q", iceberg.ErrInvalidArgument, source, sourceRef.Type) + } + if _, err := meta.SnapshotByID(sourceRef.SnapshotID); err != nil { + return fmt.Errorf("source ref %q points to an unknown snapshot: %w", source, err) + } + + 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", + 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: P0: a commit retry can undo this move and lose data. e.g., a txn fast-forwards main to audit, then appends, while someone commits to another branch. On retry, the append is rebuilt on main's old head instead of audit's. Commit succeeds, but main no longer has audit's rows. After `ReplaceBranch`, removed rows come back. could you please make `rebuildSnapshotUpdates` follow a staged ref move, and add this as a test? ########## table/manage_snapshot_refs_test.go: ########## @@ -0,0 +1,386 @@ +// 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 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 TestFastForwardBranchPinsBothRefs(t *testing.T) { + for _, tc := range []struct { + name string + move func(*MetadataBuilder) error + }{ + {"target moved", func(b *MetadataBuilder) error { return b.SetSnapshotRef("qa", 200, BranchRef) }}, + {"source moved", func(b *MetadataBuilder) error { return b.SetSnapshotRef("audit", 200, BranchRef) }}, + {"target became tag with same id", func(b *MetadataBuilder) error { return b.SetSnapshotRef("qa", 100, TagRef) }}, + } { + t.Run(tc.name, func(t *testing.T) { + base := manageSnapshotRefsFixture(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() + require.NoError(t, tx.FastForwardBranch("qa", "audit")) + _, err = tx.Commit(t.Context()) + require.ErrorIs(t, err, ErrCommitFailed) + assert.Equal(t, int32(1), cat.attempts.Load()) Review Comment: This test can't catch a missing pin. Retries are off by default, so attempts is always 1, and qa and audit are never rebased because only the commit branch is. Without the pins these tests still pass, yet a retry would overwrite another writer's main head. Could you turn on retries with assertRefRetryProps and add cases with main as target, source and no-op target, plus ReplaceBranch on main? Same in TestNoOpBranchRefOperationsFenceMixedTransactions. ########## cmd/iceberg/branch_promotion.go: ########## @@ -0,0 +1,103 @@ +// 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 main + +import ( + "context" + "fmt" + + "github.com/apache/iceberg-go/catalog" +) + +func runBranchFastForward(ctx context.Context, output Output, cat catalog.Catalog, cmd *BranchFastForwardCmd) { + if err := validateRefName(cmd.BranchName); err != nil { + output.Error(fmt.Errorf("invalid target branch: %w", err)) + osExit(1) + + return + } + if err := validateRefName(cmd.SourceRef); err != nil { + output.Error(fmt.Errorf("invalid source ref: %w", err)) + osExit(1) + + return + } + + tbl := loadTable(ctx, output, cat, cmd.TableID) + tx := tbl.NewTransaction() Review Comment: NewTransaction also checks main's head, so promoting qa fails with branch "main" has changed whenever someone appends to main, even though this command never touches main. Retries are off by default, and waiting at the prompt makes this more likely. Java's REST commit only checks the ref it changes. Could you use `tbl.NewTransactionOnBranch(cmd.BranchName)` here, and note this in the library method docs? ########## cmd/iceberg/maintenance.go: ########## @@ -97,9 +97,27 @@ type BranchDeleteCmd struct { Yes bool `arg:"--yes" help:"skip confirmation prompt"` } +// BranchFastForwardCmd advances a target branch to an ancestor-safe source ref. +type BranchFastForwardCmd struct { + TableID string `arg:"positional,required" help:"full path to a table"` + BranchName string `arg:"positional,required" help:"target branch name"` + SourceRef string `arg:"positional,required" help:"source branch or tag"` + Yes bool `arg:"--yes" help:"skip confirmation prompt"` +} + +// BranchReplaceCmd explicitly moves a branch to a selected snapshot. +type BranchReplaceCmd struct { + TableID string `arg:"positional,required" help:"full path to a table"` + BranchName string `arg:"positional,required" help:"target branch name"` + SnapshotID int64 `arg:"positional,required" help:"snapshot ID (does not require ancestry)"` + Yes bool `arg:"--yes" help:"skip confirmation prompt"` +} + type BranchCmd struct { - Create *BranchCreateCmd `arg:"subcommand:create" help:"create a branch"` - Delete *BranchDeleteCmd `arg:"subcommand:delete" help:"delete a branch"` + Create *BranchCreateCmd `arg:"subcommand:create" help:"create a branch"` + Delete *BranchDeleteCmd `arg:"subcommand:delete" help:"delete a branch"` + FastForward *BranchFastForwardCmd `arg:"subcommand:fast-forward" help:"advance a branch to an ancestor-safe source ref"` Review Comment: nit: users won't know "ancestor-safe". Same on `BranchFastForwardCmd`. Please write fast-forward a branch to a source branch or tag; fails if they have diverged. ########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,144 @@ +// 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 ( + "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 snapshot ID fails Commit. REST +// catalogs cannot check ref type or retention changes that leave the ID +// unchanged. Avoid changing the target ref concurrently with this operation. Review Comment: This sounds like only REST misses these changes, but no catalog checks retention, so another writer's retention change gets overwritten with old values and no error. Only the ref type gap is REST-specific. Could you write: "No catalog detects a concurrent retention change when the snapshot ID is unchanged, and REST catalogs also miss a ref type change."? Same in ReplaceBranch, ReplaceBranchFromRef, the README and the PR description. ########## cmd/iceberg/branch_promotion.go: ########## @@ -0,0 +1,103 @@ +// 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 main + +import ( + "context" + "fmt" + + "github.com/apache/iceberg-go/catalog" +) + +func runBranchFastForward(ctx context.Context, output Output, cat catalog.Catalog, cmd *BranchFastForwardCmd) { + if err := validateRefName(cmd.BranchName); err != nil { + output.Error(fmt.Errorf("invalid target branch: %w", err)) + osExit(1) + + return + } + if err := validateRefName(cmd.SourceRef); err != nil { + output.Error(fmt.Errorf("invalid source ref: %w", err)) + osExit(1) + + return + } + + tbl := loadTable(ctx, output, cat, cmd.TableID) + tx := tbl.NewTransaction() + if err := tx.FastForwardBranch(cmd.BranchName, cmd.SourceRef); err != nil { + output.Error(fmt.Errorf("cannot fast-forward branch: %w", err)) + osExit(1) + + return + } + + if err := confirmAction(fmt.Sprintf("Fast-forward branch %q on %s from ref %q?", + cmd.BranchName, tableIDString(tbl), cmd.SourceRef), cmd.Yes); err != nil { + output.Error(err) + osExit(1) + + return + } + + updated, err := tx.Commit(ctx) + if err != nil { + output.Error(fmt.Errorf("failed to fast-forward branch: %w", err)) + osExit(1) + + return + } + output.Text(fmt.Sprintf("Fast-forwarded branch %q on %s to snapshot %d", + cmd.BranchName, tableIDString(updated), updated.SnapshotByName(cmd.BranchName).SnapshotID)) +} + +func runBranchReplace(ctx context.Context, output Output, cat catalog.Catalog, cmd *BranchReplaceCmd) { + if err := validateRefName(cmd.BranchName); err != nil { + output.Error(fmt.Errorf("invalid branch name: %w", err)) + osExit(1) + + return + } + + tbl := loadTable(ctx, output, cat, cmd.TableID) + tx := tbl.NewTransaction() Review Comment: Same comment: better to use `tbl.NewTransactionOnBranch(cmd.BranchName)` here ########## cmd/iceberg/branch_promotion.go: ########## @@ -0,0 +1,103 @@ +// 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 main + +import ( + "context" + "fmt" + + "github.com/apache/iceberg-go/catalog" +) + +func runBranchFastForward(ctx context.Context, output Output, cat catalog.Catalog, cmd *BranchFastForwardCmd) { + if err := validateRefName(cmd.BranchName); err != nil { + output.Error(fmt.Errorf("invalid target branch: %w", err)) + osExit(1) + + return + } + if err := validateRefName(cmd.SourceRef); err != nil { + output.Error(fmt.Errorf("invalid source ref: %w", err)) + osExit(1) + + return + } + + tbl := loadTable(ctx, output, cat, cmd.TableID) + tx := tbl.NewTransaction() + if err := tx.FastForwardBranch(cmd.BranchName, cmd.SourceRef); err != nil { + output.Error(fmt.Errorf("cannot fast-forward branch: %w", err)) + osExit(1) + + return + } + + if err := confirmAction(fmt.Sprintf("Fast-forward branch %q on %s from ref %q?", + cmd.BranchName, tableIDString(tbl), cmd.SourceRef), cmd.Yes); err != nil { + output.Error(err) + osExit(1) + + return + } + + updated, err := tx.Commit(ctx) + if err != nil { + output.Error(fmt.Errorf("failed to fast-forward branch: %w", err)) + osExit(1) + + return + } + output.Text(fmt.Sprintf("Fast-forwarded branch %q on %s to snapshot %d", + cmd.BranchName, tableIDString(updated), updated.SnapshotByName(cmd.BranchName).SnapshotID)) +} + +func runBranchReplace(ctx context.Context, output Output, cat catalog.Catalog, cmd *BranchReplaceCmd) { + if err := validateRefName(cmd.BranchName); err != nil { + output.Error(fmt.Errorf("invalid branch name: %w", err)) + osExit(1) + + return + } + + tbl := loadTable(ctx, output, cat, cmd.TableID) + tx := tbl.NewTransaction() + if err := tx.ReplaceBranch(cmd.BranchName, cmd.SnapshotID); err != nil { + output.Error(fmt.Errorf("cannot replace branch: %w", err)) + osExit(1) + + return + } + + if err := confirmAction(fmt.Sprintf("Replace branch %q on %s with snapshot %d?", + cmd.BranchName, tableIDString(tbl), cmd.SnapshotID), cmd.Yes); err != nil { Review Comment: The prompt doesn't show what will change. replace can remove history from a branch, but this hides the current snapshot, while branch delete shows it. In fast-forward, a typo like mian quietly creates a new branch. Could you show both IDs, for example Replace branch "main" on db.t (snapshot 100) with snapshot 300?, and say when a new branch will be created? ########## README.md: ########## @@ -96,6 +96,36 @@ The exported env vars and `make integration-test` invocation are identical to th Tests requiring Spark 4 (currently the variant and unknown-type tests) automatically skip on the Spark 3.5 setup. +## Snapshot branch promotion + +Use a transaction to advance a branch to a source branch or tag, or to explicitly +replace a branch head with any existing snapshot. Existing branch retention Review Comment: Branch commands are documented in `website/src/cli.md` and ref APIs in `website/src/api.md`, so people reading those pages won't find this. Here it also sits between the test instructions and the feature tables. Could you move it to those two pages? Also, "Replacement requires an existing branch" isn't true for `ReplaceBranchFromRef`, which creates a missing target. ########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,144 @@ +// 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 ( + "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 snapshot ID fails Commit. REST +// catalogs cannot check ref type or retention changes that leave the ID +// unchanged. Avoid changing the target ref concurrently with this operation. +// 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. +// +// Both ref heads are pinned to their base snapshot IDs, so concurrent head +// moves fail the commit. As with FastForwardBranch, REST requirements cannot +// detect changes to ref type or retention when its snapshot ID is unchanged. +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) + } + if sourceRef.Type != BranchRef && sourceRef.Type != TagRef { + return fmt.Errorf("%w: source ref %q has invalid type %q", iceberg.ErrInvalidArgument, source, sourceRef.Type) + } + if _, err := meta.SnapshotByID(sourceRef.SnapshotID); err != nil { + return fmt.Errorf("source ref %q points to an unknown snapshot: %w", source, err) + } Review Comment: I don't think these checks can ever fire: `SnapshotRef.validate` rejects other ref types, and `checkRefsExist` rejects refs pointing at missing `snapshots. SetSnapshotRef` also checks the snapshot again anyway. ########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,144 @@ +// 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 ( + "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 snapshot ID fails Commit. REST +// catalogs cannot check ref type or retention changes that leave the ID +// unchanged. Avoid changing the target ref concurrently with this operation. +// 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. +// +// Both ref heads are pinned to their base snapshot IDs, so concurrent head +// moves fail the commit. As with FastForwardBranch, REST requirements cannot +// detect changes to ref type or retention when its snapshot ID is unchanged. +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) + } + if sourceRef.Type != BranchRef && sourceRef.Type != TagRef { + return fmt.Errorf("%w: source ref %q has invalid type %q", iceberg.ErrInvalidArgument, source, sourceRef.Type) + } + if _, err := meta.SnapshotByID(sourceRef.SnapshotID); err != nil { + return fmt.Errorf("source ref %q points to an unknown snapshot: %w", source, err) + } + + 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", + iceberg.ErrInvalidArgument, target, targetRef.SnapshotID, source, sourceRef.SnapshotID) + } + } + + update := meta.NewRetainingSnapshotRefUpdate(target, sourceRef.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, requirements, []string{target, source}) +} + +// ReplaceBranch sets the named branch to an existing snapshot without an +// ancestry restriction. Unlike FastForwardBranch, this can move a branch +// backwards or onto a different lineage. The branch must already exist; its +// retention settings are preserved from the transaction's metadata. +// +// Commit checks the target's snapshot ID. REST catalogs cannot check ref type +// or retention changes that leave the ID unchanged. Avoid changing the target +// ref concurrently with this operation. +func (t *Transaction) ReplaceBranch(name string, snapshotID int64) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if name == "" { + return fmt.Errorf("%w: branch name must not be empty", iceberg.ErrInvalidArgument) + } + existing, ok := meta.refs[name] + if !ok { + return fmt.Errorf("%w: branch %q does not exist", iceberg.ErrInvalidArgument, name) + } + if existing.Type != BranchRef { + return fmt.Errorf("%w: ref %q is a %s, not a branch", iceberg.ErrInvalidArgument, name, existing.Type) + } + if existing.SnapshotID == snapshotID { + // A no-op still fences a co-staged metadata update. + return t.applyPinned(nil, + []Requirement{assertBranchRefSnapshotID(name, t.baseRefSnapshotID(name))}, + []string{name}) + } + if _, err := meta.SnapshotByID(snapshotID); err != nil { + return fmt.Errorf("%w: cannot replace branch %q with unknown snapshot %d: %v", + iceberg.ErrInvalidArgument, name, snapshotID, err) + } Review Comment: `%v` means callers can't use `errors.Is(err, ErrSnapshotNotFound)`. `SetSnapshotRef` already does this same check, so I'd delete these lines. The error will still name the branch and snapshot, and will match both `ErrInvalidArgument` and `ErrSnapshotNotFound`. ########## cmd/iceberg/branch_promotion_test.go: ########## @@ -0,0 +1,140 @@ +// 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 main + +import ( + "context" + "strings" + "testing" + + "github.com/apache/iceberg-go" + icebergio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// CLI test catalog validates the same requirements as an optimistic catalog. +type branchPromotionCatalog struct { + *branchTagCommitCatalog + meta table.Metadata +} + +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) { + 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() + require.NoError(t, err) + + return &branchPromotionCatalog{meta: meta} +} + +type branchPromotionOutput struct { + textOutput + message string + lastErr error +} + +func (o *branchPromotionOutput) Text(s string) { o.message = s } +func (o *branchPromotionOutput) Error(err error) { o.lastErr = err } + +func TestRunBranchFastForward(t *testing.T) { + cat := newBranchPromotionTestCatalog(t) + out := &branchPromotionOutput{} + runBranchFastForward(context.Background(), out, cat, &BranchFastForwardCmd{ + TableID: "db.events", BranchName: "feature-branch", SourceRef: "audit", Yes: true, + }) + require.NoError(t, out.lastErr) + assert.Equal(t, int64(6000), cat.meta.SnapshotByName("feature-branch").SnapshotID) + assert.Equal(t, int64(6000), cat.meta.SnapshotByName("audit").SnapshotID) + assert.True(t, strings.Contains(out.message, "6000")) Review Comment: nit: `assert.Contains(t, out.message, "6000")` does the same and prints the message when it fails. strings import can go. ########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,144 @@ +// 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 ( + "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 snapshot ID fails Commit. REST +// catalogs cannot check ref type or retention changes that leave the ID +// unchanged. Avoid changing the target ref concurrently with this operation. +// 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. +// +// Both ref heads are pinned to their base snapshot IDs, so concurrent head +// moves fail the commit. As with FastForwardBranch, REST requirements cannot +// detect changes to ref type or retention when its snapshot ID is unchanged. +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) + } + if sourceRef.Type != BranchRef && sourceRef.Type != TagRef { + return fmt.Errorf("%w: source ref %q has invalid type %q", iceberg.ErrInvalidArgument, source, sourceRef.Type) + } + if _, err := meta.SnapshotByID(sourceRef.SnapshotID); err != nil { + return fmt.Errorf("source ref %q points to an unknown snapshot: %w", source, err) + } + + 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", + iceberg.ErrInvalidArgument, target, targetRef.SnapshotID, source, sourceRef.SnapshotID) + } Review Comment: With `snapshot-loading-mode=refs`, this rejects a valid fast-forward. The server sends only the snapshots that refs point to, so for an audit branch with two or more commits, the parent walk stops early and reports "not an ancestor". `ReplaceBranch` says unknown snapshot too. Java loads the full history here. Could you document this limit on these methods and mention refs mode in both errors? ########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,144 @@ +// 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 ( + "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 snapshot ID fails Commit. REST +// catalogs cannot check ref type or retention changes that leave the ID +// unchanged. Avoid changing the target ref concurrently with this operation. +// 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. +// +// Both ref heads are pinned to their base snapshot IDs, so concurrent head +// moves fail the commit. As with FastForwardBranch, REST requirements cannot +// detect changes to ref type or retention when its snapshot ID is unchanged. +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) + } + if sourceRef.Type != BranchRef && sourceRef.Type != TagRef { + return fmt.Errorf("%w: source ref %q has invalid type %q", iceberg.ErrInvalidArgument, source, sourceRef.Type) + } + if _, err := meta.SnapshotByID(sourceRef.SnapshotID); err != nil { + return fmt.Errorf("source ref %q points to an unknown snapshot: %w", source, err) + } + + 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", + iceberg.ErrInvalidArgument, target, targetRef.SnapshotID, source, sourceRef.SnapshotID) + } + } + + update := meta.NewRetainingSnapshotRefUpdate(target, sourceRef.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, requirements, []string{target, source}) +} + +// ReplaceBranch sets the named branch to an existing snapshot without an +// ancestry restriction. Unlike FastForwardBranch, this can move a branch +// backwards or onto a different lineage. The branch must already exist; its +// retention settings are preserved from the transaction's metadata. +// +// Commit checks the target's snapshot ID. REST catalogs cannot check ref type +// or retention changes that leave the ID unchanged. Avoid changing the target +// ref concurrently with this operation. +func (t *Transaction) ReplaceBranch(name string, snapshotID int64) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if name == "" { + return fmt.Errorf("%w: branch name must not be empty", iceberg.ErrInvalidArgument) + } + existing, ok := meta.refs[name] + if !ok { + return fmt.Errorf("%w: branch %q does not exist", iceberg.ErrInvalidArgument, name) + } + if existing.Type != BranchRef { + return fmt.Errorf("%w: ref %q is a %s, not a branch", iceberg.ErrInvalidArgument, name, existing.Type) + } + if existing.SnapshotID == snapshotID { + // A no-op still fences a co-staged metadata update. + return t.applyPinned(nil, + []Requirement{assertBranchRefSnapshotID(name, t.baseRefSnapshotID(name))}, + []string{name}) + } + if _, err := meta.SnapshotByID(snapshotID); err != nil { + return fmt.Errorf("%w: cannot replace branch %q with unknown snapshot %d: %v", + iceberg.ErrInvalidArgument, name, snapshotID, err) + } + + update := meta.NewRetainingSnapshotRefUpdate(name, snapshotID, BranchRef) + + return t.applyPinned([]Update{update}, + []Requirement{assertBranchRefSnapshotID(name, t.baseRefSnapshotID(name))}, + []string{name}) Review Comment: These methods now pin refs, but `concurrent-writes.md` and the comments on `Transaction.pinnedRefs` and `rewriteRefSnapshotRequirements` don't list them. Readers also won't know that a pin stops retries for other writes: `AppendTable` then `FastForwardBranch("qa", "main")` fails if someone appends to main, while the same append alone retries fine. Could you update those three places and say this in the method docs? -- 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]
