fix(reader): read the cache in cache-sized batches

[anoop-narang]

Setting datafusion.execution.batch_size to anything other than the cache's batch size makes liquid-cache scans fail or return the wrong rows.

-- cache batch size is the default 8192
SET datafusion.execution.batch_size = 2145;
SELECT id FROM t;            -- errors
SELECT id FROM t LIMIT 4290; -- succeeds, rows from the wrong offsets

LiquidCacheLocalBuilder::build pins the session batch size to the cache's, so the default configuration does not hit this. A host that changes it afterwards — per query, or by sizing batches to row width — does.

Why

LiquidCacheReader walks the row selection in windows of the session batch size, and numbers each window with current_batch_id. But that id is a cache chunk index: the cache read and the parquet fallback both turn it back into rows using the cache batch size. When the two sizes differ, batch id N names one range of rows to the reader and a different range to everything that resolves it.

• Without a LIMIT, the scan runs off the end of the row group: parquet fallback ended before batch N.
• With a LIMIT that stops it first, it returns rows from the wrong offsets and succeeds.
• On a fully cached row group, a selection mask sized by the session batch size is applied to a chunk sized by the cache's. arrow::compute::filter only rejects a mask longer than its target, so a shorter one silently selects the wrong rows.

The only guard was a debug_assert_eq! in LiquidCacheReader::new, so debug builds panic and release builds carry on.

The fix

Window the selection by the cached row group's batch size, and drop the session batch size from LiquidCacheReaderConfig and LiquidRowGroupPlanner, which no longer need it. Callers still get session-sized batches: DataSourceExec re-splits every source stream to datafusion.execution.batch_size, and session_batch_size_still_bounds_the_batches_the_caller_receives checks that.

Tests

src/datafusion-local/src/tests/batch_size_alignment.rs covers a narrowed and a widened session batch size, LIMIT, the fully cached path, multiple row groups and multiple partitions, all against a cache at the default 8192.

On main with the fix left out, in release (the debug build stops at the assert):

$ cargo test --release -p liquid-cache-datafusion-local --lib batch_size_alignment

test result: FAILED. 0 passed; 7 failed
  ... "parquet fallback read failed: Parquet error: parquet fallback ended before batch 1"
  ... narrowed_session_batch_size_with_limit_returns_aligned_rows: assertion `left == right` failed

With the fix, all 7 pass in debug and release, and cargo test -p liquid-cache-datafusion -p liquid-cache-datafusion-local passes.

One case is not covered here: page-index pruning that starts the selection partway into a row group. A scan with page-index pruning plus a pushed-down predicate panics in boolean_buffer_and_then on main even with matching batch sizes, so it cannot be tested until that is fixed; that is a separate issue.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
⚠️ Please upload report for BASE (main@53414cd). Learn more about missing BASE report.

Additional details and impacted files

@@           Coverage Diff           @@
##             main     #525   +/-   ##
=======================================
  Coverage        ?   81.14%           
=======================================
  Files           ?       80           
  Lines           ?    13447           
  Branches        ?    13447           
=======================================
  Hits            ?    10911           
  Misses          ?     2212           
  Partials        ?      324           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.