perf(encoding): batch page reads of a structural column into one I/O request

[dshepelev15]

Problem

In 2.1+ files a nested column is cut into pages by the rep/def budget of one mini-block chunk (max_repdef_levels_per_chunk), not by max_page_bytes. A list<list<list<fixed_size_list<f16, 3>>>> column with ~3000 rows per fragment ends up as 77 pages of ~250 KB each.

StructuralPrimitiveFieldSchedulingJob::schedule_next scheduled one page per call and every page scheduler submitted its own submit_request. The I/O scheduler only coalesces ranges inside one request, so a take that touches many pages costs at least one request per touched page no matter how close the pages sit in the file, and a scan costs one request per page. On this data a take of 2048 rows spread over 8 fragments never went below ~2,300 requests, whatever the coalescing gap.

Change

schedule_next schedules consecutive pages of the request through a PageReadBatch until their reads add up to 8 MiB (the writer's per-column accumulation size, so one batch is roughly what the writer meant to be one page), then submits them as one request sorted by file offset with the lowest page priority. The budget is enforced as reads are queued: a read that would push the batch past 8 MiB first submits what is queued, so a page that shards its own reads to bound buffering (blob pages) keeps every shard in a bounded request and never collapses into one read of the whole page. Each page's load future receives its own slice of the result through a Shared future. Reads submitted after the final flush, such as full-zip data behind a repetition index, pass straight through.

Measurements

Local copy of 8 fragments (24,149 rows, six nested list columns, two of them three levels deep) of a production table. Dataset::take of 2048 rows (256 random rows per fragment), requests counted with io_stats_incremental. block_size is the scheduler's coalescing gap; 64 KiB is the object store default.

file version	rows	block_size	requests before	requests after
2.1	2048 scattered	4 KiB	5,852	3,246
2.1	2048 scattered	64 KiB	4,049	1,322
2.1	2048 scattered	256 KiB	3,280	484
2.2	2048 scattered	64 KiB	4,012	1,293
2.1	2048 contiguous	4 KiB	199	11
2.0 (path unchanged)	2048 scattered	64 KiB	1,360	1,352

Bytes read are the same for a given block_size (62 MiB → 62 MiB at 4 KiB, 100 MiB → 103 MiB at 64 KiB); only the request count changes.

Tests

• page_read_batch_submits_pending_reads_as_one_request: ordering by offset, lowest priority, per-page slicing of the result, pass-through after the flush, and the early submit of a full batch when the next read would overflow it.
• schedule_next_batches_page_reads_up_to_the_budget: pages of one job are read together, untouched pages are not read, the byte budget splits batches, and a page that returns several PageLoadTasks (two 8 MiB shards) gets one bounded request per shard.
• test_read_batches_page_reads_into_one_request (lance-file): a five-page column is read with one data request.
• initialize_builds_only_requested_page_schedulers updated: the two initialized pages now come back from a single schedule_next call.

CPU on a local filesystem

On a local filesystem the change makes the I/O burstier: a batch of pages is requested at once instead of page by page. The local reader runs every pread through spawn_blocking, and on a 128-core machine the wake-ups of tokio's blocking pool contend on one futex, so at the default local io_parallelism the same take spends more CPU while wall time is unchanged (10 takes: task-clock 2.6 s → 6.0 s, wall 0.10 s → 0.10 s, perf shows the extra time in native_queued_spin_lock_slowpath under futex_wait/futex_wake of the blocking pool). With the I/O concurrency bounded the batched reader is cheaper than before (LANCE_IO_THREADS=4: 2.22 s → 2.06 s; =2: 1.80 s → 1.40 s). Object stores do not go through spawn_blocking. The blocking-pool cost is a property of the local reader and worth a change of its own.

[lance-gatekeeper]

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

The page-sharding memory bound is now preserved: queued reads flush before an additional request can overflow the batch, and the multi-task page coverage exercises that boundary.

The reported increase in local-filesystem CPU at default I/O parallelism remains a non-blocking operating risk, although wall time was unchanged and object-store reads are unaffected. Bounding local I/O concurrency is the available mitigation.

[westonpace]

This is fine. It's more of a band-aid than a proper fix though. We really shouldn't be writing small pages in the first place. The problem is that we don't account for compression when we accumulate column data. #4371 would be a better fix but is much more involved. So we can save it for a future task.

[dshepelev15]

@westonpace sounds good, thank you!