Skip to content

[opt](pipeline) Add passthrough between repeat and streaming aggregation #66339

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Mryange wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[opt](pipeline) Add passthrough between repeat and streaming aggregation #66339

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9c2a7c8
[opt](pipeline) Add passthrough between repeat and streaming aggregation
Mryange Jul 31, 2026
5b0f214
feut
Mryange Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions be/src/exec/operator/operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,8 @@ class OperatorBase {

virtual bool is_hash_join_probe() const { return false; }

virtual bool is_repeat() const { return false; }

/**
* Pipeline task is blockable means it will be blocked in the next run. So we should put the
* pipeline task into the blocking task scheduler.
Expand Down
2 changes: 2 additions & 0 deletions be/src/exec/operator/repeat_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,8 @@ class RepeatOperatorX final : public StatefulOperatorX<RepeatLocalState> {
Status pull(RuntimeState* state, Block* output_block, bool* eos) const override;
Status push(RuntimeState* state, Block* input_block, bool eos) const override;

bool is_repeat() const override { return true; }

private:
friend class RepeatLocalState;

Expand Down
5 changes: 5 additions & 0 deletions be/src/exec/operator/streaming_aggregation_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,11 @@ class StreamingAggOperatorX MOCK_REMOVE(final) : public StatefulOperatorX<Stream
_spill_streaming_agg_mem_limit = 1024 * 1024;
}
DataDistribution required_data_distribution(RuntimeState* state) const override {
// Repeat emits grouping sets as consecutive blocks. Fan them out without hashing every
// expanded row before the local streaming preaggregation.
if (_child && _child->is_repeat()) {
return {TLocalPartitionType::PASSTHROUGH};
}
if (_child && _child->is_hash_join_probe() &&
state->enable_streaming_agg_hash_join_force_passthrough()) {
return {TLocalPartitionType::PASSTHROUGH};
Expand Down
8 changes: 8 additions & 0 deletions be/test/exec/operator/streaming_agg_operator_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
#include "exec/operator/aggregation_source_operator.h"
#include "exec/operator/mock_operator.h"
#include "exec/operator/operator_helper.h"
#include "exec/operator/repeat_operator.h"
#include "exec/operator/streaming_aggregation_operator.h"
#include "testutil/column_helper.h"
#include "testutil/mock/mock_agg_fn_evaluator.h"
Expand Down Expand Up @@ -168,6 +169,13 @@ TEST_F(StreamingAggOperatorTest, require_hash_shuffle_after_non_hash_local_excha
EXPECT_EQ(TLocalPartitionType::GLOBAL_EXECUTION_HASH_SHUFFLE, distribution.distribution_type);
}

TEST_F(StreamingAggOperatorTest, require_passthrough_after_repeat) {
EXPECT_TRUE(op->set_child(std::make_shared<RepeatOperatorX>()));

const auto distribution = op->required_data_distribution(state.get());
EXPECT_EQ(TLocalPartitionType::PASSTHROUGH, distribution.distribution_type);
}

TEST_F(StreamingAggOperatorTest, test2) {
op->_aggregate_evaluators.push_back(create_mock_agg_fn_evaluator(
pool, MockSlotRef::create_mock_contexts(1, std::make_shared<DataTypeInt64>()), false,
Expand Down
6 changes: 5 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,11 @@ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
}
} else if (useStreamingPreagg) {
// StreamingAggOperatorX
if (children.get(0) instanceof HashJoinNode
// Repeat expands grouping sets before this local preaggregation. Use PASSTHROUGH
// to distribute the expanded blocks without hashing every row.
if (!needsFinalize && children.get(0) instanceof RepeatNode) {

@924060929 924060929 Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This special case unconditionally inserts a local exchange and bypasses the opt-out semantics of enable_local_exchange_before_streaming_agg.

That dedicated switch was added with a default of false specifically so StreamingAgg preserves its inherited distribution unless local exchange is explicitly enabled. In the BE path, the new Repeat check likewise runs before enable_local_exchange_before_streaming_agg() is checked. In the FE path, this branch also precedes the existing enableLeBeforeAgg fallback and does not consult the streaming-specific switch at all.

A two-phase grouping-sets query therefore gets StreamingAgg <- LE(PASSTHROUGH) <- Repeat even when the streaming-agg switch is false. This should not change query results because this is partial aggregation followed by a final merge, but it makes the extra pipeline boundary, queues, and memory overhead impossible to disable, and silently changes the default behavior established by #66222.

Please gate the Repeat PASSTHROUGH optimization with enableLocalExchangeBeforeStreamingAgg here and with enable_local_exchange_before_streaming_agg() in BE, and add FE-/BE-planned tests for both switch values. If Repeat is intentionally meant to ignore the existing switch, that needs an explicit separately named switch and documented default behavior rather than bypassing the current opt-out.

requireChild = LocalExchangeTypeRequire.requirePassthrough();
} else if (children.get(0) instanceof HashJoinNode
&& sessionVariable.enableStreamingAggHashJoinForcePassthrough) {
requireChild = LocalExchangeTypeRequire.requirePassthrough();
} else if (!needsFinalize && !enableLeBeforeAgg) {
Expand Down
40 changes: 13 additions & 27 deletions fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -532,47 +532,33 @@ public void testAnalyticPlanContainsPassthroughAndLocalHashShuffle() throws Exce
}

@Test
public void testGroupingSetsPlanContainsHashShuffle() throws Exception {
// Non-pooling grouping sets keeps the colocated BUCKET_HASH_SHUFFLE output of
// the scan all the way through Repeat→Agg; no LE(LOCAL_HASH) is needed.
public void testGroupingSetsDoesNotUseHashShuffle() throws Exception {
// Repeat→StreamingAgg uses PASSTHROUGH rather than hashing all expanded rows.
setupLocalShuffleSession(sv -> sv.setIgnoreStorageDataDistribution(false));
assertNoLocalExchangeOfType(
"select k1, k2, sum(v1) from test.t1 group by grouping sets((k1), (k1, k2))",
LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE);
}

@Test
public void testRepeatNoRequireKeepsHashLocalExchangeAboveRepeat() throws Exception {
// Behavior 1 of the RepeatNode fix — noRequire (tpcds q67, +73%).
// RepeatNode recurses with noRequire() instead of forwarding the streaming
// agg's HASH require to its child. So when the pooling scan upstream does NOT
// already provide the distribution, the parent inserts the LE(LOCAL_HASH)
// ABOVE the Repeat, never below it:
// Agg <- LE(LOCAL_HASH) <- LE(PASSTHROUGH) <- Repeat <- scan
// Pinning Repeat with repeat() (not anyTree) distinguishes the fixed plan from
// the buggy one (buggy forwarded the require, so the LE landed below the
// Repeat, hashing the pre-repeat rows by the child's single upstream key and
// collapsing them onto one instance).
setupLocalShuffleSession(null);
public void testStreamingAggAfterRepeatUsesPassthrough() throws Exception {
// Repeat expands grouping sets into consecutive blocks. A passthrough local
// exchange above Repeat distributes those blocks among streaming aggregation
// instances without hashing every expanded row. Force two-phase aggregation
// so the lower Repeat-adjacent aggregation is a streaming preaggregation.
setupLocalShuffleSession(sv -> sv.setAggPhase(2));
assertPlanShape(
"select k1, k2, count(*) from test.t1 group by grouping sets((k1), (k1, k2))",
anyTree(
agg(
localExchange(LOCAL_HASH,
localExchange(PT,
repeat(anyTree(olapScan("t1"))))))));
localExchange(PT,
repeat(anyTree(olapScan("t1")))))));
}

@Test
public void testRepeatReturnsChildDistributionSkipsRedundantHash() throws Exception {
// Behavior 2 of the RepeatNode fix — return enforceResult.second (tpcds q70).
// RepeatNode reports its child's real output distribution to the parent (not
// NOOP). With a non-pooling colocate scan, the child's BUCKET_HASH
// distribution propagates through the Repeat and already satisfies the agg's
// hash requirement, so the parent's satisfy-check SKIPS inserting any LE — no
// LOCAL_HASH appears. Had RepeatNode returned NOOP (the discarded v1), the
// satisfy-check would fail and force a redundant LE(LOCAL_HASH) that
// re-shuffles the post-repeat rows into skew.
public void testStreamingAggAfterRepeatDoesNotUseHashForColocatedScan() throws Exception {
// The dedicated PASSTHROUGH requirement also takes precedence when the scan
// provides a colocated bucket distribution.
setupLocalShuffleSession(sv -> sv.setIgnoreStorageDataDistribution(false));
assertNoLocalExchangeOfType(
"select k1, k2, count(*) from test.t1 group by grouping sets((k1), (k1, k2))",
Expand Down
Loading