Uh oh!
There was an error while loading. Please reload this page.
test: skip count distinct spill memory test under forced hash collisions - #24918
Merged
rluvaton merged 1 commit intoSep 3, 2026
Conversation
rluvatonforce-pushed
the
fix-count-distinct-spill-test-under-forced-hash-collisions
branch
2 times, most recently
from
September 3, 2026 07:29
19a64a6 to
fdd9d32CompareWith force_hash_collisions every key hashes alike, so the hash repartitioning sends all groups to a single final stage whose table cannot fit the memory limit however well memory is released. The limit is sized for the real distribution across four final stages.
rluvatonforce-pushed
the
fix-count-distinct-spill-test-under-forced-hash-collisions
branch
from
September 3, 2026 07:30
fdd9d32 to
e484973Comparerluvaton
marked this pull request as draft
September 3, 2026 07:38
codecov-commenter
commented
Sep 3, 2026
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@## main #24918 +/- ##
=======================================
Coverage 81.63% 81.63% =======================================
Files 1123 1123 Lines 409963 409963 Branches 409963 409963 =======================================
+ Hits 334685 334687 +2 + Misses 55605 55598 -7 - Partials 19673 19678 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
rluvaton
marked this pull request as ready for review
September 3, 2026 09:28
Weijun-H
approved these changes
Sep 3, 2026
Weijun-H
left a comment
Member
There was a problem hiding this comment.
LGTM
Non-blocking wording nit: filter_pushdown.rs adapts expectations under the feature rather than skipping the test, so consider removing that comparison from the PR description.
Uh oh!
There was an error while loading. Please reload this page.
rluvaton
deleted the
fix-count-distinct-spill-test-under-forced-hash-collisions
branch
September 3, 2026 10:06
pullBot
pushed a commit
to TCeason/arrow-datafusion
that referenced
this pull request
Sep 6, 2026
…TINCT) group (apache#24857) 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. ```sql -- repro.sql SET datafusion.execution.target_partitions = 1; -- 4,000 groups with 2 distinct short strings in each. -- avg() stops SingleDistinctToGroupBy from rewriting the distinct aggregate away. SELECT g, count(DISTINCT s) AS d, avg(p) AS a FROM ( SELECT v % 4000 AS g, 'v' || CAST(v AS VARCHAR) AS s, v AS p FROM generate_series(0, 7999) AS t(v) ) GROUP BY g ORDER BY g LIMIT 3; ``` ``` datafusion-cli -m 8M -f repro.sql ``` `s` is a `Utf8View` column, so this exercises `ArrowBytesViewMap`. The 8,000 rows arrive in one batch, so the aggregate builds all 4,000 accumulators before it can emit or spill. Current `main` at `20d1c56761` fails: ``` Resources exhausted: Additional allocation failed for SingleHashAggregateStream[0] with top memory consumers (across reservations) as: DataFusion-Cli#1(can spill: false) consumed 0.0 B, peak 0.0 B, SingleHashAggregateStream[0]#2(can spill: true) consumed 0.0 B, peak 48.0 B, TopK[0]#3(can spill: false) consumed 0.0 B, peak 0.0 B. Error: Failed to allocate additional 111.1 MB for SingleHashAggregateStream[0] with 0.0 B already allocated for this reservation - 8.0 MB remain available for the total memory pool: greedy(used: 0.0 B, pool_size: 8.0 MB) ``` This branch returns the rows: ``` +---+---+--------+ | g | d | a | +---+---+--------+ | 0 | 2 | 2000.0 | | 1 | 2 | 2001.0 | | 2 | 2 | 2002.0 | +---+---+--------+ 3 row(s) fetched. Elapsed 0.006 seconds. ``` 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: | value column | before | after | | --- | --- | --- | | `Utf8` | fails 34 MB, passes 36 MB | fails 1.8 MB, passes 2.0 MB | | `Utf8View` | fails 120 MB, passes 124 MB | fails 2.4 MB, passes 2.6 MB | `clickbench_extended` at `DATAFUSION_RUNTIME_MEMORY_LIMIT: 4G`, pool peak over six runs: | query | base | this branch | change | | --- | --- | --- | --- | | Q2, grouped, 4 string distincts | 98.1 to 98.6 MiB | 11.7 MiB in all six runs | -88.1% | | Q1, ungrouped string distincts | 3.4 MiB | 2.4 MiB | -29.4% | | Q0, ungrouped, high cardinality | 796.8 to 834.8 MiB | 846.5 to 885.5 MiB | +6.3% | 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 BY` on a string column. It is wrong for the distinct-count accumulators, because `GroupsAccumulatorAdapter` creates 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. `ArrowBytesViewMap` left the control bytes out. `ArrowBytesMap` charged 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_shrink` is 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? - `new` on both maps allocates nothing. A new `with_capacity` keeps the previous pre-allocating behavior. `GroupValuesBytes` and `GroupValuesBytesView` use `with_capacity`, and the two distinct-count accumulators use `new`. A map remembers how it was built, so `take` warms it back up the way it started. - `size()` reports `HashTable::allocation_size()`, the real hashbrown allocation including the control bytes, in place of the old estimate. - A new `clear_and_release` drops every allocation the map holds, and `clear_shrink` calls it. - The value buffer rounds each growth up to a power of two. A lazily grown buffer and a pre-allocated one then sit on one ladder, so a lazy map is never the larger of the two for the same values. Growth stays geometric. - `benches/arrow_bytes_map.rs` moves to `with_capacity` so 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_utf8` and `group_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 and `target_partitions` pinned 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. The `avg(payload)` in the query is load bearing; appendix C says why. Unit tests cover what the memory-limit tests cannot see: that `new` allocates nothing, that `with_capacity` reports a table size bracketed by an independently derived lower bound, that `take` preserves the configured capacity, that `clear_shrink` drops 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 the `memory_limit` module (39, which includes the `count_distinct_spill` test that arrived on `main` in apache#24888 and apache#24918). `cargo fmt --check` and `cargo clippy --all-targets -D warnings` are 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::new` and `ArrowBytesViewMap::new` no longer pre-allocate, and callers that want the previous behavior should use the new `with_capacity`. Both types also gain `clear_and_release`. This changes an existing public constructor rather than adding one, so tell me if you would like the `api change` label. For users, a grouped `COUNT(DISTINCT)` on string and binary columns uses much less memory and reports its usage to the `MemoryPool` accurately. 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 no `GROUP 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: | build | Q0 pool peak | | --- | --- | | base | 48,421,820 | | this branch with the warm-up restored | 51,048,396 | | this branch | 51,018,828 | That decomposes the increase exactly. +2,626,576 is the accounting correction: `allocation_size()` charges the real hashbrown allocation, which is `4 * buckets + 5,384` more 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: | | before, actual | before, reported | after | | --- | --- | --- | --- | | `BytesDistinctCountAccumulator` | 14,648 B | 8,240 B | 180 B | | `BytesViewDistinctCountAccumulator` | 33,920 B | 28,792 B | 260 B | The middle column is the reporting gap. The `Utf8` map 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)`, not `count(*)`. A non-distinct `count` lets `SingleDistinctToGroupBy` rewrite the distinct aggregate into a plain two-stage `GROUP 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-distinct `sum`, `min` or `max` because each re-aggregates its own partial results correctly over the deduplicated inner group by. `avg` does not, so the rule can never admit it under any extension, including the one apache#24859 proposes. The benchmark figures were measured against the previous merge base `da89c7c85b`. They have not been re-run against the current base `20d1c56761`. The commits after `84f07da` on this branch touch only `datafusion/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 - The same undercount remains at five other production `insert_accounted` call sites: `group_values/row.rs:171`, `multi_group_by/mod.rs:434,554`, `multi_group_by/dictionary.rs:197,584` and `array_agg.rs:989`. Each is one map per query, so the absolute error is bounded, and the fix is the same one-line swap. - The `count_distinct_groups` benchmarks in `datafusion/functions-aggregate/benches/count_distinct.rs` cover `Int64`, `Int32` and `UInt32` only, so this path has no criterion coverage. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
AdamGS pushed a commit
to AdamGS/arrow-datafusion
that referenced
this pull request
Sep 7, 2026
…he#25020) ## Which issue does this PR close? - Closesapache#25011. ## Rationale for this change The `cargo test hash collisions (amd64)` CI job hangs for hours (sometimes hitting the 360-minute job limit and getting cancelled) in two tests in `datafusion/functions-aggregate-common/src/aggregate/count_distinct/bytes.rs`: ``` aggregate::count_distinct::bytes::tests::ungrouped_utf8_accumulator_is_never_worse_than_a_pre_allocated_set aggregate::count_distinct::bytes::tests::ungrouped_utf8_view_accumulator_is_never_worse_than_a_pre_allocated_set ``` Root cause: both tests insert up to 500,000 distinct values into `ArrowBytesSet`/`ArrowBytesViewSet`, twice per cardinality in `CARDINALITIES` (once into a lazily-constructed set, once into a pre-allocated one), to compare their reported `.size()`. Under normal hashing this is O(n) per insert. Under `force_hash_collisions` ([`datafusion/common/src/hash_utils.rs#L1184-L1195`](https://github.com/apache/datafusion/blob/main/datafusion/common/src/hash_utils.rs#L1184-L1195)) every value hashes to the same bucket, so the underlying hash table degrades to a linear scan per insert - O(n^2) overall for a set built up to n elements. I confirmed this is quadratic, not just slow, with a throwaway local timing probe over the same insert pattern under `--features datafusion-common/force_hash_collisions` (removed before this PR, shown here for reference): | n | time | |---|---| | 100 | 297us | | 500 | 3.7ms | | 1,000 | 14ms | | 2,000 | 54ms | | 5,000 | 347ms | Each 2x step in n is roughly a 4x step in time, consistent with O(n^2), and consistent with the multi-hour runtimes reported in apache#25011 for n up to 500,000. On the question raised in apache#25011 ("what behavior or regression boundaries are the 100,000 and 500,000 cardinalities intended to protect, and what approach would preserve that coverage?"): the assertions in `assert_lazy_is_not_worse` compare allocator sizes reported by a real hash-table implementation against a pre-allocated one, at cardinalities chosen to span both sides of the warm-up capacity (`PER_GROUP_SCALE`) and the point where the two constructors converge (`UNGROUPED_SCALE`). None of that is about hash *collision* behavior - forcing every key into one bucket doesn't exercise a code path these tests are meant to protect, it just makes every insert scan the one bucket's full contents, which is why the cost goes quadratic without adding coverage. This is the same situation the `force_hash_collisions` feature already has an established answer for: `count_distinct_spill` in `datafusion/core/tests/memory_limit/mod.rs` (added in apache#24918) is gated with `#[cfg(not(feature = "force_hash_collisions"))]` because its assertions depend on a real hash distribution across partitions and don't mean anything under forced collisions. This PR applies the identical pattern here, rather than reducing cardinality or otherwise changing what real-hashing runs cover. ## What changes are included in this PR? - `datafusion/functions-aggregate-common/Cargo.toml`: add a `[features]` section declaring `force_hash_collisions = ["datafusion-common/force_hash_collisions"]`, forwarding to `datafusion-common`'s feature of the same name. This crate previously declared no features of its own. Cargo does not propagate a dependency's active feature into a consuming crate's own `cfg(feature = ...)` checks, so without this forwarding declaration, a `#[cfg(feature = "force_hash_collisions")]` inside this crate would never see the workspace-level `--features force_hash_collisions` flag the affected CI job passes (`cargo test --workspace --features=force_hash_collisions,avro`). This mirrors the exact forwarding pattern already used in `datafusion/core/Cargo.toml` for the same feature name. - `datafusion/functions-aggregate-common/src/aggregate/count_distinct/bytes.rs`: gate the whole `mod tests` block (it contains only these two tests and their shared helpers - nothing else needs to stay compiled either way) with `#[cfg(all(test, not(feature = "force_hash_collisions")))]`, with a doc comment explaining the O(n^2) mechanism and linking back to this issue. No production code changes. No reduction in cardinality or coverage for the normal (non-collision-forced) test run - both tests still run exactly as before, across all 7 cardinalities up to 500,000, whenever `force_hash_collisions` is off. ## What is the testing strategy for this PR? This is a test-only change, verified by running the tests both ways: - Without the feature: `cargo test -p datafusion-functions-aggregate-common --lib -- count_distinct::bytes` still runs and passes both tests in ~0.8s. - With the feature: `cargo test -p datafusion-functions-aggregate-common --lib --features force_hash_collisions -- count_distinct::bytes` runs 0 tests with a clean compile - confirming the gate compiles out cleanly rather than silently failing to match. - Against the exact affected CI job command (`cd datafusion && cargo test --profile ci --exclude datafusion-examples --exclude datafusion-benchmarks --exclude datafusion-sqllogictest --exclude datafusion-cli --workspace --lib --tests --features=force_hash_collisions,avro`): the `datafusion-functions-aggregate-common` test binary reports 47 tests (49 minus the 2 gated ones) all passing, with neither `ungrouped_utf8_accumulator_is_never_worse_than_a_pre_allocated_set` nor its `Utf8View` counterpart appearing anywhere in the run - confirming the workspace-level feature flag correctly reaches the new local feature via Cargo's feature unification, not just the crate-local invocation. - `cargo fmt --check` and the exact CI clippy invocation (`ci/scripts/rust_clippy.sh`, i.e. `cargo clippy --all-targets --workspace --features avro,integration-tests,extended_tests -- -D warnings`) both pass clean across the whole workspace. ## Are there any user-facing changes? None. This only changes which tests compile under a testing-only feature flag; there is no change to any public API or runtime behavior.
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 freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
3 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?
Follow-up to #24888, which broke the
cargo test hash collisionsCI job on main.Rationale for this change
The memory limit test added in #24888 fails when built with
force_hash_collisions. Every key hashes to the same value there, so the hash repartition sends all 64 groups to one final stage. That single table needs 5.3 MB against the test's 4 MB pool, and it has nothing reserved yet, so there is nothing to spill. It fails no matter how well the accumulator releases memory, which is what the test is actually about.I tried a few ways to keep it running under the feature first:
They all hit the same thing: under forced collisions the total state and a single batch are the same size, and the pool would have to sit above one and below the other.
What changes are included in this PR?
The test and its helpers move into a module gated on
not(feature = "force_hash_collisions").What is the testing strategy for this PR?
cargo test -p datafusion --features force_hash_collisions --test core_integration count_distinct_releasesruns 0 tests. Without the feature it still runs and passes.Are there any user-facing changes?
No.
🤖 Generated with Claude Code