github-actions[bot] commented on code in PR #67433:
URL: https://github.com/apache/doris/pull/67433#discussion_r4069572919
##########
fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java:
##########
@@ -206,6 +206,19 @@ private Supplier<Boolean> buildHasBucketShuffleNode() {
return true;
}
}
+ // A fused group join is a PlanNode in its own right, it does not
extend HashJoinNode, so
+ // the two loops above cannot see the bucket shuffle of a fused
INNER join. Missing it
+ // made the scan job of the bucket side an
UnassignedScanSingleOlapTableJob while the
+ // exchange into the join still claimed
BUCKET_SHFFULE_HASH_PARTITIONED, and
+ // DistributePlanner.getDestinationsByBuckets then failed to cast
that job to
+ // UnassignedScanBucketOlapTableJob.
+ List<GroupJoinNode> groupJoinNodes
Review Comment:
[P1] Please update the legacy Coordinator's bucket-shuffle detector as well.
This fragment predicate is consumed by the new `DistributePlanner`, but
`enable_nereids_distribute_planner=false` still selects
`Coordinator.BucketShuffleJoinController`, which recognizes only
`HashJoinNode`. A fused `BUCKET_SHUFFLE` GroupJoin then gets generic scan
assignment and later fails the bucket-shuffle sink assertion (or loses
bucket-aligned instances). Reuse this predicate there and run the
bucket-shuffle regression with the legacy distributor.
##########
be/src/exec/operator/groupjoin_build_sink.cpp:
##########
@@ -0,0 +1,281 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exec/operator/groupjoin_build_sink.h"
+
+#include <variant>
+
+#include "common/cast_set.h"
+#include "core/data_type/data_type_nullable.h"
+#include "exec/common/hash_table/hash_map_util.h"
+#include "exec/common/util.hpp"
+#include "exec/operator/groupjoin_operator_utils.h"
+#include "exprs/vectorized_agg_fn.h"
+#include "exprs/vexpr.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
+
+namespace doris {
+
+GroupJoinBuildSinkLocalState::GroupJoinBuildSinkLocalState(DataSinkOperatorXBase*
parent,
+ RuntimeState* state)
+ : Base(parent, state) {
+ _finish_dependency = std::make_shared<CountedFinishDependency>(
+ parent->operator_id(), parent->node_id(), parent->get_name() +
"_FINISH_DEPENDENCY");
+}
+
+Status GroupJoinBuildSinkLocalState::init(RuntimeState* state,
LocalSinkStateInfo& info) {
+ RETURN_IF_ERROR(Base::init(state, info));
+ _shared_state->memory_used_counter = _memory_used_counter;
+ auto& p = _parent->cast<GroupJoinBuildSinkOperatorX>();
+ _build_expr_ctxs.resize(p._build_expr_ctxs.size());
+ for (size_t i = 0; i < _build_expr_ctxs.size(); ++i) {
+ RETURN_IF_ERROR(p._build_expr_ctxs[i]->clone(state,
_build_expr_ctxs[i]));
+ }
+ _aggregate_evaluators.reserve(p._aggregate_evaluators.size());
+ for (auto* evaluator : p._aggregate_evaluators) {
+ _aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
+ }
+ RETURN_IF_ERROR(groupjoin::register_agg_state_layout(
+ _shared_state, p._aggregate_sides, p._sizes_of_aggregate_states,
+ p._aligns_of_aggregate_states, p._aggregate_indices,
p._aggregate_evaluators));
Review Comment:
[P1] Register the local evaluator clones in the shared layout (and make the
same change on the probe side), or reject Java UDAFs from fusion. The layout
uses these pointers to create/finalize states, while row updates use
`local_state._aggregate_evaluators`. `AggFnEvaluator::clone` creates a fresh
`AggregateJavaUdaf`, and only the instance whose `create()` runs initializes
`_exec_place`. A fused Java UDAF therefore creates through the parent evaluator
but adds through a clone whose `_exec_place` is null.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -3305,6 +3323,281 @@ private PlanFragment connectJoinNode(HashJoinNode
hashJoinNode, PlanFragment lef
return leftFragment;
}
+ private PlanFragment connectGroupJoinNode(GroupJoinNode groupJoinNode,
PlanFragment leftFragment,
+ PlanFragment rightFragment, PlanTranslatorContext context,
AbstractPlan groupJoin) {
+ groupJoinNode.setChild(0, leftFragment.getPlanRoot());
+ groupJoinNode.setChild(1, rightFragment.getPlanRoot());
+ setPlanRoot(leftFragment, groupJoinNode, groupJoin);
+ context.mergePlanFragment(rightFragment, leftFragment);
+ for (PlanFragment rightChild : rightFragment.getChildren()) {
+ leftFragment.addChild(rightChild);
+ }
+ return leftFragment;
+ }
+
+ /**
+ * V2: Try to fuse HashAggregate(HashJoin) into GroupJoin directly in the
translator stage.
+ * <p>
+ * When the aggregate's child is an INNER hash join with compatible
group-by/join-key,
+ * generate a GroupJoinNode instead of separate AggregationNode +
HashJoinNode.
+ * Returns null if conditions are not met — caller falls through to normal
translation.
+ */
+ private PlanFragment maybeTranslateToGroupJoin(
+ Plan aggregate,
+ PlanTranslatorContext context) {
+ // Gate: session variable
+ ConnectContext connectContext = ConnectContext.get();
+ if (connectContext == null
+ ||
!connectContext.getSessionVariable().isEnableGroupJoinFusion()) {
+ return null;
+ }
+ // Gate: spill not supported
+ if (connectContext.getSessionVariable().enableSpill) {
+ return null;
+ }
+
+ // Child must be PhysicalHashJoin (optionally through one
pure-passthrough
+ // PhysicalProject; see the project gate in GroupJoinFusionUtils).
+ Plan child = aggregate.child(0);
+ PhysicalProject<?> project = null;
+ if (child instanceof PhysicalProject) {
+ project = (PhysicalProject<?>) child;
+ child = child.child(0);
+ }
+ if (!(child instanceof PhysicalHashJoin)) {
+ return null;
+ }
+ PhysicalHashJoin<?, ?> join = (PhysicalHashJoin<?, ?>) child;
+
+ // Full fusion eligibility (join shape + aggregate constraints) is
decided by
+ // GroupJoinFusionUtils.alignedConjunctsForGroupJoin, the single
source of truth shared
+ // with the AlignGroupJoinConjunctOrder post-processor that pre-aligns
the child join's
+ // conjunct order with the group-by keys. The translator emits the
join's conjuncts as
+ // they are, so fusion additionally requires the conjuncts to be
listed in exactly the
+ // group-by order: the processor guarantees this for every eligible
shape, and anything
+ // that is not aligned stays on the regular HashJoinNode +
AggregationNode path.
+ List<Expression> alignedConjuncts =
GroupJoinFusionUtils.alignedConjunctsForGroupJoin(
+ (Aggregate<?>) aggregate, project, join);
+ if (alignedConjuncts == null
+ || !GroupJoinFusionUtils.sameConjunctOrder(
+ alignedConjuncts, join.getHashJoinConjuncts())) {
+ return null;
+ }
+
+ // All checks passed — generate GroupJoinNode
+ return translateToGroupJoinNode((Aggregate<?>) aggregate, join,
context);
+ }
+
+ /** Translate Aggregate(HashJoin) pattern into a GroupJoinNode fragment. */
+ private PlanFragment translateToGroupJoinNode(
+ Aggregate<?> aggregate,
+ PhysicalHashJoin<?, ?> join,
+ PlanTranslatorContext context) {
+ PhysicalHashJoin<PhysicalPlan, PhysicalPlan> physicalJoin
+ = (PhysicalHashJoin<PhysicalPlan, PhysicalPlan>) join;
+
+ // maybeTranslateToGroupJoin only lets joins without residual
conjuncts reach this
+ // point. Enforce it here as well: the fused operator has no per-pair
filtering, so
+ // a residual conjunct that slipped through would be silently dropped
and produce
+ // wrong aggregation results (the failure mode this guard exists for).
+ Preconditions.checkState(join.getOtherJoinConjuncts().isEmpty(),
+ "GroupJoin fusion requires the join to have no residual
conjuncts, got: %s",
+ join.getOtherJoinConjuncts());
+
+ // maybeTranslateToGroupJoin also only lets aggregates whose inputs
are all directly
+ // produced by the join children reach this point. Enforce it here as
well: aggregate
+ // arguments that reference slots computed by an intermediate Project
between the
+ // aggregate and the join (e.g. hoisted type-coercion casts of binary
aggregates) do
+ // not exist on the join children, and translating them would abort
fragment
+ // serialization with an NPE in GroupJoinNode.toThrift.
+ Set<Slot> joinChildrenOutputs = Sets.newHashSet();
+ joinChildrenOutputs.addAll(join.left().getOutputSet());
+ joinChildrenOutputs.addAll(join.right().getOutputSet());
+
Preconditions.checkState(joinChildrenOutputs.containsAll(aggregate.getInputSlots()),
+ "GroupJoin fusion requires all aggregate inputs to be
join-child outputs, got: %s",
+ aggregate.getInputSlots());
+
+ // Visit children right-to-left (right = build, left = probe).
+ // connectGroupJoinNode merges both child fragments into this one,
exactly like
+ // connectJoinNode does for a plain hash join, so the children must be
translated under
+ // the same fragment-merge context. Without it a child aggregate could
take the
+ // bucketed-aggregation path (shouldUseBucketedFusion), which drops
the exchange that
+ // keeps its olap scan in its own fragment and leaves two olap scans
inside the
+ // GroupJoin fragment: the scan-assignment job then rejects the
fragment with
+ // "Not supported multiple scan multiple OlapTable but not contains
colocate join or
+ // bucket shuffle join". The GroupJoin would also lose the hash
distribution its
+ // PARTITIONED input relies on, since that exchange is what enforces
it.
+ context.enterFragmentMergeChild();
+ PlanFragment rightFragment;
+ PlanFragment leftFragment;
+ try {
+ rightFragment = join.child(1).accept(this, context);
+ leftFragment = join.child(0).accept(this, context);
+ } finally {
+ context.exitFragmentMergeChild();
+ }
+ PlanNode leftPlanRoot = leftFragment.getPlanRoot();
+ PlanNode rightPlanRoot = rightFragment.getPlanRoot();
+
+ // Create GroupJoinNode
+ GroupJoinNode groupJoinNode = new GroupJoinNode(
+ context.nextPlanNodeId(), leftPlanRoot, rightPlanRoot);
+ groupJoinNode.setNereidsId(join.getId());
+ context.getNereidsIdToPlanNodeIdMap().put(join.getId(),
groupJoinNode.getId());
+
+ // Join operator
+ groupJoinNode.setJoinOp(JoinType.toJoinOperator(join.getJoinType()));
+
+ // Distribute expr lists
+ List<List<Expr>> distributeExprLists = getDistributeExprs(
+ physicalJoin.left(), physicalJoin.right());
+ groupJoinNode.setChildrenDistributeExprLists(distributeExprLists);
+
+ // Equi-join conjuncts
+ List<Expression> hashJoinConjuncts = join.getHashJoinConjuncts();
+ for (Expression hashConjunct : hashJoinConjuncts) {
+ EqualPredicate equalTo = JoinUtils.swapEqualToForChildrenOrder(
+ (EqualPredicate) hashConjunct, join.left().getOutputSet());
+ groupJoinNode.addEqJoinConjunct(
+ (BinaryPredicate) ExpressionTranslator.translate(equalTo,
context));
+ }
+
+ // Group-by expressions
+ List<Expr> groupingExprs = new ArrayList<>();
+ for (Expression e : aggregate.getGroupByExpressions()) {
+ groupingExprs.add(ExpressionTranslator.translate(e, context));
+ }
+ groupJoinNode.setGroupingExprs(groupingExprs);
+
+ // Aggregate functions with side annotations
+ List<Expr> aggFuncExprs = new ArrayList<>();
+ List<TGroupJoinAggSide> aggSides = new ArrayList<>();
+ Set<Slot> rightOutput = join.right().getOutputSet();
+ Set<AggregateExpression> seen = new HashSet<>();
+ for (NamedExpression outputExpr : aggregate.getOutputExpressions()) {
Review Comment:
[P1] Preserve the enclosing `SessionVarGuardExpr` when translating these
aggregates. The normal aggregate path's `collectAggInTree` explicitly
translates `guard(AggregateExpression)`, but this loop extracts and translates
the bare aggregate. For a view/MV created under different `enable_decimal256`
or decimal-overflow settings, the output slot keeps creation-time semantics
while the serialized aggregate signature is computed under the caller session,
which can change precision or fail the BE result-type check. Please reuse the
guard-aware collector here.
##########
be/src/exec/runtime_filter/runtime_filter_producer_helper_groupjoin.h:
##########
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <vector>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/runtime_filter/runtime_filter_producer_helper.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+// This helper is used by GroupJoin build sink. GroupJoin does not keep a full
build block, so
+// this helper caches only runtime-filter source columns and replays them
after RF size is ready.
+class RuntimeFilterProducerHelperGroupJoin final : public
RuntimeFilterProducerHelper {
+public:
+ ~RuntimeFilterProducerHelperGroupJoin() override = default;
+
+ RuntimeFilterProducerHelperGroupJoin() : RuntimeFilterProducerHelper(true,
false) {}
+
+ Status append_block(Block* block) {
+ if (_skip_runtime_filters_process) {
+ return Status::OK();
+ }
+ if (block->rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnPtr> filter_columns;
+ filter_columns.reserve(_filter_expr_contexts.size());
+ for (auto& ctx : _filter_expr_contexts) {
+ ColumnPtr column;
+ RETURN_IF_ERROR(ctx->execute(block, column));
+ column = column->convert_to_full_column_if_const();
+ filter_columns.emplace_back(std::move(column));
+ }
+
+ _build_rows += block->rows();
+ _cached_filter_columns.emplace_back(std::move(filter_columns));
Review Comment:
[P2] Account for and release this RF cache. Each build block retains one
evaluated column per filter, but GroupJoin's `MemoryUsage` counter includes
only arena/hash/container bytes, and `build_and_publish`/skip never clear
`_cached_filter_columns`. Large builds can therefore retain build-sized memory
through probe/drain while the operator profile omits it. Track the unique
cached bytes and clear the cache immediately after publish or disable.
##########
be/src/exec/runtime_filter/runtime_filter_producer_helper_groupjoin.h:
##########
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <vector>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/runtime_filter/runtime_filter_producer_helper.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+// This helper is used by GroupJoin build sink. GroupJoin does not keep a full
build block, so
+// this helper caches only runtime-filter source columns and replays them
after RF size is ready.
+class RuntimeFilterProducerHelperGroupJoin final : public
RuntimeFilterProducerHelper {
+public:
+ ~RuntimeFilterProducerHelperGroupJoin() override = default;
+
+ RuntimeFilterProducerHelperGroupJoin() : RuntimeFilterProducerHelper(true,
false) {}
+
+ Status append_block(Block* block) {
+ if (_skip_runtime_filters_process) {
+ return Status::OK();
+ }
+ if (block->rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnPtr> filter_columns;
+ filter_columns.reserve(_filter_expr_contexts.size());
+ for (auto& ctx : _filter_expr_contexts) {
+ ColumnPtr column;
+ RETURN_IF_ERROR(ctx->execute(block, column));
Review Comment:
[P1] Keep floating runtime-filter equality consistent with the GroupJoin
hash key. `do_evaluate()` canonicalizes signed zero and NaN, but this
re-evaluates and caches the raw source column; DOUBLE Bloom hashing uses the
raw bits, so a build `-0.0` filter can reject probe `+0.0` even though
GroupJoin would match them. Please canonicalize both Bloom insertion/probing
(or exclude floating Bloom filters for GroupJoin) and cover the float-special
case with a generated runtime filter.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/util/GroupJoinFusionUtils.java:
##########
@@ -0,0 +1,333 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.util;
+
+import org.apache.doris.nereids.stats.StatsCalculator;
+import org.apache.doris.nereids.trees.expressions.AggregateExpression;
+import org.apache.doris.nereids.trees.expressions.EqualPredicate;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.NullSafeEqual;
+import org.apache.doris.nereids.trees.expressions.OrderExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.plans.AggMode;
+import org.apache.doris.nereids.trees.plans.AggPhase;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.statistics.model.ColumnStatistic;
+import org.apache.doris.statistics.model.Statistics;
+
+import com.google.common.collect.Sets;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Shared eligibility and ordering rules for GroupJoin fusion
+ * (enable_group_join_fusion): fusing an INNER hash join + hash aggregation
into a single
+ * GroupJoin operator. Both consumers of these rules must stay in sync:
+ * <ul>
+ * <li>AlignGroupJoinConjunctOrder (post-processor, runs before runtime-filter
generation):
+ * when an eligible shape's GROUP BY merely permutes the join keys, it
reorders the child
+ * join's conjunct list to the group-by order (the value returned here);</li>
+ * <li>PhysicalPlanTranslator.maybeTranslateToGroupJoin (fusion decision at
translation):
+ * fuses an eligible shape only when the join's conjuncts are already listed
in exactly the
+ * group-by order (i.e. the order returned here equals the join's current
one), and emits
+ * the conjuncts as they are.</li>
+ * </ul>
+ */
+public final class GroupJoinFusionUtils {
+
+ private GroupJoinFusionUtils() {}
+
+ /**
+ * Decide whether {@code aggregate} over {@code join} is eligible for
GroupJoin fusion and,
+ * when it is, return the conjunct order the fused operator requires: a
list of the join's
+ * own conjunct instances where conjunct i produces the group-by key at
position i.
+ * <p>
+ * The fused GroupJoin operator groups rows by the shared hash key and
materializes one
+ * grouping-key column per equi-join conjunct: the BE writes the j-th
conjunct's key into
+ * the j-th output tuple slot, and the FE creates the output tuple slots
from the
+ * aggregate's group-by expressions in group-by order. The returned
columns are therefore
+ * correct iff conjunct i produces the group-by key at position i. Since
GROUP BY is
+ * unordered semantically, any GROUP BY that merely permutes the join keys
is eligible -
+ * the caller decides whether to reorder the join's conjuncts to the
returned order
+ * (AlignGroupJoinConjunctOrder) or to require it already (the translator).
+ * <p>
+ * Returns null when the shape is not eligible (not an INNER/CROSS hash
join, mark join,
+ * broadcast join, residual non-equi conjuncts, null-safe equal conjuncts,
no aggregate
+ * functions, aggregates reading both sides, aggregates with an internal
ORDER BY, an aggregate
+ * that is not the final one-phase node (GLOBAL + INPUT_TO_RESULT, per
function and node), an
+ * intermediate Project that computes columns, or intermediate project
slots) or when the
+ * group-by keys cannot be mapped one-to-one onto the conjuncts.
Session-level gates
+ * (enable_group_join_fusion, enable_spill) are checked by the callers,
not here.
+ *
+ * @param project the Project between the aggregate and the join, or null
when the aggregate
+ * directly consumes the join. Only a pure passthrough project
(each output is a bare
+ * slot already produced by one of the join children) is fusable:
the fused operator
+ * evaluates aggregates over the join children rows, so any
computation between the
+ * aggregate and the join (weighted re-multiplication of
pre-aggregated sides, hoisted
+ * casts, CSE columns, ...) must stay on the ordinary HashJoinNode
+ AggregationNode
+ * path which evaluates the Project.
+ */
+ public static List<Expression> alignedConjunctsForGroupJoin(
+ Aggregate<?> aggregate, PhysicalProject<?> project,
PhysicalHashJoin<?, ?> join) {
+ if (join.getJoinType() != JoinType.INNER_JOIN &&
!join.getJoinType().isCrossJoin()) {
+ return null;
+ }
+ if (join.isMarkJoin()) {
+ return null;
+ }
+ // The fused GroupJoin operator matches rows purely by the equi-join
key: it keeps
+ // per-key row counts and per-side aggregation states and has no
per-pair filtering
+ // stage, so a residual non-equi ON conjunct cannot be evaluated by it.
+ if (join.isBroadCastJoin() || !join.getOtherJoinConjuncts().isEmpty())
{
+ return null;
+ }
+ List<Expression> groupByExprs = aggregate.getGroupByExpressions();
+ List<Expression> hashJoinConjuncts = join.getHashJoinConjuncts();
+ if (groupByExprs.isEmpty() || hashJoinConjuncts.isEmpty()
+ || groupByExprs.size() != hashJoinConjuncts.size()) {
+ return null;
+ }
+ List<AggregateExpression> aggregateExpressions =
aggregate.getOutputExpressions().stream()
+ .flatMap(outputExpr ->
outputExpr.collect(AggregateExpression.class::isInstance).stream())
+ .map(AggregateExpression.class::cast)
+ .collect(Collectors.toList());
+ // Pure GROUP BY and DISTINCT keep the regular HashJoin + Aggregate
plan. The current
+ // GroupJoin execution path is designed around maintaining at least
one aggregate state.
+ if (aggregateExpressions.isEmpty()) {
+ return null;
+ }
+ // Phase gate: only the final one-phase aggregate is fusable. The
fused GroupJoin node
+ // materializes FINAL_RESULT and finalizes per-key aggregate state
directly, so an
+ // aggregate that is a partial/LOCAL buffer producer or a
DISTINCT/multi-phase
+ // intermediate node must stay on the ordinary HashJoinNode +
AggregationNode path.
+ // This is observable when two-phase aggregation is forced
(agg_phase=2): the LOCAL
+ // (INPUT_TO_BUFFER) phase sits directly above the join when no
exchange is inserted
+ // between them, and fusing it hard-codes FINAL_RESULT with
finalize-on evaluators
+ // while the merge-finalize aggregate above still consumes the partial
buffer - BE then
+ // aborts with "Aggregate function count result type check failed:
Column type String
+ // is not compatible with data type BIGINT". Requiring every output
aggregate function's
+ // own param to equal the node param also guards split shapes where
the node-level label
+ // alone lies (e.g. GLOBAL/INPUT_TO_RESULT node label with
per-function DISTINCT_* or
+ // buffer params), so the gate is per-function, not just node-level.
+ if (!(aggregate instanceof PhysicalHashAggregate)) {
+ return null;
+ }
+ AggregateParam nodeParam = ((PhysicalHashAggregate<?>)
aggregate).getAggregateParam();
+ if (nodeParam.aggPhase != AggPhase.GLOBAL || nodeParam.aggMode !=
AggMode.INPUT_TO_RESULT) {
+ return null;
+ }
+ for (AggregateExpression aggExpr : aggregateExpressions) {
+ AggregateParam perFunctionParam = aggExpr.getAggregateParam();
+ if (!perFunctionParam.equals(nodeParam)) {
+ return null;
+ }
+ }
+ // The fused operator evaluates aggregates over the probe/build rows
of the join
+ // children, so every group-by key and aggregate argument must be a
column one of the
+ // join children directly produces (an intermediate Project between
the aggregate and
+ // the join would translate to slots that do not exist on either
child).
+ Set<Slot> leftOutput = join.left().getOutputSet();
+ Set<Slot> rightOutput = join.right().getOutputSet();
+ Set<Slot> joinChildrenOutputs = Sets.newHashSet();
+ joinChildrenOutputs.addAll(leftOutput);
+ joinChildrenOutputs.addAll(rightOutput);
+ // Pure-passthrough gate for the intermediate Project (Scheme A): the
Project between
+ // the aggregate and the join may only forward columns the join
children already
+ // produce. A Project computing anything (eager pre-aggregation
weights such as
+ // cntL*cntR, hoisted type-coercion casts, CSE columns) cannot be
skipped by the fused
+ // operator, so such shapes fall back to the ordinary path. Note slot
ExprIds are
+ // reused by the eager-agg rewrite (the same id denotes the raw child
column below the
+ // project and the weighted value above it), so an existence test on
ids alone would
+ // let the weighted shape through; checking that every project output
IS a bare slot of
+ // a join child is the structural test that catches it.
+ if (project != null) {
+ for (NamedExpression projectOutput : project.getProjects()) {
+ if (!(projectOutput instanceof SlotReference)
+ || (!leftOutput.contains(projectOutput) &&
!rightOutput.contains(projectOutput))) {
+ return null;
+ }
+ }
+ }
+ if (!joinChildrenOutputs.containsAll(aggregate.getInputSlots())) {
+ return null;
+ }
+ // Order-sensitive aggregates (internal ORDER BY, e.g.
GROUP_CONCAT(... ORDER BY ...))
+ // are not fusable. The fused operator keeps only a per-key local
aggregate state on one
+ // side plus the other side's per-key row count, so it cannot
reconstruct the interleaved
+ // join row order such aggregates need; and TGroupJoinAggFunction
carries no per-function
+ // sort info (unlike AggregationNode's agg_sort_infos), so the
translated expression's
+ // ORDER BY column would be treated as an ordinary aggregate argument
by the BE
+ // group-join operators (they always pass an empty TSortInfo) and
abort with
+ // "Agg Function ... is not implemented". OrderExpression appears
under an output
+ // expression only inside an aggregate function's argument list.
+ for (Expression outputExpr : aggregate.getOutputExpressions()) {
+ if
(!outputExpr.collect(OrderExpression.class::isInstance).isEmpty()) {
+ return null;
+ }
+ }
+ // Aggregate functions must not reference columns from both join
sides: the per-side
+ // aggregation state is maintained by the corresponding probe/build
operator.
+ for (AggregateExpression aggExpr : aggregateExpressions) {
Review Comment:
[P1] Reject aggregate expressions containing volatile or
`NoneMovableFunction` expressions before fusion. GroupJoin evaluates arguments
per source row and then repeats aggregate state by join multiplicity, so
`SUM(random())` is no longer evaluated once per joined row. It also computes
selected-add arguments for the whole source block before skipping unmatched
places, so an `assert_true()` on an unmatched row can fail even though ordinary
HashJoin + Aggregate never evaluates that row. A
`containsVolatileOrNoneMovableExpression`-style gate is needed here.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -3305,6 +3323,281 @@ private PlanFragment connectJoinNode(HashJoinNode
hashJoinNode, PlanFragment lef
return leftFragment;
}
+ private PlanFragment connectGroupJoinNode(GroupJoinNode groupJoinNode,
PlanFragment leftFragment,
+ PlanFragment rightFragment, PlanTranslatorContext context,
AbstractPlan groupJoin) {
+ groupJoinNode.setChild(0, leftFragment.getPlanRoot());
+ groupJoinNode.setChild(1, rightFragment.getPlanRoot());
+ setPlanRoot(leftFragment, groupJoinNode, groupJoin);
+ context.mergePlanFragment(rightFragment, leftFragment);
+ for (PlanFragment rightChild : rightFragment.getChildren()) {
+ leftFragment.addChild(rightChild);
+ }
+ return leftFragment;
+ }
+
+ /**
+ * V2: Try to fuse HashAggregate(HashJoin) into GroupJoin directly in the
translator stage.
+ * <p>
+ * When the aggregate's child is an INNER hash join with compatible
group-by/join-key,
+ * generate a GroupJoinNode instead of separate AggregationNode +
HashJoinNode.
Review Comment:
[P1] Add a backend-capability/smooth-upgrade gate before emitting
`GROUP_JOIN_NODE`. An upgraded FE with this experimental switch enabled can
currently schedule the new thrift node on an old BE, whose pipeline switch has
no matching operator and rejects the fragment. Default-off does not protect an
explicitly enabled query during rolling upgrade; please fall back to HashJoin +
Aggregate until every selected BE advertises support, and add a mixed-version
planner test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -3305,6 +3323,281 @@ private PlanFragment connectJoinNode(HashJoinNode
hashJoinNode, PlanFragment lef
return leftFragment;
}
+ private PlanFragment connectGroupJoinNode(GroupJoinNode groupJoinNode,
PlanFragment leftFragment,
+ PlanFragment rightFragment, PlanTranslatorContext context,
AbstractPlan groupJoin) {
+ groupJoinNode.setChild(0, leftFragment.getPlanRoot());
+ groupJoinNode.setChild(1, rightFragment.getPlanRoot());
+ setPlanRoot(leftFragment, groupJoinNode, groupJoin);
+ context.mergePlanFragment(rightFragment, leftFragment);
+ for (PlanFragment rightChild : rightFragment.getChildren()) {
+ leftFragment.addChild(rightChild);
+ }
+ return leftFragment;
+ }
+
+ /**
+ * V2: Try to fuse HashAggregate(HashJoin) into GroupJoin directly in the
translator stage.
+ * <p>
+ * When the aggregate's child is an INNER hash join with compatible
group-by/join-key,
+ * generate a GroupJoinNode instead of separate AggregationNode +
HashJoinNode.
+ * Returns null if conditions are not met — caller falls through to normal
translation.
+ */
+ private PlanFragment maybeTranslateToGroupJoin(
+ Plan aggregate,
+ PlanTranslatorContext context) {
+ // Gate: session variable
+ ConnectContext connectContext = ConnectContext.get();
+ if (connectContext == null
+ ||
!connectContext.getSessionVariable().isEnableGroupJoinFusion()) {
+ return null;
+ }
+ // Gate: spill not supported
+ if (connectContext.getSessionVariable().enableSpill) {
+ return null;
+ }
+
+ // Child must be PhysicalHashJoin (optionally through one
pure-passthrough
+ // PhysicalProject; see the project gate in GroupJoinFusionUtils).
+ Plan child = aggregate.child(0);
+ PhysicalProject<?> project = null;
+ if (child instanceof PhysicalProject) {
+ project = (PhysicalProject<?>) child;
+ child = child.child(0);
+ }
+ if (!(child instanceof PhysicalHashJoin)) {
+ return null;
+ }
+ PhysicalHashJoin<?, ?> join = (PhysicalHashJoin<?, ?>) child;
+
+ // Full fusion eligibility (join shape + aggregate constraints) is
decided by
+ // GroupJoinFusionUtils.alignedConjunctsForGroupJoin, the single
source of truth shared
+ // with the AlignGroupJoinConjunctOrder post-processor that pre-aligns
the child join's
+ // conjunct order with the group-by keys. The translator emits the
join's conjuncts as
+ // they are, so fusion additionally requires the conjuncts to be
listed in exactly the
+ // group-by order: the processor guarantees this for every eligible
shape, and anything
+ // that is not aligned stays on the regular HashJoinNode +
AggregationNode path.
+ List<Expression> alignedConjuncts =
GroupJoinFusionUtils.alignedConjunctsForGroupJoin(
+ (Aggregate<?>) aggregate, project, join);
+ if (alignedConjuncts == null
+ || !GroupJoinFusionUtils.sameConjunctOrder(
+ alignedConjuncts, join.getHashJoinConjuncts())) {
+ return null;
+ }
+
+ // All checks passed — generate GroupJoinNode
+ return translateToGroupJoinNode((Aggregate<?>) aggregate, join,
context);
+ }
+
+ /** Translate Aggregate(HashJoin) pattern into a GroupJoinNode fragment. */
+ private PlanFragment translateToGroupJoinNode(
+ Aggregate<?> aggregate,
+ PhysicalHashJoin<?, ?> join,
+ PlanTranslatorContext context) {
+ PhysicalHashJoin<PhysicalPlan, PhysicalPlan> physicalJoin
+ = (PhysicalHashJoin<PhysicalPlan, PhysicalPlan>) join;
+
+ // maybeTranslateToGroupJoin only lets joins without residual
conjuncts reach this
+ // point. Enforce it here as well: the fused operator has no per-pair
filtering, so
+ // a residual conjunct that slipped through would be silently dropped
and produce
+ // wrong aggregation results (the failure mode this guard exists for).
+ Preconditions.checkState(join.getOtherJoinConjuncts().isEmpty(),
+ "GroupJoin fusion requires the join to have no residual
conjuncts, got: %s",
+ join.getOtherJoinConjuncts());
+
+ // maybeTranslateToGroupJoin also only lets aggregates whose inputs
are all directly
+ // produced by the join children reach this point. Enforce it here as
well: aggregate
+ // arguments that reference slots computed by an intermediate Project
between the
+ // aggregate and the join (e.g. hoisted type-coercion casts of binary
aggregates) do
+ // not exist on the join children, and translating them would abort
fragment
+ // serialization with an NPE in GroupJoinNode.toThrift.
+ Set<Slot> joinChildrenOutputs = Sets.newHashSet();
+ joinChildrenOutputs.addAll(join.left().getOutputSet());
+ joinChildrenOutputs.addAll(join.right().getOutputSet());
+
Preconditions.checkState(joinChildrenOutputs.containsAll(aggregate.getInputSlots()),
+ "GroupJoin fusion requires all aggregate inputs to be
join-child outputs, got: %s",
+ aggregate.getInputSlots());
+
+ // Visit children right-to-left (right = build, left = probe).
+ // connectGroupJoinNode merges both child fragments into this one,
exactly like
+ // connectJoinNode does for a plain hash join, so the children must be
translated under
+ // the same fragment-merge context. Without it a child aggregate could
take the
+ // bucketed-aggregation path (shouldUseBucketedFusion), which drops
the exchange that
+ // keeps its olap scan in its own fragment and leaves two olap scans
inside the
+ // GroupJoin fragment: the scan-assignment job then rejects the
fragment with
+ // "Not supported multiple scan multiple OlapTable but not contains
colocate join or
+ // bucket shuffle join". The GroupJoin would also lose the hash
distribution its
+ // PARTITIONED input relies on, since that exchange is what enforces
it.
+ context.enterFragmentMergeChild();
+ PlanFragment rightFragment;
+ PlanFragment leftFragment;
+ try {
+ rightFragment = join.child(1).accept(this, context);
+ leftFragment = join.child(0).accept(this, context);
+ } finally {
+ context.exitFragmentMergeChild();
+ }
+ PlanNode leftPlanRoot = leftFragment.getPlanRoot();
Review Comment:
[P2] Attribute this output-facing node to the aggregate, not the eliminated
join. GroupJoin emits one row per matched key, yet it receives the join's
Nereids id, `setPlanRoot`/legacy mapping, and (below) join cardinality. HBO
then records the grouped `RowsProduced` under the `PhysicalHashJoin`
fingerprint and can reuse a group count as future join cardinality (M*N joined
rows may be recorded as 1), while `StatsErrorEstimator` reports a spurious join
q-error. Map the fused output/stats to `aggregate`, or explicitly separate
pre-aggregate join feedback from GroupJoin output.
--
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]
