badalprasadsingh commented on code in PR #2384: URL: https://github.com/apache/iceberg-go/pull/2384#discussion_r4527263743
########## table/manage_snapshot_refs_retry_test.go: ########## @@ -0,0 +1,404 @@ +// 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" + "testing" + + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A source snapshot staged by this transaction must not be rebuilt onto a +// different commit-branch history after fast-forward ancestry was checked. +func TestBranchPromotionPinsCommitBranchForStagedSource(t *testing.T) { + for _, tc := range []struct { + name string + stage func(*Transaction) error + }{ + {"fast-forward", func(tx *Transaction) error { + return tx.FastForwardBranch("qa", "audit") + }}, + {"replace-from-ref", func(tx *Transaction) error { + return tx.ReplaceBranchFromRef("qa", "audit") + }}, + } { + t.Run(tc.name, func(t *testing.T) { + base := manageSnapshotRefsRetryFixture(t) + peer, err := MetadataBuilderFromBase(base, "") + require.NoError(t, err) + // Another writer rolls main back to a divergent snapshot. + require.NoError(t, peer.SetSnapshotRef(MainBranch, 900, BranchRef)) + current, err := peer.Build() + require.NoError(t, err) + + tbl, cat := newAssertRefTestTable(t, base, current) + tx := tbl.NewTransaction() + parent := int64(100) + snap := &Snapshot{ + SnapshotID: 1001, + ParentSnapshotID: &parent, + SequenceNumber: base.LastSequenceNumber() + 1, + TimestampMs: base.LastUpdatedMillis() + 2000, + ManifestList: "staged-source.avro", + Summary: &Summary{Operation: OpAppend}, + } + add := NewAddSnapshotUpdate(snap) + add.rebuildManifestList = func(_ context.Context, fresh Metadata, head *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) { + require.NotNil(t, head) + rebuilt := *snap + id := head.SnapshotID + rebuilt.ParentSnapshotID = &id + rebuilt.SequenceNumber = fresh.LastSequenceNumber() + 1 + rebuilt.ManifestList = "rebuilt-staged-source.avro" + + return &rebuilt, nil + } + require.NoError(t, tx.apply([]Update{ + add, + tx.meta.NewRetainingSnapshotRefUpdate(MainBranch, snap.SnapshotID, BranchRef), + }, nil)) + require.NoError(t, tx.ReplaceBranch("audit", snap.SnapshotID)) + require.NoError(t, tc.stage(tx)) + + _, err = tx.Commit(t.Context()) + require.ErrorIs(t, err, ErrTransactionUnusable) + require.NotErrorIs(t, err, ErrCommitFailed) Review Comment: P1: ci fails here on every run: a moved pinned ref still wraps ErrCommitFailed, as "main target moved" expects, so NotErrorIs fails better: expect ErrCommitFailed here instead, keeping the ErrTransactionUnusable check is fine ########## table/manage_snapshot_refs.go: ########## @@ -0,0 +1,170 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "errors" + "fmt" + + "github.com/apache/iceberg-go" +) + +// FastForwardBranch advances target to the snapshot referenced by source. +// Source can be a branch or tag and is never changed. If target does not +// exist, it is created with default retention settings. If target exists, +// it must be a branch and its snapshot must be an ancestor of source's +// snapshot. The target keeps its retention settings. +// +// Both refs are pinned to their snapshot IDs in the transaction's base +// metadata. A concurrent change to either ID fails Commit. This also stops +// the normal retry of a co-staged append if a pinned ref moves. When the +// source names a snapshot staged in this transaction, the commit branch is +// pinned too, so retry cannot change the source snapshot's ancestry. No catalog's +// snapshot-ID requirement detects retention-only changes, and REST catalogs +// also cannot detect a ref type change with the same snapshot ID. +// Fast-forward needs the full ancestor chain. With a REST catalog configured +// with snapshot-loading-mode=refs, a valid fast-forward can be rejected; +// reload with snapshot-loading-mode=all before calling this method. +// For non-main targets, use Table.NewTransactionOnBranch(target) so an +// unrelated main head move does not conflict with the transaction's implicit +// commit-branch requirement. +// Calls after other ref changes in this transaction use the staged ref heads. +// A transaction without updates does not contact the catalog. +func (t *Transaction) FastForwardBranch(target, source string) error { + return t.moveBranchToRef(target, source, true) +} + +// ReplaceBranchFromRef moves target to the snapshot named by source without +// checking ancestry, matching Java's ManageSnapshots.replaceBranch(from, to). +// Source may be a branch or tag and is left unchanged. If target is absent, +// it is created as a branch with default retention; otherwise the target +// branch retains its existing retention settings. +// For non-main targets, use Table.NewTransactionOnBranch(target) to avoid +// requiring main to remain unchanged during the commit. +// +// Both ref heads are pinned to their base snapshot IDs, so concurrent head +// moves fail the commit and prevent the normal retry of a co-staged append. +// A source snapshot staged in this transaction also pins the commit branch, +// preventing retry from rebuilding that snapshot on a different lineage. +// No catalog's snapshot-ID requirement detects retention-only changes, and +// REST catalogs also cannot detect ref type changes with the same snapshot ID. +func (t *Transaction) ReplaceBranchFromRef(target, source string) error { + return t.moveBranchToRef(target, source, false) +} + +func (t *Transaction) moveBranchToRef(target, source string, fastForward bool) error { + meta, err := t.txnMeta() + if err != nil { + return err + } + if target == "" || source == "" { + return fmt.Errorf("%w: target and source ref names must not be empty", iceberg.ErrInvalidArgument) + } + + sourceRef, ok := meta.refs[source] + if !ok { + return fmt.Errorf("%w: source ref %q does not exist", iceberg.ErrInvalidArgument, source) + } + + targetRef, exists := meta.refs[target] + if exists && targetRef.Type != BranchRef { + return fmt.Errorf("%w: target ref %q is a %s, not a branch", iceberg.ErrInvalidArgument, target, targetRef.Type) + } + requirements := []Requirement{assertBranchRefSnapshotID(target, t.baseRefSnapshotID(target))} + if source != target { + requirements = append(requirements, AssertRefSnapshotID(source, t.baseRefSnapshotID(source))) + } + refsToPin := []string{target, source} + for _, staged := range meta.updates { + if added, ok := staged.(*addSnapshotUpdate); ok && added.Snapshot != nil && + added.Snapshot.SnapshotID == sourceRef.SnapshotID { + // Retry can rebuild this source on a different commit-branch head, + // invalidating the ancestry check performed at staging time. + refsToPin = append(refsToPin, t.branch) + + break + } + } + 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, refsToPin) + } + lookup := func(id int64) *Snapshot { + snap, _ := meta.SnapshotByID(id) + + return snap + } + if fastForward && !IsAncestorOf(sourceRef.SnapshotID, targetRef.SnapshotID, lookup) { + return fmt.Errorf("%w: branch %q snapshot %d is not an ancestor of %q snapshot %d (REST snapshot-loading-mode=refs may omit ancestors; reload with snapshot-loading-mode=all)", + iceberg.ErrInvalidArgument, target, targetRef.SnapshotID, source, sourceRef.SnapshotID) + } + } + + update := meta.NewRetainingSnapshotRefUpdate(target, sourceRef.SnapshotID, BranchRef) + + return t.applyPinned([]Update{update}, requirements, refsToPin) +} + +// 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. +// For non-main targets, use Table.NewTransactionOnBranch(name) to avoid +// requiring main to remain unchanged during the commit. +// +// Commit pins the target's snapshot ID, preventing the normal retry of a +// co-staged append when the branch moves. No catalog's snapshot-ID requirement +// detects retention-only changes, and REST catalogs also miss ref type changes +// with the same ID. A REST catalog using snapshot-loading-mode=refs may omit a +// requested historical snapshot; reload with snapshot-loading-mode=all. +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}) + } + + update := meta.NewRetainingSnapshotRefUpdate(name, snapshotID, BranchRef) + + err = t.applyPinned([]Update{update}, + []Requirement{assertBranchRefSnapshotID(name, t.baseRefSnapshotID(name))}, + []string{name}) Review Comment: ReplaceBranch to a snapshot staged here has no commit-branch pin. After a peer rolls back main, it commits the rebuilt snapshot, while ReplaceBranchFromRef now fails. better: share the staged check in a helper, pin `t.branch` here too, and add a test ########## table/manage_snapshot_refs_retry_test.go: ########## @@ -0,0 +1,404 @@ +// 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" + "testing" + + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A source snapshot staged by this transaction must not be rebuilt onto a +// different commit-branch history after fast-forward ancestry was checked. +func TestBranchPromotionPinsCommitBranchForStagedSource(t *testing.T) { Review Comment: this only covers the rejection, no test checks that promoting an unstaged source still retries when main moves, so pinning `t.branch` every time would go unnoticed better: add that case, expecting two attempts and qa promoted -- 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]
