Row-level concurrency for OSS Delta: reconcile the writes that provably cannot conflict (umbrella)

[sezruby]

Background

Delta's documented row-level concurrency states that several concurrent-write pairs on a table with
deletion vectors cannot conflict — two DML statements editing disjoint rows, a compaction
OPTIMIZE vs a concurrent DELETE/UPDATE, a blind INSERT against anything. OSS Delta today
still aborts the loser in most of these: ConflictChecker conservatively treats a file that a
concurrent commit removed, or added a DV to, as a conflict — even when the two commits are logically
independent.

This umbrella tracks closing that gap without any protocol change and without row tracking, as a
series of small, independently reviewable, opt-in (default-off) additions to ConflictChecker.

Organizing principle

Rather than enumerating special cases, the work is driven by one property:

Reconcile a concurrent pair whenever the loser's outcome is provably reconstructable from
deletion-vector positions, reader statistics, or a derivable output offset — never from a
whole-table read, a row permutation, or row lineage.

Everything reconstructable from that information is in scope (Phase 1); anything needing a
whole-table read, a row permutation, or row tracking is backlog. This lands exactly at parity with
the documented contract, because the deferred cases are the ones the reference implementation does
not reconcile either.

Phase 1 — opt-in, no row tracking

#	Concurrent pair	Reconcile mechanism	Extra input
3	Compaction OPTIMIZE (dataChange=false) ⟂ non-blind append	exclude dataChange=false (relocation) adds from the added-files conflict check	—
1	Writers touching different files / disjoint data ranges	conflict-time data skipping: prune the loser's read set with reader stats before the file-level checks	reader stats
2	Same file, disjoint rows — DML ⟂ DML	union the two deletion vectors when the newly-deleted row sets are disjoint	DVs
4a	Compaction OPTIMIZE loses to a concurrent DML	forward offset remap: translate the winning DML's newly-deleted rows to their positions in the compacted output and union a DV onto it	derivable offsets
4b	A concurrent DML loses to a compaction OPTIMIZE	reverse offset remap: read the composition the OPTIMIZE persisted and remap the losing DML's deletions onto the compacted output	derivable offsets

Tracked issues / PRs

Each case carries its own design in its own issue.

• Case 3 — OPTIMIZE vs non-blind append — [RLC] Case 3 — exclude relocation (dataChange=false) files from the append-conflict check #6 · upstreamed as [File-level concurrency] Exclude no-data-change (OPTIMIZE) files from append-only conflict checks delta-io/delta#7331 (fork PR [RLC] Case 3 — exclude relocation (dataChange=false) files from the append-conflict check #10 closed)
• Case 1 — conflict-time data skipping — [RLC] Case 1 — conflict-time data skipping (disjoint data ranges) #4 · PR [RLC] Case 1 — conflict-time data skipping (disjoint data ranges) #8
• Case 2 — same-file disjoint DML ⟂ DML (DV union) — [RLC] Case 2 — same-file deletion-vector merge (concurrent DML ⟂ DML) #5 · PR [RLC] Case 2 — same-file deletion-vector merge (concurrent DML ⟂ DML) #2
• Case 4 (write side) — capture & persist compaction composition — [RLC] Case 4 — OPTIMIZE ⟂ DML reconciliation (compaction offset remap) #7 · PR [RLC] Case 4 — Capture & persist compaction OPTIMIZE source composition (write side) #12
• Case 4a — compaction OPTIMIZE loses to DML (forward remap) — [RLC] Case 4 — OPTIMIZE ⟂ DML reconciliation (compaction offset remap) #7 · PR [RLC] Case 4a — OPTIMIZE loses to DML (forward compaction offset remap) #9
• Case 4b — DML loses to compaction OPTIMIZE (reverse remap) — [RLC] Case 4 — OPTIMIZE ⟂ DML reconciliation (compaction offset remap) #7 · PR [RLC] Case 4b — DML loses to OPTIMIZE (reverse compaction offset remap) #11

Dependencies. Case 4 reuses Case 2's deletion-vector helpers and splits into a shared write side
(PR #12) that captures & persists the composition, plus two reconcile readers that consume it —
forward (4a) and reverse (4b) — giving the stacking chain Case 2 → 4 (capture) → 4a → 4b. Cases
1 and 3 are independent. Suggested review order: this issue → 1 → 2 → 4 (capture) → 4a → 4b.

How the compaction cases stay cheap (Case 4)

Compaction concatenates each source file's live rows into one contiguous run in the output without
permuting within a file, so the whole problem reduces to where each source's run starts in the
output. That offset isn't derivable from the read plan (scan splits are sorted before packing, a
shuffle reorders rows, a non-Spark engine's order is opaque), so it is observed at write time: a
write-stage operator reads the per-row source-file identity that input_file_name() already exposes
and records, on each removed source's RemoveFile tombstone, the run it contributed —
compactedInto (the output path) and compactionInfo ({rowOffsetInTarget, sourceNumPhysicalRecords}).
Rows pass through unchanged: no _metadata.row_index, no helper column, no row copy.

sourceNumPhysicalRecords is a physical count, kept self-consistent with the tombstone's own DV;
the live run length is sourceNumPhysicalRecords − |sourceDV|, so a source that carried a read-time
DV has its non-contiguous live rows reconstructed lazily at conflict time. The tags live on the
short-lived tombstone rather than the output AddFile (which snapshot reconstruction replays on
every read), so the read hot path is untouched; tombstone retention (~7 days default) comfortably
outlives the conflict window (a concurrent transaction's runtime), so the tag is still present when a
losing DML reads it.

Safety model

• Opt-in, default off. Every case is behind its own flag; losing a reconcile only restores
  today's abort, so a bug can never produce wrong data — only a spurious (but correct) abort.
• Abort on any doubt. A reconcile proceeds only when the layout is fully captured and every
  affected row maps within its source's run; otherwise the loser aborts exactly as it does today.
• Genuine conflicts preserved. Same file, same rows still aborts; ZORDER / reclustering still
  aborts.
• Each PR is reviewable and revertable in isolation.

Backlog — needs row tracking, not part of this proposal

#	Concurrent pair	Why deferred
5	Reclustering / ZORDER ⟂ DML	rows are permuted across files, so no derivable offset exists; a correct reconcile needs identity remap via row tracking plus a scan in the conflict path
6	MERGE net-new inserts / non-matched-by-source (full parity)	needs per-file row-tracking classification

The reference contract does not reconcile these either, so Phase 1 already reaches full parity;
closing them would require load-bearing row tracking and a scan inside the otherwise-cheap,
synchronous conflict path.

Related work

• [Feature Request][Spark] Row-level concurrency delta-io/delta#7057 — row-level concurrency tracking issue.
• [Spark] Row-level concurrency conflict rebasing delta-io/delta#7306 — concurrent DML ⟂ DML via deletion vectors (row-tracking-gated). Overlaps
  Case 2; this umbrella's Case 2 proposes a deletion-vector-only path that does not require row
  tracking, and the remaining cases (compaction, append, reader-side skipping) are additive. Happy
  to coordinate.
• [File-level concurrency] Exclude no-data-change (OPTIMIZE) files from append-only conflict checks delta-io/delta#7331 — Case 3, already open.

Notes

Clean-room: this proposal cites only Delta's public row-level-concurrency contract and our own
black-box observations of its published behavior; it copies no proprietary source. Case 4's
composition carrier uses the compactedInto / compactionInfo tombstone-tag format so an OPTIMIZE
written by another engine can be reconciled against on a shared table; that interop is modeled on the
published format and is best-effort, not a verified guarantee (an unrecognized tag only forgoes the
reconcile — never wrong data).

[sezruby]

Live e2e validation — Case 4 (OPTIMIZE ⟂ DML), both directions

Ran a concurrent end-to-end test: two engines committing to the same Delta table at the same time — one running this fork's build on Apache Spark 4.0, the other Databricks Runtime 16.4 — racing on the shared _delta_log (coordinated only by sentinel marker files, so these are real atomic-commit version races, not replayed conflicts).

Confirmed DBR-compatible: both Case 4 directions reconciled live instead of aborting, and each engine independently verified the table stayed consistent — matching row counts, no duplicate keys, every committed row-level change visible.

Case 4a — forward remap (sezruby/delta#9): a losing OPTIMIZE absorbs the winning DML's in-place deletion vector into its compacted output. Run once per winning operation:

• DELETE winner: 14 forward remaps (winningOp=DELETE). OPTIMIZE side: 48 commits / 2 aborts; DELETE side: 40 / 0 abort / 0 straggler; final rows agree (999,960), no dup — both PASS.
• merge-on-read UPDATE winner: 11 forward remaps (winningOp=UPDATE), each a single-winner remap. OPTIMIZE side: 18 commits / 0 abort; UPDATE side: 20 / 0 abort / every update visible; final rows agree (200,000), no dup — both PASS.
• merge-on-read MERGE (matched-DELETE) winner: 1 forward remap (winningOp=MERGE). OPTIMIZE side: 52 commits / 0 abort; MERGE side: 20 / 0 abort / 0 straggler; final rows agree (199,980), no dup — both PASS.

(A merge-on-read UPDATE writes a new small file each iteration, so OPTIMIZE keeps finding work to compact and remaps repeatedly; a matched-DELETE MERGE, like a DELETE, is deletion-vector-only, so OPTIMIZE converges after the first compaction and the later windows are no-ops — hence the single remap, not a limitation.)

Case 4b — reverse remap (sezruby/delta#11): a losing UPDATE rebases its deletion vector onto the winner OPTIMIZE's compacted output. 3 reverse remaps (winningOp=OPTIMIZE). UPDATE side: 33 commits / 0 abort / every update visible; OPTIMIZE side: 30 commits (8 real compactions); final rows agree (1,000,000), no dup — both PASS.

Reconcile events are the existing delta.optimize.conflictReconciliation.remapped / .reverseRemapped records.

Scope note: this reconcile is for deletion-vector-enabled tables, where merge-on-read is the default (UPDATE_USE_PERSISTENT_DELETION_VECTORS defaults on), so the winner masks its changed rows with a deletion vector in place — RemoveFile(P) + AddFile(P, dv) at the same path. Forward remap (4a) reconciles that regardless of the operation that wrote the DV — DELETE, merge-on-read UPDATE, and merge-on-read MERGE alike (the three runs above); any coexisting new file a merge-on-read UPDATE/MERGE also writes is left untouched. OptimizeConflictReconciliationSuite covers these in-unit. The genuinely-deferred cases remain the umbrella's backlog rows 5/6 — reclustering / ZORDER ⟂ DML (rows permuted across files, so no derivable offset) and MERGE net-new inserts / not-matched-by-source — which permute or reclassify rows and so need row tracking.