This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 477b5883ebe [fix](local shuffle) Prevent row loss at
parallel-to-serial pipeline boundaries (#67177)
477b5883ebe is described below
commit 477b5883ebeca2bb6df027dffddc5100c3076791
Author: 924060929 <[email protected]>
AuthorDate: Mon Sep 14 18:31:30 2026 +0800
[fix](local shuffle) Prevent row loss at parallel-to-serial pipeline
boundaries (#67177)
### What problem does this PR solve?
Issue Number: None
Related PR: #65835
Problem Summary:
FE-planned local shuffle could implicitly collapse a parallel child
subtree into a serial consumer without an explicit gather. Remote
exchange receivers outside the first task could then be left unread,
producing incomplete scalar aggregate results.
The old BE-native planner also gave `PASS_TO_ONE` two meanings: it
created a gather for a shared broadcast hash table but silently changed
it to `BROADCAST` for private hash tables. That hidden reinterpretation
made the FE planner hard to reason about.
This PR makes the final distribution explicit:
- `PlanNode.enforceRequire` inserts `PASS_TO_ONE` at every non-serial
child to serial parent boundary.
- Shared broadcast hash-table builds choose `PASS_TO_ONE`; private
builds choose `BROADCAST`, in both FE planning and BE-native planning.
- BE exchanger factories honor the requested local partition type
literally.
- An FE-planned serial gather keeps all upstream senders but creates
exactly one downstream source/task.
No execution-version gate is added. FE local-shuffle planning has not
shipped on release branches, so supported rolling upgrades continue to
use the old FE with BE-native local-shuffle planning until FE is
upgraded.
### Release note
Fix incorrect results from FE-planned local shuffle at
parallel-to-serial pipeline boundaries.
---
be/src/exec/operator/hashjoin_build_sink.h | 10 +-
be/src/exec/pipeline/pipeline_fragment_context.cpp | 88 ++++++++--------
be/test/exec/operator/hashjoin_build_sink_test.cpp | 26 ++++-
be/test/exec/pipeline/local_exchanger_test.cpp | 69 ++++++++++--
.../org/apache/doris/planner/AnalyticEvalNode.java | 16 +--
.../org/apache/doris/planner/HashJoinNode.java | 6 +-
.../java/org/apache/doris/planner/PlanNode.java | 28 ++++-
.../planner/LocalShuffleNodeCoverageTest.java | 117 +++++++++++++++++++--
.../apache/doris/qe/LocalExchangePlannerTest.java | 30 ++++++
gensrc/thrift/Partitions.thrift | 6 +-
.../test_serial_aggregation_over_parallel_join.out | 18 ++++
...st_serial_aggregation_over_parallel_join.groovy | 97 +++++++++++++++++
12 files changed, 437 insertions(+), 74 deletions(-)
diff --git a/be/src/exec/operator/hashjoin_build_sink.h
b/be/src/exec/operator/hashjoin_build_sink.h
index 67b69f2fad4..e3eed65e01f 100644
--- a/be/src/exec/operator/hashjoin_build_sink.h
+++ b/be/src/exec/operator/hashjoin_build_sink.h
@@ -132,12 +132,16 @@ public:
._should_build_hash_table;
}
- DataDistribution required_data_distribution(RuntimeState* /*state*/) const
override {
+ DataDistribution required_data_distribution(RuntimeState* state) const
override {
if (_join_op == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
return {TLocalPartitionType::NOOP};
} else if (_is_broadcast_join) {
- return _child->is_serial_operator() ?
DataDistribution(TLocalPartitionType::PASS_TO_ONE)
- :
DataDistribution(TLocalPartitionType::NOOP);
+ if (!_child->is_serial_operator()) {
+ return {TLocalPartitionType::NOOP};
+ }
+ return state->enable_share_hash_table_for_broadcast_join()
+ ? DataDistribution(TLocalPartitionType::PASS_TO_ONE)
+ : DataDistribution(TLocalPartitionType::BROADCAST);
}
return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE ||
_join_distribution ==
TJoinDistributionType::COLOCATE
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp
b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index f62bd073081..5fa5c4c2932 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -719,6 +719,7 @@ Status
PipelineFragmentContext::_build_pipelines(ObjectPool* pool, const Descrip
Status PipelineFragmentContext::_create_deferred_local_exchangers() {
for (auto& info : _deferred_exchangers) {
+ const int source_count =
cast_set<int>(info.shared_state->source_deps.size());
// DANGER ZONE — do not "fix" this line without reading the history.
//
// sender_count seeds Exchanger::_running_sink_operators, which the
source side
@@ -751,34 +752,29 @@ Status
PipelineFragmentContext::_create_deferred_local_exchangers() {
switch (info.partition_type) {
case TLocalPartitionType::LOCAL_EXECUTION_HASH_SHUFFLE:
case TLocalPartitionType::GLOBAL_EXECUTION_HASH_SHUFFLE:
- info.shared_state->exchanger = ShuffleExchanger::create_unique(
- sender_count, _num_instances, info.num_partitions,
info.free_blocks_limit,
- info.partition_type);
+ info.shared_state->exchanger =
+ ShuffleExchanger::create_unique(sender_count,
source_count, info.num_partitions,
+ info.free_blocks_limit,
info.partition_type);
break;
case TLocalPartitionType::BUCKET_HASH_SHUFFLE:
info.shared_state->exchanger =
BucketShuffleExchanger::create_unique(
- sender_count, _num_instances, info.num_partitions,
info.free_blocks_limit);
+ sender_count, source_count, info.num_partitions,
info.free_blocks_limit);
break;
case TLocalPartitionType::PASSTHROUGH:
info.shared_state->exchanger = PassthroughExchanger::create_unique(
- sender_count, _num_instances, info.free_blocks_limit);
+ sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::BROADCAST:
info.shared_state->exchanger = BroadcastExchanger::create_unique(
- sender_count, _num_instances, info.free_blocks_limit);
+ sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::PASS_TO_ONE:
- if (_runtime_state->enable_share_hash_table_for_broadcast_join()) {
- info.shared_state->exchanger =
PassToOneExchanger::create_unique(
- sender_count, _num_instances, info.free_blocks_limit);
- } else {
- info.shared_state->exchanger =
BroadcastExchanger::create_unique(
- sender_count, _num_instances, info.free_blocks_limit);
- }
+ info.shared_state->exchanger = PassToOneExchanger::create_unique(
+ sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::ADAPTIVE_PASSTHROUGH:
info.shared_state->exchanger =
AdaptivePassthroughExchanger::create_unique(
- sender_count, _num_instances, info.free_blocks_limit);
+ sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::NOOP:
case TLocalPartitionType::LOCAL_MERGE_SORT:
@@ -855,11 +851,17 @@ void
PipelineFragmentContext::_propagate_local_exchange_num_tasks() {
if (pit != id_to_pipe.end()) {
auto& pipe = pit->second;
const auto& ops = pipe->operators();
- const bool le_source =
- !ops.empty() &&
dynamic_cast<LocalExchangeSourceOperatorX*>(ops.front().get());
+ auto* le_source =
+ !ops.empty() ?
dynamic_cast<LocalExchangeSourceOperatorX*>(ops.front().get())
+ : nullptr;
const bool serial_source = !ops.empty() &&
ops.front()->is_serial_operator();
if (le_source) {
- pipe->set_num_tasks(_num_instances);
+ // PASS_TO_ONE is the explicit N-to-one boundary. Its upstream
pipeline
+ // keeps all active tasks, while only fragment instance 0
creates the
+ // downstream serial pipeline task.
+ if (le_source->exchange_type() !=
TLocalPartitionType::PASS_TO_ONE) {
+ pipe->set_num_tasks(_num_instances);
+ }
} else if (!serial_source) {
int target = pipe->num_tasks();
const auto up_it = _dag.find(id);
@@ -1072,22 +1074,12 @@ Status
PipelineFragmentContext::_add_local_exchange_impl(
: 0);
break;
case TLocalPartitionType::PASS_TO_ONE:
- if (_runtime_state->enable_share_hash_table_for_broadcast_join()) {
- // If shared hash table is enabled for BJ, hash table will be
built by only one task
- shared_state->exchanger = PassToOneExchanger::create_unique(
- cur_pipe->num_tasks(), _num_instances,
-
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
- ? cast_set<int>(_runtime_state->query_options()
-
.local_exchange_free_blocks_limit)
- : 0);
- } else {
- shared_state->exchanger = BroadcastExchanger::create_unique(
- cur_pipe->num_tasks(), _num_instances,
-
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
- ? cast_set<int>(_runtime_state->query_options()
-
.local_exchange_free_blocks_limit)
- : 0);
- }
+ shared_state->exchanger = PassToOneExchanger::create_unique(
+ cur_pipe->num_tasks(), _num_instances,
+
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
+ ? cast_set<int>(
+
_runtime_state->query_options().local_exchange_free_blocks_limit)
+ : 0);
break;
case TLocalPartitionType::ADAPTIVE_PASSTHROUGH:
shared_state->exchanger = AdaptivePassthroughExchanger::create_unique(
@@ -2046,9 +2038,12 @@ Status
PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
}
case TPlanNodeType::LOCAL_EXCHANGE_NODE: {
op = std::make_shared<LocalExchangeSourceOperatorX>(pool, tnode,
next_operator_id(), descs);
- // The downstream pipeline (containing LocalExchangeSource) must have
- // _num_instances tasks — matching BE-native
_inherit_pipeline_properties
- // which sets pipe_with_source.set_num_tasks(_num_instances).
+ const auto partition_type = tnode.local_exchange_node.partition_type;
+ const bool pass_to_one = partition_type ==
TLocalPartitionType::PASS_TO_ONE;
+ // Except at an explicit PASS_TO_ONE boundary, the downstream pipeline
+ // (containing LocalExchangeSource) must have _num_instances tasks.
This
+ // matches BE-native _inherit_pipeline_properties, which sets
+ // pipe_with_source.set_num_tasks(_num_instances).
// Without this, when the parent pipeline was reduced by a serial
operator
// (e.g., serial Exchange with use_serial_exchange=true, or
UNPARTITIONED
// Exchange), the downstream inherits the reduced num_tasks via
@@ -2057,14 +2052,23 @@ Status
PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
// sink round-robins to all channels and crashes on uninitialized ones.
RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances));
// Restore downstream pipeline's num_tasks (mirroring
_inherit_pipeline_properties:
- // downstream keeps _num_instances, upstream gets the serial/reduced
count)
- cur_pipe->set_num_tasks(_num_instances);
+ // downstream keeps _num_instances, upstream gets the serial/reduced
count).
+ // PASS_TO_ONE is the explicit parallel-to-serial boundary: its
downstream
+ // pipeline must keep the serial parent's single active task, while
the upstream
+ // pipeline is expanded below so every fragment instance keeps an
active receiver.
+ if (!pass_to_one) {
+ cur_pipe->set_num_tasks(_num_instances);
+ }
+ const int downstream_num_tasks = cur_pipe->num_tasks();
const auto downstream_pipeline_id = cur_pipe->id();
if (!_dag.contains(downstream_pipeline_id)) {
_dag.insert({downstream_pipeline_id, {}});
}
cur_pipe = add_pipeline(cur_pipe);
+ if (pass_to_one) {
+ cur_pipe->set_num_tasks(_num_instances);
+ }
// If this local exchange was inserted because of a serial scan
(is_serial_operator),
// the upstream pipeline (cur_pipe) should have num_tasks=1 (only 1
scan task).
// We set this now so the exchanger is created with the correct sender
count.
@@ -2076,7 +2080,6 @@ Status
PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
_dag[downstream_pipeline_id].push_back(cur_pipe->id());
int num_partitions = 0;
std::map<int, int> shuffle_id_to_instance_idx;
- auto partition_type = tnode.local_exchange_node.partition_type;
switch (partition_type) {
case TLocalPartitionType::BUCKET_HASH_SHUFFLE:
num_partitions = _params.num_buckets;
@@ -2116,9 +2119,10 @@ Status
PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
? cast_set<int>(
_runtime_state->query_options().local_exchange_free_blocks_limit)
: 0;
- auto shared_state =
LocalExchangeSharedState::create_shared(_num_instances);
- shared_state->create_source_dependencies(_num_instances,
local_exchange_id,
- local_exchange_id,
"LOCAL_EXCHANGE_OPERATOR");
+ const int source_count = downstream_num_tasks;
+ auto shared_state =
LocalExchangeSharedState::create_shared(source_count);
+ shared_state->create_source_dependencies(source_count,
local_exchange_id, local_exchange_id,
+ "LOCAL_EXCHANGE_OPERATOR");
shared_state->create_sink_dependency(sink_id, local_exchange_id,
"LOCAL_EXCHANGE_SINK");
_op_id_to_shared_state.insert({local_exchange_id, {shared_state,
shared_state->sink_deps}});
// Defer exchanger creation: sender count depends on final upstream
num_tasks
diff --git a/be/test/exec/operator/hashjoin_build_sink_test.cpp
b/be/test/exec/operator/hashjoin_build_sink_test.cpp
index 0cabf71258f..35cbb15fe92 100644
--- a/be/test/exec/operator/hashjoin_build_sink_test.cpp
+++ b/be/test/exec/operator/hashjoin_build_sink_test.cpp
@@ -318,6 +318,30 @@ TEST_F(HashJoinBuildSinkTest, Sink) {
run_test_block(test_block);
}
+TEST_F(HashJoinBuildSinkTest, BroadcastJoinRequiredDataDistribution) {
+ auto tnode = _helper.create_test_plan_node(TJoinOp::INNER_JOIN,
{TPrimitiveType::INT}, {false},
+ {false});
+ tnode.hash_join_node.__set_is_broadcast_join(true);
+ auto [probe_operator, sink_operator] = _helper.create_operators(tnode);
+ ASSERT_TRUE(probe_operator);
+ ASSERT_TRUE(sink_operator);
+
+
EXPECT_EQ(sink_operator->required_data_distribution(_helper.runtime_state.get())
+ .distribution_type,
+ TLocalPartitionType::NOOP);
+
+ sink_operator->child()->set_serial_operator();
+ _helper.runtime_state->_enable_share_hash_table_for_broadcast_join = true;
+
EXPECT_EQ(sink_operator->required_data_distribution(_helper.runtime_state.get())
+ .distribution_type,
+ TLocalPartitionType::PASS_TO_ONE);
+
+ _helper.runtime_state->_enable_share_hash_table_for_broadcast_join = false;
+
EXPECT_EQ(sink_operator->required_data_distribution(_helper.runtime_state.get())
+ .distribution_type,
+ TLocalPartitionType::BROADCAST);
+}
+
TEST_F(HashJoinBuildSinkTest, Terminate) {
auto test_block = [&](TJoinOp::type op_type, const
std::vector<TPrimitiveType::type>& key_types,
const std::vector<bool>& left_nullables,
@@ -695,4 +719,4 @@ TEST_F(SharedHashTableSignalTest,
MultipleNonBuildersAllReturnEOFWhenTerminated)
}
}
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/test/exec/pipeline/local_exchanger_test.cpp
b/be/test/exec/pipeline/local_exchanger_test.cpp
index 0967c6758bd..09c3dfa26c0 100644
--- a/be/test/exec/pipeline/local_exchanger_test.cpp
+++ b/be/test/exec/pipeline/local_exchanger_test.cpp
@@ -29,12 +29,14 @@
#include "exec/exchange/local_exchange_sink_operator.h"
#include "exec/exchange/local_exchange_source_operator.h"
#include "exec/pipeline/dependency.h"
+#include "exec/pipeline/pipeline_fragment_context.h"
#include "exec/pipeline/thrift_builder.h"
#include "exprs/vslot_ref.h"
+#include "runtime/descriptor_helper.h"
namespace doris {
-class LocalExchangerTest : public testing::Test {
+class LocalExchangerTest : public testing::TestWithParam<int> {
public:
LocalExchangerTest() = default;
~LocalExchangerTest() override = default;
@@ -532,9 +534,60 @@ TEST_F(LocalExchangerTest, PassthroughExchanger) {
}
}
-TEST_F(LocalExchangerTest, PassToOneExchanger) {
+TEST_F(LocalExchangerTest, FePlannedPassToOneUsesOneDownstreamSource) {
+ constexpr int num_instances = 4;
+ _query_options.__set_enable_share_hash_table_for_broadcast_join(false);
+ TPipelineFragmentParams params;
+ auto context = std::make_shared<PipelineFragmentContext>(
+ _query_id, params, _query_ctx, ExecEnv::GetInstance(),
[](RuntimeState*, Status*) {});
+ context->_num_instances = num_instances;
+ context->_total_instances = num_instances;
+ context->_runtime_state = RuntimeState::create_unique(_query_id,
_fragment_id, _query_options,
+
_query_ctx->query_globals,
+
ExecEnv::GetInstance(), _query_ctx.get());
+
+ auto downstream_pipe = context->add_pipeline();
+ downstream_pipe->set_num_tasks(1);
+ auto upstream_pipe = downstream_pipe;
+
+ TLocalExchangeNode local_exchange_node;
+ local_exchange_node.__set_partition_type(TLocalPartitionType::PASS_TO_ONE);
+ TPlanNode tnode;
+ tnode.__set_node_type(TPlanNodeType::LOCAL_EXCHANGE_NODE);
+ tnode.__set_node_id(0);
+ tnode.__set_num_children(1);
+ tnode.__set_local_exchange_node(local_exchange_node);
+ tnode.__set_row_tuples({0});
+
+ ObjectPool pool;
+ TDescriptorTableBuilder desc_builder;
+ TTupleDescriptorBuilder().build(&desc_builder);
+ DescriptorTbl* descs = nullptr;
+ ASSERT_TRUE(DescriptorTbl::create(&pool, desc_builder.desc_tbl(),
&descs).ok());
+ OperatorPtr op;
+ OperatorPtr cache_op;
+ ASSERT_TRUE(context->_create_operator(&pool, tnode, *descs, op,
upstream_pipe,
+ /*parent_idx=*/-1, /*child_idx=*/0,
+
/*followed_by_shuffled_operator=*/false,
+
/*require_bucket_distribution=*/false, cache_op)
+ .ok());
+ ASSERT_EQ(context->_deferred_exchangers.size(), 1);
+ EXPECT_EQ(downstream_pipe->num_tasks(), 1);
+ EXPECT_EQ(upstream_pipe->num_tasks(), num_instances);
+
+ auto shared_state = context->_deferred_exchangers.front().shared_state;
+ EXPECT_EQ(shared_state->source_deps.size(), 1);
+ EXPECT_EQ(shared_state->mem_counters.size(), 1);
+ ASSERT_TRUE(context->_create_deferred_local_exchangers().ok());
+ ASSERT_NE(shared_state->exchanger, nullptr);
+ EXPECT_EQ(shared_state->exchanger->get_type(),
TLocalPartitionType::PASS_TO_ONE);
+ EXPECT_EQ(shared_state->exchanger->_num_senders, num_instances);
+ EXPECT_EQ(shared_state->exchanger->_num_sources, 1);
+}
+
+TEST_P(LocalExchangerTest, PassToOneExchanger) {
int num_sink = 4;
- int num_sources = 4;
+ int num_sources = GetParam();
int free_block_limit = 0;
const auto expect_block_bytes = 128;
@@ -555,6 +608,8 @@ TEST_F(LocalExchangerTest, PassToOneExchanger) {
shared_state->create_source_dependencies(num_sources, 0, 0, "TEST");
auto* exchanger = (PassToOneExchanger*)shared_state->exchanger.get();
+ EXPECT_EQ(exchanger->_num_senders, num_sink);
+ EXPECT_EQ(exchanger->_num_sources, num_sources);
for (size_t i = 0; i < num_sink; i++) {
auto* compute_hash_value_timer =
ADD_TIMER(profile, "ComputeHashValueTime" + std::to_string(i));
@@ -585,10 +640,9 @@ TEST_F(LocalExchangerTest, PassToOneExchanger) {
"MemoryUsage" + std::to_string(i), TUnit::BYTES, "", 1);
shared_state->mem_counters[i] = _local_states[i]->_memory_used_counter;
}
-
{
- // Enqueue `num_blocks` blocks with 10 rows for each data queue.
- for (size_t i = 0; i < num_sources; i++) {
+ // Enqueue `num_blocks` blocks with 10 rows from every sender.
+ for (size_t i = 0; i < num_sink; i++) {
for (size_t j = 0; j < num_blocks; j++) {
Block in_block;
DataTypePtr int_type = std::make_shared<DataTypeInt32>();
@@ -744,6 +798,8 @@ TEST_F(LocalExchangerTest, PassToOneExchanger) {
}
}
+INSTANTIATE_TEST_SUITE_P(SourceCardinality, LocalExchangerTest,
testing::Values(4, 1));
+
TEST_F(LocalExchangerTest, BroadcastExchanger) {
int num_sink = 4;
int num_sources = 4;
@@ -1397,4 +1453,5 @@ TEST_F(LocalExchangerTest,
ShuffleExchangerRestoreOutputBlockOnAddRowsError) {
EXPECT_EQ(output_block.rows(), 1);
EXPECT_NO_THROW(output_block.check_number_of_rows());
}
+
} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
index 1698b25495d..f530875ccdd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
@@ -209,17 +209,19 @@ public class AnalyticEvalNode extends PlanNode {
LocalExchangeType outputType = null;
if (partitionExprs.isEmpty()) {
// Serial AnalyticEval (OVER() with no PARTITION BY):
- // Must NOT have any LocalExchange between AnalyticEval and its
child.
- // On BE, AnalyticSink and AnalyticSource share state
(source_deps/sink_deps).
- // A LocalExchange below would restore the AnalyticSink pipeline
to _num_instances
- // tasks while the serial AnalyticSource pipeline stays at 1 task.
+ // Do not keep a redundant LocalExchange between AnalyticEval and
an already
+ // serial child. On BE, AnalyticSink and AnalyticSource share state
+ // (source_deps/sink_deps), so restoring only the sink pipeline to
+ // _num_instances tasks would mismatch the serial source pipeline.
//
- // Use enforceRequire with noRequire to traverse children, then
strip any
- // LocalExchange the child inserted (e.g., Exchange wrapping
itself with PASSTHROUGH).
+ // PASS_TO_ONE is different: enforceRequire inserts it when the
child subtree is
+ // parallel. It is the explicit N-to-one boundary that keeps every
upstream task
+ // active while leaving the analytic sink/source pair at one task,
so retain it.
Pair<PlanNode, LocalExchangeType> enforceResult
= enforceRequire(translatorContext, children.get(0), 0,
LocalExchangeTypeRequire.noRequire());
PlanNode newChild = enforceResult.first;
- if (newChild instanceof LocalExchangeNode) {
+ if (newChild instanceof LocalExchangeNode
+ && ((LocalExchangeNode) newChild).getExchangeType() !=
LocalExchangeType.PASS_TO_ONE) {
newChild = newChild.getChild(0);
}
children = Lists.newArrayList(newChild);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
index e9d8ac63041..1fc7880e9b2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
@@ -338,7 +338,11 @@ public class HashJoinNode extends JoinNodeBase {
? LocalExchangeTypeRequire.requirePassthrough()
: LocalExchangeTypeRequire.noRequire();
buildSideRequire = buildChildSerial
- ? LocalExchangeTypeRequire.requirePassToOne()
+ ? (translatorContext.getConnectContext() == null
+ ||
translatorContext.getConnectContext().getSessionVariable()
+ .enableShareHashTableForBroadcastJoin
+ ? LocalExchangeTypeRequire.requirePassToOne()
+ : LocalExchangeTypeRequire.requireBroadcast())
: LocalExchangeTypeRequire.noRequire();
// For serial or force-passthrough probe: output is PASSTHROUGH.
// For a non-serial probe without the flag: propagate the probe's
distribution.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index b592c15f627..580dc9a4eaa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -1084,11 +1084,19 @@ public abstract class PlanNode extends
TreeNode<PlanNode> {
// serial-source mode, BE treats this operator as non-serial
regardless of isSerialNode.
// Using isSerialNode here would set the child's serial-ancestor flag
wider than BE's
// view and over-skip required LocalExchanges downstream.
- boolean childHasSerialAncestor = inheritedSerial
- || currentNodeSerialOnBe.get();
+ boolean selfSerial = currentNodeSerialOnBe.get();
+ boolean passToOneAtSerialBoundary = selfSerial
+ &&
!child.isSerialOperatorOnBe(translatorContext.getConnectContext());
+ // PASS_TO_ONE becomes the pipeline boundary between this serial
consumer and the
+ // parallel child subtree. Do not let either serial marker cross that
boundary: the
+ // child must still plan the local exchanges required by its own
parallel pipelines.
+ boolean childHasSerialAncestor = passToOneAtSerialBoundary
+ ? false : inheritedSerial || selfSerial;
+ boolean childHasSerialParentPipeline = passToOneAtSerialBoundary
+ ? false : startsNewPipeline
+ ? currentPipelineSerial :
translatorContext.hasSerialParentPipeline(this);
translatorContext.setHasSerialAncestorInPipeline(child,
childHasSerialAncestor);
- translatorContext.setHasSerialParentPipeline(child, startsNewPipeline
- ? currentPipelineSerial :
translatorContext.hasSerialParentPipeline(this));
+ translatorContext.setHasSerialParentPipeline(child,
childHasSerialParentPipeline);
// 1b. Propagate shuffle-for-correctness-ancestor flag to child.
// Mirrors BE's _followed_by_shuffled_operator: a downstream operator
needs hash
@@ -1109,6 +1117,18 @@ public abstract class PlanNode extends
TreeNode<PlanNode> {
Pair<PlanNode, LocalExchangeType> childOutput =
child.enforceAndDeriveLocalExchange(translatorContext, this,
require);
+ // A serial consumer must not implicitly reduce a non-serial subtree
to one pipeline
+ // task. Besides losing parallelism, that can make a remote Exchange
expose fewer
+ // receiver tasks than FE addresses. Keep the subtree parallel and
make the N-to-one
+ // transition explicit. PASS_TO_ONE keeps every upstream receiver task
alive and
+ // funnels their output into the serial downstream pipeline's only
task.
+ if (passToOneAtSerialBoundary && childOutput.second !=
LocalExchangeType.PASS_TO_ONE) {
+ childOutput = Pair.of(
+ createLocalExchange(translatorContext, childOutput.first,
+ LocalExchangeType.PASS_TO_ONE, null),
+ LocalExchangeType.PASS_TO_ONE);
+ }
+
// Steps 2.5 and 3 both react to a serial child but address different
concerns:
// - Step 2.5 rewrites the OUTPUT-side view (what we tell
satisfy/parent about
// the child's actual distribution). A serial pipeline runs with
1 task so
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
index 59c26166c3f..e8bcbb195aa 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
@@ -547,6 +547,26 @@ public class LocalShuffleNodeCoverageTest {
+ "even if isSerialNode()=true — BE treats the node as
non-serial.");
}
+ @Test
+ public void testPassToOneBoundaryKeepsParallelSubtreeLocalExchange() {
+ PlanTranslatorContext ctx = new PlanTranslatorContext();
+ TrackingPlanNode leaf = new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP);
+ HashRequiringPlanNode parallelSubtree = new
HashRequiringPlanNode(nextPlanNodeId(), leaf);
+ SerialPipelineBoundaryNode serialParent = new
SerialPipelineBoundaryNode(
+ nextPlanNodeId(), parallelSubtree);
+ serialParent.fragment = Mockito.mock(PlanFragment.class);
+
Mockito.when(serialParent.fragment.useSerialSource(Mockito.any())).thenReturn(true);
+
+ Pair<PlanNode, LocalExchangeType> output =
serialParent.enforceAndDeriveLocalExchange(
+ ctx, null, LocalExchangeTypeRequire.noRequire());
+
+ Assertions.assertEquals(LocalExchangeType.PASS_TO_ONE, output.second);
+ assertChildLocalExchangeType(serialParent, 0,
LocalExchangeType.PASS_TO_ONE);
+ Assertions.assertSame(parallelSubtree,
serialParent.getChild(0).getChild(0));
+ assertChildLocalExchangeType(parallelSubtree, 0,
+ LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE);
+ }
+
@Test
public void testNestedLoopJoinNodeBranches() {
PlanTranslatorContext ctx = new PlanTranslatorContext();
@@ -753,8 +773,10 @@ public class LocalShuffleNodeCoverageTest {
// Output is still PASSTHROUGH (hardcoded for useSerialSource +
ScanNode child).
SerialTrackingScanNode serialScan = new
SerialTrackingScanNode(nextPlanNodeId(), LocalExchangeType.NOOP);
SortNode scanSort = new SortNode(nextPlanNodeId(), serialScan,
sortInfo, false);
- scanSort.fragment = Mockito.mock(PlanFragment.class);
-
Mockito.when(scanSort.fragment.useSerialSource(Mockito.any())).thenReturn(true);
+ PlanFragment serialSortFragment = Mockito.mock(PlanFragment.class);
+
Mockito.when(serialSortFragment.useSerialSource(Mockito.any())).thenReturn(true);
+ scanSort.setFragment(serialSortFragment);
+ serialScan.setFragment(serialSortFragment);
Pair<PlanNode, LocalExchangeType> scanOutput =
scanSort.enforceAndDeriveLocalExchange(
ctx, null, LocalExchangeTypeRequire.noRequire());
// Non-merge, non-analytic SortNode: isSerialNode()=true,
requireChild=noRequire,
@@ -809,6 +831,24 @@ public class LocalShuffleNodeCoverageTest {
Assertions.assertEquals(LocalExchangeType.NOOP,
noPartitionOutput.second);
Assertions.assertSame(noPartitionChild, noPartition.getChild(0));
+ // A serial analytic consumer over a parallel subtree needs the generic
+ // parallel-to-serial boundary inserted by PlanNode.enforceRequire.
The analytic
+ // special case may remove a redundant exchange directly above a
serial Exchange,
+ // but must retain this PASS_TO_ONE gather.
+ TrackingPlanNode parallelChild = new
TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP);
+ AnalyticEvalNode serialOverParallel = new
AnalyticEvalNode(nextPlanNodeId(), parallelChild,
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(),
+ null, new TupleDescriptor(new
TupleId(NEXT_ID.getAndIncrement())));
+ PlanFragment serialAnalyticFragment = Mockito.mock(PlanFragment.class);
+
Mockito.when(serialAnalyticFragment.useSerialSource(Mockito.any())).thenReturn(true);
+ serialOverParallel.setFragment(serialAnalyticFragment);
+ parallelChild.setFragment(serialAnalyticFragment);
+ Pair<PlanNode, LocalExchangeType> serialOverParallelOutput
+ = serialOverParallel.enforceAndDeriveLocalExchange(
+ ctx, null, LocalExchangeTypeRequire.noRequire());
+ Assertions.assertEquals(LocalExchangeType.NOOP,
serialOverParallelOutput.second);
+ assertChildLocalExchangeType(serialOverParallel, 0,
LocalExchangeType.PASS_TO_ONE);
+
// Analytic with partition but no orderBy, non-colocated →
noRequire/NOOP.
// (Non-colocated analytic relies on parent SortNode to handle
distribution.)
TrackingScanNode hashChild = new TrackingScanNode(nextPlanNodeId(),
LocalExchangeType.NOOP);
@@ -826,12 +866,14 @@ public class LocalShuffleNodeCoverageTest {
Collections.emptyList(),
Collections.singletonList(Mockito.mock(Expr.class)),
Collections.singletonList(new
OrderByElement(Mockito.mock(Expr.class), true, true)),
null, new TupleDescriptor(new
TupleId(NEXT_ID.getAndIncrement())));
- orderedAnalytic.fragment = Mockito.mock(PlanFragment.class);
-
Mockito.when(orderedAnalytic.fragment.useSerialSource(Mockito.any())).thenReturn(true);
+ PlanFragment orderedAnalyticFragment =
Mockito.mock(PlanFragment.class);
+
Mockito.when(orderedAnalyticFragment.useSerialSource(Mockito.any())).thenReturn(true);
+ orderedAnalytic.setFragment(orderedAnalyticFragment);
+ serialScan.setFragment(orderedAnalyticFragment);
Pair<PlanNode, LocalExchangeType> orderedOutput =
orderedAnalytic.enforceAndDeriveLocalExchange(
ctx, null, LocalExchangeTypeRequire.noRequire());
- // Serial AnalyticEval returns NOOP — lets framework serial check
handle fan-out
- Assertions.assertEquals(LocalExchangeType.NOOP, orderedOutput.second);
+ Assertions.assertEquals(LocalExchangeType.PASSTHROUGH,
orderedOutput.second);
+ assertChildLocalExchangeType(orderedAnalytic, 0,
LocalExchangeType.PASSTHROUGH);
}
@Test
@@ -1210,6 +1252,69 @@ public class LocalShuffleNodeCoverageTest {
}
}
+ private static class SerialPipelineBoundaryNode extends PlanNode {
+ SerialPipelineBoundaryNode(PlanNodeId id, PlanNode child) {
+ super(id, Lists.newArrayList(new TupleId(id.asInt() + 30000)),
+ "SERIAL_PIPELINE_BOUNDARY");
+ children.add(child);
+ }
+
+ @Override
+ public boolean isSerialNode() {
+ return true;
+ }
+
+ @Override
+ protected boolean shouldResetSerialFlagForChild(int childIndex) {
+ return true;
+ }
+
+ @Override
+ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
+ PlanTranslatorContext translatorContext, PlanNode parent,
+ LocalExchangeTypeRequire parentRequire) {
+ Pair<PlanNode, LocalExchangeType> result =
enforceRequire(translatorContext,
+ children.get(0), 0, LocalExchangeTypeRequire.noRequire());
+ children = Lists.newArrayList(result.first);
+ return Pair.of(this, result.second);
+ }
+
+ @Override
+ protected void toThrift(TPlanNode msg) {
+ }
+
+ @Override
+ public String getNodeExplainString(String prefix, TExplainLevel
detailLevel) {
+ return "";
+ }
+ }
+
+ private static class HashRequiringPlanNode extends PlanNode {
+ HashRequiringPlanNode(PlanNodeId id, PlanNode child) {
+ super(id, Lists.newArrayList(new TupleId(id.asInt() + 40000)),
"HASH_REQUIRING");
+ children.add(child);
+ }
+
+ @Override
+ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
+ PlanTranslatorContext translatorContext, PlanNode parent,
+ LocalExchangeTypeRequire parentRequire) {
+ Pair<PlanNode, LocalExchangeType> result =
enforceRequire(translatorContext,
+ children.get(0), 0,
LocalExchangeTypeRequire.requireHash());
+ children = Lists.newArrayList(result.first);
+ return Pair.of(this, result.second);
+ }
+
+ @Override
+ protected void toThrift(TPlanNode msg) {
+ }
+
+ @Override
+ public String getNodeExplainString(String prefix, TExplainLevel
detailLevel) {
+ return "";
+ }
+ }
+
private static class TrackingPlanNode extends PlanNode {
private final LocalExchangeType providedType;
private LocalExchangeTypeRequire lastRequire;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
index d691028666f..7fead508744 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
@@ -75,6 +75,7 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
sv.setEnableLocalShufflePlanner(true);
sv.setEnableLocalShuffle(true);
sv.setEnableNereidsDistributePlanner(true);
+ sv.setEnableSqlCache(false);
sv.setIgnoreStorageDataDistribution(true);
sv.setPipelineTaskNum("4");
sv.setForceToLocalShuffle(false);
@@ -83,6 +84,7 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
// the default strategy) before applying this test's own tweaks.
sv.aggPhase = 0;
sv.enableBroadcastJoinForcePassthrough = false;
+ sv.enableShareHashTableForBroadcastJoin = true;
if (tweaks != null) {
tweaks.accept(sv);
}
@@ -169,6 +171,22 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
olapScan("t1")))))));
}
+ @Test
+ public void testSerialDistinctAggGathersParallelJoinOutput() throws
Exception {
+ setupLocalShuffleSession(sv -> sv.enableShareHashTableForBroadcastJoin
= false);
+ connectContext.getSessionVariable().setPipelineTaskNum("3");
+ assertPlanShape(
+ "select sum(distinct a.k1) from test.t1 a "
+ + "left join test.t2 b on a.k1 = b.k1",
+ anyTree(
+ agg(
+ localExchange(PASS_TO_ONE_LE,
+ hashJoin(
+ localExchange(PT, olapScan()),
+ localExchange(BROADCAST_LE,
+
anyTree(exchange())))))));
+ }
+
@Test
public void testCountDistinctNoGroupByRequiresHashBeforeAgg() throws
Exception {
// count(distinct k2) without group-by: the finalize merge agg emits
per-instance
@@ -286,6 +304,18 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
anyTree(exchange())))));
}
+ @Test
+ public void testPrivateBroadcastJoinBuildUsesBroadcastLocalExchange()
throws Exception {
+ setupLocalShuffleSession(sv -> sv.enableShareHashTableForBroadcastJoin
= false);
+ assertPlanShape("select * from test.t1 a join [broadcast] test.t2 b on
a.k1=b.k1",
+ anyTree(
+ hashJoin(
+ localExchange(PT,
+ olapScan()),
+ localExchange(BROADCAST_LE,
+ anyTree(exchange())))));
+ }
+
@Test
public void testNlJoinPoolingShapeDsl() throws Exception {
// doc rule "NL join / 池化": build BROADCAST, probe
ADAPTIVE_PASSTHROUGH.
diff --git a/gensrc/thrift/Partitions.thrift b/gensrc/thrift/Partitions.thrift
index da172fac735..b14e36a3f62 100644
--- a/gensrc/thrift/Partitions.thrift
+++ b/gensrc/thrift/Partitions.thrift
@@ -128,10 +128,8 @@ enum TLocalPartitionType {
// Scan(build side) -> LocalExchangeNode(BROADCAST) -> HashJoin(build)
BROADCAST = 6,
// PASS_TO_ONE: funnel all rows to a single local instance (channel 0);
every other instance gets EOS
- // immediately and produces nothing (PassToOneExchanger). used for a
broadcast join with a shared
- // hash table, where only instance 0 needs the build data and the others
share its hash table.
- // NOTE: BE only uses PassToOneExchanger when
`enable_share_hash_table_for_broadcast_join` is on;
- // when it is off the same PASS_TO_ONE type degrades to BROADCAST (each
instance keeps its own copy).
+ // immediately and produces nothing (PassToOneExchanger). Used at
parallel-to-serial boundaries and
+ // for a broadcast join with a shared hash table. A private broadcast hash
table uses BROADCAST.
PASS_TO_ONE = 7,
// LOCAL_MERGE_SORT: k-way merge of several already-sorted local inputs into
one globally sorted
// stream on a single instance (paired with LocalMergeSortSourceOperator,
for a SortNode with
diff --git
a/regression-test/data/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.out
b/regression-test/data/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.out
new file mode 100644
index 00000000000..0ef93e2a601
--- /dev/null
+++
b/regression-test/data/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.out
@@ -0,0 +1,18 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !count_distinct_left_join --
+10
+
+-- !sum_and_count_distinct_left_join --
+23 5
+
+-- !sum_distinct_broad_predicate --
+45
+
+-- !sum_distinct_reversed_join --
+45
+
+-- !sum_distinct_multi_outer_join --
+35
+
+-- !native_private_broadcast_build --
+10
diff --git
a/regression-test/suites/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.groovy
b/regression-test/suites/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.groovy
new file mode 100644
index 00000000000..ebe9c6eed63
--- /dev/null
+++
b/regression-test/suites/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.groovy
@@ -0,0 +1,97 @@
+// 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_serial_aggregation_over_parallel_join") {
+ ["serial_agg_join_probe", "serial_agg_join_left",
"serial_agg_join_right"].each { table ->
+ sql "DROP TABLE IF EXISTS ${table}"
+ }
+
+ sql """CREATE TABLE serial_agg_join_probe (
+ col_bigint BIGINT, col_v10 VARCHAR(10), col_v64 VARCHAR(64),
pk INT
+ ) ENGINE=OLAP DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num"="1")"""
+ sql """CREATE TABLE serial_agg_join_left (
+ pk INT, col_bigint BIGINT, col_v10 VARCHAR(10), col_v64
VARCHAR(64)
+ ) ENGINE=OLAP DUPLICATE KEY(pk, col_bigint, col_v10)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10 PROPERTIES
("replication_num"="1")"""
+ sql """CREATE TABLE serial_agg_join_right (
+ pk INT, col_v10 VARCHAR(10), col_bigint BIGINT, col_v64
VARCHAR(64)
+ ) ENGINE=OLAP DUPLICATE KEY(pk, col_v10)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10 PROPERTIES
("replication_num"="1")"""
+
+ sql """INSERT INTO serial_agg_join_probe VALUES
+
(-94,'had','y',0),(672609,'k','h',1),(-3766684,'a','p',2),(5070261,'on','x',3),
+
(NULL,'u','at',4),(-86,'v','c',5),(21910,'how','m',6),(-63,'that''s','go',7),
+ (-8276281,'s','a',8),(-101,'w','y',9)"""
+ sql """INSERT INTO serial_agg_join_left VALUES
+
(0,NULL,'g','i'),(1,-6138328,'z','do'),(2,-23217,'g','about'),(3,104,'you''re','z'),
+
(4,NULL,'oh','i'),(5,-54,'want','to'),(6,NULL,'x','c'),(7,NULL,'you''re','come'),
+ (8,3447,'really','from'),(9,-5459,'i','will')"""
+ sql """INSERT INTO serial_agg_join_right VALUES
+
(0,'right',NULL,'g'),(1,'on',-486256,'on'),(2,'I''ll',-1,'at'),(3,'h',29263,'don''t'),
+
(4,'a',5453,'s'),(5,'j',-119,'can''t'),(6,'one',89,'n'),(7,'s',-7227,'u'),
+ (8,'time',94,'b'),(9,'yes',1816630,'yes')"""
+
+ def variables =
"enable_local_shuffle_planner=true,enable_local_shuffle=true," +
+
"enable_bucket_shuffle_join=true,ignore_storage_data_distribution=true," +
+ "bucket_shuffle_downgrade_ratio=0.8,use_serial_exchange=false," +
+ "parallel_pipeline_task_num=3,enable_sql_cache=false," +
+ "enable_share_hash_table_for_broadcast_join=false"
+
+ order_qt_count_distinct_left_join """SELECT /*+SET_VAR(${variables})*/
COUNT(DISTINCT t1.pk)
+ FROM serial_agg_join_left t1 LEFT JOIN serial_agg_join_probe t2 ON
t2.pk=t1.pk
+ WHERE (t1.col_v64>'FVjnKolDTt' AND t1.col_v64<='z') OR t1.col_v64
IS NULL
+ OR (t1.col_v10>'me' AND t1.col_v10<='zzzz' AND t1.col_bigint
BETWEEN 3 AND 7)"""
+
+ order_qt_sum_and_count_distinct_left_join """SELECT
/*+SET_VAR(${variables})*/
+ SUM(DISTINCT t1.pk), COUNT(DISTINCT t1.pk)
+ FROM serial_agg_join_right t1 LEFT JOIN serial_agg_join_probe t2
ON t2.pk=t1.pk
+ WHERE t1.pk IN (2,9) OR t1.col_bigint IN (1,8)
+ OR (t1.col_v64>='MijtyYyxeA' AND t1.col_v64<'z'
+ AND t1.col_v64>='on' AND t1.col_v64<'zzzz')"""
+
+ order_qt_sum_distinct_broad_predicate """SELECT /*+SET_VAR(${variables})*/
SUM(DISTINCT t1.pk)
+ FROM serial_agg_join_right t1 LEFT JOIN serial_agg_join_probe t2
ON t2.pk=t1.pk
+ WHERE (t1.col_v64>='QXQpaZhWfj' AND t1.col_v64<'z')
+ OR (t1.col_v64>='fvPsFBZelL' AND t1.col_v64<='well')
+ OR (t1.pk BETWEEN 0 AND 15 AND t1.col_v10 LIKE 'a%')
+ OR (t1.pk>=3 AND t1.pk<4) OR t1.pk BETWEEN 0 AND 100 OR
(t1.pk>7 AND t1.pk<=9)"""
+
+ order_qt_sum_distinct_reversed_join """SELECT /*+SET_VAR(${variables})*/
SUM(DISTINCT t1.pk)
+ FROM serial_agg_join_probe t1 LEFT JOIN serial_agg_join_right t2
ON t1.pk=t2.pk
+ WHERE (t1.pk IS NOT NULL AND t1.pk IN (3,8,2,2)
+ AND t1.col_v64 IN
('didn''t','when','a','come','AgpEFIOTAN'))
+ OR (t1.col_v64>'HoatMBMEwP' AND t1.col_v64<='zzzz') OR t1.pk
BETWEEN 6 AND 11
+ OR (t1.pk IS NULL AND t1.pk IN (5)) OR (t1.pk<=t1.col_bigint
AND t1.pk IN (8))"""
+
+ order_qt_sum_distinct_multi_outer_join """SELECT
/*+SET_VAR(${variables})*/ SUM(DISTINCT t1.pk)
+ FROM serial_agg_join_right t1 RIGHT OUTER JOIN
serial_agg_join_probe t2 ON t2.pk=t2.pk
+ LEFT JOIN serial_agg_join_left t3 ON t3.pk=t1.pk
+ WHERE (t1.col_v10>'jHKKlhlHDn' AND t1.col_v10<'z'
+ AND t1.col_v10 NOT IN ('him','you''re'))
+ OR (t1.col_v64>='j' AND t1.col_v64<='y')
+ OR (t1.col_v10 NOT BETWEEN 'rxpMJWfBRX' AND 'z' AND
t1.col_bigint IN (1000)
+ AND t1.col_bigint IS NULL AND t1.col_bigint BETWEEN 6 AND
15)"""
+
+ def nativeVariables =
"enable_local_shuffle_planner=false,enable_local_shuffle=true," +
+ "parallel_pipeline_task_num=3,enable_sql_cache=false," +
+ "enable_share_hash_table_for_broadcast_join=false"
+
+ order_qt_native_private_broadcast_build """SELECT
/*+SET_VAR(${nativeVariables})*/ COUNT(t2.pk)
+ FROM serial_agg_join_right t1 INNER JOIN [broadcast]
serial_agg_join_probe t2 ON t2.pk=t1.pk
+ WHERE t1.pk BETWEEN 0 AND 9"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
