fix(parquet): prevent cached Mask reads from crossing unloaded sparse pages

[hhhizzz]

Proposed title

fix(parquet): prevent cached Mask reads from crossing unloaded sparse pages

Which issue does this PR close?

• Closes [parquet] Predicate caching with Mask row selection can read an unloaded sparse page #10733.

Rationale for this change

The async Parquet reader may combine page pruning, predicate caching, and a Mask-backed row selection. MaskCursor::next_chunk bounds each chunk by the current loaded row range, but previously retained all trailing skipped rows up to that boundary.

For example, if a loaded range starts with one selected row and the rest of the range is skipped, the cursor returned the whole loaded range as chunk_rows. A fixed-size cached reader can then extend that read beyond the cached segment and into a sparse page that was never loaded, producing:

Invalid offset in sparse column chunk data: ..., no matching page found

DataFusion predicate pushdown exposed this in TPC-DS queries 66, 75, and 81. Disabling the predicate cache avoided the failure, but the cache is enabled by default and should be safe with both Auto and explicit Mask selection.

What changes are included in this PR?

• Track the position immediately after the last selected row in the current loaded range.
• Use that position for both MaskCursor::position and MaskChunk::chunk_rows, instead of the scan position at the end of the loaded range.
• Leave trailing skipped rows to the next cursor step. They become part of initial_skip, allowing ArrayReader::skip_records to cross unloaded pages.
• Update the read-plan unit test to lock down the chunk boundaries.
• Add an async regression test with three pages, a cached predicate column, a sparse initial selection, and both Auto and explicit Mask policies.

The selected rows and output batch semantics are unchanged. The fix only avoids decoding trailing rows that are not selected.

Are these changes tested?

Yes.

Focused regression test:

cargo test -p parquet --features async --test arrow_reader 'row_filter::r#async::test_cached_mask_reads_sparse_pages_without_error' -- --exact --nocapture

• Test-only commit on top of bb1e6cd070: failed with the sparse-column offset error.
• This fix: 1 passed, 0 failed.
• Mask-focused arrow_reader tests: 7 passed, 0 failed.

I also ran one-iteration end-to-end correctness smoke tests with DataFusion 52964c2966e855f47b96a15d1aace01baee1f2c1. For validation only, the Arrow default policy was changed to Mask on top of this PR so the workloads could not fall back to Selectors:

• TPC-DS SF10: 99/99 queries succeeded. The 96 queries that had successful pre-fix results had identical row counts; the previously failing Q66, Q75, and Q81 also succeeded.
• TPC-H SF10: 22/22 queries succeeded, with identical row counts to the pre-fix baseline.
• ClickBench: 43/43 queries succeeded with predicate pushdown explicitly enabled, with identical row counts to the pre-fix baseline.
• No invalid offset, no matching page, panic, or query failure was found in the candidate logs.

These workload runs are correctness smoke tests, not performance claims.

Suggested CI follow-up

cc @alamb
This bug requires an interaction between predicate pushdown, page pruning, selection representation, and predicate caching, which is difficult to cover with isolated reader tests alone. I suggest adding a CI or scheduled benchmark correctness check that runs representative TPC-DS, TPC-H, and ClickBench queries with Parquet predicate pushdown enabled;

Are there any user-facing changes?

No API changes. Async Parquet scans using predicate caching and sparse page reads no longer fail when Auto or explicit Mask selection is used.

[hhhizzz]

I noticed the #10288 has been included in a release, I wondering if need a backport. @alamb
I'm not sure whether I should be happy or sad about this one 😅. The good news is that this has been released for about two weeks and nobody seems to have hit it. The bad news is that DataFusion doesn't enable Parquet predicate pushdown by default, so maybe nobody is exercising this path at all 😂.

[alamb]

This bug requires an interaction between predicate pushdown, page pruning, selection representation, and predicate caching, which is difficult to cover with isolated reader tests alone. I suggest adding a CI or scheduled benchmark correctness check that runs representative TPC-DS, TPC-H, and ClickBench queries with Parquet predicate pushdown enabled;

We can try this -- I think another potential solution would be to implement some sort of fuzz test that tests various selection patterns, etc with pushed down predicates

[alamb]

Thank you @hhhizzz -- I had one small comment suggestion, but otherwise this looks good to me

[hhhizzz]

This bug requires an interaction between predicate pushdown, page pruning, selection representation, and predicate caching, which is difficult to cover with isolated reader tests alone. I suggest adding a CI or scheduled benchmark correctness check that runs representative TPC-DS, TPC-H, and ClickBench queries with Parquet predicate pushdown enabled;

We can try this -- I think another potential solution would be to implement some sort of fuzz test that tests various selection patterns, etc with pushed down predicates

Thanks @alamb — filed both, since I think they catch different things: #10747 for the fuzz coverage, #10746 for an end-to-end CI check.

Fuzzing first, it's cheaper and lives in this repo. One thing I noticed writing it up: test_fuzz_async_reader_selection has no RowFilter at all today, so predicate pushdown and the predicate cache are unreachable from it — and the mask/sparse tests and predicate_cache.rs turn out to be completely disjoint. So this wasn't a gap in depth, it was a gap in the cross product.

That's also why I don't think fuzzing alone closes it. The parameter space is still authored by hand, and the cache axis was missing because caching and row selection look like unrelated subsystems until they aren't. Cross-module interactions are the hard ones to anticipate up front, which is what #10746 is aimed at — running what users actually run, rather than enumerating what we think matters.

#10746 is a correctness check, not a performance one, and needs no new infrastructure: dfbench already records per-query success and row_count, so it's "run the suites with pushdown on, diff against a pushdown-off run". The success check alone would have caught this one.

Happy to take #10747 first if nobody else has started on it.

[hhhizzz]

Ready to merge.

[alamb]

Thank you @hhhizzz 🚀

[alamb]

Backport PR:

• [59_maintenance] Backport fix for cached Mask reads crossing unloaded sparse pages #10766