[RLC] Case 4 — OPTIMIZE ⟂ DML reconciliation (compaction offset remap)

[sezruby]

Part of the row-level-concurrency umbrella (#3, Case 4). When a compaction OPTIMIZE and a
concurrent DELETE/UPDATE touch the same file, OSS aborts the loser even though the two are
logically independent — the OPTIMIZE only relocated rows (dataChange=false), it didn't change which
rows are live. This reconciles that instead of aborting: remap the concurrent deletion vector onto the
compacted output. Compaction (order-preserving → offset remap) is landing as three stacked PRs: a
shared capture/persist write side (PR #12) that records where each source's rows landed, plus the
two reconcile readers that consume it — forward = OPTIMIZE loses (PR #9), reverse = DML loses
(PR #11); reclustering (row-permuting → identity remap via row tracking) is backlog (umbrella Case 5).
Depends on the same-file DV-merge (#5) for the DV helpers and resolved-path plumbing.

Problem

OPTIMIZE commits RemoveFile(f1..fn) + AddFile(C, dataChange=false) — same logical rows, new
physical layout. A concurrent DV-based DELETE/UPDATE commits, for a touched source fi,
RemoveFile(fi, oldDV) + AddFile(fi, oldDV ∪ deleted). The winner's DV indexes physical positions
in fi, but fi is gone from the committed state — those rows now live in C at different
positions. OSS conservatively aborts (OptimizeConflictSuite). Databricks' documented row-level
concurrency says OPTIMIZE (non-ZORDER) does not conflict with writes; this closes that gap for OSS.

Approach

Reconcile at commit: translate the deleted rows from source-file positions to their positions in C,
union a remapped DV onto C, and commit both. The entire problem reduces to knowing where each
source row landed in C.

Compaction (order-preserving): offset remap

Compaction concatenates each source file's live rows into C without permuting within a file, so
each source occupies one contiguous run:

outputPos(i) = outputStart(sourceFile) + liveRank(i)

where i is a physical row index in the source and liveRank(i) = i minus the rows already
deleted below i when OPTIMIZE read the file. No per-row carrier and no row tracking — just, per
source file, its output start offset and live-row count.

What the write records: compactedInto / compactionInfo (write side, PR #12)

Getting the run layout right is the crux. The output order is not derivable from the plan: Spark
sorts scan splits size-descending before packing (FileSourceScanExec), repartition shuffles across
files, and a non-Spark engine's read order is opaque. So the layout is observed at write time,
cheaply:

• A write-stage operator (SourceCompositionCaptureExec) reads the source file identity per row from
  InputFileBlockHolder — the scan thread-local input_file_name() uses — and counts rows per file
  in write order. No _metadata.file_path, no _metadata.row_index, no helper column; rows pass
  through unchanged. Measured overhead ~0% (+1 ms / 4 M rows, within noise).

  Why not observe _metadata directly (the obvious alternative, built and measured at ~15–24%,
  then discarded):

  • Requesting _metadata.row_index forces the DV-aware position-tracking scan on every file;
    _metadata.file_path materializes a per-row path string on top. Together they were the bulk of
    that overhead.
  • A custom data writer to strip helper columns buys nothing: a helper column trailing the table
    columns must be narrowed away before the parquet write (ParquetWriteSupport.writeFields
    iterates row.numFields), but a zero-copy ProjectingInternalRow narrowing measured the same as
    an UnsafeProjection copy — the copy was never the bottleneck.
  • For the contiguous (coalesce) case the layout is derivable without row_index at all:
    (source file, live count) in write order fully describes it.

• On the coalesce (no-shuffle) path each source's live rows land in one contiguous output segment,
  recorded on that source's RemoveFile tombstone as:

  compactedInto  = ["<outputPath>"]
  compactionInfo = [{"rowOffsetInTarget": <outputStart>, "sourceNumPhysicalRecords": <physical count>}]
  

  rowOffsetInTarget = the source's run-start in the output (running sum in write order, evaluated on
  the driver at tag time); sourceNumPhysicalRecords = the source's physical row count. This tag
  format is modeled on the one Databricks Runtime writes, so that an OPTIMIZE written by another
  engine can be reconciled against on a shared table — best-effort interop, not a verified guarantee.
  Both directions parse the one format via parseOptimizeSourceComposition.

Entry count is O(#source files), independent of DV cardinality — a source with a large or
fragmented read-time DV still records one entry.

DV'd source files: lazy read-time-gap reconstruction

If a source carried a read-time DV, its live rows are physically non-contiguous, but the tag
still stores a single physical count. The live run length is derived at conflict time as
sourceNumPhysicalRecords − |read-time DV|, and the gaps are reconstructed there too — the reconcile
reads the source's read-time DV (sorted ascending) and, for each deleted physical row i, computes
liveRank(i) = i − rank(readTimeDv, i) by binary search. Only the incremental deletions
(winnerDv \ readTimeDv) are remapped; positions already deleted at read time were dropped from C
and are skipped. A fragmented source DV never bloats the tag.

The abort gate (what makes it safe)

Reconcile only when the layout is captured and trusted; otherwise fall back to today's abort. No
composition tag → abort when:

• the write is not on the Spark coalesce path (a native engine's writer never runs the operator;
  after a repartition shuffle InputFileBlockHolder is empty),
• the output isn't a single file / a source contributed more than one run (speculation, split), or
• a deleted row maps outside its source's live run.

Safe precisely because the feature is opt-in, default off: losing the reconcile only restores
current behavior — never wrong data.

Two directions

• Forward — OPTIMIZE loses (resolveOptimizeConflicts, PR [RLC] Case 4a — OPTIMIZE loses to DML (forward compaction offset remap) #9). For each removed source with a
  winning DV, remap the incremental deletions onto C, union into C's DV, and re-point the
  source's RemoveFile at the winner's post-image (files keyed by path and DV, so the stale
  RemoveFile must target the winner's AddFile). All-resolvable-or-abort. The OPTIMIZE reads its
  own to-be-committed composition here; PR [RLC] Case 4 — Capture & persist compaction OPTIMIZE source composition (write side) #12 also persists it, so the reverse direction can read it
  back later.
• Reverse — DML loses (resolveReverseOptimizeConflicts, PR [RLC] Case 4b — DML loses to OPTIMIZE (reverse compaction offset remap) #11, stacked on [RLC] Case 4a — OPTIMIZE loses to DML (forward compaction offset remap) #9). The losing
  DELETE/UPDATE reads the winning OPTIMIZE's persisted composition and remaps its own
  deletions onto the compacted output the same way. Config-gated
  (optimize.conflictReconciliation.reverse.enabled), default off.

The composition tag lives on the removed source's RemoveFile tombstone, not the output
AddFile — snapshot reconstruction replays every AddFile on every read, whereas a removed file's
tombstone is never materialized into the live snapshot, so the read hot path stays clean; and
tombstone retention (default ~7 days) comfortably outlives the conflict window (a concurrent
transaction's runtime), so the tag is still present when a losing DML reads it. It is persisted
to the log by the write side (PR #12; the Delta
protocol ignores unknown RemoveFile tags, so this is free and forward-compatible): an OPTIMIZE
can't predict which future DML will lose to it, so it always leaves the breadcrumb, which the reverse
direction reads back.

Reclustering (row-permuting): backlog — identity remap via row tracking

ZORDER / liquid clustering reorder rows across files, so contiguous offsets don't exist. The
order-independent solution is row tracking: translate deleted source positions → stable row IDs
(_metadata.row_id) → positions in C (row_index), then write the DV over C. This needs row
tracking enabled and preserved through OPTIMIZE, plus a scan in the conflict path (project row IDs
over the pruned output) — an architectural wrinkle since ConflictChecker is otherwise
cheap/synchronous. This is the one tier where row tracking is load-bearing (umbrella Case 5, backlog).
Lower priority: incremental ZCube clustering (Delta 3.2+) keeps clustering on new/small cubes usually
disjoint from DML on stable data, so genuine overlap is rare, and reader-side skipping (#4)
removes most false conflicts first — this handles only the rare real overlap. ZORDER may simply
stay conflicting (matches Databricks).

Config / status

• Capture / persist (write side): PR [RLC] Case 4 — Capture & persist compaction OPTIMIZE source composition (write side) #12 (base = same-file DV-merge [RLC] Case 2 — same-file deletion-vector merge (concurrent DML ⟂ DML) #5). The
  SourceCompositionCaptureExec operator and the persisted compactedInto / compactionInfo
  tombstone tags; armed whenever either reconcile flag is on. No conflict-time logic — both readers
  stack on this.
• Compaction, forward (OPTIMIZE-loser): PR [RLC] Case 4a — OPTIMIZE loses to DML (forward compaction offset remap) #9 (stacked on [RLC] Case 4 — Capture & persist compaction OPTIMIZE source composition (write side) #12). resolveOptimizeConflicts,
  optimize.conflictReconciliation.enabled, default off.
• Compaction, reverse (DML-loser): PR [RLC] Case 4b — DML loses to OPTIMIZE (reverse compaction offset remap) #11 (stacked on [RLC] Case 4a — OPTIMIZE loses to DML (forward compaction offset remap) #9). resolveReverseOptimizeConflicts,
  optimize.conflictReconciliation.reverse.enabled, default off — reads back the composition PR [RLC] Case 4 — Capture & persist compaction OPTIMIZE source composition (write side) #12
  persisted.
• Reclustering: backlog (umbrella Case 5) — design-only. Reclustering aborts today, which matches
  the documented contract; row-tracking identity-remap is the optional path to also reconcile
  permutations and would need a persisted carrier (the tag) to stay cross-cluster-correct.
• Depends on the same-file DV-merge ([RLC] Case 2 — same-file deletion-vector merge (concurrent DML ⟂ DML) #5).
• Tests: OptimizeConflictReconciliationSuite — happy path (no-DV + read-time-DV'd source),
  mixed-bin offset correctness, tag-presence guards (compaction tags present; ZORDER / repartition
  don't), and the reverse direction; regression OptimizeCompaction* / OptimizeConflictSuite /
  RowLevelConcurrencySuite green.

[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.