laskoviymishka commented on code in PR #1637:
URL: https://github.com/apache/iceberg-go/pull/1637#discussion_r3735881959
##########
table/metadata.go:
##########
@@ -390,6 +390,45 @@ func (b *MetadataBuilder) currentSnapshot() *Snapshot {
return s
}
+// currentSnapshotForRef resolves the parent for createSnapshotProducer and
mergeOverwrite.
+// An unknown branch falls back to currentSnapshot() so a new
+// branch forks from main — the opposite of currentSnapshotIDForRef,
+// which must return nil there so AssertRefSnapshotID can prove the branch is
absent;
+// never derive one from the other. A present-but-dangling ref also yields nil.
+func (b *MetadataBuilder) currentSnapshotForRef(ref string) *Snapshot {
+ if ref == "" || ref == MainBranch {
+ return b.currentSnapshot()
+ }
+
+ r, ok := b.refs[ref]
+ if !ok {
+ return b.currentSnapshot()
+ }
+
+ s, _ := b.SnapshotByID(r.SnapshotID)
+
+ return s
+}
+
+// currentSnapshotIDForRef returns the commit's AssertRefSnapshotID id (its
only caller).
+// An unknown branch returns nil so the requirement proves the branch is
absent —
+// unlike currentSnapshotForRef, whose parent lookup falls back to main's head
+// so a new branch can fork from it;
+// the two must not be conflated or new-branch creates get a false OCC
rejection.
+func (b *MetadataBuilder) currentSnapshotIDForRef(ref string) *int64 {
+ if ref == "" || ref == MainBranch {
+ return b.currentSnapshotID
Review Comment:
This is new with the assertion change, not something I raised last round.
None of the three helpers check `SnapshotRefType`, so a tag name passed as
`txn.branch` resolves like a branch. Pre-fix that was harmless by accident: the
requirement asserted main's head against the tag, so `Validate` rejected it.
Now it asserts the tag's own id, the check passes, and `commitManifests` calls
`NewRetainingSnapshotRefUpdate(branch, ..., BranchRef)` then `b.refs[name] =
ref`, so the tag advances to a new child snapshot and its immutability is gone.
Java rejects this up front in `SnapshotProducer.targetBranch()`:
`checkArgument(!refExists || base.ref(branch).isBranch(), "%s is a tag, not a
branch...")`.
I'd put the guard in `NewTransactionOnBranchWithError` so it fails at
construction rather than at commit, which also keeps all three helpers safe
without touching them.
##########
table/transaction_internal_test.go:
##########
@@ -44,6 +46,250 @@ func
TestTransactionApplyKeepsDistinctRequirementsOfSameType(t *testing.T) {
requireContainsRefSnapshotRequirement(t, txn.reqs, "feature",
&featureSnapshotID)
}
+func TestCurrentSnapshotForRefResolvesBranchHead(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+
+ main := txn.meta.currentSnapshotForRef(MainBranch)
+ require.NotNil(t, main)
+ require.Equal(t, int64(10), main.SnapshotID)
+
+ empty := txn.meta.currentSnapshotForRef("")
+ require.NotNil(t, empty)
+ require.Equal(t, int64(10), empty.SnapshotID, "empty ref must resolve
like main")
+
+ feature := txn.meta.currentSnapshotForRef("feature")
+ require.NotNil(t, feature)
+ require.Equal(t, int64(20), feature.SnapshotID, "feature branch must
resolve to its own head (20), not main (10)")
+
+ missing := txn.meta.currentSnapshotForRef("does-not-exist")
+ require.NotNil(t, missing)
+ require.Equal(t, int64(10), missing.SnapshotID, "a not-yet-created
branch falls back to main's head")
+}
+
+func TestCurrentSnapshotIDForRefResolvesBranchHead(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+
+ require.NotNil(t, txn.meta.currentSnapshotIDForRef(MainBranch))
+ require.Equal(t, int64(10),
*txn.meta.currentSnapshotIDForRef(MainBranch))
+
+ require.NotNil(t, txn.meta.currentSnapshotIDForRef("feature"))
+ require.Equal(t, int64(20),
*txn.meta.currentSnapshotIDForRef("feature"),
+ "feature branch assertion id must be the branch head (20), not
main (10)")
+
+ require.Nil(t, txn.meta.currentSnapshotIDForRef("does-not-exist"),
+ "a not-yet-created branch must assert non-existence (nil), not
main's head")
+}
+
+func TestCreateSnapshotProducerParentsOnBranchHead(t *testing.T) {
+ t.Run("feature branch parents on feature head", func(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+ txn.branch = "feature"
+ sp := createSnapshotProducer(OpAppend, txn, nil, nil, nil)
+ require.Equal(t, int64(20), sp.parentSnapshotID,
+ "append on feature must layer on the feature head (20),
not main head (10)")
+ })
+
+ t.Run("main branch still parents on main head", func(t *testing.T) {
+ txn := newTransactionWithSnapshotRefs(t)
+ txn.branch = ""
+ sp := createSnapshotProducer(OpAppend, txn, nil, nil, nil)
+ require.Equal(t, int64(10), sp.parentSnapshotID)
+ })
+}
+
+func TestBranchWriteCommitsThroughCatalogPath(t *testing.T) {
+ ctx := context.Background()
+ spec := iceberg.NewPartitionSpec()
+ ident := Identifier{"db", "tbl"}
+
+ producers := []struct {
+ name string
+ op Operation
+ newProd func(Operation, *Transaction, iceio.WriteFileIO,
*uuid.UUID, iceberg.Properties) *snapshotProducer
+ }{
+ {"fast append", OpAppend, newFastAppendFilesProducer},
+ {"merge append", OpAppend, newMergeAppendFilesProducer},
+ {"overwrite", OpOverwrite, newOverwriteFilesProducer},
+ }
+
+ for _, tc := range producers {
+ t.Run(tc.name, func(t *testing.T) {
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+
+ // 1. Create the "feature" branch on a fresh table. The
branch does
+ // not exist yet, so the snapshot has no parent and the
requirement
+ // asserts the branch is absent (nil).
+ txn.branch = "feature"
+ sp1 := newFastAppendFilesProducer(OpAppend, txn, memIO,
nil, nil)
+ sp1.appendDataFile(newTestDataFile(t, spec,
"file://feature-1.parquet", nil))
+ up1, rq1, err := sp1.commit(ctx)
+ require.NoError(t, err)
+ addSnap1, ok := up1[0].(*addSnapshotUpdate)
+ require.True(t, ok)
+ require.Nil(t, addSnap1.Snapshot.ParentSnapshotID,
"first feature snapshot has no parent")
+ requireContainsRefSnapshotRequirement(t, rq1,
"feature", nil)
+ featureHead := addSnap1.Snapshot.SnapshotID
+ require.NoError(t, txn.apply(up1, rq1))
+ meta1, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ // 2. Advance main independently so feature and main
diverge.
+ tblMain := New(ident, meta1, "metadata.json",
func(context.Context) (iceio.IO, error) { return memIO, nil }, nil)
+ txnMain := tblMain.NewTransaction()
+ spMain := newFastAppendFilesProducer(OpAppend, txnMain,
memIO, nil, nil)
+ spMain.appendDataFile(newTestDataFile(t, spec,
"file://main-1.parquet", nil))
+ upM, rqM, err := spMain.commit(ctx)
+ require.NoError(t, err)
+ addSnapM, ok := upM[0].(*addSnapshotUpdate)
+ require.True(t, ok)
+ mainHead := addSnapM.Snapshot.SnapshotID
+ require.NoError(t, txnMain.apply(upM, rqM))
+ divergedMeta, err := txnMain.meta.Build()
+ require.NoError(t, err)
+ require.NotEqual(t, featureHead, mainHead)
+
+ // 3. Write to feature again with the producer under
test and commit
+ // through the public path. Staging mirrors what the
high-level op
+ // (Append/Overwrite/Delete) does internally; Commit
runs doCommit ->
+ // CommitTable where the branch requirement is
validated.
+ cat := &headTrackingCatalog{metadata: divergedMeta}
+ tbl := New(ident, divergedMeta, "metadata.json",
func(context.Context) (iceio.IO, error) { return memIO, nil }, cat)
+ txnFeat, err :=
tbl.NewTransactionOnBranchWithError("feature")
+ require.NoError(t, err)
+
+ spFeat := tc.newProd(tc.op, txnFeat, memIO, nil, nil)
+ spFeat.appendDataFile(newTestDataFile(t, spec,
"file://feature-2.parquet", nil))
+ upF, rqF, err := spFeat.commit(ctx)
+ require.NoError(t, err)
+ require.NoError(t, txnFeat.apply(upF, rqF))
+
+ committed, err := txnFeat.Commit(ctx)
+ require.NoError(t, err, "%s: catalog must accept the
branch commit; a main-head AssertRefSnapshotID would be rejected", tc.name)
+ require.Equal(t, int32(1), cat.attempts.Load(), "no
retry expected: the branch assertion matches on the first attempt")
+
+ newFeatureHead :=
committed.Metadata().SnapshotByName("feature")
+ require.NotNil(t, newFeatureHead)
+ require.NotNil(t, newFeatureHead.ParentSnapshotID)
+ require.Equal(t, featureHead,
*newFeatureHead.ParentSnapshotID,
+ "%s on feature must be parented on the feature
head, not main", tc.name)
+ require.NotEqual(t, mainHead,
*newFeatureHead.ParentSnapshotID)
+
+ // The feature commit must leave main untouched.
+ mainRef :=
committed.Metadata().SnapshotByName(MainBranch)
+ require.NotNil(t, mainRef)
+ require.Equal(t, mainHead, mainRef.SnapshotID,
"committing to feature must not move main")
+ })
+ }
+}
+
+func liveDataFilePathsForSnapshot(t *testing.T, snap *Snapshot, fs iceio.IO)
[]string {
+ t.Helper()
+ require.NotNil(t, snap)
+ var paths []string
+ for e, err := range snap.entries(fs, iceberg.ManifestContentData) {
+ require.NoError(t, err)
+ if e.Status() == iceberg.EntryStatusDELETED {
+ continue
+ }
+ if e.DataFile().ContentType() == iceberg.EntryContentData {
+ paths = append(paths, e.DataFile().FilePath())
+ }
+ }
+
+ return paths
+}
+
+func TestBranchCreateForksFromMainHead(t *testing.T) {
+ ctx := context.Background()
+ spec := iceberg.NewPartitionSpec()
+ ident := Identifier{"db", "tbl"}
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+
+ // main -> [main.parquet]
+ spMain := newFastAppendFilesProducer(OpAppend, txn, memIO, nil, nil)
+ spMain.appendDataFile(newTestDataFile(t, spec, "file://main.parquet",
nil))
+ upM, rqM, err := spMain.commit(ctx)
+ require.NoError(t, err)
+ require.NoError(t, txn.apply(upM, rqM))
+ mainMeta, err := txn.meta.Build()
+ require.NoError(t, err)
+ mainHead := mainMeta.CurrentSnapshot().SnapshotID
+
+ // First write to the new "feature" branch.
+ cat := &headTrackingCatalog{metadata: mainMeta}
+ tbl := New(ident, mainMeta, "metadata.json", func(context.Context)
(iceio.IO, error) { return memIO, nil }, cat)
+ txnFeat, err := tbl.NewTransactionOnBranchWithError("feature")
+ require.NoError(t, err)
+ spFeat := newFastAppendFilesProducer(OpAppend, txnFeat, memIO, nil, nil)
+ spFeat.appendDataFile(newTestDataFile(t, spec,
"file://feature.parquet", nil))
+ upF, rqF, err := spFeat.commit(ctx)
+ require.NoError(t, err)
+
+ // The new-branch requirement must assert absence (nil), while the
parent must
+ // be main's head — the two halves of the split that the fallback
drives.
+ requireContainsRefSnapshotRequirement(t, rqF, "feature", nil)
+ addSnapF, ok := upF[0].(*addSnapshotUpdate)
+ require.True(t, ok)
+ require.NotNil(t, addSnapF.Snapshot.ParentSnapshotID)
+ require.Equal(t, mainHead, *addSnapF.Snapshot.ParentSnapshotID,
+ "first write to a new branch must fork from main's head")
+
+ require.NoError(t, txnFeat.apply(upF, rqF))
+ committed, err := txnFeat.Commit(ctx)
+ require.NoError(t, err)
+
+ require.ElementsMatch(t,
+ []string{"file://main.parquet", "file://feature.parquet"},
+ liveDataFilePathsForSnapshot(t,
committed.Metadata().SnapshotByName("feature"), memIO),
+ "a new branch must inherit main's data files, plus its own")
+}
+
+func TestBranchCreateForksFromMainHeadAcrossRetry(t *testing.T) {
+ ctx := context.Background()
+ spec := iceberg.NewPartitionSpec()
+ ident := Identifier{"db", "tbl"}
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+
+ spMain := newFastAppendFilesProducer(OpAppend, txn, memIO, nil, nil)
+ spMain.appendDataFile(newTestDataFile(t, spec, "file://main.parquet",
nil))
+ upM, rqM, err := spMain.commit(ctx)
+ require.NoError(t, err)
+ require.NoError(t, txn.apply(upM, rqM))
+ require.NoError(t, txn.SetProperties(iceberg.Properties{
+ CommitNumRetriesKey: "2",
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ }))
+ mainMeta, err := txn.meta.Build()
+ require.NoError(t, err)
+ mainHead := mainMeta.CurrentSnapshot().SnapshotID
+
+ // Fail attempt 0 (forcing a retry through rebuildSnapshotUpdates),
apply on 1.
+ cat := &flakyCatalog{metadata: mainMeta, failUntilAttempt: 1, failWith:
fmt.Errorf("REST: %w", ErrCommitFailed)}
Review Comment:
`flakyCatalog` returns the same `mainMeta` on the retry, so `freshMeta`
still has no `feature` ref and `latestSnapshotForBranch` falls back to main's
head, same as attempt 0. That pins the nil-parent regression, which is what I
asked for. The case it doesn't reach is a peer creating the branch between
attempts, where the fallback should be skipped and the peer's head becomes the
parent. Not a blocker, but it's the one retry path with no coverage.
##########
table/table.go:
##########
@@ -673,6 +673,20 @@ func rewriteRefSnapshotRequirements(reqs []Requirement,
branch string, fresh Met
return out
}
+func latestSnapshotForBranch(meta Metadata, branch string) *Snapshot {
+ if branch == "" || branch == MainBranch {
+ return meta.CurrentSnapshot()
+ }
+
+ for name, ref := range meta.Refs() {
Review Comment:
`Metadata` already exposes `SnapshotByName`, and its body is exactly this
lookup (`SnapshotRefs[name]` then `SnapshotByID`) as a direct map index.
`Refs()` yields through `cloneSnapshotRef` per entry, so this allocates across
the whole ref set on every retry attempt to find one name.
`rewriteRefSnapshotRequirements` a few lines up already uses `SnapshotByName`
for the same thing.
```go
if s := meta.SnapshotByName(branch); s != nil {
return s
}
return meta.CurrentSnapshot()
```
Same semantics, and it makes the missing ref-type check easier to see.
--
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]
