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

[LuQQiu]

Stack

	PR
2	#9007 — mixed FRI V1/V2 implementation (draft)
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.

[jackye1995]

I like the separation of the row-map codec and translation arithmetic in this PR, but I am less convinced that the integration should establish a second, manifest-level rewrite history alongside the existing FRI catalog. Have we considered representing ordered compaction and stable partitioning as two mapping variants in one versioned FRI ledger?

The current stacked design appears to be:

Manifest
├── index section
│   └── __lance_frag_reuse
│       └── V1 ordered-compaction versions/groups
└── stable_partition_transitions[]
    └── V2 source/destination lineage + row-map reference

MixedFragReuseIndex then loads both histories, converts them into common rewrite nodes, and topologically sorts the combined graph by fragment lineage. I think the lineage-based ordering is the right model, especially because the V1 builder version is not necessarily the installing version. My concern is that the persisted state now has two authorities for the same conceptual thing: the history needed to translate stale physical row addresses through fragment rewrites.

That split leaks into several lifecycle paths:

• Normal compaction updates the FRI system index, while stable partitioning appends a top-level manifest descriptor.
• Loading and coverage derivation must join both catalogs before they can reason about current fragments.
• Cleanup marks V2 files from manifest descriptors but uses a separate catch-up algorithm for V1.
• V1 cleanup is currently suspended whenever any V2 transition remains, because pruning the two histories independently can break a mixed chain.
• Clone, feature-flag, transaction, conflict, and cache-identity handling all need special cases for the second catalog.
• Every retained manifest carries the V2 transition descriptors even though FRI details already have an inline/external representation for growing metadata.

I wonder if the durable shape should instead be one FRI transition ledger with a tagged mapping representation. The row-map bytes should still remain external and immutable; I am only suggesting moving their ownership/reference into FRI details.

One possible protobuf shape would be:

message FragmentReuseIndexDetails {
  oneof content {
    InlineContent inline = 1;
    ExternalFile external = 2;
  }

  message InlineContent {
    // Existing data. New readers lift each legacy group into an
    // OrderedCompaction transition in memory.
    repeated Version legacy_versions = 1;

    // New format. All new mapping kinds share one lineage ledger.
    repeated Transition transitions = 2;
  }

  message Transition {
    // Snapshot used to produce the rewrite output.
    uint64 source_dataset_version = 1;

    // Actual manifest version that installed the rewrite.
    uint64 committed_version = 2;

    // Ordered for both address resolution and mapping semantics.
    repeated FragmentDigest sources = 3;
    repeated FragmentDigest destinations = 4;

    oneof mapping {
      OrderedCompaction ordered_compaction = 5;
      StablePartition stable_partition = 6;
    }
  }

  message OrderedCompaction {
    // Existing Roaring representation of rewritten/surviving source rows.
    bytes changed_row_addrs = 1;
  }

  message StablePartition {
    // Immutable dataset-level artifact:
    // _row_maps/<row_map_id>/row_map.lance
    string row_map_id = 1;
    uint64 row_map_size_bytes = 2;
    optional uint32 base_id = 3;
  }

  // Existing FragmentDigest / Version messages remain readable as V1.
}

The tagged ledger can use the index format's existing version negotiation instead of inventing a parallel format-version mechanism in the manifest:

IndexMetadata {
    name: "__lance_frag_reuse"
    index_details.type_url: "/lance.table.FragmentReuseIndexDetails"
    index_version: 1  // first version containing tagged transitions
}

index_version 0: current ordered-compaction-only FRI
index_version 1: unified transition ledger with mapping variants

IndexMetadata.index_version already records the minimum Lance index-format version required to interpret an index, and index-specific details are carried in a protobuf Any with a type URL. This is the native mechanism used by index implementations to evolve their formats and detect indices written by a newer implementation. We should be able to use it to provide the forward-compatibility contract for the unified FRI: a new reader understands both versions, while an old reader can see that FRI version 1 is newer than its supported version 0 instead of partially decoding the details.

The exact nesting and field numbers are only illustrative. The properties I think matter are:

1. The mapping kind is explicit. An absent V1 bitmap must not be confused with a partition mapping or corrupt data.
2. Sources and destinations live on the common transition because they define the rewrite graph, independent of how offsets are translated.
3. The stable-partition entry contains only an immutable row-map reference. The O(rows) labels do not belong in details.binpb.
4. Legacy V1 groups remain readable and can be adapted to OrderedCompaction nodes without rewriting historical row data.
5. New mixed histories use committed_version for validation, but execution order comes from fragment producer/consumer lineage rather than list or version order.

The read workflow would then be:

open __lance_frag_reuse
        │
        ├── decode legacy V1 versions
        │       └── lift groups to OrderedCompaction nodes
        │
        └── decode tagged transitions
                ├── OrderedCompaction
                └── StablePartition(row-map reference)
                         │
                         └── leave label file unopened
        │
        ▼
validate one producer/consumer graph
reject duplicate producers, duplicate consumers, and cycles
        │
        ▼
topologically order by fragment lineage
        │
        ▼
translate requested addresses
        ├── ordered node: synchronous bitmap/rank remap
        └── partition node: lazily open the row map and read touched blocks

This retains the important performance distinction. A V1-only dataset can keep the current CompactFragReuseIndex fast path and never construct the asynchronous machinery. A normal compaction still records the compact bitmap/layout representation; it does not write per-row labels. Only a stable partition pays for the row-map file and block reads.

For example, a mixed chain remains straightforward:

F1/F2 --OrderedCompaction--> F10 --StablePartition--> F20/F21

An old index address is translated through both nodes:

F1:7
  │ ordered bitmap/rank mapping
  ▼
F10:5
  │ stable-partition row-map lookup
  ▼
F21:2

This is logically what MixedFragReuseIndex already does. The difference is that the graph comes from one authoritative ledger instead of joining an FRI catalog and a separate manifest transition list.

The write path could also reuse most of the existing FRI lifecycle:

rewrite worker
    ├── writes destination fragments
    ├── writes immutable row_map.lance when mapping=StablePartition
    └── prepares one Transition delta
             │
             ▼
atomic transaction
    ├── install RewriteGroup fragments
    └── append FragmentReuseTransition
             │
             ▼
materialize updated __lance_frag_reuse details
    ├── inline while small
    └── external details.binpb above the existing threshold

The most important complication is concurrent updates. Today an FRI update carries a complete replacement IndexMetadata, and the conflict resolver rejects two concurrent rewrites when both produce an FRI. Simply putting the stable-partition reference into that same replacement blob would preserve this problem:

base ledger L

writer A builds L + A
writer B builds L + B

last-writer-wins would lose a transition, so one writer must currently retry

This is where Transaction V2 should make the unified approach substantially easier. Instead of treating the FRI as an opaque replacement snapshot, the transaction can express the semantic delta—append this transition—as its own sub-operation. For example:

message AppendFragmentReuseTransitions {
  repeated FragmentReuseIndexDetails.Transition transitions = 1;
}

message CompositeOperation {
  // Applied sequentially and committed as one manifest version.
  repeated Transaction transactions = 1;
}

message Transaction {
  oneof operation {
    // Existing operations omitted.
    Rewrite rewrite = 104;
    CompositeOperation composite = 116;
    AppendFragmentReuseTransitions append_fri_transitions = 117;
  }
}

A stable-partition commit becomes an ordered atomic operation:

CompositeTransaction
    1. Rewrite {
           sources: F1/F2,
           destinations: F10/F11
       }
    2. AppendFragmentReuseTransitions {
           StablePartition(row_map_id, ...)
       }

The outer transaction still owns the read version and commit identity. On conflict, Transaction V2 can delegate compatibility and rebase to each sub-operation:

latest ledger: L + A
pending delta: B
        │
        ├── rewrite sources overlap A? -> retry/reject
        ├── duplicate producer or consumer? -> reject
        ├── merged graph contains a cycle? -> reject
        └── otherwise append B
                │
                ▼
             L + A + B

For disjoint rewrites, appending transitions is naturally commutative. The rebase logic no longer has to merge two complete serialized FRI snapshots; it reloads the current ledger, applies the pending transition delta, validates the combined graph, and materializes the new FRI metadata. Cleanup can similarly be modeled as a semantic prune operation or recomputed against the latest ledger, instead of racing with opaque replacement metadata.

Transaction V2 also gives a cleaner place to separate concerns:

• Rewrite validates that the source fragments are still the exact snapshot read by the worker and that destination IDs remain valid.
• AppendFragmentReuseTransitions validates mapping identity, row-map dimensions, producer/consumer uniqueness, and lineage.
• The composite commit guarantees that fragments and their mapping become visible together.
• Rebase combines disjoint transition appends but still rejects source overlap, concurrent source mutation, duplicate lineage edges, or cycles.

Compatibility should use the native index-version mechanism and the dataset capability fence together:

open dataset
    |
    +-- no partition transition
    |      index_version=0
    |      existing V1 reader and writer remain valid
    |
    +-- tagged partition transition exists
           index_version>=1 identifies the required FRI format
           reader/writer feature flag requires dataset-level support
           unknown version or mapping kind fails closed

The index version protects the index-format contract and gives future FRI mapping representations a normal forward-compatible evolution path. The paired reader/writer feature bit proposed in #9007 remains useful as the dataset-level safety fence: merely ignoring an unsupported system index is not safe if another retained index contains stale physical addresses that require it. Protobuf unknown-field behavior is also insufficient because an old reader could otherwise partially decode the details while dropping the mapping variant. Together, index_version, the Any type URL, explicit mapping tags, and the manifest capability bit let us distinguish safe legacy V1, supported mixed FRI, and unsupported future FRI without creating a separate top-level catalog solely for compatibility.

For future extensions, the same structure works naturally:

FRI index_version 1
    OrderedCompaction
    StablePartition

FRI index_version 2
    OrderedCompaction
    StablePartition
    ArbitraryPermutation

A reader that only supports version 1 can reject or safely fall back when it encounters version 2 according to the index compatibility policy, while version 2 readers continue to load legacy ordered groups and version 1 transitions.

Cleanup would become one lineage operation:

all logical index segment provenance
        │
        ▼
determine which transition nodes are still required
        │
        ├── retain required OrderedCompaction nodes
        ├── retain required StablePartition nodes
        └── collect row_map_id roots from retained partition nodes
        │
        ▼
write pruned FRI ledger
        │
        ▼
GC unreferenced _row_maps/<id> artifacts after normal retention checks

This should allow coordinated pruning across V1 -> V2 -> V1 chains instead of the current conservative rule that disables all V1 cleanup whenever any V2 reference exists. The _row_maps namespace can remain dataset-scoped so row-map identity survives replacement of the FRI UUID and so older _indices cleanup logic cannot mistake row maps for ordinary mutable index contents.

My comparison is roughly:

Dimension	Separate manifest V2 history	Unified tagged FRI ledger
Initial implementation risk	Lower; V1 stays untouched	Higher; FRI schema and rebase logic change
Normal compaction	Existing V1 path	Same ordered fast path
Stable-partition payload	External immutable row map	Same external immutable row map
Sources of truth	Two histories joined at runtime	One transition graph
Metadata growth	V2 descriptors copied into each manifest	Reuse FRI inline/external details policy
Concurrent disjoint partitions	Easy append/rebase in the proposed design	Easy once append is a Transaction V2 delta; difficult with today's replacement metadata
Cleanup	Separate algorithms; V1 pruning suspended by V2	One coordinated reachability/catch-up pass
Long-term format complexity	More special cases across manifest/index lifecycle	More up-front migration work, simpler steady-state model

So I think the current split is reasonable as a short-lived experimental staging approach, but I would be hesitant to make the second manifest-level history the durable format. With Transaction V2 making transition appends and per-sub-operation rebasing much easier to express, I think it is probably worth doing the unified FRI ledger now: keep #8972's external row-map data plane, add a tagged mapping variant to FRI, and make FRI updates semantic append/prune operations rather than whole-catalog replacements.

What do you think? Is there a lifecycle or compatibility requirement that prevents the stable-partition descriptor from living in the FRI ledger itself?