[EPIC] Batch-granular fetch and decode in ParquetPushDecoder

Part of #6946.

What users get

Today, ParquetPushDecoder returns no rows of a row group until the caller has pushed all bytes that the row group reads. For a large row group on object storage, the first batch waits for the full row group, and the decoder holds the full row group in memory.

With this EPIC, a caller can opt in to getting each batch as soon as its pages are pushed:

let mut decoder = ParquetPushDecoderBuilder::try_new_decoder(metadata)?
    .with_fetch_granularity(FetchGranularity::Batch) // new, opt-in
    .build()?;
// `NeedsData` asks only for the pages of the next batch.

Prototype measurement (one row group of 84 MB, 50 ms per request, caller reads 8 MB ahead):

Mode	Time to the first batch	Peak buffered bytes
Today (`FetchGranularity::RowGroup`)	451 ms	84 MB
`FetchGranularity::Batch`	89 ms	8.1 MB

DataFusion can use this mode to scan large row groups with less memory (apache/datafusion#24086).

Terms

Term	Meaning
Push decoder (`ParquetPushDecoder`)	A Parquet decoder that does no I/O. `try_decode` returns `NeedsData(ranges)`; the caller fetches the byte ranges and pushes them.
Row group, column chunk, page	A row group holds one column chunk per column. A column chunk is a sequence of pages (an optional dictionary page, then data pages).
Offset index	Parquet metadata that gives the file offset, size and first row of each data page.
Predicate (`RowFilter`)	A filter that the decoder evaluates while it decodes, to skip rows of other columns.
Read-ahead	The caller fetches ranges before the decoder asks for them.

Plan

#11223 (+4,156 lines, now closed) was too large to review in one step. The PRs below replace it. Each PR is useful without the PRs after it, and no PR adds a temporary error or fallback that a later PR removes.

PR	What users get	Size	State
#11218	A test that shows the current behavior		Merged
#11235	The decoder releases the bytes of a finished row group (fix for #11233)		Merged
#11237	Faster lookups when a caller pushes many buffers		Merged
#11484	A file with a bad offset index gives an error, not a panic		Merged
#11485	Tests only: the push decoder against the sync reader, with `proptest`	+636	Ready for review
#11238	`FetchGranularity::Batch`: the first batch of a row group as soon as its pages are pushed	+2,364 (about 1,100 are tests)	Draft until #11485 merges
#11487	Batch mode: no empty decode after each batch	+70	Draft until #11238 merges
#11240	Batch mode: filtered scans decode the predicate columns one time, peak memory of about one batch, and dense row selections use the faster mask read	+690	Draft until #11487 merges

main ── 11485 tests ── 11238 batch mode ── 11487 ── 11240 batch-mode optimizations

Review order:

Ready now: test(parquet): differential tests of ParquetPushDecoder against the sync reader #11485.
Then: feat(parquet): let ParquetPushDecoder decode a batch as soon as its pages are pushed #11238. It has two commits that can be reviewed in sequence or as one diff.
Then: perf(parquet): skip the empty decode after each batch in FetchGranularity::Batch #11487 and feat(parquet): predicate cache, page release and selection policy in the batch-granular push decoder #11240.

Tangential

ParquetPushDecoder does not release pushed buffers that are larger than the requested ranges #11233 (fix: fix(parquet): release the bytes of a row group when the push decoder is done with it #11235): the decoder does not release a pushed buffer that is larger than the requested ranges. This is a bug in the default mode. feat(parquet): let ParquetPushDecoder decode a batch as soon as its pages are pushed #11238 uses the fix.

feat(parquet): release each page when the batch-granular push decoder…

feat(parquet): use the predicate cache in the batch-granular push dec…

feat(parquet): release each page when the batch-granular push decoder…

feat(parquet): use the predicate cache in the batch-granular push dec…

perf(parquet): use the configured selection policy in batch windows

feat(parquet): release each page when the batch-granular push decoder…

feat(parquet): use the predicate cache in the batch-granular push dec…

perf(parquet): use the configured selection policy in batch windows

@alamb

Note

Out of date: see #11234 (comment)

@alamb, as you asked on #11223 (comment), I split #11223 (+4,156 lines) into the PRs below. This comment gives the reason for each cut and a suggested review order.

How the split works

Rule	Result
Each PR is useful without the PRs after it	The 4 PRs on `main` fix or speed up the current decoder. Each PR on top of the core adds one improvement to batch mode.
No PR adds a temporary error or fallback that a later PR removes	Predicates must be in the core PR, because a batch mode without predicates needs a fallback for filtered scans
Later PRs only add optimizations to code that is already correct	Page release, predicate cache and selection policy are follow-ups
One decode path	The core has one windowed engine. A scan without predicates is the case with zero predicates. This removed the separate no-predicate path of #11223 and about 600 lines.

Suggested review order

Step 1 (in parallel, all on main)          Step 2          Step 3 (in parallel)
  #11218  test (yours)          ─┐
  #11235  bug fix #11233        ─┴──────▶  #11238  ──┬──▶  #11239  page release
  #11236  rebuild releases bytes                     ├──▶  #11240  predicate cache
  #11237  sorted PushBuffers                         └──▶  #11241  selection policy

Step	PR	What users get	New lines
1	#11218	A test of the current behavior	+83
1	#11235	Fixes #11233: the decoder keeps all pushed bytes until the end of the scan if the caller merges the requested ranges	+311
1	#11236	`into_builder().build()` releases the pushed bytes of the row groups that the new decoder skips	+204
1	#11237	Sorted `PushBuffers`: `Nbuf/100000ranges` goes from 574 ms to 35 ms	+225 / −41
2	#11238	`FetchGranularity::Batch`: the first batch of a row group as soon as its pages are pushed	+2,847 (about 1,580 tests)
3	#11239	Batch mode: peak memory of about one batch, not one row group	+171
3	#11240	Batch mode: filtered scans decode the predicate columns one time	+149
3	#11241	Batch mode: the configured `RowSelectionPolicy` in each window	+176

Step 2: how to read #11238

It is the largest PR, because the core cannot be smaller without a temporary fallback. About half of it is tests. Suggested reading order:

The docs of FetchGranularity in push_decoder/mod.rs: the behavior that users see, and the I/O contract.
The docs of IncrementalRowGroup and Stage in reader_builder/incremental.rs: the design.
IncrementalRowGroup::step and ColumnChunkPages: the engine.
The wiring in reader_builder/mod.rs, remaining.rs and push_decoder/mod.rs, which is small.
incremental_tests.rs: most tests compare batch mode with the default mode on the same scan.

The PR description also records two decisions from review: windows stay batch_size rows of the row group, and the caller decides how far to read ahead.

Merge notes

PRs	Note
#11235, #11236, #11237	All edit `util/push_buffers.rs`. The PR that merges last needs a small rebase: remove the duplicate `merge_ranges` and test helper, and update the maximum buffer length of #11237 in `release_ranges` and `retain_ranges`.
#11239, #11240	If both merge, the second one adds one rule: a column in the predicate cache is released at cache-batch boundaries
#11238 and step 3	I will rebase them after each merge

#11223 stays open as a reference for the full change. I will close it when the stack merges.

@sunchao

(sorry for the ping above andrew, i told it to post a summary so i could read it and then I was going to ping you after vetting, i always try to not have an ai ping a human, appoligies. that said i've now read the comment above and it is reasonable 😄)

cc @sunchao

feat(parquet): let ParquetPushDecoder decode a batch as soon as its p…