fix(encoding): split oversized primitive pages

[lance-gatefixer]

Summary

• split oversized ordinary dense primitive batches into row-aligned pages so each page is encoded by an independent CPU task
• prepare and compress an automatic dictionary once per oversized flush, slice its indices across page tasks, and reuse one physical dictionary payload
• teach V2.3 readers and writers to handle shared page-buffer offsets while preserving the stable V2.1 and V2.2 contracts
• cover page count, dictionary offset reuse, round-trip correctness, and bounded split/unsplit file-size amplification

Root cause

The structural primitive encoder planned each oversized flush as one dense page and therefore one CPU encoding task. Splitting before automatic dictionary selection then caused each page to serialize the same dictionary independently; disabling that selection avoided duplication but expanded low-cardinality binary output by orders of magnitude.

Validation

• cargo fmt --all -- --check
• cargo clippy --all --tests --benches -- -D warnings
• cargo test -p lance-file --no-fail-fast
• cargo test -p lance-encoding --no-fail-fast -- --skip test_sparse_large_string_list
• cd python && make build
• reviewer binary reproducer: round-trip, multiple pages, one dictionary offset, and split size at most twice unsplit
• cd python && uv run pytest python/tests/test_file.py::test_write_with_max_page_bytes -q
• cd python && uv run make lint
• git diff --check

Fixes #2561

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The V2.3 planner is operating at the right boundary, but two page-partitioning cases still violate the performance contract: a valid small-row split can be suppressed, and dictionary payload can be multiplied across pages.

A viable revision should partition every independently encodable ordinary row when the batch exceeds the target, while keeping dictionary storage single-owned—or leaving dictionary arrays unsplit until sharing or compaction is supported.

[lance-gatekeeper]

This equality guard leaves an oversized batch as one page and one CPU task even when every row is individually below max_page_bytes. ceil(total / max) can equal num_values simply because the batch has few rows; it does not prove a value is indivisible. Keep partitioning at available row boundaries (or base any minimum-task exception on actual per-row cost), and add this boundary regression.

Reproducer

I added the following test to writer_tests.rs and ran CARGO_TARGET_DIR=/home/agent/tmp/gate-8405-target cargo test -p lance-file gate_repro_two_sub_limit_rows_should_split -- --nocapture:

#[tokio::test]
async fn gate_repro_two_sub_limit_rows_should_split() {
    let arrow_schema = Schema::new(vec![Field::new("data", DataType::Utf8, false)]);
    let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
    let values = vec!["a".repeat(768 * 1024), "b".repeat(768 * 1024)];
    let batch = RecordBatch::try_new(
        arrow_schema.into(),
        vec![Arc::new(StringArray::from(values))],
    )
    .unwrap();
    let path = TempObjFile::default();
    let object_store = ObjectStore::local();
    let mut writer = create_writer(
        object_store.create(&path).await.unwrap(),
        lance_schema,
        ConcreteFileVersion::V2_3,
        FileWriterOptions {
            max_page_bytes: Some(1024 * 1024),
            ..Default::default()
        },
    )
    .unwrap();
    writer.write_batch(&batch).await.unwrap();
    writer.finish().await.unwrap();

    let fs = FsFixture::default();
    let file_scheduler = fs.scheduler
        .open_file(&path, &CachedFileSize::unknown()).await.unwrap();
    let file_reader = FileReader::try_open(
        file_scheduler,
        None,
        Arc::<DecoderPlugins>::default(),
        &LanceCache::no_cache(),
        FileReaderOptions::default(),
    ).await.unwrap();
    assert_eq!(file_reader.metadata().column_metadatas[0].pages.len(), 2);
}

It failed with left: 1, right: 2.

[lance-gatefixer]

Addressed in 36014cb. Oversized ordinary batches now split at every available row boundary, with a two-row UTF-8 regression covering this case.

[lance-gatekeeper]

Generic slicing here duplicates a DictionaryArray's full dictionary into every output page: Arrow slices only the keys and retains the complete values array, then each page passes through DataBlock::from_arrays and serializes those values again. Because the page count also includes dictionary bytes, a large dictionary triggers precisely this CPU, memory, and file-size amplification. Either exclude dictionary arrays from this splitter for now, or compact/share their dictionary payload so it is not encoded once per page.

Reproducer

I added this comparison to writer_tests.rs and ran CARGO_TARGET_DIR=/home/agent/tmp/gate-8405-target cargo test -p lance-file gate_repro_dictionary_split_amplifies_output -- --nocapture:

#[tokio::test]
async fn gate_repro_dictionary_split_amplifies_output() {
    use arrow_array::{DictionaryArray, types::Int32Type};

    let mut state = 0x9e3779b97f4a7c15_u64;
    let dictionary_values = StringArray::from(
        (0..2_000).map(|_| {
            let bytes = (0..1024).map(|_| {
                state ^= state << 13;
                state ^= state >> 7;
                state ^= state << 17;
                b' ' + (state % 95) as u8
            }).collect::<Vec<_>>();
            String::from_utf8(bytes).unwrap()
        }).collect::<Vec<_>>(),
    );
    let keys = Int32Array::from_iter_values((0..600_000).map(|idx| idx % 2_000));
    let dictionary = DictionaryArray::<Int32Type>::try_new(
        keys,
        Arc::new(dictionary_values),
    ).unwrap();
    let arrow_schema = Schema::new(vec![Field::new(
        "data", dictionary.data_type().clone(), false,
    )]);
    let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
    let batch = RecordBatch::try_new(
        arrow_schema.into(), vec![Arc::new(dictionary)],
    ).unwrap();
    let mut results = Vec::new();

    for max_page_bytes in [64 * 1024 * 1024, 1024 * 1024] {
        let path = TempObjFile::default();
        let object_store = ObjectStore::local();
        let mut writer = create_writer(
            object_store.create(&path).await.unwrap(),
            lance_schema.clone(),
            ConcreteFileVersion::V2_3,
            FileWriterOptions {
                max_page_bytes: Some(max_page_bytes),
                ..Default::default()
            },
        ).unwrap();
        writer.write_batch(&batch).await.unwrap();
        writer.finish().await.unwrap();

        let fs = FsFixture::default();
        let file_scheduler = fs.scheduler
            .open_file(&path, &CachedFileSize::unknown()).await.unwrap();
        let file_reader = FileReader::try_open(
            file_scheduler,
            None,
            Arc::<DecoderPlugins>::default(),
            &LanceCache::no_cache(),
            FileReaderOptions::default(),
        ).await.unwrap();
        let pages = &file_reader.metadata().column_metadatas[0].pages;
        let bytes = pages.iter()
            .flat_map(|page| page.buffer_sizes.iter())
            .sum::<u64>();
        results.push((pages.len(), bytes));
    }

    assert!(results[1].1 <= results[0].1 * 2, "results={results:?}");
}

The unsplit run produced (1 page, 2,900,236 bytes); the split run produced (5 pages, 11,162,220 bytes), so the safety assertion failed.

[lance-gatefixer]

Addressed in 36014cb. Dictionary arrays are excluded from generic page splitting so their values payload is not serialized once per page; an oversized dictionary regression now asserts the single-page behavior.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

Both earlier blockers are resolved, but two independent implementation gaps remain: the new V2.3 behavior leaves the existing cross-version page-count test failing, and low-cardinality strings can still duplicate automatically selected dictionary payloads across every split page.

Keep the stable V2.1/V2.2 expectations unchanged, update the V2.3 test contract, and make page splitting preserve one dictionary payload (or bypass splitting when automatic dictionary encoding would be selected).

[lance-gatekeeper]

Enabling splitting here changes V2.3's page-count contract, but the existing Python cross-version test still asserts that V2.3 produces one page. After building the prescribed uv environment, I ran uv run pytest python/tests/test_file.py::test_write_with_max_page_bytes -q from python/; it fails at python/tests/test_file.py:72 with assert 3 == 1. Update the test comment and expected count so V2.1/V2.2 remain one page and V2.3 records the new three-page behavior.

[lance-gatefixer]

Addressed in 917d3fc. The Python cross-version test now expects three pages for V2.3 while retaining one page for V2.1 and V2.2.

[lance-gatekeeper]

This guard only protects input DictionaryArrays. Ordinary low-cardinality StringArrays are split here before automatic dictionary selection, so every output page can build and serialize the same dictionary. On this head, the reproduction below round-trips correctly but grows from 1,057,019 bytes (1 page) to 5,180,512 bytes (5 pages) solely by lowering the maximum page size—a 4.9× file-size regression. Decide dictionary encoding before page splitting and keep its payload shared/compacted, or exclude automatic dictionary candidates from this splitter.

Reproducer

I ran this code with uv run python -c from python/ after building the local extension:

import hashlib
import os
import tempfile
import pyarrow as pa
from lance.file import LanceFileReader, LanceFileWriter

unique = [
    "".join(hashlib.sha256(f"{i}:{j}".encode()).hexdigest() for j in range(16))
    for i in range(1000)
]
table = pa.table({"a": unique * 20})
with tempfile.TemporaryDirectory() as tmp:
    results = []
    for name, limit in [("unsplit", 64 * 1024 * 1024), ("split", 4 * 1024 * 1024)]:
        path = os.path.join(tmp, f"{name}.lance")
        with LanceFileWriter(path, table.schema, max_page_bytes=limit, version="2.3") as writer:
            writer.write_batch(table)
        reader = LanceFileReader(path)
        assert reader.read_all().to_table() == table
        results.append((name, len(reader.metadata().columns[0].pages), os.path.getsize(path)))
    print(results)

Observed: [("unsplit", 1, 1057019), ("split", 5, 5180512)].

[lance-gatefixer]

Addressed in 917d3fc. Oversized automatic dictionary candidates now bypass splitting after the size check, so their payload is serialized once; a low-cardinality UTF-8 regression asserts the single-page V2.3 behavior.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The previous two findings are fixed, but the new dictionary guard treats a provisional encoding attempt as a definitive page-planning result. When dictionary construction later declines, the fallback still emits one oversized task.

Only preserve an unsplit page after the dictionary has actually been prepared and can be reused. Otherwise keep splitting at safe row boundaries, without materializing the full column synchronously in the planner.

[lance-gatekeeper]

This predicate only says dictionary encoding is worth attempting; it does not guarantee that dictionary_encode will return a payload. For 100 unique 20-KiB strings, sampling is inconclusive because there are fewer than 1,024 values, so this returns Some with a 50-entry budget and skips splitting. Actual dictionary construction aborts at the 51st unique value, and fallback writes one roughly 2-MiB page even with a 1-MiB target, restoring the one-task bottleneck. In addition, DataBlock::from_arrays copies and scans the full oversized column synchronously before any spawn_cpu task, and encoding reconstructs it. Only return unsplit after successfully preparing a dictionary and carry that payload forward, or keep the split path and prevent per-page dictionary duplication without this eager full-column conversion.

Reproducer

I ran this code with uv run python -c from python/ after building the current extension:

import os
import tempfile
import pyarrow as pa
from lance.file import LanceFileReader, LanceFileWriter

values = [f"{i:03d}-" + "x" * (20 * 1024 - 4) for i in range(100)]
table = pa.table({"a": values})
with tempfile.TemporaryDirectory() as tmp:
    path = os.path.join(tmp, "unique-strings.lance")
    with LanceFileWriter(
        path, table.schema, max_page_bytes=1024 * 1024, version="2.3"
    ) as writer:
        writer.write_batch(table)
    reader = LanceFileReader(path)
    assert reader.read_all().to_table() == table
    pages = len(reader.metadata().columns[0].pages)
    assert pages == 2, f"expected 2 independently encodable pages, got {pages}"

Observed: AssertionError: expected 2 independently encodable pages, got 1.

[lance-gatefixer]

Addressed in d6ab571. Oversized ordinary columns now split without an eager full-column dictionary probe, and split pages disable page-local automatic dictionaries to avoid duplicating payloads. The 100-by-20-KiB regression now round-trips in two pages.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The split path now restores bounded encoding work, but the replacement dictionary policy can inflate low-cardinality columns by orders of magnitude and violates the storage/I/O contract.

A viable revision should keep encoding work split while retaining a single shared or compact dictionary payload, or use another policy that demonstrably bounds output amplification.

[lance-gatekeeper]

Setting this for every multi-page result drops automatic dictionary encoding altogether, so it can replace a small dictionary with every repeated binary value. Binary and LargeBinary do not select FSST by default, making the raw payload dominant. The executed reproduction on this head round-trips correctly, but changing only max_page_bytes produces 1 page / 112,695 bytes versus 5 pages / 10,320,898 bytes (91.58×). Keep page work split while sharing or compacting one dictionary payload, or otherwise enforce a bounded output-size ratio, and add this comparison as coverage.

Reproducer

I ran this with uv run python -c from python/ after building the local extension:

import hashlib
import os
import tempfile
import pyarrow as pa
from lance.file import LanceFileReader, LanceFileWriter

unique = [
    b"".join(hashlib.sha256(f"{i}:{j}".encode()).digest() for j in range(32))
    for i in range(100)
]
table = pa.table({"data": pa.array(unique * 100, type=pa.binary())})
with tempfile.TemporaryDirectory() as tmp:
    results = []
    for name, limit in [
        ("unsplit", 64 * 1024 * 1024),
        ("split", 2 * 1024 * 1024),
    ]:
        path = os.path.join(tmp, f"{name}.lance")
        with LanceFileWriter(
            path, table.schema, max_page_bytes=limit, version="2.3"
        ) as writer:
            writer.write_batch(table)
        reader = LanceFileReader(path)
        assert reader.read_all().to_table() == table
        results.append(
            (name, len(reader.metadata().columns[0].pages), os.path.getsize(path))
        )

    print(results)
    ratio = results[1][2] / results[0][2]
    print(f"size_ratio={ratio:.2f}x")
    assert results[1][2] <= results[0][2] * 2, results

Observed:

[("unsplit", 1, 112695), ("split", 5, 10320898)]
size_ratio=91.58x
AssertionError: [("unsplit", 1, 112695), ("split", 5, 10320898)]

[lance-gatefixer]

Addressed in a0346a5. Oversized automatic dictionaries are now prepared and compressed once, their indices remain split across independent page tasks, and all pages reuse one physical dictionary payload. The binary regression round-trips and bounds split output to at most twice the unsplit size.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The writer now stores the prepared dictionary once, but the reader still treats it as page-local; scanning split data re-reads and decompresses the same payload for every page, replacing file-size amplification with read I/O, CPU, and memory amplification.

A viable revision should make the shared dictionary a reader- or column-scoped resource reused across page schedulers, while keeping page data tasks independent.

Please mark this PR with the breaking-change label.

[lance-gatekeeper]

This submits the same shared-dictionary range separately for every later page, and each scheduler then decompresses it into its own page-local MiniBlockCacheableState. On this head, a full round-trip scan of a 234,924-byte file with 65 split pages read 6,830,097 bytes through 194 physical requests (about 29× the file size), so the <=2× read bound failed. Cache dictionary initialization by physical range plus encoding metadata at reader/column scope—using one shared future or Arc—so page-local chunk indices remain independent while full and range scans fetch and decompress the dictionary once.

Reproducer

I temporarily added scheduler::IoStats to the existing lance_io import, added this test to rust/lance-file/src/writer_tests.rs, and ran:

CARGO_TARGET_DIR=/home/agent/tmp/gate-8405-head2-target cargo test -p lance-file gate_repro_shared_dictionary_read_amplification -- --nocapture

#[tokio::test]
async fn gate_repro_shared_dictionary_read_amplification() {
    let mut state = 0x9e3779b97f4a7c15_u64;
    let unique_values = (0..100)
        .map(|_| {
            (0..1024)
                .map(|_| {
                    state ^= state << 13;
                    state ^= state >> 7;
                    state ^= state << 17;
                    state as u8
                })
                .collect::<Vec<_>>()
        })
        .collect::<Vec<_>>();
    let values = BinaryArray::from_iter_values(
        (0..100_000).map(|index| unique_values[index % unique_values.len()].as_slice()),
    );
    let arrow_schema = Schema::new(vec![Field::new("data", DataType::Binary, false)]);
    let batch =
        RecordBatch::try_new(arrow_schema.into(), vec![Arc::new(values)]).unwrap();

    let (pages, file_size, file_reader) = write_batch_pages_and_size(
        &batch,
        ConcreteFileVersion::V2_3,
        2 * 1024 * 1024,
    )
    .await;
    let stats = IoStats::new();
    let reader = file_reader.with_io_stats(stats.recorder());
    let read_batches: Vec<RecordBatch> = reader
        .read_stream(
            ReadBatchParams::RangeFull,
            1024,
            4,
            FilterExpression::no_filter(),
        )
        .await
        .unwrap()
        .try_collect()
        .await
        .unwrap();
    assert_eq!(
        read_batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
        100_000
    );
    let read_stats = stats.snapshot();
    println!(
        "pages={} file_size={} bytes_read={} requests={} iops={}",
        pages[0].len(),
        file_size,
        read_stats.bytes_read,
        read_stats.requests,
        read_stats.iops
    );
    assert!(
        read_stats.bytes_read <= file_size * 2,
        "full scan read {} bytes from a {}-byte file across {} pages",
        read_stats.bytes_read,
        file_size,
        pages[0].len()
    );
}

Observed:

pages=65 file_size=234924 bytes_read=6830097 requests=194 iops=194
full scan read 6830097 bytes from a 234924-byte file across 65 pages
test result: FAILED. 0 passed; 1 failed

[lance-gatefixer]

Addressed in 5a35829. Mini-block page schedulers now share one asynchronous dictionary initialization keyed by physical range, encoding metadata, and item count, so each page keeps its own chunk index while the dictionary is fetched, decompressed, and retained once. The instrumented multi-page regression round-trips and bounds a full scan to at most twice the file size.

[lance-gatefixer]

Addressed in 5a35829; the reader-side dictionary reuse is detailed in the inline thread. No code change was needed for the separate labeling request: the breaking-change label is now applied to this PR.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The shared-dictionary cache fixes the original read-amplification problem, but dictionary initialization must stay concurrent with page metadata so stable-format scans do not gain an extra network round trip.

A viable revision can retain the shared OnceCell while polling its initialization alongside the metadata request: the first page still performs the one shared fetch, and ordinary per-page dictionaries keep their previous two-phase read path.

[lance-gatekeeper]

This dictionary read is not submitted until after io_req.await and chunk-index construction, and that applies to every mini-block dictionary—not only a dictionary shared by split V2.3 pages. Existing V2.1/V2.2 pages previously submitted their distinct dictionary I/O alongside page metadata, so latency-bound scans now require three waves (metadata → dictionary → data) instead of two (metadata + dictionary → data).

Start polling the shared OnceCell initialization before awaiting the metadata request (for example, join the two futures). One initializer will still fetch a shared dictionary, while ordinary page dictionaries remain concurrent with metadata.

Reproducer

I temporarily added gate_measure_v2_2_dictionary_io to rust/lance-file/src/writer_tests.rs. It builds the existing seeded 100×1024-byte low-cardinality binary data, writes the same 9,000-row batch eight times with ConcreteFileVersion::V2_2, asserts all eight pages use automatic dictionaries, performs a full scan through IoStats, and asserts 72,000 rows. I ran it in detached worktrees for this head and its parent:

CARGO_TARGET_DIR=/home/agent/tmp/gate-8405-head6-current-target cargo test -p lance-file gate_measure_v2_2_dictionary_io -- --nocapture
CARGO_TARGET_DIR=/home/agent/tmp/gate-8405-head6-parent-target cargo test -p lance-file gate_measure_v2_2_dictionary_io -- --nocapture

Parent a0346a5ac: pages=8 file_size=893396 bytes_read=891688 requests=16 iops=24

Current 5a35829330: pages=8 file_size=893396 bytes_read=891688 requests=24 iops=24

The byte count and physical I/Os are unchanged, but every dictionary is deferred into a new request phase. The regression expectation is that dictionary caching must not increase request batches for stable per-page dictionaries.