perf(parquet): read row groups concurrently

[XiaoHongbo-Hope]

What

• Read independent Parquet row groups concurrently for predicate-free scans.
• Preserve row-group and batch output order.
• Stream one batch per row group with backpressure instead of materializing a full row group.
• Share concurrency and projected-byte budgets across all readers and DataFusion scan partitions.
• Keep predicate and row-selection reads on the existing path.

Resource controls

• read.parquet.row-group.parallelism (default: 8; set to 1 to disable)
• read.parquet.row-group.max-inflight-bytes (default: 256 mb)

The byte budget uses projected uncompressed Parquet bytes and is shared by the whole scan.

Why

A compacted Parquet file is one scan partition, so remote row-group reads were serialized even though the ranges are independent. Per-file concurrency alone could multiply memory and object-store requests when DataFusion runs several scan partitions.

Benchmark

• Direct remote read (28 row groups, three-column projection): 4.27s -> 1.07s
• End-to-end aggregate query, four alternating runs: median 5.156s -> 3.245s (1.59x)
• Disabling row-group concurrency returned the query to about 6s.

Row count, batch count, total count, and result hashes matched.

Tests

• Ordering and overlapping-read regression test
• Multi-reader shared-budget test
• Slow-consumer batch-backpressure test
• Permit release and option-validation tests
• Paimon lib: 2054 passed, 1 ignored
• DataFusion scan tests: 29 passed
• Clippy, formatting, and diff checks passed

[QuakeWang]

+1

[JingsongLi]

Row-group producers fail to stop in a timely manner after downstream cancellation
Location: parquet.rs:517, parquet.rs:667
Issue: The producer first calls reserve() to reserve a channel, then waits for stream.next(); if the downstream drops the receiver due to LIMIT, an error, or query cancellation, the detached task waiting for I/O will not receive the cancellation notification and will continue to hold the ParquetReadPermit.
Trigger Conditions: Start forward_row_group_batches using a pending stream, then drop the receiver; if the task has not exited within 100 milliseconds, the test consistently fails.
Impact: Unnecessary object storage requests and decoding continue to run; more importantly, they continue to consume scan-shared row-group/byte permits, which may slow down DataFusion partitions that are still active.
The fix is very simple:

let next = tokio::select! {
    _ = sender.closed() => return,
    next = stream.next() => next,
};

It is recommended that the coordinator listen for row_group_tx.closed() while waiting for read_budget.acquire(...); after canceling the acquire future, the acquired semaphore permits will be automatically released. After applying the above fix, both my cancellation reproduction test and the original backpressure test passed.

[JingsongLi]

The budget passing process can be simplified as follows:

• Add an optional with_parquet_read_budget parameter to PaimonReadBuilder, which DataFusion injects before new_read().
• Avoid first creating and then discarding a default budget for each partition, only to override it later via the exposed TableRead::with_parquet_read_budget method.
• If shared resources such as caches and memory reservations are added later, they can be centralized into a single ReadContext to reduce the current field passing across a dozen or so readers.