Experimental: parquet bounded streaming of large column chunks (companion to arrow-rs#10410)#23795
Draft
adriangb wants to merge 5 commits into
Draft
Experimental: parquet bounded streaming of large column chunks (companion to arrow-rs#10410)#23795adriangb wants to merge 5 commits into
adriangb wants to merge 5 commits into
Conversation
…olumn chunks Adds datafusion.execution.parquet.bounded_streaming (default false). When enabled, the parquet push decoder decodes row groups with large projected column chunks in bounded windows, and the fetch loop consumes those chunks via single ranged requests whose bodies stream into a bounded buffer (parquet AdaptiveFetcher), instead of buffering every projected byte of a row group before decode starts. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG
github-actions
Bot
added
documentation
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Makes the PR buildable by CI and benchmark infrastructure until the parquet changes are released. Must be removed before merge. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG
Contributor
Author
|
run benchmark tpch clickbench_partitioned tpcds changed: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bounded-streaming-core (5e4bc31) to 5de7f1d (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bounded-streaming-core (5e4bc31) to 5de7f1d (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bounded-streaming-core (5e4bc31) to 5de7f1d (merge-base) diff using: tpcds File an issue against this benchmark runner |
Contributor
Author
|
run benchmark tpch clickbench_partitioned tpcds env: changed: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bounded-streaming-core (5e4bc31) to 5de7f1d (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bounded-streaming-core (5e4bc31) to 5de7f1d (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bounded-streaming-core (5e4bc31) to 5de7f1d (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG
Contributor
Author
|
run benchmark clickbench_partitioned baseline: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing main (96e8fdf) to main diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
None yet — draft companion to the arrow-rs bounded streaming RFC: apache/arrow-rs#10410. This PR will not compile in CI until that lands and is released (it was developed and benchmarked against a
[patch.crates-io]of that branch, not included here).Rationale for this change
The parquet push-decoder fetch loop materializes every projected byte of a row group before decoding. For wide-value columns (e.g. 100 MB single-row-group files whose projected blob column is ~97 MB), peak reader memory ≈ row group size × concurrent partitions, and decode start is gated on the slowest range completing. arrow-rs#10410 adds an opt-in bounded streaming mode; this PR wires it through DataFusion.
What changes are included in this PR?
datafusion.execution.parquet.bounded_streaming(defaultfalse).DecoderBuilderConfigapplieswith_bounded_streamingto the push decoder builder when enabled.PushDecoderStreamStategains aFetchState: the plainget_byte_rangesloop as today, or the parquetAdaptiveFetcher(small coalesced ranges materialized exactly as before; large column chunks consumed incrementally window by window). Falls back to the plain path when no tokio runtime is available.ParquetFileReader/CachedParquetFileReaderforwardget_bytes_streamtoParquetObjectReader(without this the fallback silently materializes).Benchmark summary (harness changes in the companion benchmarks PR)
On the motivating shape (8 × 94 MB single-row-group files, 92 MB blob column, full-scan checksum): identical results and request counts (8 GETs both), wall neutral (29.9 → 29.7 ms median SSD), peak RSS 1261 → 702 MB (−44%). TPC-H SF1 / TPC-DS SF1 / stock ClickBench: neutral within noise, zero request inflation (streaming does not engage below the 16 MiB threshold or without an offset index). Known issue inherited from the arrow-rs draft: ~+14% wall on ClickBench-with-page-indexes string queries from per-window dictionary re-decode (fix identified upstream: cache decoded dictionaries across windows).
Known CI failure:
datafusion-common'stest_cast_struct_with_array_and_map_fieldsfails when building against arrow-rs git main (Map entry field naming changed upstream:key/valuevskeys/values). This is a DataFusion-main vs arrow-main incompatibility that the next arrow upgrade PR will resolve; it is unrelated to this change and only surfaces here because of the temporary git patch.Are these changes tested?
datafusion-datasource-parquetsuite green; correctness validated off-vs-on (identical results incl. checksums) on the benchmark datasets. sqllogictestinformation_schemaand config docs updated.Are there any user-facing changes?
New experimental config option, default off.
🤖 Generated with Claude Code
https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG