Repository navigation
Conversation
ListingTable collects per-file statistics with buffer_unordered, so every footer decode and statistics conversion ran on the one thread polling the planner. In a multi-thread tokio runtime, ParquetFormat::infer_stats_and_ordering now does that work in a SpawnedTask so files are decoded in parallel. Outside one it does the work inline as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019C4oRqstvxH5CXTWnot8CD
Contributor
Author
|
run benchmark wide_schema |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark wide_schemaResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark wide_schemaCPU Details (lscpu)Details
Resource Usagewide_schema — base (merge-base)
wide_schema — branch
File an issue against this benchmark runner |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #26407 +/- ##
==========================================
- Coverage 82.87% 82.87% -0.01%
==========================================
Files 1153 1153
Lines 457101 457136 +35
Branches 457101 457136 +35
==========================================
+ Hits 378818 378844 +26
- Misses 55268 55274 +6
- Partials 23015 23018 +3 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf-parallel-parquet-stats (2f055f2) to 5009f42 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU 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
4 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?
Rationale for this change
When a
ListingTableover Parquet files collects statistics, planning time grows with the number of files multiplied by the size of their footers, and none of that work runs in parallel. On wide tables it dominates planning.ListingTable::collect_files_for_scangathers per-file statistics withbuffer_unordered(meta_fetch_concurrency). That makes the object store requests concurrent, but every future is polled by the one planning task. As a result, the footer decode and statistics conversion for each file run one file at a time, on whichever thread is polling the planner.In a profile of
wide_schemaQ02 (1024 columns, 256 files) with the metadata and statistics caches disabled, the planning thread spent about 345 ms per query, which is roughly the whole query time. About three quarters of that time was insideParquetFormat::infer_stats_and_ordering. With the default cache limits the statistics cache cannot hold the statistics for this table, so the same work happens on every query.What changes are included in this PR?
ParquetFormat::infer_stats_and_orderingresolves its session-dependent inputs (the decryption properties and the metadata cache) on the calling task. When it runs inside a multi-thread Tokio runtime, it then fetches and decodes the footer, and computes the statistics and ordering, in aSpawnedTask. With the existingbuffer_unordered, up tometa_fetch_concurrencyfiles are decoded in parallel on the runtime's threads.buffer_unorderedpolls it together with the other files. This is the same checkspawn_buffereduses.fetch_stats_and_ordering, which both paths call.This follows the convention described in the crate docs: CPU-bound work runs as tasks on the Tokio runtime, through
SpawnedTask, rather than inspawn_blocking. Dropping aSpawnedTaskaborts it, so whenget_files_with_limitstops early, outstanding work is still cancelled.ParquetFormat::infer_schemahas the same serial pattern when a table is created. It is left for a follow-up.What is the testing strategy for this PR?
The results are unchanged; only where the work runs has changed, so the existing tests cover it. The
#[tokio::test]unit tests run on a current-thread runtime and take the inline path, while the sqllogictests run on a multi-thread runtime and take the spawned path.datafusion-datasource-parquetlib testsdatafusion-catalog-listingtestsdatafusionparquet_integrationtests and the coredatasource::listingtestsThe performance change will be checked with the
wide_schemabenchmark.Are there any user-facing changes?
There are no API changes. Planning that collects Parquet statistics uses more than one thread when it runs inside a multi-thread Tokio runtime, and behaves as before otherwise.
🤖 Generated with Claude Code
https://claude.ai/code/session_019C4oRqstvxH5CXTWnot8CD