924060929 commented on code in PR #67177:
URL: https://github.com/apache/doris/pull/67177#discussion_r3870549608
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java:
##########
@@ -287,6 +316,50 @@ List<AssignedJob>
filterInstancesWhichCanReceiveDataFromRemote(
}
}
+ private boolean hasSerialReceiverPipeline(
+ PipelineDistributedPlan receiverPlan, ExchangeNode linkNode) {
+ ConnectContext connectContext = statementContext.getConnectContext();
+ if (linkNode.isSerialOperatorOnBe(connectContext)) {
+ return true;
+ }
+
+ // A non-serial Exchange does not necessarily create one receiver per
fragment
+ // instance. Build-side pipelines inherit the task count of their
parent pipeline
+ // on BE. A serial ancestor (for example, a scalar finalize
aggregation) can reduce
+ // that parent pipeline to one task per worker, and the Exchange build
pipeline
+ // inherits the same single task. Addressing the other logical
instances would
+ // send bucket rows to receivers that are never created.
+ //
+ // A LocalExchange on the path is a pipeline boundary which restores
the upstream
+ // pipeline to all fragment instances. Only serial operators after the
last such
+ // boundary can reduce the target Exchange's receiver count.
+ PlanNode receiverRoot =
receiverPlan.getFragmentJob().getFragment().getPlanRoot();
+ Optional<Boolean> serialPipeline = findSerialPipelineToExchange(
+ receiverRoot, linkNode, false, connectContext);
+ return serialPipeline.orElseThrow(() -> new IllegalStateException(
+ "Can not find receiver exchange " + linkNode.getId()
+ + " in fragment " + receiverRoot.getFragmentId()));
+ }
+
+ private Optional<Boolean> findSerialPipelineToExchange(
Review Comment:
Addressed by the final refactor. The receiver-filtering DFS and
`findSerialPipelineToExchange` logic were removed from the current diff. The
transition is detected once during the normal recursive
`PlanNode.enforceRequire` traversal, so this repeated subtree scan no longer
exists.
--
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]
