github-actions[bot] commented on code in PR #68184:
URL: https://github.com/apache/doris/pull/68184#discussion_r4045705089
##########
be/src/exec/exchange/local_exchanger.h:
##########
@@ -357,16 +358,15 @@ class AdaptivePassthroughExchanger : public
Exchanger<BlockWrapperSPtr> {
ENABLE_FACTORY_CREATOR(AdaptivePassthroughExchanger);
AdaptivePassthroughExchanger(int running_sink_operators, int
num_partitions,
int free_block_limit)
- : Exchanger<BlockWrapperSPtr>(running_sink_operators,
num_partitions,
- free_block_limit) {
+ : Exchanger<PartitionedBlock>(running_sink_operators,
num_partitions, free_block_limit,
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.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java:
##########
@@ -194,4 +200,63 @@ public int getNumInstances() {
numInstances = Math.max(1, numInstances);
return numInstances;
}
+
+ public boolean isBucketShuffle() {
+ return distributionMode.equals(DistributionMode.BUCKET_SHUFFLE);
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.
##########
be/test/exec/operator/agg_operator_test.cpp:
##########
@@ -130,11 +131,11 @@ TEST(AggOperatorRequiredDistributionTest,
require_hash_shuffle_after_non_hash_ch
sink_op->_partition_exprs.emplace_back();
sink_op->_needs_finalize = false;
OperatorPtr child =
-
std::make_shared<MockDistributionOperator>(ExchangeType::ADAPTIVE_PASSTHROUGH);
+
std::make_shared<MockDistributionOperator>(TLocalPartitionType::ADAPTIVE_PASSTHROUGH);
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/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`
+ // early return also catches FIRST_MERGE, dropping the HASH
requirement and
+ // causing wrong-result (e.g. PASSTHROUGH over serial child breaks
the
+ // group-by-key invariant — DORIS-25413).
+ if (!hasKeys) {
+ requireChild = needsFinalize
+ ? LocalExchangeTypeRequire.noRequire()
+ : baseClassRequire(connectContext);
+ } else if (!needsFinalize && !aggInfo.isMerge() &&
!enableLeBeforeAgg) {
+ // LOCAL phase (FIRST preagg / SECOND distinct local) + user
opted out
+ // of pre-agg LE → base class decides: serial child →
PASSTHROUGH
+ // (parallelism), non-serial child → NOOP (no LE).
+ requireChild = baseClassRequire(connectContext);
+ } else if (!needsFinalize || AddLocalExchange.isColocated(this)) {
+ // FIRST_MERGE (correctness) or finalize+colocate → HASH.
+ requireChild = parentRequire.autoRequireHash();
+ } else if (hasPartitionExprs(parentRequire)) {
+ // FE-only heuristic: finalize non-colocate with parent hash
requirement
+ // → inherit parent's specific hash type.
+ requireChild = parentRequire.autoRequireHash();
+ } else {
+ // FE-only heuristic: finalize non-colocate without parent
hash → skip
+ // LE (child Exchange already provides hash distribution).
+ requireChild = LocalExchangeTypeRequire.noRequire();
+ }
+ }
+
+ Pair<PlanNode, LocalExchangeType> enforceResult
+ = enforceRequire(translatorContext, children.get(0), 0,
requireChild);
+ children = Lists.newArrayList(enforceResult.first);
+ return Pair.of(this, enforceResult.second);
+ }
+
+ /** BE base class required_data_distribution: serial child → PASSTHROUGH,
else → NOOP. */
+ private LocalExchangeTypeRequire baseClassRequire(ConnectContext
connectContext) {
+ return children.get(0).isSerialOperatorOnBe(connectContext)
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]
