Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
|
Superseded by #68199, which backports the whole local shuffle series as one PR. |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Static review of the pinned head 392a6c4 against base c41194b found four blocking issues.
- The BE cannot compile because AdaptivePassthroughExchanger initializes an unrelated Exchanger specialization.
- The changed BE aggregation test still uses the removed ExchangeType enum.
- The FE cannot compile because SetOperationNode references an undeclared distributionMode field.
- FE aggregation planning can insert PASSTHROUGH LocalExchange nodes for serial non-ScanNode children, splitting the AggSink/AggSource pipeline and reproducing the empty-source-deps crash/hang documented by the added Bug 21 regression.
The supplied review focus has no additional focus points. This is a static review only; no builds or tests were run per the review contract.
There was a problem hiding this comment.
Corrected exact-head static review for 392a6c4 against c41194b. The review found four blocking issues.
- The BE cannot compile because AdaptivePassthroughExchanger initializes an unrelated Exchanger specialization.
- The changed BE aggregation test still uses the removed ExchangeType enum.
- The FE cannot compile because SetOperationNode references an undeclared distributionMode field.
- FE aggregation planning can insert PASSTHROUGH LocalExchange nodes for serial non-ScanNode children, splitting the AggSink/AggSource pipeline and reproducing the empty-source-deps crash/hang documented by the added Bug 21 regression.
The supplied review focus has no additional focus points. This is a static review only; no builds or tests were run per the review contract.
| int free_block_limit) | ||
| : Exchanger<BlockWrapperSPtr>(running_sink_operators, num_partitions, | ||
| free_block_limit) { | ||
| : Exchanger<PartitionedBlock>(running_sink_operators, num_partitions, free_block_limit, |
There was a problem hiding this comment.
AdaptivePassthroughExchanger still derives from Exchanger<BlockWrapperSPtr>, but this constructor now initializes Exchanger<PartitionedBlock>. That specialization is not a direct base, so this is a hard C++ compile error (and the implementation still consumes BlockWrapperSPtr). Please initialize Exchanger<BlockWrapperSPtr> here.
| sink_op->_needs_finalize = false; | ||
| OperatorPtr child = | ||
| std::make_shared<MockDistributionOperator>(ExchangeType::ADAPTIVE_PASSTHROUGH); | ||
| std::make_shared<MockDistributionOperator>(TLocalPartitionType::ADAPTIVE_PASSTHROUGH); |
There was a problem hiding this comment.
This test still uses the deleted ExchangeType at lines 146, 151, and 156, but this PR removes that enum from dependency.h and uses TLocalPartitionType::type instead. The test target therefore fails to compile; replace these references with the corresponding TLocalPartitionType values.
| } | ||
|
|
||
| public boolean isBucketShuffle() { | ||
| return distributionMode.equals(DistributionMode.BUCKET_SHUFFLE); |
There was a problem hiding this comment.
SetOperationNode.isBucketShuffle() uses distributionMode, but this class (and PlanNode) declares no such field; the only analogous state is HashJoinNode.distrMode. This unresolved symbol prevents the FE module from compiling. Please remove/fix this method or add the correctly initialized set-operation state.
|
|
||
| /** BE base class required_data_distribution: serial child → PASSTHROUGH, else → NOOP. */ | ||
| private LocalExchangeTypeRequire baseClassRequire(ConnectContext connectContext) { | ||
| return children.get(0).isSerialOperatorOnBe(connectContext) |
There was a problem hiding this comment.
This helper turns every serial child into a PASSTHROUGH requirement, including serial Exchange/Agg children. That causes FE to insert a PASSTHROUGH LocalExchange between the serial UNPARTITIONED Exchange/Agg and the streaming/distinct pre-agg, splitting the coupled AggSink/AggSource pipeline (the added Bug 21 regression documents the resulting empty-source-deps crash/hang). Restrict this fallback to the safe ScanNode case (or mirror the BE serial-ancestor boundary check) before inserting the LocalExchange.
2 participants
Cherry-picked from #63366