laskoviymishka commented on code in PR #1637:
URL: https://github.com/apache/iceberg-go/pull/1637#discussion_r3719187125
##########
table/metadata.go:
##########
@@ -390,6 +390,42 @@ func (b *MetadataBuilder) currentSnapshot() *Snapshot {
return s
}
+// currentSnapshotForRef returns the head snapshot for ref.
+// Empty or main refs return the current snapshot.
+// Unknown refs also fall back to the current snapshot.
+func (b *MetadataBuilder) currentSnapshotForRef(ref string) *Snapshot {
+ if ref == "" || ref == MainBranch {
+ return b.currentSnapshot()
+ }
+
+ r, ok := b.refs[ref]
+ if !ok {
+ return b.currentSnapshot()
Review Comment:
This fallback isn't mirrored on the OCC retry path, and the gap is silent.
`rebuildSnapshotUpdates` in `table/table.go:694` (outside this diff)
resolves the retry parent as `freshMeta.SnapshotByName(branch)` and stops
there, no `CurrentSnapshot()` fallback. On a first write to a not-yet-existing
branch, attempt 0 gets main's head through this helper and inherits main's
manifests, but a retry gets `freshHead == nil`, so `rebuildFn` calls
`assembleManifests` with a nil parent, `existingManifests(nil)` returns
nothing, and the rebuilt list holds only `addedContent`. The attempt-0 list
that did inherit main's manifests is then deleted as orphaned. The commit
succeeds with the branch missing all of main's data.
The asymmetry predates this PR, but pre-fix the main-head assertion failed
`Validate` on every attempt so no retry could ever commit. This change is what
makes it reachable. A couple of ways to handle, wdyt?
- give `rebuildSnapshotUpdates` the same `CurrentSnapshot()` fallback,
matching Java's `SnapshotUtil.latestSnapshot(freshBase, branch)`
- drop the fallback here and make fork-from-main an explicit choice in
`createSnapshotProducer`, so one place owns the policy
##########
table/metadata.go:
##########
@@ -390,6 +390,42 @@ func (b *MetadataBuilder) currentSnapshot() *Snapshot {
return s
}
+// currentSnapshotForRef returns the head snapshot for ref.
+// Empty or main refs return the current snapshot.
+// Unknown refs also fall back to the current snapshot.
+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 head snapshot ID for ref.
+// Empty or main refs return currentSnapshotID.
+// Unknown refs return nil so AssertRefSnapshotID asserts that the
Review Comment:
I think this works right now and it isn't a blocker, but I'd really like
both doc comments to say *why* they diverge, not just that they do.
The split is load-bearing: the parent lookup wants main's head so a new
branch forks from main, while the assertion wants nil so `AssertRefSnapshotID`
proves the branch is absent. Read cold, the two comments look like an
inconsistency. The failure mode for anyone deriving one from the other
(`currentSnapshotForRef(ref).SnapshotID` as the assertion id) is a spurious OCC
rejection on every new-branch create. One sentence in each comment naming the
caller it serves would head that off.
##########
table/transaction_internal_test.go:
##########
@@ -44,6 +45,140 @@ 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")
Review Comment:
This assertion is vacuous as written.
Step 1 builds "feature" on a fresh table, so
`currentSnapshotForRef("feature")` falls through to `currentSnapshot()`, which
is nil, and `ParentSnapshotID` comes back nil regardless of what the helper
does. It'd still pass if we deleted the `!ok` fallback entirely.
The case that actually exercises the fallback is a first write to a new
branch when main already has data: `ParentSnapshotID` should be main's head and
the snapshot's manifests should include main's files. That's the same scenario
the retry path gets wrong, so pinning it here earns its keep twice.
##########
table/transaction.go:
##########
@@ -60,7 +60,7 @@ func (s snapshotUpdate) fastAppend() *snapshotProducer {
// checks.
func (s snapshotUpdate) mergeOverwrite(commitUUID *uuid.UUID, filter
iceberg.BooleanExpression) *snapshotProducer {
op := s.operation
- if s.operation == OpOverwrite && s.txn.meta.currentSnapshot() == nil {
+ if s.operation == OpOverwrite &&
s.txn.meta.currentSnapshotForRef(s.txn.branch) == nil {
Review Comment:
Three sibling call sites in this same file still read main's head, so branch
support is only half wired.
`ReplaceDataFiles` at line 613 and `ReplaceDataFilesWithDataFiles` at line
999 both do `s := meta.currentSnapshot()`, bail with `ErrInvalidOperation` when
it's nil, then scan `s.dataFiles(fs, nil)` to build the delete set.
`AddDataFiles` does the same at line 906 for the duplicate check. On a branch
transaction that means a branch with data fails as "cannot replace files in a
table without an existing snapshot" when main is empty, and the duplicate check
misses files that live only on the branch.
I'd swap those to `currentSnapshotForRef(t.branch)` in this pass, since it's
the same one-line change you've already made here. If you'd rather keep the
diff narrow, a follow-up issue works, but then the PR description shouldn't
read as though branch writes are fixed end to end.
##########
table/metadata.go:
##########
@@ -390,6 +390,42 @@ func (b *MetadataBuilder) currentSnapshot() *Snapshot {
return s
}
+// currentSnapshotForRef returns the head snapshot for ref.
+// Empty or main refs return the current snapshot.
+// Unknown refs also fall back to the current snapshot.
Review Comment:
Small thing: "unknown refs" here means "not in `b.refs`", but a ref that
*is* present and points at a snapshot missing from `b.snapshotList` returns nil
rather than falling back.
`RemoveSnapshots` prunes dangling refs so this shouldn't happen today, and
`currentSnapshot()` at line 388 has the same shape, so I'm not asking for a
code change. Worth a line saying the fallback assumes every entry in `b.refs`
resolves, since `createSnapshotProducer` and `mergeOverwrite` both read nil as
"no snapshot exists".
##########
table/transaction_internal_test.go:
##########
@@ -44,6 +45,140 @@ 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)
+ mainHead :=
upM[0].(*addSnapshotUpdate).Snapshot.SnapshotID
Review Comment:
Worth the checked form here, for consistency with `addSnap1` fourteen lines
up. If `upM[0]` ever comes back as a different update type this panics into a
goroutine dump instead of a clean test failure.
--
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]
