[RLC] Case 4 — Capture & persist compaction OPTIMIZE source composition (write side)

[sezruby]

Part of the row-level-concurrency umbrella (#3) — the shared write side of Case 4 (OPTIMIZE ⟂ DML reconciliation; full design #7). This PR only captures and persists where each compacted source's rows landed. The two reconcile directions that consume it stack on top: forward — OPTIMIZE loses (PR #9) — and reverse — DML loses (PR #11).

What

When OPTIMIZE conflict reconciliation is enabled, record on each removed source file's RemoveFile tombstone where that source's live rows landed in the compacted output. A later conflict check can then remap a concurrent deletion vector between the source and the output instead of aborting the loser. This PR adds only the capture + the persisted tag — no conflict-time logic.

Capture: (file, count) from InputFileBlockHolder — no _metadata

The run layout can't be derived from the read plan (Spark sorts scan splits size-descending before packing; a non-Spark engine's read order is opaque), so it's observed at write time — but cheaply:

• SourceCompositionCaptureExec — a write-stage operator injected like DeltaOptimizedWriterExec — reads the current source file from InputFileBlockHolder (the thread-local input_file_name() uses) and counts rows per file, in write order, on both the row and vectorized execution paths. No helper column, no _metadata.row_index, no _metadata.file_path — rows pass through unchanged (no strip, no copy).
• On the coalesce (no-shuffle) compaction path each source file's live rows land in one contiguous output segment, so recording (outputPath, outputStart, liveCount) per source fully describes the layout.
• Overhead ≈ 0% (+1 ms / 4 M rows, within noise). Avoiding _metadata.row_index is the win: requesting it forces the DV-aware position-tracking scan path on every file, and _metadata.file_path materializes a per-row path string — together the bulk of the double-digit end-to-end overhead an earlier _metadata-observation prototype paid.

The persisted tags

On each removed source's RemoveFile tombstone:

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

• rowOffsetInTarget — the source's contiguous run-start in the compacted output (running sum in write order, evaluated on the driver at tag time).
• sourceNumPhysicalRecords — the source's physical row count. A reader derives the live run length as sourceNumPhysicalRecords − |read-time DV|, where the read-time DV is carried on the same tombstone (RemoveFile.deletionVector). Storing the physical (not live) count keeps the entry O(#source files) regardless of DV cardinality — a fragmented read-time DV never bloats the tag.
• Persisted on the tombstone (not stripped before commit, not on 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; tombstone retention (default ~7 days) comfortably outlives the conflict window. The Delta protocol ignores unknown RemoveFile tags, so persisting is free and forward-compatible — an ignored tag only forgoes a reconcile, never affects correctness.
• The tag pair is modeled on the format Databricks Runtime records; interoperating with a DBR-written OPTIMIZE on a shared table is best-effort, not a verified guarantee. CompactionInfoEntry tolerates unknown fields for forward-compatible schema drift.

Scope / guards (coalesce-only)

Captured only for order-preserving compaction on the coalesce path. Otherwise nothing is recorded and a reader falls back to today's abort:

• repartition path — a shuffle reorders rows across files and InputFileBlockHolder is empty after it.
• reclustering / ZORDER — row-permuting, no offset mapping exists.
• multi-file output, speculation, oneRunPerFile violated, or a source liveCount mismatch → not captured.

Safe because opt-in: losing the reconcile only restores current behavior, never wrong data.

Dependency

Stacked on the same-file DV-merge PoC (base branch row-level-concurrency-dv-poc, Case 2 — #5 / PR #2).

Test — SourceCompositionCaptureExecSuite

• the vectorized path folds one run per source file across batches, in write order;
• the operator stays columnar-transparent (mirrors its child).

End-to-end capture-and-reconcile behavior is exercised by the stacked reconcile PRs (#9 forward, #11 reverse).

🤖 Generated with Claude Code