github-actions[bot] commented on code in PR #68184:
URL: https://github.com/apache/doris/pull/68184#discussion_r4045694612
##########
be/src/exec/exchange/local_exchanger.h:
##########
@@ -142,7 +142,7 @@ class ExchangerBase {
SourceInfo&& source_info) = 0;
virtual Status sink(RuntimeState* state, Block* in_block, bool eos,
Profile&& profile,
SinkInfo& sink_info) = 0;
- virtual ExchangeType get_type() const = 0;
+ virtual TLocalPartitionType::type get_type() const = 0;
// Called if a local exchanger source operator are closed. Free the unused
data block in data_queue.
Review 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.
##########
be/test/exec/operator/agg_operator_test.cpp:
##########
@@ -96,7 +96,8 @@ struct MockAggSourceOperator : public AggSourceOperatorX {
class MockDistributionOperator final : public OperatorX<MockLocalState> {
public:
- MockDistributionOperator(ExchangeType exchange_type) :
_exchange_type(exchange_type) {}
+ MockDistributionOperator(TLocalPartitionType::type exchange_type)
Review 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.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java:
##########
@@ -19,6 +19,11 @@
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.TupleId;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
+import org.apache.doris.planner.HashJoinNode.DistributionMode;
Review 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.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java:
##########
@@ -282,4 +292,160 @@ public boolean isQueryCacheCandidate() {
public void setQueryCacheCandidate(boolean queryCacheCandidate) {
this.queryCacheCandidate = queryCacheCandidate;
}
+
+ @Override
+ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
+ PlanTranslatorContext translatorContext, PlanNode parent,
LocalExchangeTypeRequire parentRequire) {
+
+ ConnectContext connectContext = translatorContext.getConnectContext();
+ SessionVariable sessionVariable = connectContext.getSessionVariable();
+ // PR #62438: when false, non-finalize agg falls back to BE base class.
+ boolean enableLeBeforeAgg =
sessionVariable.enableLocalExchangeBeforeAgg;
+ boolean hasKeys = !aggInfo.getGroupingExprs().isEmpty();
+
+ // Each branch mirrors the corresponding BE operator's
required_data_distribution()
+ // check order 1:1. The helper baseClassRequire() expands BE's base
class behavior.
+ LocalExchangeTypeRequire requireChild;
+ if (canUseDistinctStreamingAgg(sessionVariable)) {
+ // DistinctStreamingAggOperatorX. Two flavors share this operator
class:
+ // - streaming preagg (useStreamingPreagg=true):
performance-only,
+ // flag controls
+ // - non-streaming dedup (useStreamingPreagg=false):
correctness-required,
+ // always HASH regardless of flag
+ // Diverges from BE: BE's `!_needs_finalize &&
!enable_local_exchange_before_agg`
+ // early return catches non-streaming dedup too, causing the same
family of
+ // wrong-result bug as AggSink (DORIS-25413).
+ if (needsFinalize && !hasKeys) {
+ requireChild = LocalExchangeTypeRequire.noRequire();
+ } else if (!needsFinalize && useStreamingPreagg &&
!enableLeBeforeAgg) {
+ requireChild = baseClassRequire(connectContext);
+ } else if (needsFinalize || (hasKeys && !useStreamingPreagg)) {
+ requireChild = AddLocalExchange.isColocated(this)
+ ? LocalExchangeTypeRequire.requireHash()
+ : parentRequire.autoRequireHash();
+ } else if
(sessionVariable.enableDistinctStreamingAggForcePassthrough) {
+ requireChild = LocalExchangeTypeRequire.requirePassthrough();
+ } else {
+ requireChild = baseClassRequire(connectContext);
+ }
+ } else if (useStreamingPreagg) {
+ // StreamingAggOperatorX
+ if (children.get(0) instanceof HashJoinNode
+ &&
sessionVariable.enableStreamingAggHashJoinForcePassthrough) {
+ requireChild = LocalExchangeTypeRequire.requirePassthrough();
+ } else if (!needsFinalize && !enableLeBeforeAgg) {
+ requireChild = baseClassRequire(connectContext);
+ } else if (!hasKeys) {
+ requireChild = needsFinalize
+ ? LocalExchangeTypeRequire.noRequire()
+ : baseClassRequire(connectContext);
+ } else {
+ requireChild = LocalExchangeTypeRequire.requireHash();
+ }
+ } else {
+ // AggSinkOperatorX — covers finalize phase AND non-finalize
phases (LOCAL
+ // preagg / FIRST_MERGE dedup). Streaming preagg goes through the
StreamingAgg
+ // branch above, not here.
+ //
+ // Phase semantics for !needsFinalize:
+ // - FIRST / SECOND (LOCAL phase, !isMerge): performance-only,
flag controls
+ // - FIRST_MERGE (correctness-required): always HASH regardless
of flag
+ //
+ // Diverges from BE here: BE's `!_needs_finalize &&
!enable_local_exchange_before_agg`
Review 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.
--
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]
