Core: Merge concurrent disjoint DVs for RowDelta commits that add data files

[sezruby]

What

Extends the concurrent deletion-vector (DV) merge added in apache#17754 to operations that also add data files (UPDATE / MERGE), not just pure deletes.

apache#17754 lets a RowDelta merge a concurrently added DV into its own DV for the same data file instead of failing, for operations that only add DVs. Because deletes are commutative, those DVs are unioned unconditionally.

An operation that also rewrites rows into new data files cannot union unconditionally. If the concurrent DV newly deletes a row position that this operation also newly deletes (the row it is rewriting), merging would let the same row be both rewritten here and deleted concurrently, which is a lost update. This change allows the merge only when the newly deleted positions are disjoint, and fails validation otherwise.

How

• BaseRowDelta.canMergeConcurrentDVs() no longer excludes operations that add data files; it returns !validateNewDeleteFiles && !deletesDataFiles(), and for operations that add data files it additionally requires the table property below.
• New hook MergingSnapshotProducer.mergedConcurrentDVsMustBeDisjoint() (default false; BaseRowDelta returns addsDataFiles()).
• When it is set, mergeConcurrentDVs calls validateDisjointConcurrentDV, which fails when (dv_self ∩ dv_concurrent) − dv_base is non-empty for the referenced data file. dv_base is the set of positions already deleted at this operation's starting snapshot (the DVs it supersedes and removes); those positions are shared by both DVs and are not a conflict, so they are subtracted before comparing. Base positions are read once and cached per referenced file so that commit retries stay correct after the superseded DVs are dropped from the pending removals.
• New table property commit.row-level.concurrent-dv-merge.enabled (default true) gates this. When disabled, an operation that adds data files falls back to failing on any concurrent DV for a data file it also added a DV to (the prior behavior). Pure delete commits are unaffected and always merge.

Where it runs and what it reads

The merge runs in the validation path (validate() → validateAddedDVs → mergeConcurrentDVs), which re-runs after refresh() on every commit attempt, so the union is recomputed against fresh table state rather than only on a retry. DV bitmaps are read on the driver only for data files that both this operation and a concurrent commit added a DV to — the file-level overlap is checked from manifest metadata first, so the read is skipped entirely when the commits touch disjoint data files. DV reads go through EncryptingFileIO so they work on encrypted tables.

Testing

TestRowDelta, format versions 3 and 4:

• a disjoint concurrent DV is merged when the operation adds data files;
• an overlapping concurrent DV fails with a clear message;
• a position shared only through the base DV does not block the merge;
• a genuine conflict on a non-base position still fails even when a base DV is present;
• the disjoint update's merged result commits once across a commit retry;
• with commit.row-level.concurrent-dv-merge.enabled=false, a disjoint concurrent DV is not merged and the commit fails.

Scope

This is the DV-merge part of the concurrency refinement for DV tables. Real Spark UPDATE/MERGE also set validateNoConflictingDeleteFiles(), whose file-granular check fails before this merge runs; relaxing that check for DV tables so end-to-end UPDATE/MERGE can reach this merge is a separate follow-up. Design context: apache#18020.

🤖 Generated with Claude Code