github-actions[bot] commented on code in PR #67177:
URL: https://github.com/apache/doris/pull/67177#discussion_r3863990520
##########
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:
[P1] Keep the existing wire value compatible across rolling upgrades. Before
this change, FE used `PASS_TO_ONE` for a serial broadcast build regardless of
the share-hash flag, and BE interpreted it as `BROADCAST` when sharing was
disabled. An old FE talking to this new BE now feeds only task 0 even though
every probe task owns a private hash table; in the reverse combination, an old
BE can broadcast the new generic scalar-gather `PASS_TO_ONE`. Since the thrift
enum is unchanged and there is no capability/version gate, mixed versions can
return wrong rows. Preserve the legacy interpretation and add a new explicit
gather type, or negotiate/gate the new semantics on both sides.
##########
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:
[P1] Recompute destinations after inserting the serial-boundary exchange.
`DistributePlanner.plan()` freezes these links before
`NereidsPlanner.addLocalExchangeAfterDistribute()` runs, so this walk cannot
see the new `PASS_TO_ONE`. For `scalar DISTINCT -> shuffle HashJoin -> HASH
Exchange`, this stores only the first instance per worker; the later rewrite
inserts `PASS_TO_ONE`, and BE expands the Exchange pipeline to
`_num_instances`. The non-first HASH/RANDOM receivers are not bucket orphans
and wait for senders/EOS that never target them. The new unit test constructs a
LocalExchange before filtering, which is the reverse of production order.
Please move the rewrite before final linking or recompute the links afterward,
and cover the real end-to-end ordering.
##########
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:
[P1] Make the `PASS_TO_ONE` source cardinality match this one-task
downstream. Task creation now instantiates only source task 0, but deferred
construction still creates `PassToOneExchanger(sender_count, _num_instances,
...)` and keeps `_num_instances` source counters/dependencies. Only
`mem_counters[0]` is initialized and only one close arrives, so
`_running_source_operators` never reaches zero/finalize; rendering the live
pipeline diagnostic then dereferences the null counters for tasks 1..N-1 and
can crash BE. Carry the downstream task count into deferred construction (one
here, N for the shared-broadcast shape), assert it matches created source
states, and add a one-source/N-sender lifecycle/diagnostic test.
--
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]
