badalprasadsingh commented on code in PR #2382:
URL: https://github.com/apache/iceberg-go/pull/2382#discussion_r4479162296
##########
catalog/rest/rest.go:
##########
@@ -2014,7 +2014,9 @@ func (r *Catalog) loadTableWithMode(ctx context.Context,
identifier table.Identi
vended := resolveStorageCredentials(ret.StorageCredentials,
ret.MetadataLoc)
mergeVendedCredentials(config, vended, ret.MetadataLoc)
- return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved),
table.WithScanPlanningDirective(r.scanPlanningDirective(ctx, identifier,
ret.Config)))
+ return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved),
+ table.WithScanPlanningDirective(r.scanPlanningDirective(ctx,
identifier, ret.Config)),
+ table.WithSnapshotHistoryLimitedToRefs(mode ==
SnapshotModeRefs))
Review Comment:
P1: `RefreshTableCredentials` builds a new table without this flag but keeps
the same partial metadata. So after a credential refresh the refs-mode check is
off, and picking an already applied source can bring deleted rows back.
please add a `WithSnapshotHistoryLimitedToRefsFrom(tbl)` option, like
`WithScanPlanningDirectiveFrom`, and pass it there?
##########
table/transaction.go:
##########
@@ -2623,7 +2645,13 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
case errors.As(err, &unknown):
t.committed = true
default:
- t.failedCommits++
+ if t.noResume {
+ // A resumed cherry-pick fast-forward
could lose the
+ // moved snapshot's files when
rebuilding later appends.
+ t.unusable = true
Review Comment:
the txn can't be used again here, but this error matches only
`ErrCommitFailed`, whose doc says committing again retries.
please wrap it with `ErrTransactionUnusable`, so callers can tell. also, add
this case to the docs on both errors and Commit.
##########
table/cherry_pick.go:
##########
@@ -0,0 +1,270 @@
+// 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.
+// Divergent replace-partitions overwrites also require ancestry, removed-file,
+// and changed-partition validation. Both follow-ups are tracked in
+// https://github.com/apache/iceberg-go/issues/2415.
+//
+// snapshotProps applies only when a divergent append is replayed. A
fast-forward
+// reuses the source snapshot unchanged and ignores these properties. On
replay,
+// source-snapshot-id and published-wap-id are derived from the source, while a
+// caller-supplied wap.id is discarded so the result is not staged.
+//
+// The target branch is pinned to its base head, and a new target branch also
+// pins main because its initial snapshot inherits main's files. A concurrent
+// move of either guarded ref fails the commit instead of rebasing.
Fast-forward
+// transactions do not retry CAS conflicts because a later staged append must
+// not lose the fast-forwarded files during manifest-list reconstruction.
+//
+// Duplicate-publication checks need complete ancestry when REST loaded only
+// referenced snapshots (snapshot-loading-mode=refs). Reload in mode=all before
+// publishing from an incomplete refs-only view. Normal snapshot expiration may
+// remove old ancestors and does not by itself prevent cherry-picking.
+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 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 && t.tbl.snapshotHistoryRefsOnly {
+ return fmt.Errorf("%w: cannot cherry-pick snapshot %d
onto branch %q with incomplete target ancestry under
snapshot-loading-mode=refs; reload with snapshot-loading-mode=all",
+ 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)
+ baseHead := t.baseRefSnapshotID(target)
+ req := AssertRefSnapshotID(target, baseHead)
+ if baseHead != nil {
+ req = assertBranchRefSnapshotID(target, baseHead)
+ }
+ reqs := []Requirement{req}
+ pins := []string{target}
+ if target != MainBranch && baseHead == nil {
+ // A new branch forks from main, so a concurrent main
move
+ // invalidates the source's directly-applicable
ancestry.
+ reqs = append(reqs, AssertRefSnapshotID(MainBranch,
t.baseRefSnapshotID(MainBranch)))
+ pins = append(pins, MainBranch)
+ }
Review Comment:
this check isn not needed, a fast-forwarded new branch points at the source
itself, so nothing forks from main, and main moving can't change its files. It
only fails good commits.
remove it and `TestCherryPickSnapshotFastForwardToMissingBranchPinsMain`?
the replay check stays, upto you.
##########
table/cherry_pick_internal_test.go:
##########
@@ -0,0 +1,110 @@
+// 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"
+ "errors"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestCherryPickSnapshotRejectsMissingSnapshot(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+ err := txn.CherryPickSnapshot(t.Context(), 99999, nil)
+ require.ErrorIs(t, err, ErrSnapshotNotFound)
+ require.NotContains(t, err.Error(), "snapshot-loading-mode=refs")
+}
+
+func TestCherryPickSnapshotRejectsNonAppendOperation(t *testing.T) {
+ for _, op := range []Operation{OpDelete, OpOverwrite, OpReplace} {
+ t.Run(string(op), func(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+ // The source is a direct child of main in the shared
fixture,
+ // which Java allows to fast-forward for any operation.
Move main
+ // onto a divergent lineage to exercise unsupported
replay.
+ source := txn.meta.snapshotList[1]
+ require.NoError(t, txn.meta.AddSnapshot(&Snapshot{
+ SnapshotID: 30,
+ ParentSnapshotID: new(int64(10)),
+ SequenceNumber: 3,
+ ManifestList:
"mem://default/table-location/metadata/manifest-30.avro",
+ Summary: &Summary{Operation: OpAppend},
+ TimestampMs: source.TimestampMs + 1,
+ }))
+ require.NoError(t, txn.meta.SetSnapshotRef(MainBranch,
30, BranchRef))
+ source.Summary.Operation = op
+ err := txn.CherryPickSnapshot(t.Context(), 20, nil)
+ require.ErrorIs(t, err, ErrInvalidOperation)
+ })
+ }
+}
+
+// REST snapshot-loading-mode=refs can provide both ref heads while omitting
+// the snapshots between them. An incomplete walk cannot prove that the
+// source snapshot or its WAP ID was never published on the target branch.
+func TestCherryPickSnapshotRejectsIncompleteTargetAncestry(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+ txn.tbl.snapshotHistoryRefsOnly = true
Review Comment:
This test sets the flag by hand, so no test checks that REST sets it.
Nothing covers `loadTableWithMode`, `Table.Refresh`, `doCommit` or
`RefreshTableCredentials` carrying it.
please add a catalog/rest test that loads in refs mode and checks the pick
is rejected after each?
##########
website/src/concurrent-writes.md:
##########
@@ -78,10 +78,14 @@ iceberg-go can refresh the table and replay the operation
against the latest
snapshot between attempts. On each retry the table metadata is reloaded, the
producer's validators re-run against the fresh snapshot, and the commit is
re-submitted only if it is still valid. `RollbackToSnapshot`, `ExpireSnapshots`
-and `AssertRefSnapshotID` pin the refs they guard, so a retry does not rebase
-those assertions. If a guarded ref has moved, the commit fails with
+and `AssertRefSnapshotID`, as well as `CherryPickSnapshot`, pin the refs
+they guard, so a retry does not rebase those assertions. If a guarded ref has
moved, the commit fails with
`ErrCommitFailed` instead of retrying. An `ExpireSnapshots` that removes
nothing
-adds no guards.
+adds no guards. A replayed cherry-pick pins its target branch; a new target
+also pins main because it forks from main. A fast-forwarded cherry-pick is
+non-replayable if another writer changes the table, even when the target branch
+does not move, because later snapshots in the same transaction must retain
+the fast-forwarded files. Rebuild the transaction from a fresh table on
conflict.
Review Comment:
this reads as if any other writer fails a fast-forward, but alone it still
commits when another branch moves.
what's true is that such a txn never retries a conflict. Could you say that,
and update the noReplay comments in transaction.go and table.go?
##########
table/transaction.go:
##########
@@ -90,13 +90,19 @@ type Transaction struct {
// noReplay marks the commit as non-replayable: on a CAS conflict
// doCommit fails instead of refreshing and replaying. Set by
- // snapshot producers whose staged snapshot carries delete-file
- // removals; see the flag site in commitManifests.
+ // snapshot producers whose staged snapshot carries delete-file removals
+ // and by cherry-pick fast-forwards that precede later staged writes.
noReplay bool
+ // noResume is set for staged cherry-pick fast-forwards. An automatic
+ // retry is already disabled by noReplay, but resuming a failed Commit
+ // could rebuild later snapshots against the head before the ref move.
+ // Such a transaction must be reconstructed after any failed commit.
+ noResume bool
Review Comment:
as zeroshade noted, this limit can go once #2384 lands, along with
`applyPinnedNoReplay` and the hint.
#2415 doesn't list that yet. Could you add it there and cite it here?
##########
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:
thanks, most of these are in, still missing: a source that adds two files, a
fast-forward keeping the branch's retention settings, and a pick onto an
existing branch while main moves, which should succeed
can you please add them?
--
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]
