Repository navigation
feat: aggregate a large final hash table as hash buckets (experimental, off by default) - #25724
jayzhan211 wants to merge 5 commits into
Conversation
bdd382c to
f9df89e
Compare
|
@Rachelint I couldn't get rid of all the regressions, but given the size of the win compared to the loss, do you think it's still worth adding this as opt-in? 🤔 |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25724 +/- ##
==========================================
+ Coverage 82.64% 82.65% +0.01%
==========================================
Files 1147 1149 +2
Lines 445440 446512 +1072
Branches 445440 446512 +1072
==========================================
+ Hits 368123 369067 +944
- Misses 54970 55002 +32
- Partials 22347 22443 +96 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
run benchmark clickbench_partitioned |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (f9df89e) to fb42cf5 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (f9df89e) to fb42cf5 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
run benchmark clickbench_partitioned |
|
Hi @Rachelint, your benchmark configuration could not be parsed (#25724 (comment)). Error: Usage: Any benchmark name is accepted: Per-side configuration ( env:
# shared env is inherited by BOTH the build and the run, so build
# flags go here. Builds default to no debuginfo for speed; opt back
# in for hung-job gdb dumps and cap jobs to stay within memory:
CARGO_PROFILE_RELEASE_DEBUG: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: v45.0.0
env:
# per-side env only reaches the benchmark run, not the build
DATAFUSION_RUNTIME_MEMORY_LIMIT: 1G
changed:
ref: v46.0.0
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: 2GPod size and CPU architecture (each key is optional and keeps the default when absent; requests and limits are set equal): resources:
cpu: "16"
memory: "128Gi"
arch: arm64 # arm64 or amd64File an issue against this benchmark runner |
|
run benchmark clickbench_partitioned |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (f9df89e) to fb42cf5 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (f9df89e) to fb42cf5 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
52a76ba to
7b7c506
Compare
|
run benchmarks clickbench_partitioned tpch10 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (7b7c506) to 22a0fdb (merge-base) diff Run configurationrun benchmark tpch10
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (7b7c506) to 22a0fdb (merge-base) diff Run configurationrun benchmark clickbench_partitioned
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (7b7c506) to 22a0fdb (merge-base) diff Run configurationrun benchmark tpch10
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (7b7c506) to 22a0fdb (merge-base) diff Run configurationrun benchmark clickbench_partitioned
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
On the Q25 "2.04x slower" in the latest run: I don't think it comes from this PR.
In this run, main's Q24 stall is larger than the branch's Q25 one. Which query absorbs the stall changes between runs; what stays the same is that it follows Q23. Q19 and Q29 get similar outliers on both sides.
To separate the option from the binary, I can ask for a run where both sides use this commit and only the threshold differs (0 vs 262144). |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (88d36d4) to 7b1e55d diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"
baseline:
ref: "7b1e55d"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
I agree that we should start by applying the bucket approach only to fixed-width types. Despite trying several ideas over the past few days, I haven’t yet found a way to reduce the routing overhead enough to make the approach broadly beneficial... I also have a few thoughts we could explore as follow-ups:
|
|
@Rachelint I'll take the Partial flush out of this PR and leave it to #26124 / #26116, so this PR only buckets the Final stage. On compaction, I'd rather keep it. Without it, a bucket holds every row it receives until the end. When the Partial stage passes rows through mostly unreduced (skipped, or few rows per group), Final memory then grows with the input rows instead of the groups. Before I added compaction, peak RSS with bucketing on was 2.3x of bucketing off on 20M random integer keys and 2.26x on TPC-H SF10 I answered the question about #24704 on the issue. |
The heavy ClickBench GROUP BY statements run one plan shape: a partial aggregate that skips, a 32-way hash repartition, and a FinalPartitioned aggregate whose group tables are far larger than any cache. On q33 the final never hits (99,997,493 groups over as many rows), 35.5% of server cycles are stalled on memory, and the cache-miss profile puts 0.40 of the misses in the final's group-table probe and 0.36 in the exchange's scatter. DataFusion 55.2 has no hook to give the final a different group table, and the upstream bucketed aggregate (apache/datafusion#25724) is a draft against a later release. Record the decision to add a Ravel ExecutionPlan, behind a SqlConfig field that is off by default, that aggregates one hash bucket at a time in a cache-sized table once a partition's final passes 65,536 groups, and cuts each completed bucket to its exact top k when a SortExec with a fetch sits above it. The exec charges its retained input to its own consumer before retention, never spills, and refuses string, float, boolean, Decimal128 and Duration keys. It is enabled only after a pre-registered paired A/B on the reference machine, with the bands in the ADR. Fused routing, which would take back the router copy the exec adds, is a later amendment measured against this decision. The memory gate rests on a new factor-free peakAskedIntermediateBytes query stat, because the ledger's peak_intermediate_bytes charges the stock consumers at the reserve-ahead factor and the new one at 1.0, and process VmHWM is a lifetime figure once the allocator keeps its arenas. Refs: NOFireAI#3565 Signed-off-by: Panagiotis Moustafellos <[email protected]>
|
Just a suggestion since I fell for it. don't benchmark on macbooks since they have very different mem and cpu behavior than linux machines (especially in memory performance once it pass L1 or something) |
|
Thanks @jayzhan211 I was actually about experimenting the same direction. I can benchmark it on 1TB if you resolve conflicts |
88d36d4 to
8f2fb69
Compare
…, off by default)
…ting buckets of unique groups
8f2fb69 to
ee5866b
Compare
|
run benchmarks clickbench_partitioned clickbench_extended tpch10 |
|
@comphead the conflicts are resolved. Could you run your 1TB benchmark with What to expect:
|
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (ee5866b) to ee5866b diff Run configurationrun benchmark clickbench_extended
baseline:
ref: "ee5866bc6fd319baf7bde1085ff9de0dc9095f30"
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "0"
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (ee5866b) to ee5866b diff Run configurationrun benchmark tpch10
baseline:
ref: "ee5866bc6fd319baf7bde1085ff9de0dc9095f30"
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "0"
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing agg-buckets-final-stage (ee5866b) to ee5866b diff Run configurationrun benchmark clickbench_partitioned
baseline:
ref: "ee5866bc6fd319baf7bde1085ff9de0dc9095f30"
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "0"
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (ee5866b) to ee5866b diff Run configurationrun benchmark tpch10
baseline:
ref: "ee5866bc6fd319baf7bde1085ff9de0dc9095f30"
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "0"
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (ee5866b) to ee5866b diff Run configurationrun benchmark clickbench_partitioned
baseline:
ref: "ee5866bc6fd319baf7bde1085ff9de0dc9095f30"
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "0"
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing agg-buckets-final-stage (ee5866b) to ee5866b diff Run configurationrun benchmark clickbench_extended
baseline:
ref: "ee5866bc6fd319baf7bde1085ff9de0dc9095f30"
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "0"
changed:
env:
DATAFUSION_EXECUTION_HASH_AGGREGATE_BUCKET_THRESHOLD: "262144"CPU Details (lscpu)Details
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
I'm turning this PR back to draft: with buckets on, it makes some aggregations faster and others slower, and I don't see a fix yet that doesn't add more tuned limits. On the benchmark runner (same commit, threshold 0 vs 262144):
My laptop also shows costs the description understates:
|
Which issue does this PR close?
Rationale for this change
Once the table of a final hash aggregation outgrows the CPU caches, every probe and accumulator update is a cache miss, so the cost per row grows with the number of groups. With
datafusion.execution.hash_aggregate_bucket_threshold = 262144, integer keys with 4M-60M groups run in 0.68x-0.76x of the time and ClickBench Q32 in 0.75x. Queries whose final table stays below the threshold run as before. The option is experimental and off by default.When the final table of one partition reaches the threshold, the stream moves its state and all further input into 64 hash buckets, then aggregates the buckets one after another with one small, reused table. Each bucket is emitted, released, spilled or split again on its own.
Three constraints shaped the design:
M4 Pro, 12 partitions, same binary off and on, median of 5-7 interleaved rounds:
BIGINTWatchID, ClientIPUserIDControls and known costs
Controls (no bucketing): ClickBench Q5 1.00x, Q11 0.99x, Q13 0.98x, Q17 0.96x; 1M uniform keys (83k per final partition) 0.99x; near-unique integer keys ending at 104k-174k groups per partition 0.99-1.00x.
Known costs:
GROUP BY l_partkey, l_suppkey) run in 0.74x of the time with 1.13x peak RSS (622 to 702 MB).What changes are included in this PR?
Start reading at
BucketedAggregationinaggregates/bucketed_aggregation.rs:split,compact,reserveandoutput_streamare the whole life of the buckets.aggregates/final_buckets.rsstores them;FinalHashAggregateStreamonly decides when to switch.The decision I most want reviewed is keeping compaction in this PR rather than adding it later: without it, buckets hold every row that arrives after the switch until the input ends, so memory grows with the input instead of with the groups.
Behavior of each piece
LIMITpushed into the aggregation), spilling enabledWhat is the testing strategy for this PR?
hash_stream.rsunit tests: buckets return the same groups as one table, split again, spill under a memory limit, compact repeated groups, and stay off without a disk manager.final_buckets.rsandbucketed_aggregation.rstest routing and eligibility.aggregate_bucketed.sltruns each query off and at 100 groups with identical results: integer, nullable,Utf8,Utf8Viewand mixed keys, AVG, DISTINCT, COUNT(DISTINCT), median,array_agg.memory_limittests and the whole sqllogictest suite pass, exceptinformation_schema.slt, which prints the default.Are there any user-facing changes?
A new experimental option, off by default and documented in
configs.mdwith its costs, and two metrics onAggregateExec:bucket_splits,bucket_compactions.