laskoviymishka commented on code in PR #1419:
URL: https://github.com/apache/iceberg-go/pull/1419#discussion_r3539732342
##########
table/arrow_utils.go:
##########
@@ -1848,7 +1853,25 @@ func positionDeleteRecordsToDataFilesDV(ctx
context.Context, rootLocation string
return metadata.PartitionSpecByID(int(id))
})
+ // Seed the writer with any deletion vector the referenced data
file
+ // already carries so its previously deleted positions survive
into the
+ // merged DV. The read path that produced args.itr already
applied these
Review Comment:
This isn't quite right — `makePositionDeleteRecordsForFilter` builds the
`FileScanTask` without `DeleteFiles`, so the scan does not apply the existing
DVs; the incoming batches can carry already-deleted positions. Correctness
still holds, but only because `bitmap.Set` is idempotent, so re-setting old
positions on top of the seed is a no-op.
I'd reword to say the scan does not apply existing deletes and the seed is
what guarantees the old positions survive. As written, someone later reads
this, concludes the seed is redundant, removes it, and the original bug is back.
##########
table/transaction.go:
##########
@@ -1913,9 +1939,50 @@ func (t *Transaction) writePositionDeletesForFiles(ctx
context.Context, fs io.IO
}
}
+ // Supersede the old DVs only after the merged replacements are staged,
so a
+ // write failure above leaves the existing DVs untouched.
+ for _, dvFile := range existingDVs {
+ updater.removeDeletionVector(dvFile)
+ }
+
return referenced, nil
}
+// collectExistingDVs returns the live deletion vectors that reference any of
+// files, keyed by the referenced data file path. Used by the v3 merge-on-read
+// delete path to fold prior deletes into a new DV and remove the superseded
+// entries, preserving the spec's one-DV-per-data-file invariant.
+func (t *Transaction) collectExistingDVs(fs io.IO, files []iceberg.DataFile)
(map[string]iceberg.DataFile, error) {
+ s := t.meta.currentSnapshot()
+ if s == nil {
+ return nil, nil
+ }
+
+ wanted := make(map[string]struct{}, len(files))
+ for _, df := range files {
+ wanted[df.FilePath()] = struct{}{}
+ }
+
+ result := make(map[string]iceberg.DataFile)
+ for df, err := range s.dataFiles(fs, nil) {
Review Comment:
`s.dataFiles(fs, nil)` runs with `discardDeleted=false`, so this pulls in
DELETED-status entries too. After a second delete, the snapshot's
deleted-entries manifest still carries the superseded old DV as a DELETED entry
for the same referenced data file. On a third delete this loop can yield both
the live ADDED merged DV and the stale DELETED one for the same path, and
`result[*ref] = df` is last-write-wins into a map — with the current manifest
concat order (added, then position deletes, then deleted-files) the DELETED
ghost lands last and wins. That seeds the new DV from a stale bitmap
(resurrecting positions deleted in the 2nd commit), calls
`removeDeletionVector` on a DV that no longer exists, and writes a Puffin
missing the merged positions.
The guard is already in the PR — `liveDVCount` in the test skips
`entry.Status() == EntryStatusDELETED`. I'd give `collectExistingDVs` the same
treatment: iterate `s.entries(fs, iceberg.ManifestContentDeletes)` and
`continue` on DELETED status, which also narrows the scan to delete manifests
instead of walking everything. wdyt?
##########
table/dv/dv_writer.go:
##########
@@ -108,6 +108,31 @@ func (w *DVWriter) Add(dataFilePath string, positions
[]int64, specID int32, par
}
}
+// Load seeds the writer with an already-written deletion vector for a data
+// file so subsequent Add calls merge new positions into it. The spec permits
+// at most one DV per data file per snapshot, so when a data file that already
+// has a DV receives more deletes the old positions must be carried into the
+// replacement DV (and the old DV superseded) rather than dropped. Load must be
+// called before any Add for the same path; specID and partitionData are
+// captured exactly as Add's first call would, and the supplied bitmap is
+// unioned into the entry. A nil bitmap simply registers the entry so it is
+// re-emitted. Callers retain ownership of the passed bitmap; only its set
+// positions are copied.
+func (w *DVWriter) Load(dataFilePath string, bitmap *RoaringPositionBitmap,
specID int32, partitionData map[int]any) {
Review Comment:
The doc says Load "must be called before any Add for the same path," but
nothing enforces that — call it after Add, or twice, and it silently ORs into
the live entry and discards the later `specID`/`partitionData`. Idempotent OR
makes it harmless today, but the doc promises a precondition the code doesn't
hold. I'd either error when the entry already exists, or soften the doc to
describe what actually happens.
Also "a nil bitmap simply registers the entry" — `ReadDV` always returns
non-nil, so that path never runs, and if it did it'd register a cardinality-0
DV. I'd document that `bitmap` must be non-nil instead.
##########
table/mor_delete_pruning_test.go:
##########
@@ -95,6 +103,79 @@ func newMergeOnReadTestTable(t *testing.T) *table.Table {
return table.New(table.Identifier{"db", "mor_delete_test"}, meta,
metaLoc, fsF, cat)
}
+// TestV3MergeOnReadDoubleDeleteMergesDeletionVector guards against a second
+// merge-on-read DELETE on a data file that already has a deletion vector
+// writing a second live DV. The spec permits at most one DV per data file, so
+// the new deletes must merge into the existing DV (which is then superseded),
+// not accumulate as a parallel one. A second live DV corrupts the table:
+// buildDVIndex rejects it at scan time with "can't index multiple deletion
+// vectors". See issue #1372.
+func TestV3MergeOnReadDoubleDeleteMergesDeletionVector(t *testing.T) {
+ ctx := context.Background()
+ tbl := newMergeOnReadTestTableVersion(t, "3")
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{
+
`[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"},{"id":4,"data":"d"},{"id":5,"data":"e"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ // First delete writes DV #1 against the single data file.
+ tbl, err = tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(2)), nil)
+ require.NoError(t, err)
+ require.Equal(t, []int64{1, 3, 4, 5}, idsInTable(t, tbl))
+ require.Equal(t, 1, liveDVCount(t, tbl), "first delete must write
exactly one DV")
+
+ // Second delete on the same data file must merge into DV #1, not add a
+ // second live DV.
+ tbl, err = tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(4)), nil)
+ require.NoError(t, err)
+
+ require.Equal(t, 1, liveDVCount(t, tbl),
+ "the two deletes must collapse into a single merged deletion
vector")
+ assert.Equal(t, []int64{1, 3, 5}, idsInTable(t, tbl),
+ "both id=2 and id=4 must be deleted and no previously deleted
row may resurrect")
Review Comment:
This stops at two deletes, but the ghost-DV bug above only triggers on the
third — the second delete is what writes the DELETED entry the third then picks
up. Could we add a triple-delete variant (append, delete id=2, delete id=4,
delete id=1, assert `liveDVCount == 1` and rows `[3, 5]`)? That's the test that
would've caught this.
While we're here, a partitioned / multi-data-file case would exercise the
per-file `pCtx` capture in the seed loop, which the single unpartitioned file
doesn't touch today.
##########
table/arrow_utils.go:
##########
@@ -1848,7 +1853,25 @@ func positionDeleteRecordsToDataFilesDV(ctx
context.Context, rootLocation string
return metadata.PartitionSpecByID(int(id))
})
+ // Seed the writer with any deletion vector the referenced data
file
+ // already carries so its previously deleted positions survive
into the
+ // merged DV. The read path that produced args.itr already
applied these
+ // existing deletes, so the incoming batches hold only the new
positions;
+ // without the seed the replacement DV would silently resurrect
rows. The
+ // spec allows at most one DV per data file, so the caller
supersedes the
+ // old DV once this merged one is written.
hasEntries := false
+ for filePath, bitmap := range args.existingDVs {
+ pCtx, ok := partitionContextByFilePath[filePath]
+ if !ok {
+ yield(nil, fmt.Errorf("unexpected missing
partition context for path %s", filePath))
+
+ return
+ }
+ writer.Load(filePath, bitmap, pCtx.specID,
pCtx.partitionData)
+ hasEntries = true
Review Comment:
`hasEntries` is set true for every existing-DV path even when that file gets
no new delete records this commit. If we seed a file but `posDeleteRecIter`
yields nothing for it, we still flush an identical-replacement Puffin and
remove the old DV — churn for no change. Gating the flush on whether records
were actually added (or `!bitmap.IsEmpty()`) would avoid rewriting an unchanged
DV. Latent, not a correctness bug.
##########
table/transaction.go:
##########
@@ -1913,9 +1939,50 @@ func (t *Transaction) writePositionDeletesForFiles(ctx
context.Context, fs io.IO
}
}
+ // Supersede the old DVs only after the merged replacements are staged,
so a
+ // write failure above leaves the existing DVs untouched.
+ for _, dvFile := range existingDVs {
+ updater.removeDeletionVector(dvFile)
+ }
+
return referenced, nil
}
+// collectExistingDVs returns the live deletion vectors that reference any of
+// files, keyed by the referenced data file path. Used by the v3 merge-on-read
+// delete path to fold prior deletes into a new DV and remove the superseded
+// entries, preserving the spec's one-DV-per-data-file invariant.
+func (t *Transaction) collectExistingDVs(fs io.IO, files []iceberg.DataFile)
(map[string]iceberg.DataFile, error) {
+ s := t.meta.currentSnapshot()
+ if s == nil {
+ return nil, nil
+ }
+
+ wanted := make(map[string]struct{}, len(files))
+ for _, df := range files {
+ wanted[df.FilePath()] = struct{}{}
+ }
+
+ result := make(map[string]iceberg.DataFile)
+ for df, err := range s.dataFiles(fs, nil) {
+ if err != nil {
+ return nil, err
Review Comment:
Every other error return in this function wraps with `%w` — this one returns
the iterator error bare. `fmt.Errorf("scanning existing deletion vectors: %w",
err)` keeps it consistent.
##########
table/arrow_utils.go:
##########
@@ -1848,7 +1853,25 @@ func positionDeleteRecordsToDataFilesDV(ctx
context.Context, rootLocation string
return metadata.PartitionSpecByID(int(id))
})
+ // Seed the writer with any deletion vector the referenced data
file
+ // already carries so its previously deleted positions survive
into the
+ // merged DV. The read path that produced args.itr already
applied these
+ // existing deletes, so the incoming batches hold only the new
positions;
+ // without the seed the replacement DV would silently resurrect
rows. The
+ // spec allows at most one DV per data file, so the caller
supersedes the
+ // old DV once this merged one is written.
hasEntries := false
+ for filePath, bitmap := range args.existingDVs {
+ pCtx, ok := partitionContextByFilePath[filePath]
+ if !ok {
+ yield(nil, fmt.Errorf("unexpected missing
partition context for path %s", filePath))
Review Comment:
This message is identical to the one in the batch path just below, so in
logs there's no telling which fired. A distinguishing string for the seed path
would save some head-scratching later.
##########
table/transaction.go:
##########
@@ -1891,11 +1892,36 @@ func (t *Transaction) writePositionDeletesForFiles(ctx
context.Context, fs io.IO
partitionContextByFilePath[df.FilePath()] =
partitionContext{partitionData: df.Partition(), specID: df.SpecID()}
}
+ // V3 tables store deletes as deletion vectors, and the spec allows at
most
+ // one DV per data file. A data file being partially deleted may already
+ // carry a DV from an earlier delete; those existing positions are
folded
+ // into the new DV (the scan that produced the records already applied
them,
+ // so the new records hold only the incremental deletes) and the
superseded
+ // DVs are removed. Without this a second delete on the same file would
+ // write a second live DV, corrupting the table (buildDVIndex rejects
it).
+ var existingDVs map[string]iceberg.DataFile
+ if t.meta.formatVersion >= 3 {
+ existingDVs, err = t.collectExistingDVs(fs, files)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ existingDVBitmaps := make(map[string]*dv.RoaringPositionBitmap,
len(existingDVs))
Review Comment:
The `formatVersion >= 3` guard covers the `collectExistingDVs` call, but
this alloc + `ReadDV` loop still runs on v1/v2 (harmless — `existingDVs` is nil
so the loop is len-0). Moving the `make` and loop inside the v3 block keeps the
whole existing-DV machinery behind one guard.
--
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]
