[Priya2996]

Rationale for this change

The asynchronous Parquet record batch generator decoded an entire row group into memory before slicing it into record batches. For files containing large row groups, this caused high peak memory usage even when a smaller batch_size was configured.

What changes are included in this PR?

• Reuse the existing RecordBatchReader logic to decode row groups incrementally.
• Yield batches as they are decoded instead of materializing the complete row group first.
• Preserve asynchronous execution and row-group readahead.
• Avoid deadlocks when the same executor is configured for CPU and I/O work by disabling parallel column decoding in that case.
• Keep the existing non-pre-buffered fallback unchanged.

Are these changes tested?

Yes. Tests verify that:

• Row groups are returned as multiple batches of the configured size.
• Reassembling the batches produces the original table.
• Empty-column projections preserve row counts and batch sizes.
• A single-thread CPU executor completes successfully.
• A single executor shared by CPU and I/O work does not deadlock.
• Dataset scans with multiple files, row groups, and readahead complete with a single I/O thread.

Are there any user-facing changes?

There are no public API changes.

For pre-buffered Parquet dataset scans, the existing batch_size option now also limits the number of rows decoded per batch instead of only controlling how an already-decoded row group is sliced. This reduces peak memory usage for large row groups.

Was AI used for this PR?

AI was used to help trace the relevant call chain, review concurrency behavior, suggest tests, and draft portions of the implementation and PR description. The changes were reviewed before submission.

Benchmark

Workload: one 262,144-row group with 128 int64 columns and pre-buffering enabled. Results are medians of five runs using Google Benchmark wall-clock time.

Implementation	Batch size	Rows/s	Peak reader memory
Current eager path	Full row group	139.3M	268.4 MB
Incremental, serial	32,768	61.1M	38.3 MB
Incremental, threaded	32,768	70.3M	34.9 MB
Incremental, threaded	65,536	103.0M	69.2 MB
Incremental, threaded	131,072	120.6M	138.4 MB
Incremental, threaded	262,144	127.1M	268.4 MB

The existing use_threads option remains supported. At a 32,768-row batch size, threaded column decoding improved incremental throughput from 61.1M to 70.3M rows/s in this workload.

These results show that incremental decoding turns the existing batch_size setting into a direct memory-throughput tuning knob. Smaller batches reduce peak decoded memory, while larger batches recover throughput.

With the eager path, batch_size controlled only the emitted batches; the complete row group was still decoded first, so peak decoded memory remained determined by row-group size. The incremental path lets users choose an
appropriate point in this tradeoff instead of always accepting a row-group-sized decoded allocation.

A 1,481,181-row, 650-column Parquet file with one row group was scanned in a fresh process for each case.

Path	Batch size	Arrow peak	RSS peak
Eager Dataset decode	65,536	8.290 GiB	5.420 GiB
Incremental Dataset decode	65,536	0.877 GiB	1.144 GiB
Eager Dataset decode	131,072	8.290 GiB	6.834 GiB
Incremental Dataset decode	131,072	1.739 GiB	1.873 GiB

The incremental path reduced Arrow peak memory by 89.4% at a 65,536-row batch size and by 79.0% at 131,072 rows. The eager path's peak was unchanged across batch sizes, while incremental memory scaled with the configured batch size.

Associated Arrow Issue Links

#36778
#20473

Downstream relevance

We encountered this behavior while investigating high memory usage in PyIceberg’s to_arrow_batch_reader(). Tracing the call path into PyArrow showed that the asynchronous Parquet scanner decoded an entire row group before applying the configured batch_size. This Arrow-level behavior can amplify memory usage when Parquet files contain large row groups. The change in this PR makes decoding incremental for the pre-buffered scanner path. This will also help address concerns similar to ours for decoding large row groups using PyIceberg’s to_arrow_batch_reader

PR code and description written by:

• Human
• AI

Reviewed before submission by:

• Human
• AI
• Not reviewed

• GitHub Issue: [C++] Scanner slicing large row groups leads to inefficient RAM usage #20473

[github-actions]

⚠️ GitHub issue #20473 has been automatically assigned in GitHub to PR creator.

[dmitry-chirkov-dremio]

Comment update needed?
Perhaps even benchmark the change of behavior since new benchmark does not cover it?

[Priya2996]

Thanks for pointing this out! I updated the comment and adjusted the benchmark to cover the new incremental batching behavior.

[dmitry-chirkov-dremio]

This test prints OK, then the process segfaults in ProxyMemoryPool::Free. The generator has already started another batch, but the test ends and destroys both memory pools.

Please wait for the background work before leaving this scope. Draining and discarding the remaining batches fixed my local run. Please check the benchmark's reader_pool lifetime too.

[Priya2996]

Thanks for catching this! I’ve drained the generator and verified the benchmark’s pool lifetime.