diff --git a/be/src/exec/operator/hashjoin_build_sink.h b/be/src/exec/operator/hashjoin_build_sink.h index ea8e53a872681f..75bdd109998739 100644 --- a/be/src/exec/operator/hashjoin_build_sink.h +++ b/be/src/exec/operator/hashjoin_build_sink.h @@ -133,12 +133,16 @@ class HashJoinBuildSinkOperatorX MOCK_REMOVE(final) ._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 bccb92fbf5dddd..14ef0fd572e04f 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp @@ -716,6 +716,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(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 @@ -748,34 +749,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: @@ -852,11 +848,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(ops.front().get()); + auto* le_source = + !ops.empty() ? dynamic_cast(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); @@ -1069,22 +1071,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(_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(_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( + _runtime_state->query_options().local_exchange_free_blocks_limit) + : 0); break; case TLocalPartitionType::ADAPTIVE_PASSTHROUGH: shared_state->exchanger = AdaptivePassthroughExchanger::create_unique( @@ -1987,9 +1979,12 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo } case TPlanNodeType::LOCAL_EXCHANGE_NODE: { op = std::make_shared(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 @@ -1998,14 +1993,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. @@ -2017,7 +2021,6 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo _dag[downstream_pipeline_id].push_back(cur_pipe->id()); int num_partitions = 0; std::map 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; @@ -2057,9 +2060,10 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo ? cast_set( _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 38de2389352343..b779dd580e53e3 100644 --- a/be/test/exec/operator/hashjoin_build_sink_test.cpp +++ b/be/test/exec/operator/hashjoin_build_sink_test.cpp @@ -266,6 +266,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& key_types, const std::vector& left_nullables, @@ -358,4 +382,4 @@ TEST_F(HashJoinBuildSinkTest, Terminate) { run_test_block(test_block); } -} // 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 4245d87bf3824d..f54a903e263c35 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 { public: LocalExchangerTest() = default; ~LocalExchangerTest() override = default; @@ -526,9 +528,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( + _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; @@ -549,6 +602,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)); @@ -579,10 +634,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(); @@ -735,6 +789,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; @@ -1386,4 +1442,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 c16b62010868e2..62c54761e8c4fb 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 @@ -253,17 +253,19 @@ public Pair enforceAndDeriveLocalExchange(PlanTrans 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 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 4ae1d480b5fb67..b2700a05af9f8e 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 @@ -434,7 +434,11 @@ public Pair enforceAndDeriveLocalExchange( ? 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 18556071315bd4..c2b4e316139a8f 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 @@ -1216,8 +1216,14 @@ protected Pair enforceRequire( // 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 - || isSerialOperatorOnBe(translatorContext.getConnectContext()); + boolean selfSerial = isSerialOperatorOnBe(translatorContext.getConnectContext()); + 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; translatorContext.setHasSerialAncestorInPipeline(child, childHasSerialAncestor); // 1b. Propagate shuffle-for-correctness-ancestor flag to child. @@ -1239,6 +1245,18 @@ protected Pair enforceRequire( Pair 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 59c26166c3f51b..e8bcbb195aa1d3 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 void testLayer1SkipUsesIsSerialOperatorOnBeNotIsSerialNode() { + "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 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 void testSortNodeBranches() { // 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 scanOutput = scanSort.enforceAndDeriveLocalExchange( ctx, null, LocalExchangeTypeRequire.noRequire()); // Non-merge, non-analytic SortNode: isSerialNode()=true, requireChild=noRequire, @@ -809,6 +831,24 @@ public void testAnalyticEvalNodeBranches() { 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 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 void testAnalyticEvalNodeBranches() { 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 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 String getNodeExplainString(String prefix, TExplainLevel detailLevel) { } } + 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 enforceAndDeriveLocalExchange( + PlanTranslatorContext translatorContext, PlanNode parent, + LocalExchangeTypeRequire parentRequire) { + Pair 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 enforceAndDeriveLocalExchange( + PlanTranslatorContext translatorContext, PlanNode parent, + LocalExchangeTypeRequire parentRequire) { + Pair 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 d691028666f3c4..7fead508744382 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 @@ protected void setupLocalShuffleSession(java.util.function.Consumer 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 void testBroadcastJoinPoolingShapeDsl() throws Exception { 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 9b57a7078f76f6..5cff33eaa1e70e 100644 --- a/gensrc/thrift/Partitions.thrift +++ b/gensrc/thrift/Partitions.thrift @@ -129,10 +129,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 00000000000000..0ef93e2a60108e --- /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 00000000000000..ebe9c6eed63190 --- /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""" +}