fix: stop pre-allocating and undercounting a hash table per COUNT(DISTINCT) group - #24857
fix: stop pre-allocating and undercounting a hash table per COUNT(DISTINCT) group#24857adriangb wants to merge 7 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24857 +/- ##
==========================================
+ Coverage 81.64% 81.65% +0.01%
==========================================
Files 1123 1123
Lines 410248 410542 +294
Branches 410248 410542 +294
==========================================
+ Hits 334940 335233 +293
Misses 55617 55617
- Partials 19691 19692 +1 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
run benchmark clickbench_partitioned external_aggr |
|
run benchmark clickbench_extended |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (84f07da) to da89c7c (merge-base) diff Run configurationrun benchmark external_aggr
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (84f07da) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (84f07da) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (84f07da) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (84f07da) to da89c7c (merge-base) diff Run configurationrun benchmark external_aggr
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageexternal_aggr — base (merge-base)
external_aggr — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (84f07da) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
Summary of benchmarks and measurementsConsolidating the evidence for this PR in one place, including the parts that are not yet settled. 1. Minimum viable memory budget (strongest evidence)Two tests added in
The tests are pinned at 8 MB and 16 MB, at least 4x clear of both cliffs, and were verified to The 2. Per-accumulator footprintOne per-group accumulator holding a single 24-byte value, measured on the merge-base and on this
Two separate defects show up in that table. The pre-allocation is the large one. The reporting 3. Benchmarks, memory
Everything else moves by less than 5% in either direction. Worth stating plainly: this rests on a Other suites at the same limit: 4. Benchmarks, latencyNo effect, which is the expected result for an allocation-sizing change. 5. Open: peak RSSPeak process RSS rose on all three runs: 9.3 to 9.6 GiB on It is not yet clear whether that is real. A companion accounting-only change that cannot alter If the effect is real, the likely mechanism is allocator retention rather than live memory: |
|
run benchmark clickbench_extended clickbench_extended clickbench_extended Noise floor, not a comparison of this PR.Both sides are pinned to da89c7c, the merge-base the earlierclickbench_extended run used, so the two sides compile identical code andevery difference reported below is harness noise. Three jobs give threeindependent A/A readings of peak RSS and of the per-query pool peaks.env: |
|
run benchmark clickbench_extended clickbench_extended clickbench_extended Three repeats of the comparison in#24857 (comment),with both sides pinned so the repeats measure the same two commits.84f07da is the last commit on this branch that changes benchmarked code;the two commits after it touch onlydatafusion/core/tests/memory_limit/mod.rs.env: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing da89c7c (da89c7c) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 84f07da (84f07da) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "84f07dadabe68e25900b59909c9c0c28e99c7c56"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 84f07da (84f07da) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "84f07dadabe68e25900b59909c9c0c28e99c7c56"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing da89c7c (da89c7c) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 84f07da (84f07da) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "84f07dadabe68e25900b59909c9c0c28e99c7c56"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing da89c7c (da89c7c) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 84f07da (84f07da) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "84f07dadabe68e25900b59909c9c0c28e99c7c56"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 84f07da (84f07da) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "84f07dadabe68e25900b59909c9c0c28e99c7c56"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 84f07da (84f07da) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "84f07dadabe68e25900b59909c9c0c28e99c7c56"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing da89c7c (da89c7c) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing da89c7c (da89c7c) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing da89c7c (da89c7c) to da89c7c diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
baseline:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"
changed:
ref: "da89c7c85b1c888ba4038317eacbdcfe03ab9b1a"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
Repeat runs: RSS question resolved, and two correctionsFollowing up on the open item in the summary above. Six further runs of Peak RSS: no effect. The earlier increase was a single high draw.Peak RSS at the merge-base across 11 readings of identical code: 9.4, 9.5, 9.7, 9.9, 10.0, 10.1, Within-run RSS delta on identical code: 0.0%, -2.9%, -4.0%. On the three base-vs-branch repeats: There is also a mild order effect: the base side runs first in each pod and the second side reads Query 2 reproduces exactly. The single-run caveat is retired.Null spread on identical code, 6 readings: 98.1, 98.1, 98.2, 98.2, 98.2, 98.6 MiB, a mean delta of Across the three repeats the changed side read 11.7 MiB every time, giving -88.12%, -88.09%, It is also mechanistically predicted rather than only measured. Query 1 also holds: 3.4 MiB base and 2.4 MiB changed in every run, with zero variance on either Correction 1: the Query 12 number was noise. Retracting it.The summary above cited Q12 at -21.2%. That does not survive. The null spread on Q12 is -7.1% to Correction 2: Query 0's pool peak genuinely increases, by about 4.2%This is a real, opposite-signed cost that the first run did not surface, and it should be on the Base commit, 11 readings: 796.8 to 840.8 MiB, mean 821.9. Branch, 4 readings: 846.5, 850.5, 858.5, Q0 is The overall picture is coherent. Ungrouped and low cardinality (Q1) improves because the A note on this harness that applies beyond this PRPool peak is a trustworthy instrument here and peak RSS is not. Pool peak reproduced to under 1% on Result comments: base-vs-base |
|
run benchmark clickbench_extended clickbench_extended clickbench_extended |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
run benchmark clickbench_extended clickbench_extended |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
|
run benchmark clickbench_extended |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/bytes-map-initial-capacity-accounting (b9e6252) to da89c7c (merge-base) diff Run configurationrun benchmark clickbench_extended
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_extended — base (merge-base)
clickbench_extended — branch
File an issue against this benchmark runner |
Query 0: six runs, and a correction to the mechanism I gave earlierAn earlier comment on this PR put Query 0's pool-peak increase at +4.2% from four readings, and Six runs on the current headAll six ran
Mean +6.3%. The base readings span 796.8 to 834.8 MiB and the changed readings span 846.5 to It is the accounting correction, not new allocationMeasured locally on a deterministic 3-file subset with
That decomposes the increase exactly:
The transient explanation fails on two counts. The pool samples Query 0 is Why this is not a regression to fixThe extra 6.3% is memory the query always held and the pool could not see. Reverting the change Restoring the warm-up was tested and makes Query 0 worse, so the "capacity hint from Query 2 across the same six runsThe changed side reported 11.7 MiB in every run, against a base of 98.1 to 98.6 MiB. Six |
…wBytesViewMap Both maps tracked their hash table footprint in a `map_size` field that was only ever incremented by `HashTableAllocExt::insert_accounted`, which charges `capacity * size_of::<Entry>()` on growth and nothing else. That undercounts in two ways. `ArrowBytesViewMap::new` seeded `map_size` with `capacity() * size_of::<Entry<V>>()`, which ignores the control bytes and the trailing group that hashbrown allocates alongside the entry array, so the reported size was roughly half the real allocation. `ArrowBytesMap::new` seeded `map_size` with 0 despite pre-allocating a table for 128 entries. Since `insert_accounted` only charges when the table grows, any map holding fewer entries than the pre-allocated capacity reported its hash table as free forever. Drop the field and ask hashbrown for the exact figure with `HashTable::allocation_size`, which covers entries, control bytes and the trailing group. It is a constant time layout calculation, so `size()` stays cheap, and it cannot drift out of sync with the table the way an incrementally maintained counter can.
`ArrowBytesMap` and `ArrowBytesViewMap` always pre-allocated their hash table, and `ArrowBytesMap` also pre-allocated an 8 KiB value buffer. That is the right trade for the single map that backs a `GROUP BY` on one string column, which goes on to hold every group value in the query. It is the wrong trade for `BytesDistinctCountAccumulator` and `BytesViewDistinctCountAccumulator`, because `GroupsAccumulatorAdapter` creates one accumulator per group: a grouped `COUNT(DISTINCT)` over a high cardinality key holds hundreds of thousands of them at once, and most see only a handful of values, so the pre-allocation dwarfs the data. Split the constructors. `new` no longer allocates anything, and `with_capacity` keeps the previous behavior for the callers that want it. The capacity is stored so `take` re-creates the map the way it was built. The `GroupValuesBytes` and `GroupValuesBytesView` call sites move to `with_capacity`; the two distinct-count accumulators stay on `new`. The `arrow_bytes_map` benchmark also moves to `with_capacity`: its `long_low_cardinality` case is defined by the distinct values fitting inside the pre-allocated buffer.
Keep the comment about what `HashTable::allocation_size` covers next to the value it describes, and say what the test helper's lower bound is derived from.
`GroupValuesBytes::clear_shrink` and `GroupValuesBytesView::clear_shrink` reset their map with `take()`, which restores the capacity the map was configured with so the emptied map stays warm. That is what the emit path wants, but `clear_shrink` exists to hand memory back before spilling and before the spilled batch is sorted, so it left roughly 16 KiB (string and binary) and 34 KiB (view) reserved instead of releasing it. Add `clear_and_release` to `ArrowBytesMap` and `ArrowBytesViewMap`, which empties the map and drops its allocations while remembering the configured capacities so a later `take()` still warms the map up, and call it from the two `clear_shrink` implementations. The pre-allocation stays at construction, where the hot single column string `GROUP BY` path earns it.
A grouped `COUNT(DISTINCT <string>)` gets one accumulator per group, and each of those owns a hash set of the distinct values it has seen. Those sets were created pre-allocated, so the query's memory use tracked the number of groups rather than the amount of data. Add two `memory_limit` tests that turn that into a binary observable, one for `Utf8` and one for `Utf8View`, over a new scenario of 4,000 groups holding 2 distinct values each. Measured against this branch's base commit with spilling disabled and `target_partitions` pinned to 1: | value column | budget needed before | budget needed after | | ------------ | -------------------- | ------------------- | | `Utf8` | ~35.5 MB | ~1.9 MB | | `Utf8View` | ~123 MB | ~2.7 MB | The tests run at 8 MB and 16 MB respectively, so each sits at least 4x above what the branch needs and at least 4x below what the base needs. Both fail on the base commit with `Resources exhausted` and pass here.
The two grouped `COUNT(DISTINCT <string>)` memory limit tests only reach the per group accumulators while `single_distinct_aggregation_to_group_by` declines to rewrite the query. They leant on `count(*)` for that, which the rule rejects only because `count` is missing from the `sum`/`min`/`max` allow list. #24859 proposes adding `count` to that list, which would rewrite the query, remove the accumulators, and leave both tests passing at any memory limit while still looking like they test something. Aggregate `avg(payload)` over a new `Int64` column instead. `avg` cannot be added to that list: the rule re-aggregates its own partial results over the deduplicated inner group by, and averaging per group averages of different sizes gives the wrong answer. That is why ClickBench Q9 keeps its distinct aggregate under #24859. Verified from the physical plan with #24859 cherry-picked on top of this branch: the `avg` query still plans as `aggr=[count(DISTINCT t.value), avg(t.payload)]`, while the `count(*)` query becomes `aggr=[count(alias1), sum(alias2)]` over an inner `GROUP BY group_key, value`, and drops from needing ~1.9 MB to ~0.9 MB. Re-swept both thresholds against the base commit. `Utf8` needs ~35.5 MB before and ~1.9 MB after; `Utf8View` needs ~123 MB before and ~2.5 MB after, so the 8 MB and 16 MB limits keep at least 4x margin on each side and are unchanged.
`ArrowBytesMap::new` starts its value buffer empty and `ArrowBytesMap::with_capacity` starts it at `INITIAL_BUFFER_CAPACITY`. `Vec` then doubles from wherever its first allocation landed, so the two sit on different ladders and can hold the same values at capacities differing by up to 2x, in either direction depending on the value lengths. Measured on 500,000 distinct 28 byte values, the lazily grown map reported 52,494,344 bytes against 45,154,312 for a pre-allocated one, 16% more for identical contents. That matters because the ungrouped `COUNT(DISTINCT <string>)` accumulator is the caller that had a use for the warm up: it builds one map and grows it to hold every distinct value in the input. Rounding every buffer growth up to a power of two puts both constructors on one ladder, so a lazily allocated map is never larger than a pre-allocated one holding the same values. Growth stays geometric, so appending is still amortized constant time. `ArrowBytesViewMap` has no such buffer and is unaffected. Two new tests cover the ungrouped path, which had none: `ungrouped_utf8_accumulator_is_never_worse_than_a_pre_allocated_set` and its `Utf8View` counterpart drive an accumulator to 0 through 500,000 distinct values and assert it is strictly cheaper than a pre-allocated set at per group cardinalities and exactly equal at ungrouped ones. The `Utf8` one fails without this change, at 1,000 distinct values, with the lazy set reporting 110,408 bytes against 96,072. Two map level tests pin the ladder itself.
b9e6252 to
f31fc51
Compare
|
@2010YOUY01 @kosiew would either of you be interested in reviewing this change? |
A grouped
COUNT(DISTINCT <string>)over 4,000 groups holding 2 short strings each needs a 36 MB memory budget. It needs 2.0 MB after this change.Every group gets its own hash table, and each table is allocated at warm-up size before the group holds anything, so the memory the query needs tracks the number of groups rather than the amount of data. The query also reports less memory than it holds, so a memory limit does not stop it at the right point.
Reproduction
This needs only
datafusion-cli. There is no patch, no custom allocator and no data file.sis aUtf8Viewcolumn, so this exercisesArrowBytesViewMap. The 8,000 rows arrive in one batch, so the aggregate builds all 4,000 accumulators before it can emit or spill.Current
mainat20d1c56761fails:This branch returns the rows:
Those 4,000 accumulators hold 8,000 short strings, which is about 100 KB of data. The base asks the pool for 111.1 MB to hold it. This branch runs the same query inside
-m 3M. Both builds return the same rows, and the base does so at-m 200M. Each run takes well under a second, and the outcome repeats exactly over three runs on each side.How much it improves
The minimum memory limit at which that query completes, bisected on each side:
Utf8Utf8Viewclickbench_extendedatDATAFUSION_RUNTIME_MEMORY_LIMIT: 4G, pool peak over six runs:Q2 is the only query in any benchmark suite that puts a grouped
COUNT(DISTINCT)on a non-integer column.Q0 costs more, and it is the one disclosed cost of this PR. Those extra bytes are memory Q0 always held and the pool could not see, not new allocation; appendix A has the three-build decomposition that separates the two. Latency does not move anywhere, which is what an allocation-sizing change should do.
Which issue does this PR close?
No existing issue. I found this when I investigated a production out of memory. I can file an issue if you want it in the changelog.
Rationale for this change
Pre-allocating is right for the one long-lived map behind a
GROUP BYon a string column. It is wrong for the distinct-count accumulators, becauseGroupsAccumulatorAdaptercreates one accumulator per group and most groups hold a handful of values. There the warm-up dwarfs the data.Both maps also under-report the table they hold.
ArrowBytesViewMapleft the control bytes out.ArrowBytesMapcharged the table only when it grew, so a map that stayed inside its pre-allocation reported its table as free forever. A memory limit acted on a number that was too small.clear_shrinkis the third part. The aggregate stream calls it to hand memory back before it spills and before a downstream sort. It restored the warm-up capacity instead of releasing it, so nothing came back.What changes are included in this PR?
newon both maps allocates nothing. A newwith_capacitykeeps the previous pre-allocating behavior.GroupValuesBytesandGroupValuesBytesViewusewith_capacity, and the two distinct-count accumulators usenew. A map remembers how it was built, sotakewarms it back up the way it started.size()reportsHashTable::allocation_size(), the real hashbrown allocation including the control bytes, in place of the old estimate.clear_and_releasedrops every allocation the map holds, andclear_shrinkcalls it.benches/arrow_bytes_map.rsmoves towith_capacityso it keeps measuring the pre-allocating constructor.What is the testing strategy for this PR?
Two tests in
datafusion/core/tests/memory_limit/mod.rs,group_by_count_distinct_utf8andgroup_by_count_distinct_utf8_view, turn the headline claim into a pass or a fail rather than a number. They run the reproduction query over 4,000 groups with spilling disabled andtarget_partitionspinned to 1, so completing means the query fits the budget rather than spills out of it. The limits are 8 MB and 16 MB, at least 4x clear of both cliffs in the table above. Both tests fail on the merge base and pass here, over five consecutive runs. Theavg(payload)in the query is load bearing; appendix C says why.Unit tests cover what the memory-limit tests cannot see: that
newallocates nothing, thatwith_capacityreports a table size bracketed by an independently derived lower bound, thattakepreserves the configured capacity, thatclear_shrinkdrops the reported size to near zero, and that a lazily grown buffer never exceeds a pre-allocated one holding the same values.Run locally on the rebased head, all passing:
datafusion-physical-expr-common(87 lib, 8 doc),datafusion-functions-aggregate-common(49),datafusion-functions-aggregate -- count_distinct(2),datafusion-physical-plan -- group_values(96) and thememory_limitmodule (39, which includes thecount_distinct_spilltest that arrived onmainin #24888 and #24918).cargo fmt --checkandcargo clippy --all-targets -D warningsare clean on the changed crates. CI has not yet run this branch against the new base.No query results change.
Are there any user-facing changes?
Yes, in
datafusion-physical-expr-common.ArrowBytesMap::newandArrowBytesViewMap::newno longer pre-allocate, and callers that want the previous behavior should use the newwith_capacity. Both types also gainclear_and_release. This changes an existing public constructor rather than adding one, so tell me if you would like theapi changelabel.For users, a grouped
COUNT(DISTINCT)on string and binary columns uses much less memory and reports its usage to theMemoryPoolaccurately. A query that previously hit a memory limit may now succeed.Appendix A: Query 0 costs 6.3% more
Q0 is
COUNT(DISTINCT)over three high-cardinality strings with noGROUP BY. It is a handful of maps that each grow to millions of entries, which is the opposite population from the one this PR targets. The pre-allocation was never the dominant cost there, so removing it buys nothing.Over six runs the base spans 796.8 to 834.8 MiB and this branch spans 846.5 to 885.5 MiB. The ranges do not overlap, so the effect is real and not run-to-run noise.
Three local builds on a deterministic subset separate the two changes. The middle build differs from the base only in the accounting, because restoring the warm-up makes the constructors byte-identical to base:
That decomposes the increase exactly. +2,626,576 is the accounting correction:
allocation_size()charges the real hashbrown allocation, which is4 * buckets + 5,384more than the old formula, being the control bytes plus the 7/8 load-factor slack. -29,568 is the lazy constructor, which makes Q0 slightly better.Reverting the accounting would restore an under-report of about 19% on this path. That under-report is the bug this PR exists to fix, and the memory-limit result above depends on fixing it.
Appendix B: what one accumulator costs
One per-group accumulator holding a single 24-byte value:
BytesDistinctCountAccumulatorBytesViewDistinctCountAccumulatorThe middle column is the reporting gap. The
Utf8map really held 14,648 bytes and reported 8,240, because the whole hash table was invisible to the old accounting.These are measured directly rather than asserted in a test, since the exact numbers follow the hashbrown layout.
Appendix C: notes on the tests and the benchmarks
The memory-limit query uses
avg(payload), notcount(*). A non-distinctcountletsSingleDistinctToGroupByrewrite the distinct aggregate into a plain two-stageGROUP BY. The per-group accumulators would then never exist, and the tests would pass by construction on the base commit too. That rule accepts a non-distinctsum,minormaxbecause each re-aggregates its own partial results correctly over the deduplicated inner group by.avgdoes not, so the rule can never admit it under any extension, including the one #24859 proposes.The benchmark figures were measured against the previous merge base
da89c7c85b. They have not been re-run against the current base20d1c56761. The commits after84f07daon this branch touch onlydatafusion/core/tests/memory_limit/mod.rs, so nothing on this branch since then can move a benchmark, but the base itself has moved.Pool peak is the instrument here, not peak RSS. Pool peak reproduces to under 1% on a 98 MiB query and exactly on the 3.4 and 11.7 MiB ones. Peak RSS on this harness has a 4.1% standard deviation over 11 readings of identical code plus a 2.3% order bias, and shows no effect from this change once that null is accounted for.
Follow-ups, not in this PR
insert_accountedcall sites:group_values/row.rs:171,multi_group_by/mod.rs:434,554,multi_group_by/dictionary.rs:197,584andarray_agg.rs:989. Each is one map per query, so the absolute error is bounded, and the fix is the same one-line swap.count_distinct_groupsbenchmarks indatafusion/functions-aggregate/benches/count_distinct.rscoverInt64,Int32andUInt32only, so this path has no criterion coverage.