laskoviymishka commented on code in PR #1427:
URL: https://github.com/apache/iceberg-go/pull/1427#discussion_r3553319684
##########
table/snapshot_producers.go:
##########
@@ -812,125 +886,207 @@ func (sp *snapshotProducer) writeAddedManifest(content
iceberg.ManifestContent,
}
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error)
{
+ delta, err := sp.accumulateSummaryDelta(nil)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSnapshot *Snapshot
+ if sp.parentSnapshotID > 0 {
+ previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+ if err != nil {
+ return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ }
+ }
+
+ var previousSummary iceberg.Properties
+ if previousSnapshot != nil && previousSnapshot.Summary != nil {
+ previousSummary = previousSnapshot.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// summaryOnRetry recomputes the snapshot summary against the fresh parent for
an
+// OCC retry. Additions are always counted; a removal of a delete file /
deletion
+// vector is counted only when it is still present on parent (present.counts),
+// because a peer may have already dropped it — replaying the captured removal
+// delta on top of a fresh parent that no longer counts the file undercounts
+// total-delete-files. Removed data files are always counted: a successful
retry
+// has already asserted they are present (assertRemovedFilesPresent).
+func (sp *snapshotProducer) summaryOnRetry(props iceberg.Properties, parent
*Snapshot, present *removedFilePresence) (Summary, error) {
+ delta, err := sp.accumulateSummaryDelta(present.counts)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSummary iceberg.Properties
+ if parent != nil && parent.Summary != nil {
+ previousSummary = parent.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// accumulateSummaryDelta builds the added/removed file delta for this
producer.
+// countDeleteRemoval, when non-nil, gates removal of delete files / deletion
+// vectors (returning false skips that removal); data-file removals are always
+// counted. nil counts every removal (the initial commit).
+func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval
func(iceberg.DataFile) bool) (iceberg.Properties, error) {
var ssc SnapshotSummaryCollector
partitionSummaryLimit := sp.txn.meta.props.
GetInt(WritePartitionSummaryLimitKey,
WritePartitionSummaryLimitDefault)
ssc.setPartitionSummaryLimit(partitionSummaryLimit)
- var err error
currentSchema := sp.txn.meta.CurrentSchema()
partitionSpec, err := sp.txn.meta.CurrentSpec()
if err != nil || partitionSpec == nil {
- return Summary{}, fmt.Errorf("could not get current partition
spec: %w", err)
+ return nil, fmt.Errorf("could not get current partition spec:
%w", err)
}
for _, df := range sp.addedFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
for _, df := range sp.addedDeleteFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
- if len(sp.deletedFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ specs := sp.txn.meta.specs
+ for _, df := range sp.deletedFiles {
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- if len(sp.deletedDeleteFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDeleteFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ for _, df := range sp.deletedDeleteFiles {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
}
- }
-
- if len(sp.deletedDVsByRef) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDVsByRef {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- var previousSnapshot *Snapshot
- if sp.parentSnapshotID > 0 {
- previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ for _, df := range sp.deletedDVsByRef {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
+ }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
- var previousSummary iceberg.Properties
- if previousSnapshot != nil && previousSnapshot.Summary != nil {
- previousSummary = previousSnapshot.Summary.Properties
- }
+ return ssc.build(), nil
+}
- summaryProps := ssc.build()
- maps.Copy(summaryProps, props)
+func (sp *snapshotProducer) rebaseSummary(delta, previousSummary, props
iceberg.Properties) (Summary, error) {
+ maps.Copy(delta, props)
return updateSnapshotSummaries(Summary{
Operation: sp.op,
- Properties: summaryProps,
+ Properties: delta,
}, previousSummary)
}
-// computeOwnManifests returns the subset of allManifests that were written
-// by this producer (i.e. not inherited from the parent snapshot). These are
-// preserved across OCC retry attempts when the manifest list is rebuilt
-// against a fresh parent.
-func (sp *snapshotProducer) computeOwnManifests(allManifests
[]iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
- if sp.parentSnapshotID <= 0 {
- // No parent means all manifests are new — nothing to exclude.
- return allManifests, nil
+// removedFilePresence records which of a producer's to-be-removed delete files
+// and deletion vectors are still live on the retry's fresh parent. counts is
the
+// gate passed to accumulateSummaryDelta so the retry summary only subtracts
+// removals the parent still counts.
+type removedFilePresence struct {
+ deleteFiles map[string]struct{} // pos/eq delete FilePath()
+ dvRefs map[string]struct{} // deletion-vector ReferencedDataFile()
+}
+
+func (p *removedFilePresence) counts(df iceberg.DataFile) bool {
+ if isDeletionVector(df) {
+ ref := df.ReferencedDataFile()
+ if ref == nil {
+ return false
+ }
+ _, ok := p.dvRefs[*ref]
+
+ return ok
}
+ _, ok := p.deleteFiles[df.FilePath()]
- parent, err := sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ return ok
+}
+
+// checkRemovedFiles scans parent once and (1) fails terminally if any data
file
+// this producer intends to rewrite is no longer live — a concurrent writer
+// already removed it, so proceeding would leave our replacement beside the
+// concurrent result and duplicate rows (Java's failMissingDeletePaths) — and
+// (2) returns which removed delete files / deletion vectors are still present
so
+// the retry summary does not double-subtract a peer's prior removal.
+func (sp *snapshotProducer) checkRemovedFiles(parent *Snapshot)
(*removedFilePresence, error) {
Review Comment:
`rebuildFn` calls this unconditionally on every retry, including for
`fastAppendFiles`/`mergeAppendFiles`, which never populate any of the
removed-file maps. For those producers we now read `parent.Manifests` and
iterate every entry on each attempt just to conclude there's nothing to check.
That's a scaling regression precisely where it hurts — under append
contention the retry loop turns into repeated full-table manifest scans, where
the old path only re-read the parent's manifest list. Quick fix, short-circuit
at the top before touching `parent.Manifests`:
```go
if len(sp.deletedFiles) == 0 && len(sp.deletedDeleteFiles) == 0 &&
len(sp.deletedDVsByRef) == 0 {
return present, nil
}
```
Longer term it'd be nice to share one parent scan across
`checkRemovedFiles`, `deletedEntries`, and `existingManifests` — an overwrite
retry currently walks the same entries up to three times — but the early-out is
the part I'd want for this PR.
##########
table/snapshot_producers.go:
##########
@@ -812,125 +886,207 @@ func (sp *snapshotProducer) writeAddedManifest(content
iceberg.ManifestContent,
}
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error)
{
+ delta, err := sp.accumulateSummaryDelta(nil)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSnapshot *Snapshot
+ if sp.parentSnapshotID > 0 {
+ previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+ if err != nil {
+ return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ }
+ }
+
+ var previousSummary iceberg.Properties
+ if previousSnapshot != nil && previousSnapshot.Summary != nil {
+ previousSummary = previousSnapshot.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// summaryOnRetry recomputes the snapshot summary against the fresh parent for
an
+// OCC retry. Additions are always counted; a removal of a delete file /
deletion
+// vector is counted only when it is still present on parent (present.counts),
+// because a peer may have already dropped it — replaying the captured removal
+// delta on top of a fresh parent that no longer counts the file undercounts
+// total-delete-files. Removed data files are always counted: a successful
retry
+// has already asserted they are present (assertRemovedFilesPresent).
+func (sp *snapshotProducer) summaryOnRetry(props iceberg.Properties, parent
*Snapshot, present *removedFilePresence) (Summary, error) {
+ delta, err := sp.accumulateSummaryDelta(present.counts)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSummary iceberg.Properties
+ if parent != nil && parent.Summary != nil {
+ previousSummary = parent.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// accumulateSummaryDelta builds the added/removed file delta for this
producer.
+// countDeleteRemoval, when non-nil, gates removal of delete files / deletion
+// vectors (returning false skips that removal); data-file removals are always
+// counted. nil counts every removal (the initial commit).
+func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval
func(iceberg.DataFile) bool) (iceberg.Properties, error) {
var ssc SnapshotSummaryCollector
partitionSummaryLimit := sp.txn.meta.props.
GetInt(WritePartitionSummaryLimitKey,
WritePartitionSummaryLimitDefault)
ssc.setPartitionSummaryLimit(partitionSummaryLimit)
- var err error
currentSchema := sp.txn.meta.CurrentSchema()
partitionSpec, err := sp.txn.meta.CurrentSpec()
if err != nil || partitionSpec == nil {
- return Summary{}, fmt.Errorf("could not get current partition
spec: %w", err)
+ return nil, fmt.Errorf("could not get current partition spec:
%w", err)
}
for _, df := range sp.addedFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
for _, df := range sp.addedDeleteFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
- if len(sp.deletedFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ specs := sp.txn.meta.specs
+ for _, df := range sp.deletedFiles {
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- if len(sp.deletedDeleteFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDeleteFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ for _, df := range sp.deletedDeleteFiles {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
}
- }
-
- if len(sp.deletedDVsByRef) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDVsByRef {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- var previousSnapshot *Snapshot
- if sp.parentSnapshotID > 0 {
- previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ for _, df := range sp.deletedDVsByRef {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
+ }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
- var previousSummary iceberg.Properties
- if previousSnapshot != nil && previousSnapshot.Summary != nil {
- previousSummary = previousSnapshot.Summary.Properties
- }
+ return ssc.build(), nil
+}
- summaryProps := ssc.build()
- maps.Copy(summaryProps, props)
+func (sp *snapshotProducer) rebaseSummary(delta, previousSummary, props
iceberg.Properties) (Summary, error) {
+ maps.Copy(delta, props)
return updateSnapshotSummaries(Summary{
Operation: sp.op,
- Properties: summaryProps,
+ Properties: delta,
}, previousSummary)
}
-// computeOwnManifests returns the subset of allManifests that were written
-// by this producer (i.e. not inherited from the parent snapshot). These are
-// preserved across OCC retry attempts when the manifest list is rebuilt
-// against a fresh parent.
-func (sp *snapshotProducer) computeOwnManifests(allManifests
[]iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
- if sp.parentSnapshotID <= 0 {
- // No parent means all manifests are new — nothing to exclude.
- return allManifests, nil
+// removedFilePresence records which of a producer's to-be-removed delete files
+// and deletion vectors are still live on the retry's fresh parent. counts is
the
+// gate passed to accumulateSummaryDelta so the retry summary only subtracts
+// removals the parent still counts.
+type removedFilePresence struct {
+ deleteFiles map[string]struct{} // pos/eq delete FilePath()
+ dvRefs map[string]struct{} // deletion-vector ReferencedDataFile()
+}
+
+func (p *removedFilePresence) counts(df iceberg.DataFile) bool {
+ if isDeletionVector(df) {
+ ref := df.ReferencedDataFile()
+ if ref == nil {
+ return false
+ }
+ _, ok := p.dvRefs[*ref]
+
+ return ok
}
+ _, ok := p.deleteFiles[df.FilePath()]
- parent, err := sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ return ok
+}
+
+// checkRemovedFiles scans parent once and (1) fails terminally if any data
file
+// this producer intends to rewrite is no longer live — a concurrent writer
+// already removed it, so proceeding would leave our replacement beside the
+// concurrent result and duplicate rows (Java's failMissingDeletePaths) — and
+// (2) returns which removed delete files / deletion vectors are still present
so
+// the retry summary does not double-subtract a peer's prior removal.
+func (sp *snapshotProducer) checkRemovedFiles(parent *Snapshot)
(*removedFilePresence, error) {
+ present := &removedFilePresence{
+ deleteFiles: map[string]struct{}{},
+ dvRefs: map[string]struct{}{},
+ }
+
+ missingData := make(map[string]struct{}, len(sp.deletedFiles))
+ for path := range sp.deletedFiles {
+ missingData[path] = struct{}{}
}
+
if parent == nil {
- return nil, fmt.Errorf("%w: computeOwnManifests parent id %d",
ErrSnapshotNotFound, sp.parentSnapshotID)
+ return sp.missingDataError(present, missingData)
}
- parentManifests, err := parent.Manifests(sp.io)
+ manifests, err := parent.Manifests(sp.io)
if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: read parent
manifests: %w", err)
+ return nil, err
+ }
+ for _, m := range manifests {
+ for entry, err := range sp.iterManifestEntries(m, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if m.ManifestContent() == iceberg.ManifestContentData {
+ delete(missingData, df.FilePath())
+
+ continue
+ }
+ if ref := df.ReferencedDataFile(); isDeletionVector(df)
&& ref != nil {
Review Comment:
Two things on this branch. A deletion vector with a nil
`ReferencedDataFile()` falls into the `else` and gets matched against
`deletedDeleteFiles` by path — the wrong map for a DV. Low-probability (a v3 DV
should always carry a ref) but it silently misclassifies instead of erroring;
I'd split `isDeletionVector(df) && ref == nil` into its own case that skips or
errors.
The other is a question on the matching itself: presence is keyed off the
referenced data file, not the DV's own path. If a peer superseded the same data
file's DV with a new one (same ref, different path), we'd see "some DV for this
ref still exists" and report our stale DV as present — which `summaryOnRetry`
would then count as a live removal for a DV that's no longer physically in the
manifest. I couldn't fully trace whether an isolation validator rejects this
first. If it doesn't, keying presence off the exact DV path (falling back to
ref only for the summary decision) plus a concurrent-DV-replacement test would
settle it. wdyt?
##########
table/snapshot_producers.go:
##########
@@ -812,125 +886,207 @@ func (sp *snapshotProducer) writeAddedManifest(content
iceberg.ManifestContent,
}
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error)
{
+ delta, err := sp.accumulateSummaryDelta(nil)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSnapshot *Snapshot
+ if sp.parentSnapshotID > 0 {
+ previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+ if err != nil {
+ return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ }
+ }
+
+ var previousSummary iceberg.Properties
+ if previousSnapshot != nil && previousSnapshot.Summary != nil {
+ previousSummary = previousSnapshot.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// summaryOnRetry recomputes the snapshot summary against the fresh parent for
an
+// OCC retry. Additions are always counted; a removal of a delete file /
deletion
+// vector is counted only when it is still present on parent (present.counts),
+// because a peer may have already dropped it — replaying the captured removal
+// delta on top of a fresh parent that no longer counts the file undercounts
+// total-delete-files. Removed data files are always counted: a successful
retry
+// has already asserted they are present (assertRemovedFilesPresent).
+func (sp *snapshotProducer) summaryOnRetry(props iceberg.Properties, parent
*Snapshot, present *removedFilePresence) (Summary, error) {
+ delta, err := sp.accumulateSummaryDelta(present.counts)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSummary iceberg.Properties
+ if parent != nil && parent.Summary != nil {
+ previousSummary = parent.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// accumulateSummaryDelta builds the added/removed file delta for this
producer.
+// countDeleteRemoval, when non-nil, gates removal of delete files / deletion
+// vectors (returning false skips that removal); data-file removals are always
+// counted. nil counts every removal (the initial commit).
+func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval
func(iceberg.DataFile) bool) (iceberg.Properties, error) {
var ssc SnapshotSummaryCollector
partitionSummaryLimit := sp.txn.meta.props.
GetInt(WritePartitionSummaryLimitKey,
WritePartitionSummaryLimitDefault)
ssc.setPartitionSummaryLimit(partitionSummaryLimit)
- var err error
currentSchema := sp.txn.meta.CurrentSchema()
partitionSpec, err := sp.txn.meta.CurrentSpec()
if err != nil || partitionSpec == nil {
- return Summary{}, fmt.Errorf("could not get current partition
spec: %w", err)
+ return nil, fmt.Errorf("could not get current partition spec:
%w", err)
}
for _, df := range sp.addedFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
for _, df := range sp.addedDeleteFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
- if len(sp.deletedFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ specs := sp.txn.meta.specs
+ for _, df := range sp.deletedFiles {
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- if len(sp.deletedDeleteFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDeleteFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ for _, df := range sp.deletedDeleteFiles {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
}
- }
-
- if len(sp.deletedDVsByRef) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDVsByRef {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- var previousSnapshot *Snapshot
- if sp.parentSnapshotID > 0 {
- previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ for _, df := range sp.deletedDVsByRef {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
+ }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
- var previousSummary iceberg.Properties
- if previousSnapshot != nil && previousSnapshot.Summary != nil {
- previousSummary = previousSnapshot.Summary.Properties
- }
+ return ssc.build(), nil
+}
- summaryProps := ssc.build()
- maps.Copy(summaryProps, props)
+func (sp *snapshotProducer) rebaseSummary(delta, previousSummary, props
iceberg.Properties) (Summary, error) {
+ maps.Copy(delta, props)
return updateSnapshotSummaries(Summary{
Operation: sp.op,
- Properties: summaryProps,
+ Properties: delta,
}, previousSummary)
}
-// computeOwnManifests returns the subset of allManifests that were written
-// by this producer (i.e. not inherited from the parent snapshot). These are
-// preserved across OCC retry attempts when the manifest list is rebuilt
-// against a fresh parent.
-func (sp *snapshotProducer) computeOwnManifests(allManifests
[]iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
- if sp.parentSnapshotID <= 0 {
- // No parent means all manifests are new — nothing to exclude.
- return allManifests, nil
+// removedFilePresence records which of a producer's to-be-removed delete files
+// and deletion vectors are still live on the retry's fresh parent. counts is
the
+// gate passed to accumulateSummaryDelta so the retry summary only subtracts
+// removals the parent still counts.
+type removedFilePresence struct {
+ deleteFiles map[string]struct{} // pos/eq delete FilePath()
+ dvRefs map[string]struct{} // deletion-vector ReferencedDataFile()
+}
+
+func (p *removedFilePresence) counts(df iceberg.DataFile) bool {
+ if isDeletionVector(df) {
+ ref := df.ReferencedDataFile()
+ if ref == nil {
+ return false
+ }
+ _, ok := p.dvRefs[*ref]
+
+ return ok
}
+ _, ok := p.deleteFiles[df.FilePath()]
- parent, err := sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ return ok
+}
+
+// checkRemovedFiles scans parent once and (1) fails terminally if any data
file
+// this producer intends to rewrite is no longer live — a concurrent writer
+// already removed it, so proceeding would leave our replacement beside the
+// concurrent result and duplicate rows (Java's failMissingDeletePaths) — and
+// (2) returns which removed delete files / deletion vectors are still present
so
+// the retry summary does not double-subtract a peer's prior removal.
+func (sp *snapshotProducer) checkRemovedFiles(parent *Snapshot)
(*removedFilePresence, error) {
+ present := &removedFilePresence{
+ deleteFiles: map[string]struct{}{},
+ dvRefs: map[string]struct{}{},
+ }
+
+ missingData := make(map[string]struct{}, len(sp.deletedFiles))
+ for path := range sp.deletedFiles {
+ missingData[path] = struct{}{}
}
+
if parent == nil {
- return nil, fmt.Errorf("%w: computeOwnManifests parent id %d",
ErrSnapshotNotFound, sp.parentSnapshotID)
+ return sp.missingDataError(present, missingData)
}
- parentManifests, err := parent.Manifests(sp.io)
+ manifests, err := parent.Manifests(sp.io)
if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: read parent
manifests: %w", err)
+ return nil, err
+ }
+ for _, m := range manifests {
+ for entry, err := range sp.iterManifestEntries(m, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if m.ManifestContent() == iceberg.ManifestContentData {
+ delete(missingData, df.FilePath())
+
+ continue
+ }
+ if ref := df.ReferencedDataFile(); isDeletionVector(df)
&& ref != nil {
+ if _, want := sp.deletedDVsByRef[*ref]; want {
+ present.dvRefs[*ref] = struct{}{}
+ }
+ } else if _, want :=
sp.deletedDeleteFiles[df.FilePath()]; want {
+ present.deleteFiles[df.FilePath()] = struct{}{}
+ }
+ }
}
- inherited := make(map[string]bool, len(parentManifests))
- for _, m := range parentManifests {
- inherited[m.FilePath()] = true
+ return sp.missingDataError(present, missingData)
+}
+
+func (sp *snapshotProducer) missingDataError(present *removedFilePresence,
missingData map[string]struct{}) (*removedFilePresence, error) {
+ if len(missingData) == 0 {
+ return present, nil
}
- own := make([]iceberg.ManifestFile, 0, len(allManifests))
- for _, m := range allManifests {
- if !inherited[m.FilePath()] {
- own = append(own, m)
- }
+ missing := make([]string, 0, len(missingData))
+ for path := range missingData {
+ missing = append(missing, path)
}
+ sort.Strings(missing)
Review Comment:
We already import `slices` and target Go 1.25, so `slices.Sort(missing)`
drops the `sort` import entirely. While we're here, `%v` on a `[]string` prints
Go slice syntax (`[path1 path2]`) in a user-facing error — `strings.Join` or
`%q` reads cleaner.
##########
table/snapshot_producers.go:
##########
@@ -812,125 +886,207 @@ func (sp *snapshotProducer) writeAddedManifest(content
iceberg.ManifestContent,
}
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error)
{
+ delta, err := sp.accumulateSummaryDelta(nil)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSnapshot *Snapshot
+ if sp.parentSnapshotID > 0 {
+ previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+ if err != nil {
+ return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ }
+ }
+
+ var previousSummary iceberg.Properties
+ if previousSnapshot != nil && previousSnapshot.Summary != nil {
+ previousSummary = previousSnapshot.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// summaryOnRetry recomputes the snapshot summary against the fresh parent for
an
+// OCC retry. Additions are always counted; a removal of a delete file /
deletion
+// vector is counted only when it is still present on parent (present.counts),
+// because a peer may have already dropped it — replaying the captured removal
+// delta on top of a fresh parent that no longer counts the file undercounts
+// total-delete-files. Removed data files are always counted: a successful
retry
+// has already asserted they are present (assertRemovedFilesPresent).
Review Comment:
`assertRemovedFilesPresent` doesn't exist — the function that does this is
`checkRemovedFiles`. Worth fixing so the comment doesn't send someone hunting
for a symbol that isn't there.
##########
table/snapshot_producers.go:
##########
@@ -812,125 +886,207 @@ func (sp *snapshotProducer) writeAddedManifest(content
iceberg.ManifestContent,
}
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error)
{
+ delta, err := sp.accumulateSummaryDelta(nil)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSnapshot *Snapshot
+ if sp.parentSnapshotID > 0 {
+ previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+ if err != nil {
+ return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ }
+ }
+
+ var previousSummary iceberg.Properties
+ if previousSnapshot != nil && previousSnapshot.Summary != nil {
+ previousSummary = previousSnapshot.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// summaryOnRetry recomputes the snapshot summary against the fresh parent for
an
+// OCC retry. Additions are always counted; a removal of a delete file /
deletion
+// vector is counted only when it is still present on parent (present.counts),
+// because a peer may have already dropped it — replaying the captured removal
+// delta on top of a fresh parent that no longer counts the file undercounts
+// total-delete-files. Removed data files are always counted: a successful
retry
+// has already asserted they are present (assertRemovedFilesPresent).
+func (sp *snapshotProducer) summaryOnRetry(props iceberg.Properties, parent
*Snapshot, present *removedFilePresence) (Summary, error) {
+ delta, err := sp.accumulateSummaryDelta(present.counts)
+ if err != nil {
+ return Summary{}, err
+ }
+
+ var previousSummary iceberg.Properties
+ if parent != nil && parent.Summary != nil {
+ previousSummary = parent.Summary.Properties
+ }
+
+ return sp.rebaseSummary(delta, previousSummary, props)
+}
+
+// accumulateSummaryDelta builds the added/removed file delta for this
producer.
+// countDeleteRemoval, when non-nil, gates removal of delete files / deletion
+// vectors (returning false skips that removal); data-file removals are always
+// counted. nil counts every removal (the initial commit).
+func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval
func(iceberg.DataFile) bool) (iceberg.Properties, error) {
var ssc SnapshotSummaryCollector
partitionSummaryLimit := sp.txn.meta.props.
GetInt(WritePartitionSummaryLimitKey,
WritePartitionSummaryLimitDefault)
ssc.setPartitionSummaryLimit(partitionSummaryLimit)
- var err error
currentSchema := sp.txn.meta.CurrentSchema()
partitionSpec, err := sp.txn.meta.CurrentSpec()
if err != nil || partitionSpec == nil {
- return Summary{}, fmt.Errorf("could not get current partition
spec: %w", err)
+ return nil, fmt.Errorf("could not get current partition spec:
%w", err)
}
for _, df := range sp.addedFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
for _, df := range sp.addedDeleteFiles {
if err = ssc.addFile(df, currentSchema,
sp.spec(int(df.SpecID()))); err != nil {
- return Summary{}, err
+ return nil, err
}
}
- if len(sp.deletedFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ specs := sp.txn.meta.specs
+ for _, df := range sp.deletedFiles {
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- if len(sp.deletedDeleteFiles) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDeleteFiles {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ for _, df := range sp.deletedDeleteFiles {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
}
- }
-
- if len(sp.deletedDVsByRef) > 0 {
- specs := sp.txn.meta.specs
- for _, df := range sp.deletedDVsByRef {
- if err = ssc.removeFile(df, currentSchema,
specs[df.SpecID()]); err != nil {
- return Summary{}, err
- }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
-
- var previousSnapshot *Snapshot
- if sp.parentSnapshotID > 0 {
- previousSnapshot, err =
sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return Summary{}, fmt.Errorf("summary: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ for _, df := range sp.deletedDVsByRef {
+ if countDeleteRemoval != nil && !countDeleteRemoval(df) {
+ continue
+ }
+ if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]);
err != nil {
+ return nil, err
}
}
- var previousSummary iceberg.Properties
- if previousSnapshot != nil && previousSnapshot.Summary != nil {
- previousSummary = previousSnapshot.Summary.Properties
- }
+ return ssc.build(), nil
+}
- summaryProps := ssc.build()
- maps.Copy(summaryProps, props)
+func (sp *snapshotProducer) rebaseSummary(delta, previousSummary, props
iceberg.Properties) (Summary, error) {
+ maps.Copy(delta, props)
return updateSnapshotSummaries(Summary{
Operation: sp.op,
- Properties: summaryProps,
+ Properties: delta,
}, previousSummary)
}
-// computeOwnManifests returns the subset of allManifests that were written
-// by this producer (i.e. not inherited from the parent snapshot). These are
-// preserved across OCC retry attempts when the manifest list is rebuilt
-// against a fresh parent.
-func (sp *snapshotProducer) computeOwnManifests(allManifests
[]iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
- if sp.parentSnapshotID <= 0 {
- // No parent means all manifests are new — nothing to exclude.
- return allManifests, nil
+// removedFilePresence records which of a producer's to-be-removed delete files
+// and deletion vectors are still live on the retry's fresh parent. counts is
the
+// gate passed to accumulateSummaryDelta so the retry summary only subtracts
+// removals the parent still counts.
+type removedFilePresence struct {
+ deleteFiles map[string]struct{} // pos/eq delete FilePath()
+ dvRefs map[string]struct{} // deletion-vector ReferencedDataFile()
+}
+
+func (p *removedFilePresence) counts(df iceberg.DataFile) bool {
+ if isDeletionVector(df) {
+ ref := df.ReferencedDataFile()
+ if ref == nil {
+ return false
+ }
+ _, ok := p.dvRefs[*ref]
+
+ return ok
}
+ _, ok := p.deleteFiles[df.FilePath()]
- parent, err := sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
- if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: lookup parent
snapshot %d: %w", sp.parentSnapshotID, err)
+ return ok
+}
+
+// checkRemovedFiles scans parent once and (1) fails terminally if any data
file
+// this producer intends to rewrite is no longer live — a concurrent writer
+// already removed it, so proceeding would leave our replacement beside the
+// concurrent result and duplicate rows (Java's failMissingDeletePaths) — and
+// (2) returns which removed delete files / deletion vectors are still present
so
+// the retry summary does not double-subtract a peer's prior removal.
+func (sp *snapshotProducer) checkRemovedFiles(parent *Snapshot)
(*removedFilePresence, error) {
+ present := &removedFilePresence{
+ deleteFiles: map[string]struct{}{},
+ dvRefs: map[string]struct{}{},
+ }
+
+ missingData := make(map[string]struct{}, len(sp.deletedFiles))
+ for path := range sp.deletedFiles {
+ missingData[path] = struct{}{}
}
+
if parent == nil {
- return nil, fmt.Errorf("%w: computeOwnManifests parent id %d",
ErrSnapshotNotFound, sp.parentSnapshotID)
+ return sp.missingDataError(present, missingData)
}
- parentManifests, err := parent.Manifests(sp.io)
+ manifests, err := parent.Manifests(sp.io)
if err != nil {
- return nil, fmt.Errorf("computeOwnManifests: read parent
manifests: %w", err)
+ return nil, err
+ }
+ for _, m := range manifests {
+ for entry, err := range sp.iterManifestEntries(m, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if m.ManifestContent() == iceberg.ManifestContentData {
+ delete(missingData, df.FilePath())
+
+ continue
+ }
+ if ref := df.ReferencedDataFile(); isDeletionVector(df)
&& ref != nil {
+ if _, want := sp.deletedDVsByRef[*ref]; want {
+ present.dvRefs[*ref] = struct{}{}
+ }
+ } else if _, want :=
sp.deletedDeleteFiles[df.FilePath()]; want {
+ present.deleteFiles[df.FilePath()] = struct{}{}
+ }
+ }
}
- inherited := make(map[string]bool, len(parentManifests))
- for _, m := range parentManifests {
- inherited[m.FilePath()] = true
+ return sp.missingDataError(present, missingData)
+}
+
+func (sp *snapshotProducer) missingDataError(present *removedFilePresence,
missingData map[string]struct{}) (*removedFilePresence, error) {
+ if len(missingData) == 0 {
+ return present, nil
}
- own := make([]iceberg.ManifestFile, 0, len(allManifests))
- for _, m := range allManifests {
- if !inherited[m.FilePath()] {
- own = append(own, m)
- }
+ missing := make([]string, 0, len(missingData))
+ for path := range missingData {
+ missing = append(missing, path)
}
+ sort.Strings(missing)
- return own, nil
+ return nil, fmt.Errorf("%w: %d data file(s) to rewrite are no longer on
the branch head, e.g. %v",
+ ErrCommitDiverged, len(missing), missing[:min(len(missing), 5)])
Review Comment:
This abort — `ErrCommitDiverged` when a rewrite target is already gone from
head — is the one behavior in this PR that changes what callers see. Previously
they got silent resurrection; now they get a terminal, non-retryable error so
they can re-plan. That's the right call and matches Java's
`failMissingDeletePaths`, but nothing exercises it —
`TestSummaryOnRetry_Skips/SubtractsPresentDeleteFile` cover the delete-file
presence gating, not the `missingDataError` branch.
A unit test calling `checkRemovedFiles` with a parent missing one of
`sp.deletedFiles` and asserting `errors.Is(err, ErrCommitDiverged)` would lock
this down. An end-to-end scenario where the peer removes the exact file writer
A is replacing, asserting the commit fails diverged rather than resurrecting,
would be even better.
##########
table/occ_scenario_test.go:
##########
@@ -451,6 +451,129 @@ func (s *OCCScenarioTestSuite)
TestV3MergeAppendRowLineageAfterOCCRetry() {
}
}
+// TestReplaceFilesNoResurrectionAfterConflict is a regression test for the OCC
+// retry-replay resurrection bug. When a replace/overwrite commit retries after
+// a conflict, the rebuild must re-apply the file removal against the FRESH
+// parent — not graft attempt-0's stale "own" manifests onto a fresh parent
+// that still lists the removed files. Otherwise the removed files come back
+// alive next to their replacement and their rows are duplicated.
+//
+// Scenario:
+// - Base: file0 = rows [1,2,3] committed.
+// - Concurrent peer: appends filePeer = row [4]; catalog head is file0+peer.
+// - Writer A (stale base = only file0) replaces file0 with a compacted file
+// holding the same rows [1,2,3]. First attempt conflicts; the retry
rebases
+// onto the fresh head.
+//
+// Correct: live rows = {1,2,3,4} (compacted + peer), file0 dropped → 4 rows.
+// Buggy: file0 resurrected → rows [1,2,3] duplicated → 7 rows, 3 live files.
+func (s *OCCScenarioTestSuite) TestReplaceFilesNoResurrectionAfterConflict() {
+ props := iceberg.Properties{
+ table.CommitMinRetryWaitMsKey: "0",
+ table.CommitMaxRetryWaitMsKey: "0",
+ table.CommitTotalRetryTimeoutMsKey: "60000",
+ table.CommitNumRetriesKey: "2",
+ // Snapshot isolation lets the replace commit past the
concurrent
+ // append the same way compaction's rewrite semantics do; the
bug we
+ // are guarding lives in the retry rebuild, not the validator.
+ table.WriteUpdateIsolationLevelKey:
string(table.IsolationSnapshot),
+ }
+
+ localFS := func(_ context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }
+
+ // Step 1: base table with file0 = rows [1,2,3].
+ baseTbl, cat := s.makeTable(0, props)
+ arrowSc, err := table.SchemaToArrowSchema(baseTbl.Schema(), nil, false,
false)
+ s.Require().NoError(err)
+
+ file0Path := s.location + "/data/file0.parquet"
+ writeParquetFile(s.T(), file0Path, arrowSc,
+ `[{"id":1,"val":"a"},{"id":2,"val":"b"},{"id":3,"val":"c"}]`)
+
+ tx := baseTbl.NewTransaction()
+ s.Require().NoError(tx.AddFiles(s.ctx, []string{file0Path}, nil, false))
+ _, err = tx.Commit(s.ctx)
+ s.Require().NoError(err)
+ metaBase := cat.current
+
+ // Step 2: a concurrent peer appends filePeer = row [4] onto metaBase.
+ peerCat := &occScenarioCatalog{current: metaBase, location: s.location}
+ peerTbl := table.New(baseTbl.Identifier(), metaBase,
+ s.location+"/metadata/base.metadata.json", localFS, peerCat)
+
+ peerPath := s.location + "/data/file_peer.parquet"
+ writeParquetFile(s.T(), peerPath, arrowSc, `[{"id":4,"val":"d"}]`)
+
+ ptx := peerTbl.NewTransaction()
+ s.Require().NoError(ptx.AddFiles(s.ctx, []string{peerPath}, nil, false))
+ _, err = ptx.Commit(s.ctx)
+ s.Require().NoError(err)
+ metaPeer := peerCat.current
+
+ // Step 3: Writer A replaces file0 -> compacted, starting from the stale
+ // metaBase; its catalog is at metaPeer and returns one 409 before
success.
+ catA := &occScenarioCatalog{current: metaPeer, conflictsLeft: 1,
location: s.location}
Review Comment:
Every OCC scenario here uses `conflictsLeft: 1`, so we only ever exercise a
single rebuild. The reused-added-content path is most interesting across more
than one retry — that's where orphaned-manifest churn and cumulative removal
tracking would show up if the reuse logic were wrong.
A `conflictsLeft: 2` variant (two distinct peer commits) asserting the same
no-resurrection invariant would lock in retry idempotency, which is the
property the added-content reuse is really claiming. Nice-to-have, not blocking.
##########
table/snapshot_producers.go:
##########
@@ -1115,17 +1271,13 @@ func (sp *snapshotProducer) commit(ctx context.Context)
(_ []Update, _ []Require
rebuilt.AddedRows = &addedRows
}
- // Recompute snapshot summary against the fresh parent so that
totals
- // (total-records, total-data-files, total-files-size) are not
regressed
- // to the stale values captured at attempt 0. The per-operation
delta
- // (added-data-files, added-records, etc.) is preserved in
- // capturedSnapshot.Summary and is replayed on top of the fresh
base.
+ // Recompute the snapshot summary against the fresh parent so
totals are
+ // not regressed to the stale attempt-0 values. Removals of
delete files
+ // / DVs are gated by present (see checkRemovedFiles) so a
peer's prior
+ // removal is not subtracted twice, which would undercount
+ // total-delete-files.
if freshParent != nil && freshParent.Summary != nil &&
capturedSnapshot.Summary != nil {
- deltaSummary := Summary{
- Operation: capturedSnapshot.Summary.Operation,
- Properties:
maps.Clone(capturedSnapshot.Summary.Properties),
- }
- if s, sumErr := updateSnapshotSummaries(deltaSummary,
freshParent.Summary.Properties); sumErr == nil {
+ if s, sumErr := sp.summaryOnRetry(sp.snapshotProps,
freshParent, present); sumErr == nil {
Review Comment:
If `summaryOnRetry` errors, we silently keep the stale attempt-0 summary —
`sumErr` is discarded. The old `updateSnapshotSummaries` path swallowed its
error too so this isn't a regression, but this PR reworks exactly this code and
it's a natural spot to fix. I'd return the error from `rebuildFn` so the commit
loop surfaces it rather than committing a summary we know is wrong.
--
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]
