laskoviymishka commented on code in PR #2046:
URL: https://github.com/apache/iceberg-go/pull/2046#discussion_r4286741813
##########
table/transaction.go:
##########
@@ -2538,17 +2573,41 @@ func (t *Transaction) performMergeOnReadDeletion(ctx
context.Context, snapshotPr
updater := t.updateSnapshot(wfs, snapshotProps,
OpDelete).mergeOverwrite(&commitUUID, filter)
updater.setManifestConcurrency(concurrency)
- filesToDelete, withPartialDeletions, _, err :=
t.classifyFilesForDeletions(ctx, fs, filter, caseSensitive, concurrency)
+ filesToDelete, withPartialDeletions, err :=
t.classifyFilesForDeletions(ctx, fs, filter, caseSensitive, concurrency)
+ if err != nil {
+ return nil, err
+ }
+
+ // SetProperties above may have staged a name mapping, which replaces
t.meta, so re-read it.
+ meta, err = t.txnMeta()
if err != nil {
return nil, err
}
+ // On v3 every removed file takes its deletion vector with it, and a
partially deleted file's vector is
+ // merged into its replacement. One read serves both. DVs only exist
from v3, so v2 skips it.
+ var deletes planningDeletes
+ if meta.formatVersion >= 3 && (len(filesToDelete) > 0 ||
len(withPartialDeletions) > 0) {
+ builtMeta, err := meta.Build()
+ if err != nil {
+ return nil, fmt.Errorf("failed to build metadata: %w",
err)
+ }
+ deletes, err = t.readPlanningDeletes(fs, builtMeta)
Review Comment:
MoR only uses `deletes.dvs`, but this builds the positional and equality
indexes too. So a v3 merge-on-read delete can now fail on something in the
equality index (an equality delete whose field ids no longer resolve after
schema evolution, say) that `collectExistingDVs` never looked at. I'd either
split out a DV-only read or at least extend the #2160 comment to say MoR pays
for this now. Fine as a follow-up.
##########
table/transaction.go:
##########
@@ -2846,17 +2893,78 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
}
if err := g.Wait(); err != nil {
- return nil, nil, nil, err
+ return nil, nil, err
+ }
+
+ return filesToDelete, filesWithPartialDeletes, nil
+}
+
+// planningDeletes indexes the live delete files of the planning snapshot the
same way scan planning indexes them.
+type planningDeletes struct {
+ positional *positionalDeleteIndex
+ dvs map[string]iceberg.ManifestEntry
+ equality *equalityDeleteIndex
+}
+
+// readPlanningDeletes reads and indexes every live delete file of the
planning snapshot.
+//
+// Delete manifests are not pruned by the row filter: with filter id=2 the
rewrite keeps id=1,
+// so a delete on id=1 must still be applied.
+//
+// Known limitation (#2160): every live delete manifest of the snapshot is
read on each call,
+// with no partition pruning, so a narrow delete or rewrite on a delete-heavy
table pays for all of them.
+func (t *Transaction) readPlanningDeletes(fs io.IO, builtMeta Metadata)
(planningDeletes, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return planningDeletes{}, err
+ }
+
+ var liveDeletes []iceberg.ManifestEntry
+ if s := t.planningSnapshot(meta); s != nil {
+ // Skip DELETED entries: a replaced DV stays behind as one, and
merging it
+ // into a new DV would bring back deleted rows. See #1372.
+ for entry, err := range s.entries(fs,
iceberg.ManifestContentDeletes, true) {
+ if err != nil {
+ return planningDeletes{}, fmt.Errorf("failed to
read delete manifests: %w", err)
+ }
+ liveDeletes = append(liveDeletes, entry)
+ }
+ }
+
+ classified, err := classifyManifestEntries(liveDeletes)
+ if err != nil {
+ return planningDeletes{}, err
+ }
+ positional, err :=
buildPositionalDeleteIndex(classified.positionalDeleteEntries)
+ if err != nil {
+ return planningDeletes{}, err
+ }
+ dvs, err := buildDVIndex(classified.dvEntries)
+ if err != nil {
+ return planningDeletes{}, err
+ }
+ equality, err :=
buildEqualityDeleteIndex(classified.equalityDeleteEntries, builtMeta,
meta.CurrentSchema())
+ if err != nil {
+ return planningDeletes{}, err
}
- return filesToDelete, filesWithPartialDeletes, fileSeqByPath, nil
+ return planningDeletes{positional: positional, dvs: dvs, equality:
equality}, nil
+}
+
+// removeDataFile removes df and, as the spec requires, the deletion vector
that references it.
+// Removing a deletion vector makes the commit non-replayable (see
snapshotProducer.commitManifests).
Review Comment:
Fine with the stricter behavior for this PR, but I think it's tighter than
it needs to be here. The stranded-two-DVs case that `noReplay` protects against
comes from replacing a DV while its data file stays live. On this path the data
file goes too, and on copy-on-write a concurrent DV on that file already trips
`validateNoNewDeletesForRewrittenFiles` through the `rewriteValidator(removed)`
registered in `performCopyOnWriteDeletion`, so replay looks safe there.
Worth a follow-up issue to set `noReplay` only for DV removals that aren't
paired with removing their data file. That gives plain CoW deletes their retry
back, while the MoR partial-delete merge stays non-replayable.
--
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]
