You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/12 20:59:34 UTC
(iceberg-go) branch main updated: feat(table): RowDelta.RemoveDeletes for deletion-vector supersession (#1785)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 19cbebea3 feat(table): RowDelta.RemoveDeletes for deletion-vector supersession (#1785)
19cbebea3 is described below
commit 19cbebea3c487583d00fe90e2c705aeaf928e95e
Author: ryanworl <ry...@gmail.com>
AuthorDate: Wed Aug 12 15:59:28 2026 -0500
feat(table): RowDelta.RemoveDeletes for deletion-vector supersession (#1785)
* feat(table): add Positions iterator to RoaringPositionBitmap
RoaringPositionBitmap exposes Set, Contains, and Cardinality but no way
to enumerate the positions it holds. Consumers that need the contents
of an existing deletion vector (e.g. to merge it into a superseding DV,
as the v3 spec requires when a data file's DV is replaced) currently
have to Serialize() the bitmap and re-parse the Iceberg portable wire
format by hand.
Add Positions() iter.Seq[uint64], yielding every set position in
ascending 64-bit order. The yielded values are the same ones passed to
Set, mirroring Java's RoaringPositionBitmap#forEach.
Tests cover the empty bitmap, a single position above the 32-bit
boundary, ascending order across multiple bitmap keys with agreement
against Cardinality, and early termination by the consumer.
Co-authored-by: Cursor <cu...@cursor.com>
* feat(table): support deletion vector supersession in RowDelta
The v3 spec allows at most one live deletion vector per data file. A
writer adding a DV for a data file that already carries one must write
a DV containing the previous DV's positions and remove the previous DV
entry in the same snapshot. Java's RowDelta supports this through
removeDeletes; the Go RowDelta could only add delete files, forcing
consumers into a two-snapshot ReplaceFiles + RowDelta workaround whose
intermediate snapshot briefly resurrects the previously deleted rows.
The engine-internal merge-on-read delete path already supersedes DVs
this way (writePositionDeletesForFiles); this exposes the same
capability to library users who write their own deletion vectors.
Add RowDelta.RemoveDeletes(files ...iceberg.DataFile), mirroring Java's
RowDelta#removeDeletes, restricted to deletion vectors. Removal
identity is (file path, referenced data file), not path alone: a
multi-blob Puffin file legally carries DVs for several data files under
one path, one manifest entry each, and superseding one blob must not
drop its siblings' entries. The snapshot producer's existing
removeDeletionVector plumbing already keys removals this way, so only
the matching entry is recorded as DELETED while other live entries at
the same Puffin path survive.
Commit first resolves every removed file to its live entry in the
current snapshot's delete manifests — when several live entries share
the removed file's path, its referenced data file selects among them,
and the removal is rejected as ambiguous if it does not identify
exactly one — then validates against the RESOLVED entries, never the
caller's copies, so a stale or forged DataFile with a real path but a
wrong referenced_data_file cannot pair a removal with an unrelated
replacement:
- every resolved entry must be a DV (Puffin pos-delete with
referenced_data_file); expunging plain pos/eq-delete files remains
the rewrite path's job,
- every removed DV's referenced data file — as recorded by the live
entry — gets a replacement DV among the added delete files, so a
delta cannot silently resurrect rows (content coverage stays the
writer's responsibility, as in Java),
- a removed file that does not resolve to a live delete entry is an
error, as are two removals resolving to live entries that reference
the same data file (the producer keys DV removals by referenced
data file, and two live DVs on one data file is a spec violation to
surface, not paper over); the DELETED record is produced from the
manifest's own entry so it keeps its spec ID, partition, and
sequence numbers. Blob offset and size are not part of the
identity: the spec's one live DV per data file makes (path,
referenced data file) unique among live entries, and every produced
record comes from the live entry, so stale caller-side offsets are
harmless.
When removals are present the delta commits through the overwrite
producer, which already drops removed entries from inherited delete
manifests and records them with status DELETED; the overwrite
producer's default conflict validator is suppressed in favor of
RowDelta's own, the same way rewrites do. Without removals the
fast-append path is unchanged.
Also correct the stale comments on commitOpts.validators and RowDelta
claiming pre-flight validators run only on the first attempt; doCommit
runs them on every attempt of the retry loop against the freshly
refreshed catalog state.
The end-to-end test builds a v3 table with a live DV, commits a
superseding RowDelta, and asserts single-snapshot shape (operation
delete, added and removed delete file counts), exactly one live DV
with the superseded entry marked DELETED, and row visibility at plan
level (the scanner plans only the replacement DV and its bitmap hides
the union of positions) and through materialized Arrow records. A
shared multi-blob Puffin test supersedes one of two DVs sharing a
Puffin path and asserts the sibling entry stays live (its rows still
hidden) alongside the DELETED record for the same path, plus
superseding both blobs in one delta. Validation failures are covered
for a missing replacement, a non-DV live entry behind a DV-shaped
caller copy, a forged reference targeting an unrelated DV, stale
caller metadata pairing correctly per the live entry, ambiguous and
unmatched shared-path removals, duplicate removals, an unknown file,
and an empty table.
Co-authored-by: Cursor <cu...@cursor.com>
* docs(table): pin Positions iteration-safety semantics
Document that RoaringPositionBitmap.Positions captures the key set up
front while consulting each bucket lazily, so the bitmap must not be
mutated during iteration — consistent with the rest of the type, which
is not safe for concurrent use.
Co-authored-by: Cursor <cu...@cursor.com>
* fix(table): address review feedback on RowDelta DV supersession
Close the converse validation gap: the delta validated that every
removed DV has a replacement among the added delete files, but not
that every added replacement's data file has its live DV removed. A
supersession delta could add a replacement DV for a data file whose
live DV was not in RemoveDeletes, committing two live DVs on that
file — a v3 spec violation that scan planning then rejects. The
manifest walk the removals already perform now also collects every
live DV whose referenced data file gets a replacement in this delta,
and validation requires each to be among the resolved removals.
Deltas without removals do not pay for (or get) this check; plain
AddDeletes stays unvalidated, matching Java, and is called out as
follow-up work. The same pass also rejects two added replacement DVs
referencing one data file — guaranteed corruption regardless of
removals — and detects tables already corrupted with duplicate live
DV entries, both at one Puffin path (where the producer would
tombstone every matching entry while the summary counts one removal)
and across distinct paths (where the sibling duplicate would survive
next to the replacement).
Also from review:
- a removal-only delta now reaches the replacement-required error
instead of failing the empty-delta guard with a message that never
mentions removals,
- removals require format version >= 3 up front, instead of failing
resolution with a confusing does-not-belong lookup error on v2,
- a live Puffin DV entry without a recorded referenced data file
gets its own error instead of being reported as a non-DV,
- RemoveDeletes documents that a single live entry at the removed
path is selected regardless of the caller copy's reference (the
reference only disambiguates shared paths), and both RemoveDeletes
and Commit document that removal-carrying commits are
non-replayable: a CAS conflict fails with ErrCommitFailed and the
caller must rebuild against a reloaded table,
- the post-commit validator comment no longer claims the producer is
always fast-append.
New tests cover the surviving-live-DV rejection, duplicate
replacements for one data file, removal-only deltas, both
duplicate-live-DV corruption shapes, the v2 guard, and an end-to-end
RowDelta supersession race pinning fail-on-first-CAS-conflict with the
peer's DV as the sole survivor.
Co-authored-by: Cursor <cu...@cursor.com>
---------
Co-authored-by: Cursor <cu...@cursor.com>
---
table/dv/roaring_bitmap.go | 25 ++
table/dv/roaring_bitmap_test.go | 80 +++++
table/row_delta.go | 359 ++++++++++++++++++-
table/row_delta_test.go | 773 ++++++++++++++++++++++++++++++++++++++++
table/table.go | 7 +-
5 files changed, 1225 insertions(+), 19 deletions(-)
diff --git a/table/dv/roaring_bitmap.go b/table/dv/roaring_bitmap.go
index d01dda0c4..1436d1c32 100644
--- a/table/dv/roaring_bitmap.go
+++ b/table/dv/roaring_bitmap.go
@@ -22,6 +22,9 @@ import (
"encoding/binary"
"fmt"
"io"
+ "iter"
+ "maps"
+ "slices"
"sort"
"github.com/RoaringBitmap/roaring/v2"
@@ -137,6 +140,28 @@ func (b *RoaringPositionBitmap) Contains(pos uint64) bool {
return bm.Contains(low)
}
+// Positions returns an iterator over every set position in ascending
+// order. The positions yielded are the same 64-bit values passed to Set,
+// mirroring Java's RoaringPositionBitmap#forEach.
+//
+// The bitmap must not be modified while iterating: the key set is
+// captured up front while each bucket is consulted lazily, so a
+// concurrent or re-entrant Set/Or produces an inconsistent view (as
+// with the rest of the type, which is not safe for concurrent use).
+func (b *RoaringPositionBitmap) Positions() iter.Seq[uint64] {
+ return func(yield func(uint64) bool) {
+ for _, key := range slices.Sorted(maps.Keys(b.bitmaps)) {
+ high := uint64(key) << 32
+ it := b.bitmaps[key].Iterator()
+ for it.HasNext() {
+ if !yield(high | uint64(it.Next())) {
+ return
+ }
+ }
+ }
+ }
+}
+
// IsEmpty returns true if no positions are set. Returns true both for the
// no-bucket case and for the (currently impossible-via-public-API) case
// where a bucket exists but its inner roaring bitmap has zero cardinality
diff --git a/table/dv/roaring_bitmap_test.go b/table/dv/roaring_bitmap_test.go
index 9588c5652..11f963978 100644
--- a/table/dv/roaring_bitmap_test.go
+++ b/table/dv/roaring_bitmap_test.go
@@ -618,3 +618,83 @@ func TestRoaringBitmapEmptyRoundTrip(t *testing.T) {
assert.True(t, got.IsEmpty())
assert.Equal(t, int64(0), got.Cardinality())
}
+
+// Why: Positions is the supported way to enumerate a deletion vector's
+// contents; without it consumers must round-trip the serialized format.
+// Condition: iterate an empty bitmap.
+// Assertion: no positions are yielded.
+func TestRoaringBitmapPositionsEmpty(t *testing.T) {
+ bm := NewRoaringPositionBitmap()
+
+ for pos := range bm.Positions() {
+ t.Fatalf("empty bitmap yielded position %d", pos)
+ }
+}
+
+// Why: the single-position case pins down that Positions yields exactly what Set stored.
+// Condition: set one position above the 32-bit boundary and iterate.
+// Assertion: exactly that position is yielded.
+func TestRoaringBitmapPositionsSingle(t *testing.T) {
+ bm := NewRoaringPositionBitmap()
+ want := (uint64(7) << 32) | 123
+ bm.Set(want)
+
+ var got []uint64
+ for pos := range bm.Positions() {
+ got = append(got, pos)
+ }
+
+ assert.Equal(t, []uint64{want}, got)
+}
+
+// Why: positions spanning multiple 32-bit keys must come back in ascending
+// 64-bit order, with the count agreeing with Cardinality.
+// Condition: set positions across keys 0, 1, and 5 (crossing the 32-bit
+// boundary) in shuffled insertion order, then iterate.
+// Assertion: yielded positions are the sorted originals and their count
+// equals Cardinality.
+func TestRoaringBitmapPositionsAscendingAcrossKeys(t *testing.T) {
+ positions := []uint64{
+ (uint64(5) << 32) | 1,
+ 0,
+ (uint64(1) << 32) | 42,
+ 65535,
+ uint64(5) << 32,
+ (uint64(1) << 32) | 9999,
+ 1,
+ }
+
+ bm := NewRoaringPositionBitmap()
+ for _, pos := range positions {
+ bm.Set(pos)
+ }
+
+ var got []uint64
+ for pos := range bm.Positions() {
+ got = append(got, pos)
+ }
+
+ want := slices.Clone(positions)
+ slices.Sort(want)
+ assert.Equal(t, want, got)
+ assert.Equal(t, bm.Cardinality(), int64(len(got)))
+}
+
+// Why: iter.Seq consumers may stop early (e.g. range-over-func with break);
+// the iterator must honor that instead of continuing to yield.
+// Condition: set positions in two keys and break after the first yield.
+// Assertion: exactly one position (the smallest) is observed.
+func TestRoaringBitmapPositionsEarlyBreak(t *testing.T) {
+ bm := NewRoaringPositionBitmap()
+ bm.Set(10)
+ bm.Set((uint64(2) << 32) | 3)
+
+ var got []uint64
+ for pos := range bm.Positions() {
+ got = append(got, pos)
+
+ break
+ }
+
+ assert.Equal(t, []uint64{10}, got)
+}
diff --git a/table/row_delta.go b/table/row_delta.go
index 90db6c6c3..09d2125f3 100644
--- a/table/row_delta.go
+++ b/table/row_delta.go
@@ -24,11 +24,14 @@ import (
"maps"
"github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
)
// RowDelta encodes a set of row-level changes to a table: new data files
-// (inserts) and delete files (equality or position deletes). All changes
-// are committed atomically in a single snapshot.
+// (inserts) and delete files (equality or position deletes). A delta
+// that replaces a data file's deletion vector can remove the superseded
+// DV via [RowDelta.RemoveDeletes]. All changes are committed atomically
+// in a single snapshot.
//
// The operation type of the produced snapshot is determined automatically:
// - Data files only → OpAppend
@@ -53,8 +56,11 @@ import (
// any concurrent append is a conflict). Opt out by setting
// write.delete.isolation-level=snapshot.
//
-// Refresh-and-replay between retries is deferred to a follow-up PR;
-// today the pre-flight runs once on the first attempt.
+// The pre-flight runs before cat.CommitTable on every commit attempt.
+// On the first attempt the writer's view and the catalog state are the
+// same, so there are no concurrent snapshots to inspect and the checks
+// short-circuit; on retries doCommit refreshes the catalog state and
+// the checks run against the freshly loaded branch head.
//
// Usage:
//
@@ -63,10 +69,11 @@ import (
// rd.AddDeletes(equalityDeleteFile1)
// err := rd.Commit(ctx)
type RowDelta struct {
- txn *Transaction
- dataFiles []iceberg.DataFile
- delFiles []iceberg.DataFile
- props iceberg.Properties
+ txn *Transaction
+ dataFiles []iceberg.DataFile
+ delFiles []iceberg.DataFile
+ removedDels []iceberg.DataFile
+ props iceberg.Properties
}
// NewRowDelta creates a new RowDelta for committing row-level changes
@@ -96,17 +103,67 @@ func (rd *RowDelta) AddDeletes(files ...iceberg.DataFile) *RowDelta {
return rd
}
+// RemoveDeletes marks superseded deletion vectors as removed in the
+// snapshot produced by this RowDelta, mirroring Java's
+// RowDelta#removeDeletes.
+//
+// The v3 spec allows at most one live deletion vector per data file: a
+// writer adding a DV for a data file that already carries one must
+// write a DV containing all of the previous DV's positions and remove
+// the previous DV in the same snapshot. RemoveDeletes is the removal
+// half of that contract; Commit validates that every removed DV's
+// referenced data file gets a replacement DV among the files passed to
+// AddDeletes, so the delta cannot silently resurrect deleted rows.
+//
+// Only deletion vectors (Puffin position deletes with a referenced
+// data file) can be removed here. Expunging fully-applied position or
+// equality delete files belongs to the rewrite path
+// ([Transaction.ReplaceFiles] / [RewriteFiles]), which owns the
+// data-file replacement validation that makes such removals safe.
+//
+// Removal identity is (file path, referenced data file): a multi-blob
+// Puffin file may carry deletion vectors for several data files under
+// one path, and removing one of its entries leaves the others live.
+// When exactly one live entry carries the removed file's path, it is
+// selected regardless of the caller copy's (possibly stale) reference;
+// when several live entries share the path, the removed file's
+// ReferencedDataFile selects which entry to remove, and Commit rejects
+// the removal as ambiguous if it does not identify exactly one.
+//
+// Removal resolution is snapshot-relative, so a commit carrying
+// removals is never replayed on an optimistic-concurrency conflict: if
+// a concurrent writer advances the branch first, Commit's transaction
+// fails with ErrCommitFailed instead of retrying, and the caller must
+// reload the table and rebuild the delta against the fresh snapshot.
+func (rd *RowDelta) RemoveDeletes(files ...iceberg.DataFile) *RowDelta {
+ rd.removedDels = append(rd.removedDels, files...)
+
+ return rd
+}
+
// Commit validates and commits all accumulated row-level changes as a
// single atomic snapshot. Returns an error if there are no files to
-// commit, if any file has an unexpected content type, or if the table
-// format version does not support delete files.
+// commit, if any file has an unexpected content type, if a removed
+// deletion vector lacks a replacement or is not referenced by the
+// current snapshot, if the delta would leave two live deletion vectors
+// on one data file (a replacement added while the superseded live DV
+// is not removed, or two added replacements for one data file), or if
+// the table format version does not support delete files.
+//
+// With removals present the staged commit is non-replayable: an
+// optimistic-concurrency conflict fails the transaction's Commit with
+// ErrCommitFailed rather than refreshing and retrying (see
+// RemoveDeletes).
func (rd *RowDelta) Commit(ctx context.Context) error {
meta, err := rd.txn.txnMeta()
if err != nil {
return err
}
- if len(rd.dataFiles) == 0 && len(rd.delFiles) == 0 {
+ // A removal-only delta deliberately passes this guard: it fails
+ // later in validateRemovedDeletes with the replacement-required
+ // error, which tells the caller what is actually missing.
+ if len(rd.dataFiles) == 0 && len(rd.delFiles) == 0 && len(rd.removedDels) == 0 {
return errors.New("row delta must have at least one data file or delete file")
}
@@ -116,6 +173,13 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
meta.formatVersion)
}
+ // Deletion vectors — the only removable delete files — exist only
+ // in format version >= 3.
+ if len(rd.removedDels) > 0 && meta.formatVersion < 3 {
+ return fmt.Errorf("removing deletion vectors requires table format version >= 3, got v%d",
+ meta.formatVersion)
+ }
+
for _, f := range rd.dataFiles {
if f.ContentType() != iceberg.EntryContentData {
return fmt.Errorf("expected data file, got content type %s: %s",
@@ -157,7 +221,43 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
}
op := rd.Operation()
- producer := newFastAppendFilesProducer(op, rd.txn, wfs, nil, rd.props)
+
+ var producer *snapshotProducer
+ if len(rd.removedDels) > 0 {
+ // Resolve the removed files against the current snapshot's
+ // delete manifests so both validation and the produced DELETED
+ // entries work from the manifest's own DataFile (correct
+ // content type, referenced data file, spec ID, partition, and
+ // sequence numbers) rather than the caller's copy, which may
+ // carry stale or forged metadata. replacedLive collects the
+ // live DVs whose referenced data file this delta adds a
+ // replacement for; validateRemovedDeletes checks each is
+ // actually removed.
+ resolvedRemovals, replacedLive, err := rd.resolveRemovedDeletes(fs, meta)
+ if err != nil {
+ return err
+ }
+
+ if err := rd.validateRemovedDeletes(resolvedRemovals, replacedLive); err != nil {
+ return err
+ }
+
+ // The overwrite producer knows how to drop the removed entries
+ // from inherited delete manifests and to record them with
+ // status DELETED in the produced snapshot. Its default
+ // conflict validator (any concurrent append conflicts under
+ // serializable isolation) implements overwrite semantics, not
+ // row-delta semantics; RowDelta registers its own validator
+ // below, so suppress the default the same way rewrites do.
+ producer = newOverwriteFilesProducer(op, rd.txn, wfs, nil, rd.props)
+ producer.producerImpl.(*overwriteFiles).skipDefaultValidator = true
+
+ for _, live := range resolvedRemovals {
+ producer.removeDeletionVector(live)
+ }
+ } else {
+ producer = newFastAppendFilesProducer(op, rd.txn, wfs, nil, rd.props)
+ }
for _, f := range rd.dataFiles {
producer.appendDataFile(f)
@@ -173,17 +273,242 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
}
// Register RowDelta's pre-commit conflict validator. The underlying
- // fast-append producer's validator is a no-op; RowDelta semantics
- // (pos-delete references, eq-delete predicate) require a dedicated
- // check that snapshot_producers does not know about.
+ // producer contributes no conflict check of its own — fast-append's
+ // validator is a no-op, and the overwrite producer's default is
+ // suppressed above; RowDelta semantics (pos-delete references,
+ // eq-delete predicate) require a dedicated check that
+ // snapshot_producers does not know about.
rd.txn.addValidator(rd.validate)
return rd.txn.apply(updates, reqs)
}
+// pathRefKey is the removal identity of a delete-manifest entry: its
+// file path plus the data file its deletion vector references (""
+// for delete files that record none).
+type pathRefKey struct{ path, ref string }
+
+// explicitReferencedDataFile returns df's referenced_data_file field,
+// or "" when unset. Unlike referencedDataFilePath it does not fall back
+// to file_path column bounds: it is used for removal identity, where a
+// bounds-derived guess must not stand in for the recorded reference.
+func explicitReferencedDataFile(df iceberg.DataFile) string {
+ if ref := df.ReferencedDataFile(); ref != nil {
+ return *ref
+ }
+
+ return ""
+}
+
+// validateRemovedDeletes enforces the DV-supersession contract against
+// the RESOLVED live entries returned by resolveRemovedDeletes. The
+// caller's copies are deliberately not consulted: a stale or forged
+// DataFile with a real path but a wrong referenced_data_file must not
+// be able to pair a removal with an unrelated replacement.
+//
+// - at most one added DV may reference a given data file; two
+// replacements for one data file would commit two live DVs, which
+// the v3 spec forbids and scan planning rejects.
+// - every removed entry must be a deletion vector (a Puffin
+// pos-delete file with referenced_data_file set); other delete
+// files are removed through the rewrite path, which validates them
+// differently.
+// - every removed DV's referenced data file — as recorded by the
+// live manifest entry — must get a replacement DV among the added
+// delete files. The v3 spec requires the superseding DV to be
+// committed in the same snapshot as the removal; without a
+// replacement the delta would resurrect the rows the removed DV
+// was hiding. The replacement must contain the removed DV's
+// positions — that is the writer's responsibility, as in Java;
+// only the metadata pairing is validated here.
+// - conversely, every live DV whose referenced data file gets a
+// replacement in this delta (replacedLive) must be among the
+// removals — otherwise the commit would leave both the old and the
+// new DV live on one data file.
+func (rd *RowDelta) validateRemovedDeletes(resolved, replacedLive []iceberg.DataFile) error {
+ addedDVs := make(map[string]struct{}, len(rd.delFiles))
+ for _, f := range rd.delFiles {
+ if ref := explicitReferencedDataFile(f); IsDeletionVector(f) && ref != "" {
+ if _, ok := addedDVs[ref]; ok {
+ return fmt.Errorf("multiple added deletion vectors reference data file %s; at most one live deletion vector may exist per data file",
+ ref)
+ }
+ addedDVs[ref] = struct{}{}
+ }
+ }
+
+ removedKeys := make(map[pathRefKey]struct{}, len(resolved))
+ for _, live := range resolved {
+ ref := explicitReferencedDataFile(live)
+ if !IsDeletionVector(live) {
+ return fmt.Errorf("only deletion vectors can be removed by a row delta, got %s file with content type %s: %s",
+ live.FileFormat(), live.ContentType(), live.FilePath())
+ }
+ if ref == "" {
+ return fmt.Errorf("cannot remove %s: the live deletion vector entry does not record a referenced data file",
+ live.FilePath())
+ }
+
+ if _, ok := addedDVs[ref]; !ok {
+ return fmt.Errorf("cannot remove deletion vector %s: no replacement deletion vector for data file %s in this row delta",
+ live.FilePath(), ref)
+ }
+ removedKeys[pathRefKey{path: live.FilePath(), ref: ref}] = struct{}{}
+ }
+
+ for _, live := range replacedLive {
+ key := pathRefKey{path: live.FilePath(), ref: explicitReferencedDataFile(live)}
+ if _, ok := removedKeys[key]; !ok {
+ return fmt.Errorf("cannot add a replacement deletion vector for data file %s: live deletion vector %s is not removed by this row delta; the superseded entry must be removed in the same snapshot",
+ key.ref, key.path)
+ }
+ }
+
+ return nil
+}
+
+// resolveRemovedDeletes walks the current snapshot's delete manifests
+// and resolves each removed file to its live manifest entry, returned
+// in the order the removals were registered.
+//
+// Removal identity is (file path, referenced data file), not path
+// alone: a multi-blob Puffin file legally carries deletion vectors for
+// several data files under one path, one manifest entry each, and
+// removing one blob's entry must not drop its siblings. When a single
+// live entry carries the path, the caller's copy resolves to it
+// regardless of its (possibly stale) metadata; when several live
+// entries share the path, the caller's referenced data file selects
+// among them and the removal is rejected as ambiguous if it does not
+// identify exactly one. Blob offset and size are not part of the
+// identity: the spec allows at most one live DV per data file, so
+// (path, referenced data file) is unique among live entries, and every
+// record this commit produces is built from the live entry's own
+// DataFile, so stale caller-side offsets cannot corrupt the removal.
+//
+// A removed file that does not resolve to a live delete entry is an
+// error: the producer would otherwise silently skip the removal and
+// the commit would strand two live DVs on one data file. Two removals
+// resolving to live entries that reference the same data file are also
+// an error — the producer keys DV removals by referenced data file, so
+// letting both through would silently drop one, and two live DVs on
+// one data file is a spec violation to surface, not paper over. For
+// the same reason a removal whose referenced data file matches more
+// than one live entry at the path is rejected: the producer would
+// tombstone every matching entry while the snapshot summary counts
+// only one removal.
+//
+// The walk also collects replacedLive: every live deletion vector
+// whose referenced data file this delta adds a replacement DV for.
+// validateRemovedDeletes requires each to be among the removals, so a
+// delta cannot commit a replacement while leaving the superseded DV
+// live. This piggybacks on the manifest walk the removals already
+// need; deltas without removals do not pay for it (nor get it — see
+// AddDeletes).
+func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta *MetadataBuilder) (resolved, replacedLive []iceberg.DataFile, _ error) {
+ snap := meta.currentSnapshot()
+ if snap == nil {
+ return nil, nil, errors.New("cannot remove delete files from a table without an existing snapshot")
+ }
+
+ want := make(map[string]struct{}, len(rd.removedDels))
+ for _, f := range rd.removedDels {
+ want[f.FilePath()] = struct{}{}
+ }
+
+ addedRefs := make(map[string]struct{}, len(rd.delFiles))
+ for _, f := range rd.delFiles {
+ if ref := explicitReferencedDataFile(f); IsDeletionVector(f) && ref != "" {
+ addedRefs[ref] = struct{}{}
+ }
+ }
+
+ liveByPath := make(map[string][]iceberg.DataFile, len(rd.removedDels))
+ for entry, err := range snap.entries(fs, iceberg.ManifestContentDeletes) {
+ if err != nil {
+ return nil, nil, err
+ }
+ if entry.Status() == iceberg.EntryStatusDELETED {
+ continue
+ }
+ df := entry.DataFile()
+ if _, ok := want[df.FilePath()]; ok {
+ liveByPath[df.FilePath()] = append(liveByPath[df.FilePath()], df)
+ }
+ if ref := explicitReferencedDataFile(df); IsDeletionVector(df) && ref != "" {
+ if _, ok := addedRefs[ref]; ok {
+ replacedLive = append(replacedLive, df)
+ }
+ }
+ }
+
+ seenKeys := make(map[pathRefKey]struct{}, len(rd.removedDels))
+ seenRefs := make(map[string]string, len(rd.removedDels)) // live ref → live path
+ resolved = make([]iceberg.DataFile, len(rd.removedDels))
+ for i, f := range rd.removedDels {
+ candidates := liveByPath[f.FilePath()]
+
+ var live iceberg.DataFile
+ switch {
+ case len(candidates) == 0:
+ return nil, nil, fmt.Errorf("cannot remove delete files that do not belong to the table: %s", f.FilePath())
+ case len(candidates) == 1:
+ live = candidates[0]
+ default:
+ ref := explicitReferencedDataFile(f)
+ if ref == "" {
+ return nil, nil, fmt.Errorf("ambiguous removal: %d live delete entries share path %s; the removed file must declare a referenced data file to identify one",
+ len(candidates), f.FilePath())
+ }
+ matches := 0
+ for _, c := range candidates {
+ if explicitReferencedDataFile(c) == ref {
+ if live == nil {
+ live = c
+ }
+ matches++
+ }
+ }
+ switch {
+ case matches == 0:
+ return nil, nil, fmt.Errorf("cannot remove delete file %s: no live delete entry references data file %s",
+ f.FilePath(), ref)
+ case matches > 1:
+ return nil, nil, fmt.Errorf("found %d live delete entries at %s referencing data file %s; the table has duplicate live deletion vectors for one data file",
+ matches, f.FilePath(), ref)
+ }
+ }
+
+ liveRef := explicitReferencedDataFile(live)
+ key := pathRefKey{path: live.FilePath(), ref: liveRef}
+ if _, ok := seenKeys[key]; ok {
+ return nil, nil, fmt.Errorf("removed delete files must be unique: %s (referenced data file %q)",
+ live.FilePath(), liveRef)
+ }
+ seenKeys[key] = struct{}{}
+ // The producer keys DV removals by referenced data file, so two
+ // distinct live entries sharing a ref (a spec violation: two
+ // live DVs on one data file) would silently collapse to one
+ // removal. Surface the corruption instead. Non-DV entries (ref
+ // "") are rejected by validateRemovedDeletes with a clearer
+ // error, so they are exempt here.
+ if liveRef != "" {
+ if prevPath, ok := seenRefs[liveRef]; ok {
+ return nil, nil, fmt.Errorf("removed delete files %s and %s both reference data file %q; the table has two live deletion vectors for one data file",
+ prevPath, live.FilePath(), liveRef)
+ }
+ seenRefs[liveRef] = live.FilePath()
+ }
+ resolved[i] = live
+ }
+
+ return resolved, replacedLive, nil
+}
+
// validate is the client-side conflict check for a RowDelta commit. It
-// runs against cc, which reflects the branch state at the first commit
-// attempt. Two invariants are enforced:
+// runs on every commit attempt: on attempt 0 cc's base and current
+// state coincide (nothing concurrent to check), and on retries cc
+// reflects the freshly refreshed branch state. Two invariants are
+// enforced:
//
// - Every data file referenced by a position-delete in this RowDelta
// must still be reachable from the branch head. A concurrent
diff --git a/table/row_delta_test.go b/table/row_delta_test.go
index 5cb30a4e2..3705371b8 100644
--- a/table/row_delta_test.go
+++ b/table/row_delta_test.go
@@ -33,6 +33,7 @@ import (
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/dv"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -631,6 +632,778 @@ func TestRowDeltaIntegrationPosDeleteRoundTrip(t *testing.T) {
assert.Equal(t, []string{"alpha", "gamma", "epsilon"}, data)
}
+// buildDVFile builds a deletion-vector manifest entry (puffin pos-delete
+// with a referenced data file) without writing any bytes; for tests that
+// only exercise metadata validation.
+func buildDVFile(t *testing.T, path, refDataFile string) iceberg.DataFile {
+ t.Helper()
+
+ b, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ path, iceberg.PuffinFile, nil, nil, nil, 2, 128)
+ require.NoError(t, err)
+
+ return b.ReferencedDataFile(refDataFile).
+ ContentOffset(4).
+ ContentSizeInBytes(64).
+ Build()
+}
+
+// writeDV writes a single-blob deletion vector puffin file for dataPath
+// and returns its manifest entry.
+func writeDV(t *testing.T, location, name, dataPath string, positions []int64) iceberg.DataFile {
+ t.Helper()
+
+ w := dv.NewDVWriter(iceio.LocalFS{}, func(int32) *iceberg.PartitionSpec {
+ return iceberg.UnpartitionedSpec
+ })
+ require.NoError(t, w.Add(dataPath, positions, 0, nil))
+ files, err := w.Flush(t.Context(), location+"/data/"+name)
+ require.NoError(t, err)
+ require.Len(t, files, 1)
+
+ return files[0]
+}
+
+// snapshotDeleteEntryFiles walks snap's delete manifests and returns
+// the DataFiles of live (non-DELETED) and removed (DELETED) entries.
+func snapshotDeleteEntryFiles(t *testing.T, snap *table.Snapshot, fs iceio.IO) (live, removed []iceberg.DataFile) {
+ t.Helper()
+
+ manifests, err := snap.Manifests(fs)
+ require.NoError(t, err)
+
+ for _, m := range manifests {
+ if m.ManifestContent() != iceberg.ManifestContentDeletes {
+ continue
+ }
+ for e, err := range m.Entries(fs, false) {
+ require.NoError(t, err)
+ if e.Status() == iceberg.EntryStatusDELETED {
+ removed = append(removed, e.DataFile())
+ } else {
+ live = append(live, e.DataFile())
+ }
+ }
+ }
+
+ return live, removed
+}
+
+// snapshotDVEntries walks snap's delete manifests and returns the paths
+// of live (non-DELETED) and removed (DELETED) entries.
+func snapshotDVEntries(t *testing.T, snap *table.Snapshot, fs iceio.IO) (live, removed []string) {
+ t.Helper()
+
+ liveFiles, removedFiles := snapshotDeleteEntryFiles(t, snap, fs)
+ for _, df := range liveFiles {
+ live = append(live, df.FilePath())
+ }
+ for _, df := range removedFiles {
+ removed = append(removed, df.FilePath())
+ }
+
+ return live, removed
+}
+
+// refOf returns df's referenced data file, or "" when unset.
+func refOf(df iceberg.DataFile) string {
+ if r := df.ReferencedDataFile(); r != nil {
+ return *r
+ }
+
+ return ""
+}
+
+// pathRefPairs maps entries to [path, referenced data file] pairs for
+// order-insensitive assertions on shared-Puffin manifests.
+func pathRefPairs(files []iceberg.DataFile) [][2]string {
+ pairs := make([][2]string, 0, len(files))
+ for _, df := range files {
+ pairs = append(pairs, [2]string{df.FilePath(), refOf(df)})
+ }
+
+ return pairs
+}
+
+// newTableWithLiveDV builds a v3 table containing one data file with two
+// rows and a live deletion vector hiding position 0, committed across
+// two snapshots. Returns the table, its location, the data file path,
+// and the live DV's manifest entry.
+func newTableWithLiveDV(t *testing.T) (*table.Table, string, string, iceberg.DataFile) {
+ t.Helper()
+
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := location + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[
+ {"id": 1, "data": "alpha"},
+ {"id": 2, "data": "beta"}
+ ]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ dv1 := writeDV(t, location, "dv-001.puffin", dataPath, []int64{0})
+ tx2 := tbl.NewTransaction()
+ require.NoError(t, tx2.NewRowDelta(nil).AddDeletes(dv1).Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+
+ return tbl, location, dataPath, dv1
+}
+
+// newTableWithSharedPuffinDVs builds a v3 table with two 2-row data
+// files whose deletion vectors (each hiding position 0) share a single
+// multi-blob Puffin file: two live manifest entries with the same
+// FilePath and distinct referenced data files. Returns the table, its
+// location, the two data paths, and the two DV entries.
+func newTableWithSharedPuffinDVs(t *testing.T) (*table.Table, string, string, string, iceberg.DataFile, iceberg.DataFile) {
+ t.Helper()
+
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ pathA := location + "/data/data-a.parquet"
+ pathB := location + "/data/data-b.parquet"
+ writeParquetFile(t, pathA, arrowSc, `[{"id": 1, "data": "alpha"}, {"id": 2, "data": "beta"}]`)
+ writeParquetFile(t, pathB, arrowSc, `[{"id": 3, "data": "gamma"}, {"id": 4, "data": "delta"}]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{pathA, pathB}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ w := dv.NewDVWriter(iceio.LocalFS{}, func(int32) *iceberg.PartitionSpec {
+ return iceberg.UnpartitionedSpec
+ })
+ require.NoError(t, w.Add(pathA, []int64{0}, 0, nil))
+ require.NoError(t, w.Add(pathB, []int64{0}, 0, nil))
+ files, err := w.Flush(t.Context(), location+"/data/dv-shared.puffin")
+ require.NoError(t, err)
+ require.Len(t, files, 2)
+
+ var dvA, dvB iceberg.DataFile
+ for _, f := range files {
+ switch refOf(f) {
+ case pathA:
+ dvA = f
+ case pathB:
+ dvB = f
+ }
+ }
+ require.NotNil(t, dvA)
+ require.NotNil(t, dvB)
+ require.Equal(t, dvA.FilePath(), dvB.FilePath(), "both DVs must share one Puffin path")
+
+ tx2 := tbl.NewTransaction()
+ require.NoError(t, tx2.NewRowDelta(nil).AddDeletes(dvA, dvB).Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+
+ return tbl, location, pathA, pathB, dvA, dvB
+}
+
+// scanSingleDVTask plans a scan of tbl and returns its single task's
+// deletion-vector file paths.
+func scanSingleDVTask(t *testing.T, tbl *table.Table) []string {
+ t.Helper()
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ dvPaths := make([]string, 0, len(tasks[0].DeletionVectorFiles))
+ for _, df := range tasks[0].DeletionVectorFiles {
+ dvPaths = append(dvPaths, df.FilePath())
+ }
+
+ return dvPaths
+}
+
+// Why: the v3 spec allows one live DV per data file and requires a writer
+// replacing a DV to add the superseding DV and remove the old one in the
+// SAME snapshot; two-snapshot workarounds briefly resurrect rows.
+// Condition: a v3 table with 5 rows and a live DV hiding position 1; a
+// RowDelta adds a superseding DV for positions {1,3} and removes the old
+// DV, then the transaction commits.
+// Assertion: exactly one snapshot is produced with operation delete, the
+// delete manifests carry one live DV (the replacement) and the superseded
+// DV with status DELETED, the scanner plans exactly the replacement DV,
+// whose contents hide positions 1 and 3, and the materialized scan
+// returns only the three surviving rows.
+func TestRowDeltaDVSupersessionSingleSnapshot(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := location + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[
+ {"id": 1, "data": "alpha"},
+ {"id": 2, "data": "beta"},
+ {"id": 3, "data": "gamma"},
+ {"id": 4, "data": "delta"},
+ {"id": 5, "data": "epsilon"}
+ ]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ assertRowCount(t, tbl, 5)
+
+ // First DV: hide position 1 ("beta").
+ dv1 := writeDV(t, location, "dv-001.puffin", dataPath, []int64{1})
+ tx2 := tbl.NewTransaction()
+ require.NoError(t, tx2.NewRowDelta(nil).AddDeletes(dv1).Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, []string{dv1.FilePath()}, scanSingleDVTask(t, tbl))
+ assertRowCount(t, tbl, 4)
+
+ snapsBefore := len(tbl.Metadata().Snapshots())
+
+ // Superseding DV: previous position 1 plus new position 3 ("delta").
+ dv2 := writeDV(t, location, "dv-002.puffin", dataPath, []int64{1, 3})
+ tx3 := tbl.NewTransaction()
+ require.NoError(t, tx3.NewRowDelta(nil).AddDeletes(dv2).RemoveDeletes(dv1).Commit(t.Context()))
+ tbl, err = tx3.Commit(t.Context())
+ require.NoError(t, err)
+
+ require.Len(t, tbl.Metadata().Snapshots(), snapsBefore+1,
+ "supersession must produce a single snapshot")
+
+ snap := tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ assert.Equal(t, table.OpDelete, snap.Summary.Operation)
+ assert.Equal(t, "1", snap.Summary.Properties["added-delete-files"])
+ assert.Equal(t, "1", snap.Summary.Properties["removed-delete-files"])
+
+ fs := iceio.LocalFS{}
+ liveDVs, removedDVs := snapshotDVEntries(t, snap, fs)
+ assert.Equal(t, []string{dv2.FilePath()}, liveDVs,
+ "the data file must carry exactly one live DV: the replacement")
+ assert.Equal(t, []string{dv1.FilePath()}, removedDVs,
+ "the superseded DV must be recorded as DELETED in the same snapshot")
+
+ // Row visibility: the scanner plans exactly the replacement DV, its
+ // bitmap hides positions 1 and 3, and the materialized scan returns
+ // the three surviving rows — ids 1/3/5.
+ assert.Equal(t, []string{dv2.FilePath()}, scanSingleDVTask(t, tbl))
+
+ bm, err := dv.ReadDV(fs, dv2)
+ require.NoError(t, err)
+ var hidden []uint64
+ for pos := range bm.Positions() {
+ hidden = append(hidden, pos)
+ }
+ assert.Equal(t, []uint64{1, 3}, hidden)
+
+ assertRowCount(t, tbl, 3)
+}
+
+// Why: removing a DV without publishing a superseding DV for the same
+// data file would resurrect the rows the removed DV was hiding.
+// Condition: a RowDelta removes a live DV while adding either no DV at
+// all or a DV that references a different data file.
+// Assertion: Commit fails with a replacement-required error before any
+// snapshot work happens.
+func TestRowDeltaRemoveDeletesRequiresReplacement(t *testing.T) {
+ t.Run("no added deletes", func(t *testing.T) {
+ tbl, _, _, dv1 := newTableWithLiveDV(t)
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddRows(buildDataFile(t, "s3://bucket/data/new.parquet")).
+ RemoveDeletes(dv1)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "no replacement deletion vector")
+ })
+
+ t.Run("replacement references a different data file", func(t *testing.T) {
+ tbl, location, _, dv1 := newTableWithLiveDV(t)
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(buildDVFile(t, location+"/data/dv-002.puffin", "s3://bucket/data/other.parquet")).
+ RemoveDeletes(dv1)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "no replacement deletion vector")
+ })
+
+ t.Run("removal-only delta", func(t *testing.T) {
+ tbl, _, _, dv1 := newTableWithLiveDV(t)
+ rd := tbl.NewTransaction().NewRowDelta(nil).RemoveDeletes(dv1)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "no replacement deletion vector",
+ "a removal-only delta must fail with the replacement-required error, not the empty-delta error")
+ })
+}
+
+// Why: committing a replacement DV while the data file's current live
+// DV stays live would put two live DVs on one data file, which the v3
+// spec forbids and scan planning rejects; the same holds for two added
+// replacements targeting one data file in a single delta.
+// Condition: a supersession delta that (a) adds a replacement DV for a
+// data file whose live DV is not in RemoveDeletes, or (b) adds two
+// replacement DVs referencing the same data file.
+// Assertion: Commit fails identifying the surviving live DV (a) or the
+// duplicated reference (b).
+func TestRowDeltaRemoveDeletesRejectsSurvivingLiveDV(t *testing.T) {
+ t.Run("replacement added while live DV not removed", func(t *testing.T) {
+ tbl, location, pathA, pathB, dvA, _ := newTableWithSharedPuffinDVs(t)
+
+ // Supersede A's DV, but also add a replacement for B without
+ // removing B's live DV.
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ dvB2 := writeDV(t, location, "dv-b2.puffin", pathB, []int64{0, 1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(dvA2, dvB2).
+ RemoveDeletes(dvA)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "is not removed by this row delta")
+ assert.Contains(t, err.Error(), pathB)
+ })
+
+ t.Run("two replacements for one data file", func(t *testing.T) {
+ tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+
+ rep1 := writeDV(t, location, "dv-002.puffin", dataPath, []int64{0, 1})
+ rep2 := buildDVFile(t, location+"/data/dv-003.puffin", dataPath)
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(rep1, rep2).
+ RemoveDeletes(dv1)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "multiple added deletion vectors reference data file "+dataPath)
+ })
+}
+
+// Why: a table that already violates the one-live-DV-per-data-file
+// invariant must fail a supersession commit loudly — the producer keys
+// DV removals by referenced data file, so proceeding would either
+// tombstone several entries while counting one removal (same path) or
+// leave the sibling duplicate live next to the replacement (different
+// paths).
+// Condition: v3 tables corrupted through the unvalidated plain
+// AddDeletes path so one data file carries two live DV entries, first
+// as duplicate entries at one Puffin path, then at two distinct paths;
+// a RowDelta then supersedes one of them.
+// Assertion: Commit fails naming the duplicate entries rather than
+// committing.
+func TestRowDeltaRemoveDeletesCorruptDuplicateLiveDVs(t *testing.T) {
+ t.Run("duplicate entries at one path", func(t *testing.T) {
+ tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+
+ // Corrupt the table: a second snapshot re-adds the same DV
+ // entry (same path, same referenced data file).
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(dv1).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ replacement := writeDV(t, location, "dv-002.puffin", dataPath, []int64{0, 1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(replacement).
+ RemoveDeletes(dv1)
+
+ err = rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "duplicate live deletion vectors")
+ })
+
+ t.Run("duplicate entries at two paths", func(t *testing.T) {
+ tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+
+ // Corrupt the table: a second live DV for the same data file
+ // at a different Puffin path.
+ dv1b := writeDV(t, location, "dv-001b.puffin", dataPath, []int64{0})
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(dv1b).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ replacement := writeDV(t, location, "dv-002.puffin", dataPath, []int64{0, 1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(replacement).
+ RemoveDeletes(dv1)
+
+ err = rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "is not removed by this row delta",
+ "the sibling duplicate at the other path must be flagged as surviving")
+ assert.Contains(t, err.Error(), dv1b.FilePath())
+ })
+}
+
+// Why: RemoveDeletes resolves removals against the snapshot the writer
+// built on, so a commit carrying them must fail on a CAS conflict
+// instead of refresh-and-replaying — a replay would inherit the peer's
+// replacement DV from the fresh base while the stale removal replays
+// as a no-op, stranding two live DVs on one data file.
+// Condition: two writers stage supersessions of the same live DV from
+// the same view via RowDelta.RemoveDeletes; the peer commits first;
+// the stale writer commits with retries enabled.
+// Assertion: the stale commit fails wrapping ErrCommitFailed after
+// exactly one CommitTable attempt, and the committed table carries
+// exactly one live DV — the peer's.
+func TestRowDeltaRemoveDeletesFailsInsteadOfReplaying(t *testing.T) {
+ ctx := t.Context()
+ tbl, cat := newNoReplayV3Table(t)
+ tbl = appendTenRows(t, tbl)
+ location := tbl.Location()
+
+ tasks, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ dataPath := tasks[0].File.FilePath()
+
+ // Baseline DV hides position 0.
+ dv1 := writeDV(t, location, "dv-001.puffin", dataPath, []int64{0})
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(dv1).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ // Stage the stale writer's supersession first, from the current view.
+ dvStale := writeDV(t, location, "dv-stale.puffin", dataPath, []int64{0, 2})
+ staleTxn := tbl.NewTransaction()
+ require.NoError(t, staleTxn.NewRowDelta(nil).AddDeletes(dvStale).RemoveDeletes(dv1).Commit(ctx))
+
+ // Peer supersedes dv1 from the same view and wins the race.
+ dvPeer := writeDV(t, location, "dv-peer.puffin", dataPath, []int64{0, 1})
+ peerTxn := tbl.NewTransaction()
+ require.NoError(t, peerTxn.NewRowDelta(nil).AddDeletes(dvPeer).RemoveDeletes(dv1).Commit(ctx))
+ _, err = peerTxn.Commit(ctx)
+ require.NoError(t, err)
+
+ attemptsBefore := cat.attempts.Load()
+ _, err = staleTxn.Commit(ctx)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrCommitFailed)
+ assert.Equal(t, attemptsBefore+1, cat.attempts.Load(),
+ "a RowDelta carrying DV removals must fail on the first CAS conflict, not refresh-and-replay")
+
+ // The table is uncorrupted: exactly one live DV — the peer's.
+ committed, err := cat.LoadTable(ctx, table.Identifier{"db", "no_replay_removals"})
+ require.NoError(t, err)
+ assert.Equal(t, []string{dvPeer.FilePath() + " -> " + dataPath},
+ deleteEntriesReferencing(t, committed, map[string]struct{}{dataPath: {}}),
+ "the data file must carry exactly one live DV: the peer's")
+}
+
+// Why: deletion vectors exist only in format v3; a v2 table cannot
+// carry the entries RemoveDeletes targets, and the error should say so
+// instead of failing resolution with a confusing lookup error.
+// Condition: a v2 table commits a RowDelta with a removal.
+// Assertion: Commit fails with a format-version error.
+func TestRowDeltaRemoveDeletesRequiresV3(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 2)
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(buildPosDeleteFile(t, "s3://bucket/data/pos-del.parquet")).
+ RemoveDeletes(buildDVFile(t, "s3://bucket/data/dv-001.puffin", "s3://bucket/data/data-001.parquet"))
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "format version >= 3")
+}
+
+// Why: only DV supersession is safe to express through a RowDelta;
+// expunging plain position or equality delete files needs the rewrite
+// path's data-file validation. The check must classify the LIVE entry,
+// not the caller's copy, so a stale copy cannot smuggle a non-DV
+// removal through.
+// Condition: the table's live delete entry is a plain Parquet
+// pos-delete file; a RowDelta removes it (caller copy claims it is a
+// DV) while adding a valid DV.
+// Assertion: Commit fails identifying the live entry as not a DV.
+func TestRowDeltaRemoveDeletesRejectsNonDV(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ posDelPath := location + "/data/pos-del-001.parquet"
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).
+ AddDeletes(buildPosDeleteFile(t, posDelPath)).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ // The caller's copy carries DV-shaped metadata for the live plain
+ // pos-delete's path; classification must follow the live entry.
+ disguised := buildDVFile(t, posDelPath, "s3://bucket/data/data-001.parquet")
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(buildDVFile(t, location+"/data/dv-001.puffin", "s3://bucket/data/data-001.parquet")).
+ RemoveDeletes(disguised)
+
+ err = rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "only deletion vectors can be removed")
+}
+
+// Why: replacement pairing must be validated against the referenced
+// data file recorded by the LIVE manifest entry; trusting the caller's
+// copy lets a stale or forged DataFile with a real path but a wrong
+// referenced_data_file remove an unrelated live DV.
+// Condition: removed files whose paths resolve to live DVs but whose
+// caller-side referenced_data_file metadata is wrong.
+// Assertion: a stale copy still pairs correctly per the live entry and
+// the commit succeeds; a forged reference cannot satisfy pairing with a
+// replacement for a different data file, so the unrelated DV survives.
+func TestRowDeltaRemoveDeletesStaleReference(t *testing.T) {
+ t.Run("stale metadata pairs against the live entry", func(t *testing.T) {
+ tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+
+ // Same path as the live DV, wrong referenced data file.
+ stale := buildDVFile(t, dv1.FilePath(), "s3://bucket/data/bogus.parquet")
+ replacement := writeDV(t, location, "dv-002.puffin", dataPath, []int64{0, 1})
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).
+ AddDeletes(replacement).RemoveDeletes(stale).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ live, removed := snapshotDVEntries(t, tbl.CurrentSnapshot(), iceio.LocalFS{})
+ assert.Equal(t, []string{replacement.FilePath()}, live,
+ "the live DV pairs with the replacement per the live entry's reference")
+ assert.Equal(t, []string{dv1.FilePath()}, removed)
+ })
+
+ t.Run("forged reference cannot remove an unrelated DV", func(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ pathA := location + "/data/data-a.parquet"
+ pathB := location + "/data/data-b.parquet"
+ writeParquetFile(t, pathA, arrowSc, `[{"id": 1, "data": "alpha"}, {"id": 2, "data": "beta"}]`)
+ writeParquetFile(t, pathB, arrowSc, `[{"id": 3, "data": "gamma"}, {"id": 4, "data": "delta"}]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{pathA, pathB}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ dvA := writeDV(t, location, "dv-a.puffin", pathA, []int64{0})
+ dvB := writeDV(t, location, "dv-b.puffin", pathB, []int64{0})
+ tx2 := tbl.NewTransaction()
+ require.NoError(t, tx2.NewRowDelta(nil).AddDeletes(dvA, dvB).Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+
+ // Forged copy: dvB's real path, but claims to reference data
+ // file A — for which the delta does add a replacement. Pairing
+ // against the caller's copy would remove dvB and strand data
+ // file B's deletes.
+ forged := buildDVFile(t, dvB.FilePath(), pathA)
+ replacementA := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(replacementA).RemoveDeletes(forged)
+
+ err = rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "no replacement deletion vector for data file "+pathB,
+ "pairing must use the live entry's referenced data file")
+
+ // The unrelated DV is untouched: still live on the branch head.
+ live, removed := snapshotDVEntries(t, tbl.CurrentSnapshot(), iceio.LocalFS{})
+ assert.ElementsMatch(t, []string{dvA.FilePath(), dvB.FilePath()}, live)
+ assert.Empty(t, removed)
+ })
+}
+
+// Why: a multi-blob Puffin file legally carries DVs for several data
+// files under one path, one manifest entry each; removal identity must
+// be (file path, referenced data file), or superseding one blob would
+// silently drop its siblings' entries and resurrect their deleted rows.
+// Condition: a v3 table where data files A and B carry DVs in one
+// shared Puffin file, then RowDeltas that supersede one blob, both
+// blobs, or misidentify a blob.
+// Assertion: superseding A's DV alone produces a single snapshot where
+// A's old entry is DELETED while B's entry at the SAME path stays live
+// and B's rows stay hidden; both blobs can be superseded in one delta;
+// removals that cannot identify exactly one live entry fail.
+func TestRowDeltaRemoveDeletesSharedPuffin(t *testing.T) {
+ fs := iceio.LocalFS{}
+
+ t.Run("superseding one blob leaves siblings live", func(t *testing.T) {
+ tbl, location, pathA, pathB, dvA, dvB := newTableWithSharedPuffinDVs(t)
+ snapsBefore := len(tbl.Metadata().Snapshots())
+
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(dvA2).RemoveDeletes(dvA).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tbl.Metadata().Snapshots(), snapsBefore+1,
+ "supersession must produce a single snapshot")
+
+ live, removed := snapshotDeleteEntryFiles(t, tbl.CurrentSnapshot(), fs)
+ assert.ElementsMatch(t, [][2]string{
+ {dvA2.FilePath(), pathA},
+ {dvB.FilePath(), pathB},
+ }, pathRefPairs(live),
+ "B's entry at the shared Puffin path must survive A's supersession")
+ assert.Equal(t, [][2]string{{dvA.FilePath(), pathA}}, pathRefPairs(removed),
+ "only A's entry may be recorded as DELETED")
+
+ // B's rows stay hidden: the scanner still plans B's blob of the
+ // shared Puffin, and its bitmap still hides position 0.
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+ for _, task := range tasks {
+ require.Len(t, task.DeletionVectorFiles, 1)
+ planned := task.DeletionVectorFiles[0]
+ switch task.File.FilePath() {
+ case pathA:
+ assert.Equal(t, dvA2.FilePath(), planned.FilePath())
+ case pathB:
+ assert.Equal(t, dvB.FilePath(), planned.FilePath())
+ assert.Equal(t, pathB, refOf(planned))
+ bm, err := dv.ReadDV(fs, planned)
+ require.NoError(t, err)
+ assert.True(t, bm.Contains(0), "B's deleted row must stay hidden")
+ assert.EqualValues(t, 1, bm.Cardinality())
+ default:
+ t.Fatalf("unexpected data file in plan: %s", task.File.FilePath())
+ }
+ }
+
+ // A hides {0,1} of its 2 rows, B hides {0} of its 2 rows.
+ assertRowCount(t, tbl, 1)
+ })
+
+ t.Run("all blobs can be superseded in one delta", func(t *testing.T) {
+ tbl, location, pathA, pathB, dvA, dvB := newTableWithSharedPuffinDVs(t)
+
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ dvB2 := writeDV(t, location, "dv-b2.puffin", pathB, []int64{0, 1})
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).
+ AddDeletes(dvA2, dvB2).RemoveDeletes(dvA, dvB).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ live, removed := snapshotDeleteEntryFiles(t, tbl.CurrentSnapshot(), fs)
+ assert.ElementsMatch(t, [][2]string{
+ {dvA2.FilePath(), pathA},
+ {dvB2.FilePath(), pathB},
+ }, pathRefPairs(live))
+ assert.ElementsMatch(t, [][2]string{
+ {dvA.FilePath(), pathA},
+ {dvB.FilePath(), pathB},
+ }, pathRefPairs(removed),
+ "two removals at the shared path with distinct references are legal")
+
+ assertRowCount(t, tbl, 0)
+ })
+
+ t.Run("removal without a referenced data file is ambiguous", func(t *testing.T) {
+ tbl, location, pathA, _, dvA, _ := newTableWithSharedPuffinDVs(t)
+
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(dvA2).
+ RemoveDeletes(buildPosDeleteFile(t, dvA.FilePath()))
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "ambiguous removal")
+ })
+
+ t.Run("reference matching no live entry at the path", func(t *testing.T) {
+ tbl, location, pathA, _, dvA, _ := newTableWithSharedPuffinDVs(t)
+
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(dvA2).
+ RemoveDeletes(buildDVFile(t, dvA.FilePath(), "s3://bucket/data/nonexistent.parquet"))
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "no live delete entry references data file")
+ })
+
+ t.Run("duplicate removals of one entry are rejected", func(t *testing.T) {
+ tbl, location, pathA, _, dvA, _ := newTableWithSharedPuffinDVs(t)
+
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(dvA2).
+ RemoveDeletes(dvA, buildDVFile(t, dvA.FilePath(), pathA))
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "removed delete files must be unique")
+ })
+}
+
+// Why: a removal the producer cannot match against a live delete entry
+// would be silently dropped, leaving two live DVs on one data file.
+// Condition: remove a DV that no snapshot references, on an empty table
+// and on a table whose snapshot has no delete files.
+// Assertion: Commit fails naming the missing file (or the missing
+// snapshot).
+func TestRowDeltaRemoveDeletesUnknownFile(t *testing.T) {
+ dataPath := "s3://bucket/data/data-001.parquet"
+ oldDV := buildDVFile(t, "s3://bucket/data/dv-001.puffin", dataPath)
+ newDV := buildDVFile(t, "s3://bucket/data/dv-002.puffin", dataPath)
+
+ t.Run("no current snapshot", func(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(newDV).
+ RemoveDeletes(oldDV)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "without an existing snapshot")
+ })
+
+ t.Run("not referenced by current snapshot", func(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+ realDataPath := location + "/data/data-001.parquet"
+ writeParquetFile(t, realDataPath, arrowSc, `[{"id": 1, "data": "alpha"}]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{realDataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ rd := tbl.NewTransaction().NewRowDelta(nil).
+ AddDeletes(newDV).
+ RemoveDeletes(oldDV)
+
+ err = rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "do not belong to the table")
+ })
+}
+
func assertRowCount(t *testing.T, tbl *table.Table, expected int64) {
t.Helper()
diff --git a/table/table.go b/table/table.go
index be3ff08d2..3fc3319a0 100644
--- a/table/table.go
+++ b/table/table.go
@@ -494,8 +494,11 @@ type commitOpts struct {
// may leave this empty; Transaction.Commit always sets it.
branch string
- // validators runs once before cat.CommitTable on the first attempt
- // only. Refresh-and-replay across retries is deferred to PR 2.5.
+ // validators run before cat.CommitTable on every attempt of the
+ // retry loop. On attempt 0 the writer's metadata and the catalog
+ // state coincide, so the conflict context has no concurrent
+ // snapshots and validators short-circuit; on retries they run
+ // against the freshly refreshed catalog state (refresh-and-replay).
validators []conflictValidatorFunc
// noReplay makes a CAS conflict terminal: doCommit returns the