V2 Format: Compaction outputs one page per input fragment

Problem

compact_files (re-encode mode) merges many fragments into one file but does not coalesce their data into larger pages: the output gets roughly one page per input fragment. A dataset produced as many small fragments (e.g. a producer doing micro-appends) keeps that fine page granularity after compaction. batch_size has no effect.

Since point-take cost scales with pages touched, this makes scattered takes far slower than a dataset written from large batches. Production case (~1 TB, list<int16> ~370 KB/row, indexed 8-row take): ~7.5 s via write_dataset vs ~540 s via compact_files.

Root cause

This is a v2 regression from v1. do_write_fragments in rust/lance/src/dataset/write.rs (L572–581) coalesces in v1 but not v2:

let mut buffered_reader = if storage_version == LanceFileVersion::Legacy {
    chunk_stream(data, params.max_rows_per_group)   // v1: COALESCES small batches up to target
} else {
    break_stream(data, params.max_rows_per_file)     // v2: only SPLITS oversized; never merges
        .map_ok(|batch| vec![batch]).boxed()
};

The v2 writer forms ~one page per input batch. Two things combine:

1. The compaction scanner emits one batch per source fragment and never coalesces across fragment boundaries (LanceScanExec::try_new_v2 flattens per-fragment streams via try_flatten, rust/lance/src/io/exec/scan.rs:338). batch_size can only split a fragment finer, never merge across fragments — so it is inert here.
2. do_write_fragments (v2) then passes each of those per-fragment batches straight through break_stream to the writer, which emits one page each.

v1's chunk_stream would have merged those small per-fragment batches up to max_rows_per_group before page formation. v2 dropped that step, so compaction merges files but not pages.

Reproduction (pylance 7.0.0)

import os, shutil, tempfile
from collections import Counter
import numpy as np, pyarrow as pa, lance
from lance.file import LanceFileReader

VPR, NFRAG, ROWS_PER_FRAG, BIG = 20_000, 20, 300, 6_000  # 20k int16 = ~40 KB/row
N = NFRAG * ROWS_PER_FRAG
ROOT = tempfile.mkdtemp()

def make_table(n):
    rng = np.random.default_rng(0)
    flat = rng.integers(-100, 100, size=n * VPR, dtype=np.int16)
    offs = np.arange(0, (n + 1) * VPR, VPR, dtype=np.int32)
    feat = pa.ListArray.from_arrays(pa.array(offs), pa.array(flat))
    return pa.table({"feat": feat, "id": pa.array(np.arange(n), pa.int64())})

def dfp(b, df):
    return next(c for c in (os.path.join(b, "data", df.path()), os.path.join(b, df.path())) if os.path.exists(c))

def pages(label, ds):
    n = sum(len(LanceFileReader(dfp(ds.uri, df)).metadata().columns[0].pages)
            for f in ds.get_fragments() for df in f.data_files())
    print(f"{label:<40} frags={len(ds.get_fragments()):<4} feat_pages={n}")

def write_many_small(p):  # NFRAG fragments, each one page
    lance.write_dataset(make_table(N).to_reader(max_chunksize=ROWS_PER_FRAG), p,
                        max_rows_per_file=ROWS_PER_FRAG, max_rows_per_group=ROWS_PER_FRAG)

p = os.path.join(ROOT, "small"); write_many_small(p); ds = lance.dataset(p)
print("scanner(batch_size=6000) emits:",
      dict(Counter(b.num_rows for b in ds.scanner(batch_size=BIG).to_batches())))
pages("many small fragments UNCOMPACTED", ds)
ds.optimize.compact_files(compaction_mode="reencode", batch_size=BIG)
pages(f"compact batch_size={BIG}", lance.dataset(p)); shutil.rmtree(p)

p2 = os.path.join(ROOT, "big")
lance.write_dataset(make_table(N).to_reader(max_chunksize=BIG), p2, max_rows_per_file=1_000_000)
pages(f"write_dataset {BIG}-row batches", lance.dataset(p2)); shutil.rmtree(ROOT)

scanner(batch_size=6000) emits: {300: 20}   # one batch per fragment; batch_size ignored
many small fragments UNCOMPACTED   frags=20  feat_pages=20
compact batch_size=6000            frags=1   feat_pages=20   # merged to 1 file, still 1 page/fragment
write_dataset 6000-row batches     frags=1   feat_pages=1    # 20x fewer pages

The scanner emits one batch per fragment even at batch_size=6000; compaction merges 20 fragments into 1 file but keeps 20 pages; write_dataset from large batches produces 1 page.

Proposed fix

Restore the v1 coalescing in the v2 write path: route the v2 stream through the existing chunk_stream coalescer (already used by v1, already tested) so per-fragment batches merge up to max_rows_per_group before page formation. This re-forms pages during compaction, makes max_rows_per_group meaningful for v2, and restores v1 parity — no new tuning knob.

Caveat: coalescing buffers in memory, so for fat columns the merge should be bounded by both rows and bytes (flush at min(max_rows_per_group, byte budget), reusing data_cache_bytes/max_page_bytes) to avoid OOM.

Related

#6634 (page-size control for fat nested lists), #4374 (v2.1 point-lookup regression), #4090 (take perf gap).

Environment: pylance 7.0.0, file format v2.1.

Start in rust/lance/src/dataset/write.rs, especially do_write_fragments and the existing chunk_stream path, then inspect rust/lance/src/io/exec/scan.rs around LanceScanExec::try_new_v2. Run the supplied Python reproduction to compare page counts before and after compaction. Done means v2 compaction coalesces fragment batches up to the configured row and byte limits without unbounded buffering.