badalprasadsingh commented on code in PR #2382: URL: https://github.com/apache/iceberg-go/pull/2382#discussion_r4459163339
########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + + source, err := meta.SnapshotByID(snapshotID) + if err != nil { + return fmt.Errorf("cherry-pick source snapshot %d may be unavailable with snapshot-loading-mode=refs: %w", + snapshotID, err) + } + // Check both direct ancestry and earlier cherry-picks. A later snapshot + // may have removed the source files, so checking live paths is not enough. + wapID := "" + if source.Summary != nil { + wapID = source.Summary.Properties["wap.id"] + } + head := t.planningSnapshot(meta) + if head != nil { + lookup := func(id int64) *Snapshot { + snapshot, _ := meta.SnapshotByID(id) + + return snapshot + } + ancestors, complete := AncestorsOfChecked(head.SnapshotID, lookup) + for _, ancestor := range ancestors { + if ancestor.SnapshotID == snapshotID || + (ancestor.Summary != nil && ancestor.Summary.Properties["source-snapshot-id"] == strconv.FormatInt(snapshotID, 10)) { + return fmt.Errorf("%w: snapshot %d has already been applied to target branch %q", ErrInvalidOperation, snapshotID, t.branch) + } + if wapID != "" && ancestor.Summary != nil && + (ancestor.Summary.Properties["wap.id"] == wapID || ancestor.Summary.Properties["published-wap-id"] == wapID) { + return fmt.Errorf("%w: WAP ID %q has already been published", ErrInvalidOperation, wapID) + } + } + if !complete { + return fmt.Errorf("%w: cannot cherry-pick snapshot %d onto branch %q with incomplete target ancestry; snapshot-loading-mode=refs may omit snapshots, retry with snapshot-loading-mode=all if history has not expired", + ErrInvalidOperation, snapshotID, t.branch) + } + } + + // Java's CherryPickOperation returns the source snapshot itself when its + // parent matches the target head. This is valid for all operations and + // all table versions: no file is copied, so row lineage stays intact. + // A fast-forward intentionally has no new snapshot summary properties. + fastForward := head == nil && source.ParentSnapshotID == nil + if head != nil && source.ParentSnapshotID != nil { + fastForward = *source.ParentSnapshotID == head.SnapshotID + } + if fastForward { + target := t.branch + if target == "" { + target = MainBranch + } + if ref, ok := meta.refs[target]; ok && ref.Type != BranchRef { + return fmt.Errorf("%w: target ref %q is not a branch", ErrInvalidOperation, target) + } + update := meta.NewRetainingSnapshotRefUpdate(target, source.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, + []Requirement{assertBranchRefSnapshotID(target, t.baseRefSnapshotID(target))}, + []string{target}) Review Comment: P0: a retry can undo this fast-forward. If a txn picks A, then picks B or appends, while someone commits to another branch, commit succeeds but main loses A's rows. Your #2384 fixes rebuildSnapshotUpdates. could you land this after it, and add a test here? ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + + source, err := meta.SnapshotByID(snapshotID) + if err != nil { + return fmt.Errorf("cherry-pick source snapshot %d may be unavailable with snapshot-loading-mode=refs: %w", + snapshotID, err) + } + // Check both direct ancestry and earlier cherry-picks. A later snapshot + // may have removed the source files, so checking live paths is not enough. + wapID := "" + if source.Summary != nil { + wapID = source.Summary.Properties["wap.id"] + } + head := t.planningSnapshot(meta) + if head != nil { + lookup := func(id int64) *Snapshot { + snapshot, _ := meta.SnapshotByID(id) + + return snapshot + } + ancestors, complete := AncestorsOfChecked(head.SnapshotID, lookup) + for _, ancestor := range ancestors { + if ancestor.SnapshotID == snapshotID || + (ancestor.Summary != nil && ancestor.Summary.Properties["source-snapshot-id"] == strconv.FormatInt(snapshotID, 10)) { + return fmt.Errorf("%w: snapshot %d has already been applied to target branch %q", ErrInvalidOperation, snapshotID, t.branch) + } + if wapID != "" && ancestor.Summary != nil && + (ancestor.Summary.Properties["wap.id"] == wapID || ancestor.Summary.Properties["published-wap-id"] == wapID) { + return fmt.Errorf("%w: WAP ID %q has already been published", ErrInvalidOperation, wapID) + } + } + if !complete { + return fmt.Errorf("%w: cannot cherry-pick snapshot %d onto branch %q with incomplete target ancestry; snapshot-loading-mode=refs may omit snapshots, retry with snapshot-loading-mode=all if history has not expired", + ErrInvalidOperation, snapshotID, t.branch) + } Review Comment: P1: this blocks every table with expired history, not just refs mode, expiry leaves the oldest snapshot's parent missing, so complete stays false forever, and both fast-forward and replay fail on any catalog. Can you reject only tables REST loaded in refs mode? ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + + source, err := meta.SnapshotByID(snapshotID) + if err != nil { + return fmt.Errorf("cherry-pick source snapshot %d may be unavailable with snapshot-loading-mode=refs: %w", + snapshotID, err) + } + // Check both direct ancestry and earlier cherry-picks. A later snapshot + // may have removed the source files, so checking live paths is not enough. + wapID := "" + if source.Summary != nil { + wapID = source.Summary.Properties["wap.id"] + } + head := t.planningSnapshot(meta) + if head != nil { + lookup := func(id int64) *Snapshot { + snapshot, _ := meta.SnapshotByID(id) + + return snapshot + } + ancestors, complete := AncestorsOfChecked(head.SnapshotID, lookup) + for _, ancestor := range ancestors { + if ancestor.SnapshotID == snapshotID || + (ancestor.Summary != nil && ancestor.Summary.Properties["source-snapshot-id"] == strconv.FormatInt(snapshotID, 10)) { + return fmt.Errorf("%w: snapshot %d has already been applied to target branch %q", ErrInvalidOperation, snapshotID, t.branch) + } + if wapID != "" && ancestor.Summary != nil && + (ancestor.Summary.Properties["wap.id"] == wapID || ancestor.Summary.Properties["published-wap-id"] == wapID) { + return fmt.Errorf("%w: WAP ID %q has already been published", ErrInvalidOperation, wapID) + } + } + if !complete { + return fmt.Errorf("%w: cannot cherry-pick snapshot %d onto branch %q with incomplete target ancestry; snapshot-loading-mode=refs may omit snapshots, retry with snapshot-loading-mode=all if history has not expired", + ErrInvalidOperation, snapshotID, t.branch) + } + } + + // Java's CherryPickOperation returns the source snapshot itself when its + // parent matches the target head. This is valid for all operations and + // all table versions: no file is copied, so row lineage stays intact. + // A fast-forward intentionally has no new snapshot summary properties. + fastForward := head == nil && source.ParentSnapshotID == nil + if head != nil && source.ParentSnapshotID != nil { + fastForward = *source.ParentSnapshotID == head.SnapshotID + } + if fastForward { + target := t.branch + if target == "" { + target = MainBranch + } + if ref, ok := meta.refs[target]; ok && ref.Type != BranchRef { + return fmt.Errorf("%w: target ref %q is not a branch", ErrInvalidOperation, target) + } + update := meta.NewRetainingSnapshotRefUpdate(target, source.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, + []Requirement{assertBranchRefSnapshotID(target, t.baseRefSnapshotID(target))}, + []string{target}) + } + + if source.Summary == nil || source.Summary.Operation != OpAppend { + return fmt.Errorf("%w: snapshot %d is not an append and cannot be replayed onto a divergent branch", + ErrInvalidOperation, snapshotID) + } + if meta.formatVersion >= 3 { + return fmt.Errorf("%w: replaying v3 append snapshots requires row ID reassignment", + iceberg.ErrNotImplemented) + } Review Comment: iceberg-java replays v3 appends with new row IDs, and replace-partitions overwrites too. Rejecting both for now is fair, but nothing tracks it. can you file an issue for each and cite them here and in the method doc? ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { Review Comment: on a fast-forward, `snapshotProps` are silently dropped. On a replay, `source-snapshot-id`, `wap.id` and `published-wap-id` are replaced or removed. The method doc says none of this. Can you document it? Also, iceberg-java's cherrypick takes no properties, so removing this parameter also works. Upto you. ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + + source, err := meta.SnapshotByID(snapshotID) + if err != nil { + return fmt.Errorf("cherry-pick source snapshot %d may be unavailable with snapshot-loading-mode=refs: %w", + snapshotID, err) + } + // Check both direct ancestry and earlier cherry-picks. A later snapshot + // may have removed the source files, so checking live paths is not enough. + wapID := "" + if source.Summary != nil { + wapID = source.Summary.Properties["wap.id"] + } + head := t.planningSnapshot(meta) + if head != nil { + lookup := func(id int64) *Snapshot { + snapshot, _ := meta.SnapshotByID(id) + + return snapshot + } + ancestors, complete := AncestorsOfChecked(head.SnapshotID, lookup) + for _, ancestor := range ancestors { + if ancestor.SnapshotID == snapshotID || + (ancestor.Summary != nil && ancestor.Summary.Properties["source-snapshot-id"] == strconv.FormatInt(snapshotID, 10)) { + return fmt.Errorf("%w: snapshot %d has already been applied to target branch %q", ErrInvalidOperation, snapshotID, t.branch) + } + if wapID != "" && ancestor.Summary != nil && + (ancestor.Summary.Properties["wap.id"] == wapID || ancestor.Summary.Properties["published-wap-id"] == wapID) { + return fmt.Errorf("%w: WAP ID %q has already been published", ErrInvalidOperation, wapID) + } + } + if !complete { + return fmt.Errorf("%w: cannot cherry-pick snapshot %d onto branch %q with incomplete target ancestry; snapshot-loading-mode=refs may omit snapshots, retry with snapshot-loading-mode=all if history has not expired", + ErrInvalidOperation, snapshotID, t.branch) + } + } + + // Java's CherryPickOperation returns the source snapshot itself when its + // parent matches the target head. This is valid for all operations and + // all table versions: no file is copied, so row lineage stays intact. + // A fast-forward intentionally has no new snapshot summary properties. + fastForward := head == nil && source.ParentSnapshotID == nil + if head != nil && source.ParentSnapshotID != nil { + fastForward = *source.ParentSnapshotID == head.SnapshotID + } + if fastForward { + target := t.branch + if target == "" { + target = MainBranch + } + if ref, ok := meta.refs[target]; ok && ref.Type != BranchRef { + return fmt.Errorf("%w: target ref %q is not a branch", ErrInvalidOperation, target) + } + update := meta.NewRetainingSnapshotRefUpdate(target, source.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, + []Requirement{assertBranchRefSnapshotID(target, t.baseRefSnapshotID(target))}, + []string{target}) + } + + if source.Summary == nil || source.Summary.Operation != OpAppend { + return fmt.Errorf("%w: snapshot %d is not an append and cannot be replayed onto a divergent branch", + ErrInvalidOperation, snapshotID) + } + if meta.formatVersion >= 3 { + return fmt.Errorf("%w: replaying v3 append snapshots requires row ID reassignment", + iceberg.ErrNotImplemented) + } + + props := make(iceberg.Properties, len(snapshotProps)+2) + maps.Copy(props, snapshotProps) + props["source-snapshot-id"] = strconv.FormatInt(snapshotID, 10) + // The cherry-picked snapshot is published, not staged. Never carry a + // caller-supplied WAP ID into its summary: a later publish could + // mistake it for an unpublished staged snapshot. + delete(props, "wap.id") + // Publication provenance must describe the source snapshot, not a + // caller-supplied marker that could falsely block a future WAP publish. + delete(props, "published-wap-id") + if wapID != "" { + props["published-wap-id"] = wapID + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return err + } + + // A source snapshot references its ancestors' files too. Select only + // entries added by this snapshot, not every live file in its manifests. + // A valid append snapshot never adds delete files or removes files. + added := make([]iceberg.DataFile, 0) + seen := make(map[string]struct{}) + // Manifests are immutable. An entry added by the source snapshot can + // only appear in a manifest created by that snapshot. Skip inherited + // manifests rather than scanning the entire table history. + manifests, err := source.Manifests(fs) + if err != nil { + return fmt.Errorf("read cherry-pick source snapshot %d: %w", snapshotID, err) + } + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return err + } + if manifest.SnapshotID() != snapshotID { + continue + } + for entry, err := range manifest.Entries(fs, false) { + if err != nil { + return fmt.Errorf("read cherry-pick source snapshot %d: %w", snapshotID, err) + } + if err := ctx.Err(); err != nil { + return err + } + if entry.SnapshotID() != snapshotID { + continue + } + switch entry.Status() { + case iceberg.EntryStatusDELETED: + return fmt.Errorf("%w: append snapshot %d contains a removed file %q", ErrInvalidOperation, snapshotID, entry.DataFile().FilePath()) + case iceberg.EntryStatusADDED: + file := entry.DataFile() + if file.ContentType() != iceberg.EntryContentData { + return fmt.Errorf("%w: append snapshot %d contains an added delete file %q", ErrInvalidOperation, snapshotID, file.FilePath()) + } + if _, exists := seen[file.FilePath()]; exists { + return fmt.Errorf("%w: source snapshot %d adds file %q more than once", iceberg.ErrInvalidArgument, snapshotID, file.FilePath()) + } + seen[file.FilePath()] = struct{}{} + added = append(added, file) + } + } + } + + // Unlike a normal fast append, replaying old files is not safe to rebase + // after the target branch moves. Recheck its original head on commit. + if err := t.AssertRefSnapshotID(""); err != nil { + return err + } Review Comment: here, `AddDataFiles` can still fail after this pin is set. If a caller handles that error and keeps the transaction, the pin stays, and a later commit fails when main moves, even though the pick never happened. we should pin after the add succeeds. ########## table/cherry_pick_test.go: ########## @@ -0,0 +1,309 @@ +// 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_test + +import ( + "strconv" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCherryPickSnapshotFromDivergentBranch(t *testing.T) { + tbl := newBranchPlanningTable(t, nil) Review Comment: no test covers a root snapshot fast-forwarded onto an empty target, `wap.id` on an ancestor, a missing target branch, or a merge-append source. Fixtures are also unpartitioned with one file per snapshot. can you add these, plus a partitioned source under an older spec? ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. Review Comment: This operation pins refs, but the docs don't list it. Could you add it to `website/src/concurrent-writes.md`, `website/src/api.md`, and the comments on `Transaction.pinnedRefs` and `rewriteRefSnapshotRequirements`? Please also say it fails, rather than retries, when the branch moves. ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + + source, err := meta.SnapshotByID(snapshotID) + if err != nil { + return fmt.Errorf("cherry-pick source snapshot %d may be unavailable with snapshot-loading-mode=refs: %w", + snapshotID, err) + } + // Check both direct ancestry and earlier cherry-picks. A later snapshot + // may have removed the source files, so checking live paths is not enough. + wapID := "" + if source.Summary != nil { + wapID = source.Summary.Properties["wap.id"] + } + head := t.planningSnapshot(meta) + if head != nil { + lookup := func(id int64) *Snapshot { + snapshot, _ := meta.SnapshotByID(id) + + return snapshot + } + ancestors, complete := AncestorsOfChecked(head.SnapshotID, lookup) + for _, ancestor := range ancestors { + if ancestor.SnapshotID == snapshotID || + (ancestor.Summary != nil && ancestor.Summary.Properties["source-snapshot-id"] == strconv.FormatInt(snapshotID, 10)) { + return fmt.Errorf("%w: snapshot %d has already been applied to target branch %q", ErrInvalidOperation, snapshotID, t.branch) + } + if wapID != "" && ancestor.Summary != nil && + (ancestor.Summary.Properties["wap.id"] == wapID || ancestor.Summary.Properties["published-wap-id"] == wapID) { + return fmt.Errorf("%w: WAP ID %q has already been published", ErrInvalidOperation, wapID) + } + } + if !complete { + return fmt.Errorf("%w: cannot cherry-pick snapshot %d onto branch %q with incomplete target ancestry; snapshot-loading-mode=refs may omit snapshots, retry with snapshot-loading-mode=all if history has not expired", + ErrInvalidOperation, snapshotID, t.branch) + } + } + + // Java's CherryPickOperation returns the source snapshot itself when its + // parent matches the target head. This is valid for all operations and + // all table versions: no file is copied, so row lineage stays intact. + // A fast-forward intentionally has no new snapshot summary properties. + fastForward := head == nil && source.ParentSnapshotID == nil + if head != nil && source.ParentSnapshotID != nil { + fastForward = *source.ParentSnapshotID == head.SnapshotID + } + if fastForward { + target := t.branch + if target == "" { + target = MainBranch + } + if ref, ok := meta.refs[target]; ok && ref.Type != BranchRef { + return fmt.Errorf("%w: target ref %q is not a branch", ErrInvalidOperation, target) + } + update := meta.NewRetainingSnapshotRefUpdate(target, source.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, + []Requirement{assertBranchRefSnapshotID(target, t.baseRefSnapshotID(target))}, + []string{target}) + } + + if source.Summary == nil || source.Summary.Operation != OpAppend { + return fmt.Errorf("%w: snapshot %d is not an append and cannot be replayed onto a divergent branch", + ErrInvalidOperation, snapshotID) + } + if meta.formatVersion >= 3 { + return fmt.Errorf("%w: replaying v3 append snapshots requires row ID reassignment", + iceberg.ErrNotImplemented) + } + + props := make(iceberg.Properties, len(snapshotProps)+2) + maps.Copy(props, snapshotProps) + props["source-snapshot-id"] = strconv.FormatInt(snapshotID, 10) + // The cherry-picked snapshot is published, not staged. Never carry a + // caller-supplied WAP ID into its summary: a later publish could + // mistake it for an unpublished staged snapshot. + delete(props, "wap.id") + // Publication provenance must describe the source snapshot, not a + // caller-supplied marker that could falsely block a future WAP publish. + delete(props, "published-wap-id") + if wapID != "" { + props["published-wap-id"] = wapID + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return err + } + + // A source snapshot references its ancestors' files too. Select only + // entries added by this snapshot, not every live file in its manifests. + // A valid append snapshot never adds delete files or removes files. + added := make([]iceberg.DataFile, 0) + seen := make(map[string]struct{}) + // Manifests are immutable. An entry added by the source snapshot can + // only appear in a manifest created by that snapshot. Skip inherited + // manifests rather than scanning the entire table history. + manifests, err := source.Manifests(fs) + if err != nil { + return fmt.Errorf("read cherry-pick source snapshot %d: %w", snapshotID, err) + } + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return err + } + if manifest.SnapshotID() != snapshotID { + continue + } + for entry, err := range manifest.Entries(fs, false) { + if err != nil { + return fmt.Errorf("read cherry-pick source snapshot %d: %w", snapshotID, err) + } + if err := ctx.Err(); err != nil { + return err + } + if entry.SnapshotID() != snapshotID { + continue + } + switch entry.Status() { + case iceberg.EntryStatusDELETED: + return fmt.Errorf("%w: append snapshot %d contains a removed file %q", ErrInvalidOperation, snapshotID, entry.DataFile().FilePath()) + case iceberg.EntryStatusADDED: + file := entry.DataFile() + if file.ContentType() != iceberg.EntryContentData { + return fmt.Errorf("%w: append snapshot %d contains an added delete file %q", ErrInvalidOperation, snapshotID, file.FilePath()) + } + if _, exists := seen[file.FilePath()]; exists { + return fmt.Errorf("%w: source snapshot %d adds file %q more than once", iceberg.ErrInvalidArgument, snapshotID, file.FilePath()) + } + seen[file.FilePath()] = struct{}{} + added = append(added, file) + } + } + } + + // Unlike a normal fast append, replaying old files is not safe to rebase + // after the target branch moves. Recheck its original head on commit. + if err := t.AssertRefSnapshotID(""); err != nil { + return err + } Review Comment: P1: this pin doesn't cover a branch that doesn't exist yet, because the new snapshot builds on main. If someone publishes the same source to main first, a retry lists that file twice. can you also pin main when `t.baseRefSnapshotID(t.branch)` is nil, and add a test? ########## table/cherry_pick.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "context" + "fmt" + "maps" + "strconv" + + "github.com/apache/iceberg-go" +) + +// CherryPickSnapshot applies the additions from an existing append snapshot to +// the transaction's target branch. The original snapshot and its branch are +// not modified. If the source directly follows the target head, the target +// fast-forwards to the existing source snapshot (including for non-append +// operations and v3). Otherwise, a supported append is replayed into a new +// snapshot and sequence number without copying the underlying data files. +// +// Source snapshots may come from a different branch or an unreferenced +// snapshot still present in table metadata. A source that is already an +// ancestor of the target branch is rejected to avoid replaying old changes. +// Only append snapshots can be replayed onto a divergent target. Other +// operations require the direct-parent fast-forward described above. +// +// Replaying file additions is currently limited to v1 and v2. In v3, +// copying files into another snapshot may reuse row IDs; fast-forwarding an +// existing snapshot is safe because the lineage is not duplicated. +// +// The target branch is pinned to its base head to prevent a concurrent append +// from introducing duplicate file references between validation and commit. +// A conflict must be retried with a new transaction and refreshed table. +// Duplicate-publication checks require complete target ancestry. Refs-only +// loads and expired ancestors can both truncate it. Reload with +// snapshot-loading-mode=all if the missing snapshots have not expired. +func (t *Transaction) CherryPickSnapshot(ctx context.Context, snapshotID int64, snapshotProps iceberg.Properties) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + + source, err := meta.SnapshotByID(snapshotID) + if err != nil { + return fmt.Errorf("cherry-pick source snapshot %d may be unavailable with snapshot-loading-mode=refs: %w", + snapshotID, err) Review Comment: this blames `snapshot-loading-mode=refs` for every missing source, even a wrong ID or a catalog with no such setting, like SQL. It also repeats the ID. Better: return the plain error here, since the method doc already explains refs mode. ########## table/cherry_pick_test.go: ########## @@ -0,0 +1,309 @@ +// 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_test + +import ( + "strconv" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCherryPickSnapshotFromDivergentBranch(t *testing.T) { + tbl := newBranchPlanningTable(t, nil) + tbl = appendRowsOnRef(t, tbl, table.MainBranch, 1) + base := tbl.CurrentSnapshot().SnapshotID + tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 2) + source := tbl.SnapshotByName(branchPlanningRef) + require.NotNil(t, source) + require.Equal(t, table.OpAppend, source.Summary.Operation) + + tbl = appendRowsOnRef(t, tbl, table.MainBranch, 3) + targetBefore := tbl.CurrentSnapshot().SnapshotID + require.NotEqual(t, base, targetBefore) + require.NotContains(t, tbl.Metadata().Properties(), table.DefaultNameMappingKey) + + tx := tbl.NewTransaction() + require.NoError(t, tx.CherryPickSnapshot(t.Context(), source.SnapshotID, iceberg.Properties{ + "cherry-picked": "yes", "published-wap-id": "forged", "source-snapshot-id": "forged", "wap.id": "forged", + })) + result, err := tx.Commit(t.Context()) + require.NoError(t, err) + + require.NotEqual(t, targetBefore, result.CurrentSnapshot().SnapshotID) + require.NotEqual(t, source.SnapshotID, result.CurrentSnapshot().SnapshotID) + require.Equal(t, table.OpAppend, result.CurrentSnapshot().Summary.Operation) + assert.Equal(t, "yes", result.CurrentSnapshot().Summary.Properties["cherry-picked"]) + assert.Equal(t, strconv.FormatInt(source.SnapshotID, 10), result.CurrentSnapshot().Summary.Properties["source-snapshot-id"]) + assert.NotContains(t, result.CurrentSnapshot().Summary.Properties, "published-wap-id", + "a snapshot without WAP must not acquire a caller-supplied publication marker") + assert.NotContains(t, result.CurrentSnapshot().Summary.Properties, "wap.id", + "the published cherry-pick snapshot must not carry a staged WAP marker") + assert.NotContains(t, result.Metadata().Properties(), table.DefaultNameMappingKey, + "cherry-picking existing files must not add a name mapping") + assert.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, result, table.MainBranch)) + assert.ElementsMatch(t, []int64{1, 2}, idsOnRef(t, result, branchPlanningRef)) + + // The source snapshot's inherited file must not be cherry-picked twice. + // Replaying the same source after a successful commit also fails rather + // than creating a snapshot with two live references to the same file. + duplicate := result.NewTransaction() + err = duplicate.CherryPickSnapshot(t.Context(), source.SnapshotID, nil) + require.ErrorContains(t, err, "already been applied") + + // Removing the original file must not make the same source replayable. + overwrite := result.NewTransaction() + reader := idRecordReader(t, result, 10) + require.NoError(t, overwrite.Overwrite(t.Context(), reader, nil)) + reader.Release() + result, err = overwrite.Commit(t.Context()) + require.NoError(t, err) + retry := result.NewTransaction() + err = retry.CherryPickSnapshot(t.Context(), source.SnapshotID, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "already been applied") +} + +func TestCherryPickSnapshotFastForwardsDirectChild(t *testing.T) { + tbl := appendRowsOnRef(t, newBranchPlanningTable(t, nil), table.MainBranch, 1) + tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 2) + source := tbl.SnapshotByName(branchPlanningRef) + require.NotNil(t, source) + require.Equal(t, tbl.CurrentSnapshot().SnapshotID, *source.ParentSnapshotID) + + tx := tbl.NewTransaction() + require.NoError(t, tx.CherryPickSnapshot(t.Context(), source.SnapshotID, + iceberg.Properties{"ignored-on-fast-forward": "true"})) + result, err := tx.Commit(t.Context()) + require.NoError(t, err) + require.Equal(t, source.SnapshotID, result.CurrentSnapshot().SnapshotID, + "fast-forward reuses the source snapshot ID") + assert.NotContains(t, result.CurrentSnapshot().Summary.Properties, "ignored-on-fast-forward") + assert.ElementsMatch(t, []int64{1, 2}, idsOnRef(t, result, table.MainBranch)) + assert.ElementsMatch(t, []int64{1, 2}, idsOnRef(t, result, branchPlanningRef)) +} + +func TestCherryPickSnapshotFastForwardsNonAppendSnapshot(t *testing.T) { + tbl := appendRowsOnRef(t, newBranchPlanningTable(t, nil), table.MainBranch, 1) + branchTxn, err := tbl.NewTransactionOnBranch(branchPlanningRef) + require.NoError(t, err) + reader := idRecordReader(t, tbl, 2) + require.NoError(t, branchTxn.Overwrite(t.Context(), reader, nil)) + reader.Release() + tbl, err = branchTxn.Commit(t.Context()) + require.NoError(t, err) + + source := tbl.SnapshotByName(branchPlanningRef) + require.NotNil(t, source) + require.Equal(t, table.OpOverwrite, source.Summary.Operation) + require.NotNil(t, source.ParentSnapshotID) + require.Equal(t, tbl.CurrentSnapshot().SnapshotID, *source.ParentSnapshotID) + + tx := tbl.NewTransaction() + require.NoError(t, tx.CherryPickSnapshot(t.Context(), source.SnapshotID, nil)) + result, err := tx.Commit(t.Context()) + require.NoError(t, err) + require.Equal(t, source.SnapshotID, result.CurrentSnapshot().SnapshotID) + assert.ElementsMatch(t, []int64{2}, idsOnRef(t, result, table.MainBranch)) +} + +func TestCherryPickSnapshotRejectsAncestor(t *testing.T) { + tbl := appendRowsOnRef(t, newBranchPlanningTable(t, nil), table.MainBranch, 1) + id := tbl.CurrentSnapshot().SnapshotID + + tx := tbl.NewTransaction() + err := tx.CherryPickSnapshot(t.Context(), id, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "already been applied") +} + +func TestCherryPickSnapshotAllowsEmptyAppend(t *testing.T) { + tbl := newBranchPlanningTable(t, nil) + tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 11) + oldBranch := tbl.SnapshotByName(branchPlanningRef) + require.NotNil(t, oldBranch) + + empty := tbl.NewTransaction() + require.NoError(t, empty.AddDataFilesByPath(t.Context(), nil, nil, false)) + tbl, err := empty.Commit(t.Context()) + require.NoError(t, err) + source := tbl.CurrentSnapshot() + require.NotNil(t, source) + require.Equal(t, table.OpAppend, source.Summary.Operation) + + // AddDataFilesByPath used to create the empty source snapshot installs + // name mapping. Remove it to verify cherry-picking does not re-add it. + removeMapping := tbl.NewTransaction() + require.NoError(t, removeMapping.RemoveProperties([]string{table.DefaultNameMappingKey})) + tbl, err = removeMapping.Commit(t.Context()) + require.NoError(t, err) + require.NotContains(t, tbl.Metadata().Properties(), table.DefaultNameMappingKey) + + tx, err := tbl.NewTransactionOnBranch(branchPlanningRef) + require.NoError(t, err) + require.NoError(t, tx.CherryPickSnapshot(t.Context(), source.SnapshotID, nil)) + result, err := tx.Commit(t.Context()) + require.NoError(t, err) + + newBranch := result.SnapshotByName(branchPlanningRef) + require.NotNil(t, newBranch) + assert.NotEqual(t, oldBranch.SnapshotID, newBranch.SnapshotID) + assert.Equal(t, oldBranch.SnapshotID, *newBranch.ParentSnapshotID) + assert.ElementsMatch(t, []int64{11}, idsOnRef(t, result, branchPlanningRef)) + assert.NotContains(t, result.Metadata().Properties(), table.DefaultNameMappingKey, + "empty cherry-pick must not change the name mapping") + + retry, err := result.NewTransactionOnBranch(branchPlanningRef) + require.NoError(t, err) + err = retry.CherryPickSnapshot(t.Context(), source.SnapshotID, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "already been applied") +} + +// A file already live on the target must not be added again even when the +// source snapshot itself has not been cherry-picked before. +func TestCherryPickSnapshotRejectsSourceFileAlreadyOnTarget(t *testing.T) { + tbl := appendRowsOnRef(t, newBranchPlanningTable(t, nil), table.MainBranch, 1) + mainPaths, _ := liveFilesOnRef(t, tbl, table.MainBranch) + require.Len(t, mainPaths, 1) + tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 2) + source := tbl.SnapshotByName(branchPlanningRef) + require.NotNil(t, source) + + branchPaths, _ := liveFilesOnRef(t, tbl, branchPlanningRef) + require.Len(t, branchPaths, 2) + var sourcePath string + for _, path := range branchPaths { + if path != mainPaths[0] { + sourcePath = path + + break + } + } + require.NotEmpty(t, sourcePath) + + // Import the same physical file directly, but not the source snapshot. + duplicate := tbl.NewTransaction() + require.NoError(t, duplicate.AddDataFilesByPath(t.Context(), []string{sourcePath}, nil, false)) + var err error + tbl, err = duplicate.Commit(t.Context()) + require.NoError(t, err) + + replay := tbl.NewTransaction() + err = replay.CherryPickSnapshot(t.Context(), source.SnapshotID, nil) + require.ErrorContains(t, err, "already referenced") + require.ElementsMatch(t, []int64{1, 2}, idsOnRef(t, tbl, table.MainBranch)) +} + +func TestCherryPickSnapshotRejectsConcurrentTargetMove(t *testing.T) { + tbl := appendRowsOnRef(t, newBranchPlanningTable(t, nil), table.MainBranch, 1) + tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 2) + source := tbl.SnapshotByName(branchPlanningRef) + require.NotNil(t, source) + + stale := tbl.NewTransaction() + require.NoError(t, stale.CherryPickSnapshot(t.Context(), source.SnapshotID, nil)) + + // A concurrent commit changes main after the cherry-pick validated its + // duplicate-file set. The pinned head forbids an unsafe automatic rebase. + tbl = appendRowsOnRef(t, tbl, table.MainBranch, 3) + _, err := stale.Commit(t.Context()) + require.ErrorIs(t, err, table.ErrCommitFailed) Review Comment: P1: this test can't catch a missing pin. Retries are off by default, and it takes the fast-forward path, so no test reaches the replay pin. can you set `CommitNumRetriesKey` and add one fast-forward and one replay case where another writer moves main first? -- 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]
