englefly commented on code in PR #67433:
URL: https://github.com/apache/doris/pull/67433#discussion_r4421289239
##########
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:
现在的HBO 要重新实现, 不用考虑待删除的逻辑
--
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]
