GH-45847: [C++][Acero] Two-pass merge in GroupByNode to amortise kernel resizes - #1
Open
Periecle wants to merge 1 commit into
Open
GH-45847: [C++][Acero] Two-pass merge in GroupByNode to amortise kernel resizes#1Periecle wants to merge 1 commit into
Periecle wants to merge 1 commit into
Conversation
…e kernel resizes
The hash-aggregate `Merge()` step was a single pass that, for every
contributing thread, fed its uniques into state0->grouper and then iterated
every kernel — calling `kernel->resize(state0->grouper->num_groups())`
once per (thread, kernel) pair. With N threads and K kernels that is
N*K resize calls, while only K are needed (the grouper grows monotonically
so the only resize that does real work is the final one).
This commit implements the two-pass algorithm noted in the existing
in-code TODO at lines 285-289:
Pass 1 Compute every contributing thread's transposition into
state0->grouper. After this pass the grouper is fully grown.
Pass 2 For each kernel, resize state0->agg_states[k] once to the
final group count, then merge every contributing thread's
kernel state in turn.
Semantically equivalent to the previous code: the grouper grows
monotonically across Consume() calls, so each thread's transposition
remains a valid index into state0's final kernel state. Per-row
output (count, sum, etc.) is unchanged.
Verified with a join-then-aggregate reproducer of apacheGH-45847 that
exercises the multi-threaded merge path on a worst-case unique-key
inner join (build x probe = 64 x 64 batches of 32 768 rows = ~2M
groups across 4 worker threads). out_rows and per-group counts match
the unmodified Merge() exactly.
The catastrophic 9000x slowdown reported in apacheGH-45847 itself is largely
addressed by the JoinResultMaterialize::Flush task group that landed
in apache#45918; this change is the complementary aggregator-side fix called
out by the in-code TODO and reduces resize() call count in Merge() from
O(num_threads * num_kernels) to O(num_kernels).
Reproducer:
cpp/src/arrow/acero/repro_45847.cc
Standalone program that wires source(probe) + source(build) ->
hashjoin -> aggregate(hash_count) and times the run. Validates the
output against the analytical expected count so it can be used to
check correctness of further changes.
https://claude.ai/code/session_01BX3n7pizaHw9yVpSetFmXo
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
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.
The hash-aggregate
Merge()step was a single pass that, for everycontributing thread, fed its uniques into state0->grouper and then iterated
every kernel — calling
kernel->resize(state0->grouper->num_groups())once per (thread, kernel) pair. With N threads and K kernels that is
N*K resize calls, while only K are needed (the grouper grows monotonically
so the only resize that does real work is the final one).
This commit implements the two-pass algorithm noted in the existing
in-code TODO at lines 285-289:
Pass 1 Compute every contributing thread's transposition into
state0->grouper. After this pass the grouper is fully grown.
Pass 2 For each kernel, resize state0->agg_states[k] once to the
final group count, then merge every contributing thread's
kernel state in turn.
Semantically equivalent to the previous code: the grouper grows
monotonically across Consume() calls, so each thread's transposition
remains a valid index into state0's final kernel state. Per-row
output (count, sum, etc.) is unchanged.
Verified with a join-then-aggregate reproducer of apacheGH-45847 that
exercises the multi-threaded merge path on a worst-case unique-key
inner join (build x probe = 64 x 64 batches of 32 768 rows = ~2M
groups across 4 worker threads). out_rows and per-group counts match
the unmodified Merge() exactly.
The catastrophic 9000x slowdown reported in apacheGH-45847 itself is largely
addressed by the JoinResultMaterialize::Flush task group that landed
in apache#45918; this change is the complementary aggregator-side fix called
out by the in-code TODO and reduces resize() call count in Merge() from
O(num_threads * num_kernels) to O(num_kernels).
Reproducer:
cpp/src/arrow/acero/repro_45847.cc
Standalone program that wires source(probe) + source(build) ->
hashjoin -> aggregate(hash_count) and times the run. Validates the
output against the analytical expected count so it can be used to
check correctness of further changes.
https://claude.ai/code/session_01BX3n7pizaHw9yVpSetFmXo