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

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.