This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 723c7ebf1db [branch-4.1][improvement](agg) Add a knob to control local
exchange before aggregation (#66104)
723c7ebf1db is described below
commit 723c7ebf1db08f0f8f273c9ac4cef80a4e599fe9
Author: Jerry Hu <[email protected]>
AuthorDate: Thu Jul 30 21:03:23 2026 +0800
[branch-4.1][improvement](agg) Add a knob to control local exchange before
aggregation (#66104)
### What problem does this PR solve?
Issue Number: None
Related PR: #62438, #63529, #63766
Problem Summary:
This safely backports #62438 to `branch-4.1`, adding the
`enable_local_exchange_before_agg` session variable so users can skip
unnecessary local exchanges before eligible non-finalizing aggregation.
A literal cherry-pick of #62438 is unsafe: when a serial or another
non-hash local exchange precedes aggregation, disabling the exchange can
split equal group or DISTINCT keys across local pipeline tasks and
produce incorrect results. This backport therefore also includes the
final correctness guard from #63766 and the serial-exchange regression
coverage from #63529:
- Skip the local exchange only when the child preserves local key
distribution.
- Require a hash exchange after serial and other non-hash local
exchanges.
- Handle a non-hash `LocalExchangeSource` when the downstream target
requires hash distribution.
- Cover the session-variable Thrift propagation, distribution decisions,
serial DISTINCT aggregation, and nested-loop-join DISTINCT aggregation.
`branch-4.1` already contains the compatible
`enable_local_exchange_before_agg` Thrift field, so this PR keeps the
existing field id and adds no duplicate IDL field.
### Release note
Add the `enable_local_exchange_before_agg` session variable. Disabling
it can reduce local exchanges before eligible non-finalizing aggregation
while preserving required hash exchanges for correctness.
### Check List (For Author)
- Test
- [x] Regression test
- Worktree-isolated config:
`output/local-regression/regression-conf-11600.groovy` (FE query port
`20630`, BE heartbeat port `20650`)
- `doris-local-regression --network 10.26.20.3/24 --offset 11600 all -d
nereids_syntax_p0 -s agg_4_phase` (1 suite passed; also completed
`./build.sh --be --fe`)
- `doris-local-regression --network 10.26.20.3/24 --offset 11600 start
&& doris-local-regression --network 10.26.20.3/24 --offset 11600 run -d
query_p0/join -s test_agg_after_nested_loop_join_local_exchange` (1
suite passed)
- [x] Unit Test
- `./run-be-ut.sh --run
--filter='StreamingAggOperatorTest.require_hash_shuffle_after_non_hash_local_exchange:DistinctStreamingAggOperatorTest.require_hash_shuffle_after_non_hash_local_exchange:AggOperatorRequiredDistributionTest.*:AggOperatorGroupByLimitOptTestWithGroupBy.*:AggOperatorTestWithGroupBy.test_2_phase:AggOperatorTestWithGroupBy.other_case_3:AggOperatorTestWithOutGroupBy.*'
-j 64` (21 tests passed)
- `./run-fe-ut.sh --run org.apache.doris.qe.VariableMgrTest` (12 tests
passed)
- `build-support/clang-format.sh`, `build-support/check-format.sh`, and
`git diff --check` passed
- `build-support/run-clang-tidy.sh` is absent on `branch-4.1`; a direct
focused `clang-tidy` attempt was blocked by pre-existing
toolchain/header diagnostics (`stddef.h` lookup and
`be/src/core/types.h` unmatched `NOLINTEND`), with no changed-line
diagnostic reported before the blocker
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Add a session variable that controls eligible local exchanges
before aggregation while retaining correctness-required hash exchanges.
- Does this need documentation?
- [x] No. This is a branch backport of an existing session variable.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
Co-authored-by: Gabriel <[email protected]>
Co-authored-by: Pxl <[email protected]>
---
.../exec/exchange/local_exchange_source_operator.h | 5 +
be/src/exec/operator/aggregation_sink_operator.h | 5 +-
.../distinct_streaming_aggregation_operator.h | 4 +-
be/src/exec/operator/operator.cpp | 17 +++
be/src/exec/operator/operator.h | 3 +
.../exec/operator/streaming_aggregation_operator.h | 3 +-
be/src/exec/pipeline/pipeline.cpp | 10 +-
be/test/exec/operator/agg_operator_test.cpp | 65 +++++++++++
...istinct_streaming_aggregation_operator_test.cpp | 15 +++
.../exec/operator/streaming_agg_operator_test.cpp | 14 +++
.../java/org/apache/doris/qe/VariableMgrTest.java | 12 ++
.../data/nereids_syntax_p0/agg_4_phase.out | 2 +
...t_agg_after_nested_loop_join_local_exchange.out | 3 +
.../suites/nereids_syntax_p0/agg_4_phase.groovy | 25 +++-
...gg_after_nested_loop_join_local_exchange.groovy | 130 +++++++++++++++++++++
15 files changed, 308 insertions(+), 5 deletions(-)
diff --git a/be/src/exec/exchange/local_exchange_source_operator.h
b/be/src/exec/exchange/local_exchange_source_operator.h
index d99dd57a378..c315680901f 100644
--- a/be/src/exec/exchange/local_exchange_source_operator.h
+++ b/be/src/exec/exchange/local_exchange_source_operator.h
@@ -82,6 +82,11 @@ public:
bool is_source() const override { return true; }
+ DataDistribution required_data_distribution(RuntimeState* /*state*/) const
override {
+ return {_exchange_type};
+ }
+ ExchangeType exchange_type() const { return _exchange_type; }
+
private:
friend class LocalExchangeSourceLocalState;
diff --git a/be/src/exec/operator/aggregation_sink_operator.h
b/be/src/exec/operator/aggregation_sink_operator.h
index 35823d075a2..5925f15ce53 100644
--- a/be/src/exec/operator/aggregation_sink_operator.h
+++ b/be/src/exec/operator/aggregation_sink_operator.h
@@ -160,7 +160,10 @@ public:
:
DataSinkOperatorX<AggSinkLocalState>::required_data_distribution(
state);
}
- if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
+
+ const bool child_breaks_distribution =
child_breaks_local_key_distribution(state);
+ if (!_needs_finalize && !state->enable_local_exchange_before_agg() &&
+ !child_breaks_distribution) {
return
DataSinkOperatorX<AggSinkLocalState>::required_data_distribution(state);
}
return _is_colocate && _require_bucket_distribution
diff --git a/be/src/exec/operator/distinct_streaming_aggregation_operator.h
b/be/src/exec/operator/distinct_streaming_aggregation_operator.h
index e0b175d93c5..6aecaa209f6 100644
--- a/be/src/exec/operator/distinct_streaming_aggregation_operator.h
+++ b/be/src/exec/operator/distinct_streaming_aggregation_operator.h
@@ -118,7 +118,9 @@ public:
if (_needs_finalize && _probe_expr_ctxs.empty()) {
return {ExchangeType::NOOP};
}
- if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
+
+ if (!_needs_finalize && !state->enable_local_exchange_before_agg() &&
+ !child_breaks_local_key_distribution(state)) {
return
StatefulOperatorX<DistinctStreamingAggLocalState>::required_data_distribution(
state);
}
diff --git a/be/src/exec/operator/operator.cpp
b/be/src/exec/operator/operator.cpp
index b6bbd64f96b..a7325981fb3 100644
--- a/be/src/exec/operator/operator.cpp
+++ b/be/src/exec/operator/operator.cpp
@@ -153,6 +153,23 @@ DataDistribution
OperatorBase::required_data_distribution(RuntimeState* /*state*
: DataDistribution(ExchangeType::NOOP);
}
+bool OperatorBase::is_hash_shuffle(ExchangeType exchange_type) {
+ return exchange_type == ExchangeType::HASH_SHUFFLE ||
+ exchange_type == ExchangeType::BUCKET_HASH_SHUFFLE;
+}
+
+bool OperatorBase::child_breaks_local_key_distribution(RuntimeState* state)
const {
+ if (!_child) {
+ return false;
+ }
+ if (_child->is_serial_operator()) {
+ return true;
+ }
+ const auto child_distribution = _child->required_data_distribution(state);
+ return child_distribution.need_local_exchange() &&
+ !is_hash_shuffle(child_distribution.distribution_type);
+}
+
const RowDescriptor& OperatorBase::row_desc() const {
return _child->row_desc();
}
diff --git a/be/src/exec/operator/operator.h b/be/src/exec/operator/operator.h
index 8c63b9d8a71..676a8b67b18 100644
--- a/be/src/exec/operator/operator.h
+++ b/be/src/exec/operator/operator.h
@@ -193,6 +193,9 @@ public:
}
protected:
+ [[nodiscard]] static bool is_hash_shuffle(ExchangeType exchange_type);
+ [[nodiscard]] bool child_breaks_local_key_distribution(RuntimeState*
state) const;
+
OperatorPtr _child = nullptr;
bool _is_closed;
diff --git a/be/src/exec/operator/streaming_aggregation_operator.h
b/be/src/exec/operator/streaming_aggregation_operator.h
index 18dcb4ca906..5abdbd40fbd 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.h
+++ b/be/src/exec/operator/streaming_aggregation_operator.h
@@ -222,7 +222,8 @@ public:
}
// Keep streaming aggregation on its inherited distribution unless the
dedicated switch
// explicitly enables a local hash exchange.
- if (!state->enable_local_exchange_before_streaming_agg()) {
+ if (!state->enable_local_exchange_before_streaming_agg() &&
+ !child_breaks_local_key_distribution(state)) {
return
StatefulOperatorX<StreamingAggLocalState>::required_data_distribution(state);
}
if (_partition_exprs.empty()) {
diff --git a/be/src/exec/pipeline/pipeline.cpp
b/be/src/exec/pipeline/pipeline.cpp
index 78a2cffafad..b2a6ca57596 100644
--- a/be/src/exec/pipeline/pipeline.cpp
+++ b/be/src/exec/pipeline/pipeline.cpp
@@ -21,6 +21,7 @@
#include <string>
#include <utility>
+#include "exec/exchange/local_exchange_source_operator.h"
#include "exec/operator/operator.h"
#include "exec/pipeline/pipeline_fragment_context.h"
#include "exec/pipeline/pipeline_task.h"
@@ -53,7 +54,14 @@ bool Pipeline::need_to_local_exchange(const DataDistribution
target_data_distrib
// If non-serial operators exist, we should improve parallelism for
those.
return true;
}
-
+ if (auto local_exchange_source =
+
std::dynamic_pointer_cast<LocalExchangeSourceOperatorX>(_operators.front());
+ local_exchange_source &&
is_hash_exchange(target_data_distribution.distribution_type)) {
+ const auto source_exchange_type =
local_exchange_source->exchange_type();
+ if (source_exchange_type != ExchangeType::NOOP &&
!is_hash_exchange(source_exchange_type)) {
+ return true;
+ }
+ }
if (target_data_distribution.distribution_type !=
ExchangeType::BUCKET_HASH_SHUFFLE &&
target_data_distribution.distribution_type !=
ExchangeType::HASH_SHUFFLE) {
// Always do local exchange if non-hash-partition exchanger is
required.
diff --git a/be/test/exec/operator/agg_operator_test.cpp
b/be/test/exec/operator/agg_operator_test.cpp
index b3178af7a8d..1934de3b59f 100644
--- a/be/test/exec/operator/agg_operator_test.cpp
+++ b/be/test/exec/operator/agg_operator_test.cpp
@@ -22,12 +22,14 @@
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
+#include "exec/exchange/local_exchange_source_operator.h"
#include "exec/operator/aggregation_sink_operator.h"
#include "exec/operator/aggregation_source_operator.h"
#include "exec/operator/assert_num_rows_operator.h"
#include "exec/operator/mock_operator.h"
#include "exec/operator/operator_helper.h"
#include "exec/pipeline/dependency.h"
+#include "exec/pipeline/pipeline.h"
#include "testutil/column_helper.h"
#include "testutil/mock/mock_agg_fn_evaluator.h"
#include "testutil/mock/mock_slot_ref.h"
@@ -119,6 +121,69 @@ std::shared_ptr<AggSinkOperatorX>
create_agg_sink_op(OperatorContext& ctx, bool
return op;
}
+TEST(AggOperatorRequiredDistributionTest,
require_hash_shuffle_after_non_hash_child_exchange) {
+ OperatorContext ctx;
+ TQueryOptions query_options;
+ query_options.__set_enable_local_exchange_before_agg(false);
+ ctx.state.set_query_options(query_options);
+ auto sink_op = std::make_shared<MockAggsinkOperator>();
+ sink_op->_partition_exprs.emplace_back();
+ sink_op->_needs_finalize = false;
+ OperatorPtr child =
+
std::make_shared<MockDistributionOperator>(ExchangeType::ADAPTIVE_PASSTHROUGH);
+ sink_op->_child = child;
+
+ const auto distribution = sink_op->required_data_distribution(&ctx.state);
+ EXPECT_EQ(ExchangeType::HASH_SHUFFLE, distribution.distribution_type);
+}
+
+TEST(AggOperatorRequiredDistributionTest, toggle_hash_shuffle_for_safe_child) {
+ OperatorContext ctx;
+ auto sink_op = std::make_shared<MockAggsinkOperator>();
+ sink_op->_partition_exprs.emplace_back();
+ sink_op->_needs_finalize = false;
+ sink_op->_child =
std::make_shared<MockDistributionOperator>(ExchangeType::NOOP);
+
+ TQueryOptions query_options;
+ query_options.__set_enable_local_exchange_before_agg(false);
+ ctx.state.set_query_options(query_options);
+ EXPECT_EQ(ExchangeType::NOOP,
+
sink_op->required_data_distribution(&ctx.state).distribution_type);
+
+ query_options.__set_enable_local_exchange_before_agg(true);
+ ctx.state.set_query_options(query_options);
+ EXPECT_EQ(ExchangeType::HASH_SHUFFLE,
+
sink_op->required_data_distribution(&ctx.state).distribution_type);
+}
+
+TEST(AggOperatorRequiredDistributionTest,
require_hash_shuffle_after_non_hash_local_exchange) {
+ OperatorContext ctx;
+ TQueryOptions query_options;
+ query_options.__set_enable_local_exchange_before_agg(false);
+ ctx.state.set_query_options(query_options);
+ auto sink_op = std::make_shared<MockAggsinkOperator>();
+ sink_op->_needs_finalize = false;
+ OperatorPtr child = std::make_shared<LocalExchangeSourceOperatorX>();
+ EXPECT_TRUE(child->init(ExchangeType::ADAPTIVE_PASSTHROUGH).ok());
+ sink_op->_child = child;
+
+ TExpr distinct_agg_expr;
+ distinct_agg_expr.nodes.emplace_back();
+ distinct_agg_expr.nodes[0].fn.name.function_name = "multi_distinct_count";
+ TPlanNode tnode;
+ tnode.agg_node.aggregate_functions.push_back(distinct_agg_expr);
+ tnode.__set_distribute_expr_lists({{TExpr {}}});
+ sink_op->update_operator(tnode, false, false);
+
+ const auto distribution = sink_op->required_data_distribution(&ctx.state);
+ EXPECT_EQ(ExchangeType::HASH_SHUFFLE, distribution.distribution_type);
+
+ Pipeline pipeline(0, 4, 4);
+ EXPECT_TRUE(pipeline.add_operator(child, 0).ok());
+
pipeline.set_data_distribution(DataDistribution(ExchangeType::HASH_SHUFFLE));
+ EXPECT_TRUE(pipeline.need_to_local_exchange(distribution, 1));
+}
+
std::shared_ptr<AggSourceOperatorX> create_agg_source_op(OperatorContext& ctx,
bool without_key,
bool needs_finalize) {
auto op = std::make_shared<MockAggSourceOperator>();
diff --git
a/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
b/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
index 88434e47fd7..17282356625 100644
--- a/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
+++ b/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
@@ -22,6 +22,7 @@
#include <memory>
#include "core/block/block.h"
+#include "exec/exchange/local_exchange_source_operator.h"
#include "exec/operator/mock_operator.h"
#include "exec/operator/operator_helper.h"
#include "testutil/column_helper.h"
@@ -97,6 +98,20 @@ TEST_F(DistinctStreamingAggOperatorTest, test1) {
}
}
+TEST_F(DistinctStreamingAggOperatorTest,
require_hash_shuffle_after_non_hash_local_exchange) {
+ state->_query_options.__set_enable_local_exchange_before_agg(false);
+ op->_is_streaming_preagg = false;
+ op->_partition_exprs.emplace_back();
+ op->_probe_expr_ctxs = MockSlotRef::create_mock_contexts(0,
std::make_shared<DataTypeInt64>());
+
+ OperatorPtr child = std::make_shared<LocalExchangeSourceOperatorX>();
+ EXPECT_TRUE(child->init(ExchangeType::ADAPTIVE_PASSTHROUGH).ok());
+ op->_child = child;
+
+ const auto distribution = op->required_data_distribution(state.get());
+ EXPECT_EQ(ExchangeType::HASH_SHUFFLE, distribution.distribution_type);
+}
+
TEST_F(DistinctStreamingAggOperatorTest, test2) {
op->_is_streaming_preagg = false;
op->_limit = 3;
diff --git a/be/test/exec/operator/streaming_agg_operator_test.cpp
b/be/test/exec/operator/streaming_agg_operator_test.cpp
index d56e88ec5a0..4596f040b03 100644
--- a/be/test/exec/operator/streaming_agg_operator_test.cpp
+++ b/be/test/exec/operator/streaming_agg_operator_test.cpp
@@ -23,6 +23,7 @@
#include "core/data_type/data_type_bitmap.h"
#include "core/data_type/data_type_number.h"
#include "core/value/bitmap_value.h"
+#include "exec/exchange/local_exchange_source_operator.h"
#include "exec/operator/aggregation_sink_operator.h"
#include "exec/operator/aggregation_source_operator.h"
#include "exec/operator/mock_operator.h"
@@ -154,6 +155,19 @@ TEST_F(StreamingAggOperatorTest, test1) {
{ EXPECT_TRUE(local_state->close(state.get()).ok()); }
}
+TEST_F(StreamingAggOperatorTest,
require_hash_shuffle_after_non_hash_local_exchange) {
+ state->_query_options.__set_enable_local_exchange_before_agg(false);
+ op->_needs_finalize = false;
+ op->_partition_exprs.emplace_back();
+
+ OperatorPtr child = std::make_shared<LocalExchangeSourceOperatorX>();
+ EXPECT_TRUE(child->init(ExchangeType::ADAPTIVE_PASSTHROUGH).ok());
+ EXPECT_TRUE(op->set_child(child));
+
+ const auto distribution = op->required_data_distribution(state.get());
+ EXPECT_EQ(ExchangeType::HASH_SHUFFLE, 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,
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java
index 2cc9d43b311..d24edd52a20 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java
@@ -203,6 +203,18 @@ public class VariableMgrTest {
Assert.assertEquals(8388608L, options.getPreferredBlockSizeBytes());
}
+ @Test
+ public void testLocalExchangeBeforeAggToThrift() throws Exception {
+ SessionVariable var = new SessionVariable();
+ Assert.assertTrue(var.enableLocalExchangeBeforeAgg);
+ Assert.assertTrue(var.toThrift().isEnableLocalExchangeBeforeAgg());
+
+ VariableMgr.setVar(var, new SetVar(SetType.SESSION,
+ SessionVariable.ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, new
StringLiteral("false")));
+ Assert.assertFalse(var.enableLocalExchangeBeforeAgg);
+ Assert.assertFalse(var.toThrift().isEnableLocalExchangeBeforeAgg());
+ }
+
@Test
public void testAdaptiveBatchSizeSessionVariables() throws Exception {
SessionVariable var = new SessionVariable();
diff --git a/regression-test/data/nereids_syntax_p0/agg_4_phase.out
b/regression-test/data/nereids_syntax_p0/agg_4_phase.out
index 97a97b8816d..0d22b7039e9 100644
--- a/regression-test/data/nereids_syntax_p0/agg_4_phase.out
+++ b/regression-test/data/nereids_syntax_p0/agg_4_phase.out
@@ -17,3 +17,5 @@
2 -4,-4 1 b 1
3 -4 1 f 1
+-- !serial_exchange_distinct_sum --
+45 44850
diff --git
a/regression-test/data/query_p0/join/test_agg_after_nested_loop_join_local_exchange.out
b/regression-test/data/query_p0/join/test_agg_after_nested_loop_join_local_exchange.out
new file mode 100644
index 00000000000..cd357b13436
--- /dev/null
+++
b/regression-test/data/query_p0/join/test_agg_after_nested_loop_join_local_exchange.out
@@ -0,0 +1,3 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !agg_after_nlj_local_exchange --
+10 5070261
diff --git a/regression-test/suites/nereids_syntax_p0/agg_4_phase.groovy
b/regression-test/suites/nereids_syntax_p0/agg_4_phase.groovy
index e38c92018ff..137a92b3f41 100644
--- a/regression-test/suites/nereids_syntax_p0/agg_4_phase.groovy
+++ b/regression-test/suites/nereids_syntax_p0/agg_4_phase.groovy
@@ -85,11 +85,27 @@ suite("agg_4_phase") {
from agg_4_phase_tbl2
group by id
order by id"""
+
+ sql "drop table if exists agg_4_phase_serial_exchange_tbl"
+ sql """
+ create table agg_4_phase_serial_exchange_tbl (
+ pk int,
+ x int,
+ y int
+ ) engine=olap
+ duplicate key(pk)
+ distributed by hash(pk) buckets 10
+ properties("replication_num"="1");
+ """
+ sql """
+ insert into agg_4_phase_serial_exchange_tbl
+ select number, number % 10, number from numbers("number" = "300");
+ """
multi_sql """
set runtime_filter_type= "BLOOM_FILTER,MIN_MAX";
set enable_runtime_filter_prune= "false";
set exchange_multi_blocks_byte_size= "4722978";
-set parallel_pipeline_task_num= "3";
+set parallel_pipeline_task_num= "4";
set experimental_parallel_scan_min_rows_per_scanner= "256";
set enable_strong_consistency_read= "true";
set runtime_filter_wait_infinitely= "true";
@@ -97,6 +113,7 @@ set enable_share_hash_table_for_broadcast_join= "false";
set experimental_parallel_scan_max_scanners_count= "8";
set disable_streaming_preaggregations= "true";
set experimental_use_serial_exchange= "true";
+set enable_local_exchange_before_agg= "false";
"""
qt_phase4_multi_distinct """
select
@@ -108,4 +125,10 @@ set experimental_use_serial_exchange= "true";
from agg_4_phase_tbl2
group by id
order by id"""
+
+ order_qt_serial_exchange_distinct_sum """
+ select
+ sum(distinct x),
+ sum(y)
+ from agg_4_phase_serial_exchange_tbl"""
}
diff --git
a/regression-test/suites/query_p0/join/test_agg_after_nested_loop_join_local_exchange.groovy
b/regression-test/suites/query_p0/join/test_agg_after_nested_loop_join_local_exchange.groovy
new file mode 100644
index 00000000000..91889b319c8
--- /dev/null
+++
b/regression-test/suites/query_p0/join/test_agg_after_nested_loop_join_local_exchange.groovy
@@ -0,0 +1,130 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_agg_after_nested_loop_join_local_exchange", "query_p0") {
+ sql "DROP TABLE IF EXISTS test_agg_after_nlj_local_exchange_t1"
+ sql "DROP TABLE IF EXISTS test_agg_after_nlj_local_exchange_t2"
+
+ sql """
+ CREATE TABLE test_agg_after_nlj_local_exchange_t1 (
+ col_bigint_undef_signed BIGINT,
+ col_varchar_10__undef_signed VARCHAR(10),
+ col_varchar_64__undef_signed VARCHAR(64),
+ pk INT
+ )
+ ENGINE=OLAP
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES("replication_num" = "1")
+ """
+
+ sql """
+ INSERT INTO test_agg_after_nlj_local_exchange_t1
+ (pk, col_bigint_undef_signed, col_varchar_10__undef_signed,
col_varchar_64__undef_signed)
+ VALUES
+ (0, -94, 'had', 'y'),
+ (1, 672609, 'k', 'h'),
+ (2, -3766684, 'a', 'p'),
+ (3, 5070261, 'on', 'x'),
+ (4, NULL, 'u', 'at'),
+ (5, -86, 'v', 'c'),
+ (6, 21910, 'how', 'm'),
+ (7, -63, 'that''s', 'go'),
+ (8, -8276281, 's', 'a'),
+ (9, -101, 'w', 'y')
+ """
+
+ sql """
+ CREATE TABLE test_agg_after_nlj_local_exchange_t2 (
+ pk INT,
+ col_varchar_10__undef_signed VARCHAR(10),
+ col_bigint_undef_signed BIGINT,
+ col_varchar_64__undef_signed VARCHAR(64)
+ )
+ ENGINE=OLAP
+ DUPLICATE KEY(pk, col_varchar_10__undef_signed)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES("replication_num" = "1")
+ """
+
+ sql """
+ INSERT INTO test_agg_after_nlj_local_exchange_t2
+ (pk, col_bigint_undef_signed, col_varchar_10__undef_signed,
col_varchar_64__undef_signed)
+ VALUES
+ (0, NULL, 'right', 'g'),
+ (1, -486256, 'on', 'on'),
+ (2, -1, 'I''ll', 'at'),
+ (3, 29263, 'h', 'don''t'),
+ (4, 5453, 'a', 's'),
+ (5, -119, 'j', 'can''t'),
+ (6, 89, 'one', 'n'),
+ (7, -7227, 's', 'u'),
+ (8, 94, 'time', 'b'),
+ (9, 1816630, 'yes', 'yes')
+ """
+
+ sql "SYNC"
+
+ sql "SET default_variant_doc_hash_shard_count = 0"
+ sql "SET default_variant_max_subcolumns_count = 4"
+ sql "SET default_variant_sparse_hash_shard_count = 4"
+ sql "SET disable_join_reorder = true"
+ sql "SET disable_streaming_preaggregations = true"
+ sql "SET enable_common_expr_pushdown = false"
+ sql "SET enable_common_expr_pushdown_for_inverted_index = false"
+ sql "SET enable_distinct_streaming_agg_force_passthrough = false"
+ sql "SET enable_function_pushdown = true"
+ sql "SET enable_local_exchange_before_agg = false"
+ sql "SET enable_runtime_filter_partition_prune = false"
+ sql "SET enable_runtime_filter_prune = false"
+ sql "SET enable_strong_consistency_read = true"
+ sql "SET enable_sync_runtime_filter_size = false"
+ sql "SET exchange_multi_blocks_byte_size = 5563624"
+ sql "SET experimental_enable_parallel_scan = false"
+ sql "SET parallel_pipeline_task_num = 4"
+ sql "SET parallel_prepare_threshold = 28"
+ sql "SET query_timeout = 600"
+ sql "SET runtime_filter_type = 'IN,MIN_MAX'"
+ sql "SET runtime_filter_wait_time_ms = 5000"
+ sql "SET topn_opt_limit_threshold = 1000"
+ sql "SET agg_phase = 4"
+
+ order_qt_agg_after_nlj_local_exchange """
+ SELECT
+ COUNT(DISTINCT table1.`pk`) AS field1,
+ MAX(table1.col_bigint_undef_signed) AS field2
+ FROM
+ test_agg_after_nlj_local_exchange_t1 AS table1
+ LEFT OUTER JOIN test_agg_after_nlj_local_exchange_t2 AS table2
+ ON table2.col_varchar_10__undef_signed =
table2.col_varchar_64__undef_signed
+ LEFT JOIN test_agg_after_nlj_local_exchange_t1 AS table3
+ ON table2.col_varchar_10__undef_signed =
table2.col_varchar_64__undef_signed
+ WHERE
+ table1.`pk` > 3
+ AND table1.`pk` < (3 + 25)
+ OR table1.col_varchar_64__undef_signed > 'cnvUBxJyCp'
+ AND table1.col_varchar_64__undef_signed <= 'z'
+ OR table1.col_bigint_undef_signed != 2
+ OR table1.`pk` NOT BETWEEN 2 AND (2 + 1)
+ AND table1.`pk` > 7
+ AND table1.`pk` < (7 + 2)
+ AND table1.`pk` IN (2, 10, 2)
+ ORDER BY
+ field1,
+ field2
+ LIMIT 1000
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
