924060929 commented on code in PR #67177:
URL: https://github.com/apache/doris/pull/67177#discussion_r3868412933
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java:
##########
@@ -278,7 +280,7 @@ List<AssignedJob>
filterInstancesWhichCanReceiveDataFromRemote(
// decoupled an exchange's serial flag from the fragment's serial
scan, a local-shuffle
// fragment can host a non-serial RANDOM/HASH exchange, and only the
BUCKET_SHUFFLE path
// re-spreads its destinations (see getDestinationsByBuckets).
- if
(linkNode.isSerialOperatorOnBe(statementContext.getConnectContext())) {
+ if (hasSerialReceiverPipeline(receiverPlan, linkNode)) {
Review Comment:
Fixed in a21d33a95ea. FE local-exchange rewriting now runs after instance
assignment and before DistributePlanner.linkPlans(), so PASS_TO_ONE is present
when remote receiver destinations are selected. The receiver-selection test
covers the serial-root boundary, and all five reported RQG cases plus the
complete nereids_p0/local_shuffle suite pass on a four-BE cluster.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -769,13 +769,8 @@ Status
PipelineFragmentContext::_create_deferred_local_exchangers() {
sender_count, _num_instances, 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(
Review Comment:
Fixed in a21d33a95ea. Execution version 14 introduces unconditional
PASS_TO_ONE semantics; version 13 preserves the legacy PASS_TO_ONE/BROADCAST
interpretation on both FE and BE. New FE-to-old-BE queries force the compatible
shared-build semantic only when FE local shuffle is actually planned, and
destination linking uses the same effective value. Plan type, destinations, and
Coordinator options also share one query-level execution-version snapshot.
Tests cover v13/v14, share=false, local-shuffle disabled, and a global version
change after planning.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2058,14 +2050,22 @@ 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 task, while the
upstream
+ // pipeline is expanded below so every remote receiver remains live.
+ if (partition_type != TLocalPartitionType::PASS_TO_ONE) {
Review Comment:
Fixed in a21d33a95ea. Deferred exchanger construction now carries the actual
downstream pipeline and sizes the exchanger source count, source dependencies,
and memory counters from downstream_pipe->num_tasks(), with consistency checks
before construction. The upstream sender count remains tied to the final
upstream pipeline task count. Added a 4-sender/1-source PASS_TO_ONE lifecycle
and diagnostic test; the full BE build passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
