feat(core,index): fragment reuse row map for reordered rewrites

[LuQQiu]

Stack

	PR
2	#8978 — transition + commit path
1	👉 #8972 — row map format (base of this stack)

---

What

Part 1 of adding first-class support for reordered rewrites to the fragment reuse machinery. A reordered rewrite (e.g. reclustering rows by a sort or clustering key) reads n source fragments in scan order and distributes their live rows across m destination fragments. Compaction's existing remapping (CompactRowAddrRemap) derives the old-to-new address mapping from row order alone, which only works because compaction preserves relative row order; a reordered rewrite breaks that assumption, so the mapping must be recorded explicitly.

This PR adds the standalone format capability that records and replays that mapping. It has no transaction or read-path integration yet (that comes in follow-up PRs); everything here is a self-contained module with its own tests and benchmarks.

Design

What is recorded

A reordered rewrite is a stable partition: it scans the source fragments in order and appends each live row to exactly one destination fragment, so within every destination, rows keep their relative source order. As the rewrite job scans, it makes one routing decision per row — which destination does this row go to — and that decision is the only information the mapping needs, because the row's offset inside its destination is derivable (see below). The recorded decision is called the row's label.

Worked example used throughout. Sources F1 (5 physical rows, row 2 deleted) then F2 (3 rows); the rewrite produced two destinations, listed in order as [F10, F11]. The job's routing decisions, in scan order:

F1 row 0  -> written to F10      label 0
F1 row 1  -> written to F11      label 1
F1 row 2  -> deleted, not moved  label NULL
F1 row 3  -> written to F10      label 0
F1 row 4  -> written to F10      label 0
F2 row 0  -> written to F11      label 1
F2 row 1  -> written to F10      label 0
F2 row 2  -> written to F11      label 1

A label is the destination's index in the ordered destination list (0 = F10, 1 = F11), not the fragment id itself: u16 indices cover up to 65,536 destinations per rewrite, and the small id list is stored once elsewhere.

The label column is that right-hand column persisted as one Lance file: a single nullable u16 column, one row per physical source row (deleted rows included), in concatenated scan order. With an illustrative block size of 4 rows (real block size: 64K):

source row     0    1    2     3    4  |  5    6    7        (F1 rows 0-4, F2 rows 0-2)
label          0    1    NULL  0    0  |  1    0    1
               └────── block 0 ──────┘   └── block 1 ──┘

Reading the labels left to right and counting per destination replays the whole rewrite: F10 received source rows 0, 3, 4, 6 (as its rows 0-3), F11 received rows 1, 5, 7.

The counts, in the same file's global buffer. Translating one row must not require replaying the file from the start. So for every 64K-row block boundary the file stores, per destination, the cumulative "rows so far" count. For the example (block = 4 rows):

                  F10  F11
after block 0:     2    1        (rows 0-3: two F10 labels, one F11, one NULL)
after block 1:     4    3        (final row = per-destination totals)

This is a dense num_blocks x m grid of u32s — exact and data-independent: ~1.5 MB for a 50M-row rewrite across 500 destinations, ~61 MB at a 1B-row rewrite across 1000, read once at open. The encoded header names the representation, so sparser encodings (e.g. per-destination postings for strongly local redistributions) can be added later without breaking readers; unknown tags fail with a clear error. Deleted counts are implied (block length − sum of the block's deltas), and the final row doubles as the per-destination totals for conservation checks.

The arithmetic rests on an explicit ordering contract (documented in the module): labels are recorded in source physical-row order, each destination receives its rows in that same order and is never re-sorted, and the destination list is fixed for the whole rewrite. A rewrite that routes rows through parallel writers must restore per-destination source order before recording the mapping.

That contract is deliberate scope: this format represents stable partitions only. A rewrite that sorts rows within a destination is not representable by destination labels (two rows with equal labels rank in source order, not output order); such a rewrite needs a per-row final-offset (permutation) encoding, which would be a separate format rather than a relaxation of this one.

How it is read

Open = one tail read (file footer + the counts global buffer) plus an in-memory consistency check of the counts, so a corrupt buffer fails at open instead of mistranslating rows. No label IO.

Point lookup ("where did source row g go?") reads one block:

1. block = g / 64K, pos = g mod 64K; read that one block of labels.
2. If the label at pos is NULL → the row was deleted, done.
3. Otherwise destination offset = counts[block-1][label] + (labels equal to it in this block before pos).

Example: source row 6 lands in block 1 at position 2 with label 0 (= F10). Base = counts after block 0 for F10 = 2. Within block 1, one earlier row (row 4) is labeled F10. So row 6 is (F10, row 3) — matching the replay above. The in-block count scans at most 64K u16s (3.8 µs measured); the block read is one aligned range read.

Sweep ("translate everything from here on") seeds per-destination counters from a block boundary (counters = counts[block-1], zeros for block 0), then for each row: offset = counter[label]++. O(1) per row, no rank computation — used for bulk translation, and it works from any block boundary, not just the file start (~170M rows/s measured).

Where the code lives

• lance-core/src/utils/stable_partition.rs — the arithmetic above, no IO: CountsMatrix (+ builder, codec, validate()), translate_in_block, SweepTranslator. Sibling of row_addr_remap.rs, which is the order-preserving (compaction) counterpart.
• lance-index/src/frag_reuse/row_map.rs — the file: RowMapWriter / RowMapReader on the existing IndexStore / IndexWriter / IndexReader traits, no new IO plumbing. The writer takes labels for live rows only (a rewrite job scans with deletions applied, so it never sees a deleted row) and interleaves the NULLs itself from the source deletion vectors. The reader offers point, coalesced-batch (read_ranges) and sweep translation.

Counts blocks are logical 64K-row blocks addressed by row-number range reads, deliberately decoupled from physical page boundaries: correctness never depends on how the encoder cut pages.

Benchmarks

2M rows, 1000 destinations, ~1/8 deleted, local FS (benches/stable_partition_row_map.rs):

metric	result
encoded size, uniform-random labels (nominal width is 10 bits)	11.15 bits/row
encoded size, labels with block locality (each 64K block draws from ~16 destinations)	6.27 bits/row
sweep translation throughput	~170M rows/s
label rank over a full 64K block (in-memory point-lookup cost)	3.8 µs
end-to-end point translation (one block read + decode + rank)	~240 µs (V2_1)

Uniform-random is the number to budget with: a first-pass reclustering exists precisely because arrival order does not correlate with the clustering key, so a 64K-row source block scatters across all destinations and page dictionaries cannot beat the nominal width (11.15 = 10 bits + validity + page overhead; ~1.4 GB per billion source rows). The locality row is the upside case — time-correlated clustering keys, or incremental re-clustering of mostly-sorted data — where per-page dictionaries kick in. Format version matters either way: the 2.1 miniblock path is what reaches these numbers at all (2.0 stores plain u16, ~17 bits/row).

Tests

• Property tests: point lookup and sweep independently checked against a per-destination-counter reference map over seeded-random labels with deletions, across block boundaries, short final blocks, and mid-block sweep starts.
• Writer conservation: too many / too few labels vs live source rows, deleted offsets outside a fragment's physical rows, per-destination totals vs reference.
• Counts codec: round trip; decode rejects bad magic, truncation, and unknown representation tags; validate() rejects non-monotone counts and over-budget blocks.
• Open hardening: RowMapReader::open enforces the schema contract (exactly one column, named label, u16, nullable), runs the full counts consistency check, and reconciles the label row count, so a corrupt or foreign file errors instead of panicking or silently mistranslating.
• Deterministic NULL/empty shapes: a fully-deleted source, a zero-physical-row source, a deleted tail drained by finish(), and empty translate_many / sweep inputs.

cargo test -p lance-core -p lance-index: 1556 passed, 0 failed. Clippy clean for the new code (--no-deps; the two pre-existing nightly-drift lints in lance-encoding and inverted/cross_column.rs are untouched).

Follow-ups (separate PRs)

1. Transition record + commit path: attach {sources, destinations, row_map} to Operation::Rewrite, conservation validation at commit, legalize deferred index remap for reordered groups.
2. Read integration: coverage derivation and decode-time translation through the existing RowIdRemapper seams.
3. Conflict handling: deletion-vector fold on rebase via the sweep translator, and combining disjoint concurrent rewrites.

🤖 Generated with Claude Code

[lance-gatekeeper]

⚠️ Gate recommendation: approve with a non-blocking risk.

This revision now makes the intended stable-partition-only contract explicit. Arbitrary per-destination permutations remain unsupported by the accepted scope and require a separate encoding; no further change is requested for that limitation.

Batch lookup still expands every touched 64K block and repeats prefix scans, so follow-up integrations should use the linear sweep for dense ranges and bound sparse batches to keep decoded memory predictable.