badalprasadsingh commented on code in PR #2046:
URL: https://github.com/apache/iceberg-go/pull/2046#discussion_r4263742770
##########
table/transaction.go:
##########
@@ -2833,17 +2845,75 @@ 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, fileSeqByPath, nil
+ 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 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 {
+ 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 planningDeletes{positional: positional, dvs: dvs, equality:
equality}, nil
+}
+
+// removeDataFile removes df together with the deletion vector that references
it.
+// Position and equality delete files are left alone, since they can still
apply to other files.
+func (d planningDeletes) removeDataFile(updater *snapshotProducer, df
iceberg.DataFile) {
+ updater.deleteDataFile(df)
+ if dv, ok := d.dvs[df.FilePath()]; ok {
+ updater.removeDeletionVector(dv.DataFile())
Review Comment:
Documented on Delete, Overwrite and OverwriteTable.
`TestCoWDeleteRemovingDVFailsInsteadOfReplaying` pins ErrCommitFailed, with
a no-DV control that retries.
--
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]
