github-actions[bot] commented on code in PR #63690:
URL: https://github.com/apache/doris/pull/63690#discussion_r3433728384
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -70,77 +84,200 @@
* ->T2(D)
*/
public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ public static final int BIG_JOIN_BUILD_SIZE = 400_000;
private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private static final String JOIN_CNT = "joinCnt";
private final StatsDerive derive = new StatsDerive(false);
@Override
public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan>
join, PushDownAggContext context) {
- boolean toLeft = false;
- boolean toRight = false;
- boolean pushHere = false;
- if (join.getJoinType().isAsofJoin()) {
- // do nothing for asof join
- return join;
+ Pair<Boolean, Boolean> pushSide = decideJoinPushSide(join, context);
+ boolean toLeft = pushSide.first;
+ boolean toRight = pushSide.second;
+ if (!toLeft && !toRight) {
+ if (SessionVariable.isEagerAggregationOnJoin()) {
+ return genAggregate(join, context);
+ } else {
+ return join;
+ }
}
- if (context.getAggFunctions().isEmpty()) {
- // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
- // if no agg function, try to push agg to the child which contains
all group keys
- // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id,
t2.v).ndv to determine push target
- if
(join.left().getOutputSet().containsAll(context.getGroupKeys())) {
- toLeft = true;
- } else if
(join.right().getOutputSet().containsAll(context.getGroupKeys())) {
- toRight = true;
+
+ // construct left and right group by keys
+ List<SlotReference> leftChildGroupByKeys = new ArrayList<>();
+ List<SlotReference> rightChildGroupByKeys = new ArrayList<>();
+ if (toLeft) {
+ fillGroupByKeys(join, join.left(), context, leftChildGroupByKeys);
+ }
+ if (toRight) {
+ fillGroupByKeys(join, join.right(), context,
rightChildGroupByKeys);
+ }
+ // construct left and right aggFuncs and aliasMap
+ List<AggregateFunction> leftFuncs = new ArrayList<>();
+ List<AggregateFunction> rightFuncs = new ArrayList<>();
+ Map<AggregateFunction, Alias> leftAliasMap = new IdentityHashMap<>();
+ Map<AggregateFunction, Alias> rightAliasMap = new IdentityHashMap<>();
+ for (AggregateFunction f : context.getAggFunctions()) {
+ Set<Slot> inputs = f.getInputSlots();
+ Alias a = context.getAliasMap().get(f);
+ if (inputs.isEmpty()) {
+ if (join.getJoinType().isRightSemiOrAntiJoin()) {
+ rightFuncs.add(f);
+ rightAliasMap.put(f, a);
+ } else {
+ leftFuncs.add(f);
+ leftAliasMap.put(f, a);
+ }
+ continue;
+ }
+ if (join.left().getOutputSet().containsAll(inputs)) {
+ leftFuncs.add(f);
+ leftAliasMap.put(f, a);
+ } else if (join.right().getOutputSet().containsAll(inputs)) {
+ rightFuncs.add(f);
+ rightAliasMap.put(f, a);
} else {
- pushHere = true;
+ return join;
}
+ }
+
+ boolean passThroughBigJoin = isPassThroughBigJoin(join, context);
+ boolean leftNeedOutputCount = needOutputCountForJoinChild(join,
toLeft, toRight,
+ context.needOutputCount(), rightFuncs);
+ boolean rightNeedOutputCount = needOutputCountForJoinChild(join,
toRight, toLeft,
+ context.needOutputCount(), leftFuncs);
+ Optional<PushDownAggContext> leftChildContext = toLeft ?
Optional.of(context.forOneBranch(leftFuncs,
+ leftAliasMap, leftChildGroupByKeys, passThroughBigJoin,
leftNeedOutputCount)) : Optional.empty();
+ Optional<PushDownAggContext> rightChildContext = toRight ?
Optional.of(context.forOneBranch(rightFuncs,
+ rightAliasMap, rightChildGroupByKeys, passThroughBigJoin,
rightNeedOutputCount)) : Optional.empty();
+
+ Plan newLeft = join.left();
+ Plan newRight = join.right();
+ if (leftChildContext.isPresent() &&
!leftChildContext.get().noGroupKeyAndNoAggFunc()) {
+ newLeft = join.left().accept(this, leftChildContext.get());
+ }
+ if (rightChildContext.isPresent() &&
!rightChildContext.get().noGroupKeyAndNoAggFunc()) {
+ newRight = join.right().accept(this, rightChildContext.get());
+ }
+
+ if (newLeft == join.left() && newRight == join.right()) {
+ context.getBilateralState().registerNoCountSlot(join);
+ return join;
+ }
+ Optional<Slot> leftChildCountSlot =
context.getBilateralState().getCountSlot(newLeft);
+ Optional<Slot> rightChildCountSlot =
context.getBilateralState().getCountSlot(newRight);
+ LogicalJoin<? extends Plan, ? extends Plan> newJoin = (LogicalJoin<?
extends Plan, ? extends Plan>)
+ join.withChildren(newLeft, newRight);
+
+ if (leftChildCountSlot.isPresent() || rightChildCountSlot.isPresent())
{
+ return buildCanonicalJoinProject(newJoin, context,
leftChildContext, rightChildContext,
+ leftChildCountSlot, rightChildCountSlot);
+ }
+ context.getBilateralState().registerNoCountSlot(newJoin);
+ return newJoin;
+ }
+
+ private Pair<Boolean, Boolean> decideJoinPushSide(
+ LogicalJoin<? extends Plan, ? extends Plan> join,
PushDownAggContext context) {
+ if (join.getJoinType().isAsofJoin() || join.isMarkJoin()) {
+ // do nothing for asof join and mark join
+ return Pair.of(false, false);
+ }
+
+ boolean deduplicateOnly = context.getAggFunctions().isEmpty();
+ boolean toLeft = false;
+ boolean toRight = false;
+ if (deduplicateOnly) {
+ toLeft = true;
+ toRight = true;
} else {
for (AggregateFunction aggFunc : context.getAggFunctions()) {
- if
(join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ Set<Slot> inputs = aggFunc.getInputSlots();
+ if (inputs.isEmpty()) {
+ if (join.getJoinType().isRightSemiOrAntiJoin()) {
+ toRight = true;
+ } else {
+ toLeft = true;
+ }
+ continue;
+ }
+ if (join.left().getOutputSet().containsAll(inputs)) {
toLeft = true;
- } else if
(join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ } else if (join.right().getOutputSet().containsAll(inputs)) {
toRight = true;
} else {
- pushHere = true;
+ toLeft = false;
+ toRight = false;
+ break;
}
}
}
+ if (!toLeft && !toRight) {
+ return Pair.of(false, false);
+ }
+ if (deduplicateOnly) {
+ return adjustPushSideForCaseWhen(join, context, toLeft, toRight);
+ }
+ if (toLeft && toRight) {
+ return join.getJoinType().isInnerOrCrossJoin()
+ ? Pair.of(true, true)
Review Comment:
The bilateral path should not run through joins whose ON predicates contain
volatile expressions. Other rewrite rules keep volatile ON conjuncts at
join-pair scope because `random()`, `uuid()`, or volatile UDFs must be
evaluated per joined row pair. This branch now returns `(true, true)` for any
inner/cross join once aggregates are found on both sides, and
`fillGroupByKeys()` only groups by the conjunct input slots. For a query like
`sum(l.v), sum(r.v)` over `l JOIN r ON l.k = r.k AND random() < 0.5`,
pre-aggregating both sides changes how many times the volatile predicate is
evaluated and can change the result. Please bail out when any hash/other join
conjunct contains a volatile expression before enabling bilateral pushdown, and
add coverage for that path.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -533,22 +738,452 @@ public Plan visitLogicalRelation(LogicalRelation
relation, PushDownAggContext co
return genAggregate(relation, context);
}
+ private Optional<Alias> findCountStarAlias(PushDownAggContext context) {
+ for (AggregateFunction func : context.getAggFunctions()) {
+ if (func instanceof Count && ((Count) func).isCountStar()) {
+ return Optional.of(context.getAliasMap().get(func));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private Optional<Integer> findCountStarAggFunctionIndex(PushDownAggContext
context) {
+ for (int i = 0; i < context.getAggFunctions().size(); i++) {
+ AggregateFunction func = context.getAggFunctions().get(i);
+ if (func instanceof Count && ((Count) func).isCountStar()) {
+ return Optional.of(i);
+ }
+ }
+ return Optional.empty();
+ }
+
private Plan genAggregate(Plan child, PushDownAggContext context) {
- if (context.isValid() && checkStats(child, context)) {
+ if (isPushDisabledByVariable(context)) {
+ context.getBilateralState().registerNoCountSlot(child);
+ return child;
+ }
+ if (checkStats(child, context) || isPushEnabledByVariable(context)) {
List<NamedExpression> aggOutputExpressions = new ArrayList<>();
for (AggregateFunction func : context.getAggFunctions()) {
aggOutputExpressions.add(context.getAliasMap().get(func));
}
+ Optional<Alias> countStarAlias = findCountStarAlias(context);
+ Optional<Alias> outputCountAlias = Optional.empty();
+ if (context.needOutputCount()) {
+ if (countStarAlias.isPresent()) {
+ outputCountAlias = countStarAlias;
+ } else {
+ outputCountAlias = Optional.of(new Alias(new Count(),
+ "cnt" +
context.getCascadesContext().getStatementContext().generateColumnName()));
+ }
+ }
aggOutputExpressions.addAll(context.getGroupKeys());
+ if (outputCountAlias.isPresent() && !countStarAlias.isPresent()) {
+ aggOutputExpressions.add(outputCountAlias.get());
+ }
LogicalAggregate genAgg = new
LogicalAggregate(context.getGroupKeys(), aggOutputExpressions, child);
NormalizeAggregate normalizeAggregate = new NormalizeAggregate();
- return normalizeAggregate.normalizeAgg(genAgg, Optional.empty(),
+ Plan normalized = normalizeAggregate.normalizeAgg(genAgg,
Optional.empty(),
context.getCascadesContext());
+
+ for (AggregateFunction func : context.getAggFunctions()) {
+ Alias a = context.getAliasMap().get(func);
+ Slot pushedSlot = normalized.getOutput().stream()
+ .filter(slot -> slot.getExprId().equals(a.getExprId()))
+ .findFirst()
+ .orElse(a.toSlot());
+
context.getBilateralState().registerPushedAggFuncSlot(a.getExprId(),
pushedSlot);
+ }
+
+ if (outputCountAlias.isPresent()) {
+ context.getBilateralState().registerCountSlot(normalized,
outputCountAlias.get().toSlot());
+ } else {
+ context.getBilateralState().registerNoCountSlot(normalized);
+ }
+ return normalized;
+ } else {
+ context.getBilateralState().registerNoCountSlot(child);
+ return child;
+ }
+ }
+
+ // Build the canonical project above a rewritten join after
eager-aggregation pushdown.
+ // Responsibilities:
+ // 1. Restore the outputs expected by the parent rollup. If a join side
has a childContext, materialize
+ // that side's aggregate current values and group keys; otherwise
forward the original join outputs.
+ // 2. For inner joins, recover join multiplicity by multiplying
non-MIN/MAX aggregate current values by
+ // the opposite side's count slot when that side contributes rows to
the parent aggregate.
+ // 3. Append and register a synthetic join-count slot `cnt` (logical jcnt)
for upper-level rollup.
+ //
+ // The examples below are schematic. The real project may keep extra
forwarded slots such as join keys.
+ //
+ // Inner join + sum, single-side rewrite:
+ // Before:
+ // agg(sum(t1.a), sum(t2.a), gby t2.k)
+ // -> inner join(k = k)
+ // -> scan(t1)
+ // -> scan(t2)
+ // After:
+ // agg(sum(s1), sum(s2), gby t2.k)
+ // -> project(s1, t2.a * cnt1 as s2, t2.k, cnt1)
+ // -> inner join(k = k)
+ // -> agg(sum(t1.a) as s1, count(*) as cnt1, gby k)
+ // -> scan(t1)
+ // -> scan(t2)
+ //
+ // Inner join + sum, bilateral rewrite:
+ // Before:
+ // agg(sum(t1.a), sum(t2.a), gby t2.k)
+ // -> inner join(k = k)
+ // -> scan(t1)
+ // -> scan(t2)
+ // After:
+ // agg(sum(s1'), sum(s2'), gby t2.k)
+ // -> project(s1 * cnt2 as s1', s2 * cnt1 as s2', t2.k, cnt1 * cnt2
as cnt)
+ // -> inner join(k = k)
+ // -> agg(sum(t1.a) as s1, count(*) as cnt1, gby k)
+ // -> scan(t1)
+ // -> agg(sum(t2.a) as s2, count(*) as cnt2, gby k)
+ // -> scan(t2)
+ //
+ // Inner join + count(col), single-side rewrite:
+ // Before:
+ // agg(count(t1.a), count(t2.a), gby t2.k)
+ // -> inner join(k = k)
+ // -> scan(t1)
+ // -> scan(t2)
+ // After:
+ // agg(sum0(c1), sum0(c2), gby t2.k)
+ // -> project(c1, if(t2.a is null, 0, 1) * cnt1 as c2, t2.k, cnt1 as
cnt)
+ // -> inner join(k = k)
+ // -> agg(count(t1.a) as c1, count(*) as cnt1, gby k)
+ // -> scan(t1)
+ // -> scan(t2)
+ //
+ // Inner join + count(col), bilateral rewrite:
+ // Before:
+ // agg(count(t1.a), count(t2.a), gby t2.k)
+ // -> inner join(k = k)
+ // -> scan(t1)
+ // -> scan(t2)
+ // After:
+ // agg(sum0(c1'), sum0(c2'), gby t2.k)
+ // -> project(c1 * cnt2 as c1', c2 * cnt1 as c2', t2.k, cnt1 * cnt2
as cnt)
+ // -> inner join(k = k)
+ // -> agg(count(t1.a) as c1, count(*) as cnt1, gby k)
+ // -> scan(t1)
+ // -> agg(count(t2.a) as c2, count(*) as cnt2, gby k)
+ // -> scan(t2)
+ // For count(*), the current row value is 1 instead of if(col is null,
0, 1).
+ //
+ // Semi/anti join:
+ // The project does not multiply by the opposite-side count
+ //
+ // Outer join:
+ // Aggregate outputs are not multiplied by the opposite-side count
either; only `cnt` changes:
+ // left outer join with left push -> project(s1, t2.k, cnt1 as cnt)
+ // right outer join with left push -> project(s1, t2.k, nvl(cnt1, 1)
as cnt)
+ private Plan buildCanonicalJoinProject(LogicalJoin<? extends Plan, ?
extends Plan> join, PushDownAggContext context,
+ Optional<PushDownAggContext> leftChildContext,
Optional<PushDownAggContext> rightChildContext,
+ Optional<Slot> leftCountSlot, Optional<Slot> rightCountSlot) {
+ List<NamedExpression> projections = new ArrayList<>();
+ Set<ExprId> outputIds = new HashSet<>();
+ boolean remainLeft = join.getJoinType().isRemainLeftJoin();
+ boolean remainRight = join.getJoinType().isRemainRightJoin();
+ boolean shouldAdjustLeft =
shouldUseJoinOppositeCntAdjustAggOutput(join, leftChildContext, rightCountSlot);
+ boolean shouldAdjustRight =
shouldUseJoinOppositeCntAdjustAggOutput(join, rightChildContext, leftCountSlot);
+
+ if (remainLeft) {
+ appendJoinSideOutputs(projections, outputIds, join.left(),
leftChildContext, context,
+ rightCountSlot, shouldAdjustLeft);
+ }
+ if (remainRight) {
+ appendJoinSideOutputs(projections, outputIds, join.right(),
rightChildContext, context,
+ leftCountSlot, shouldAdjustRight);
+ }
+
+ Optional<? extends NamedExpression> joinCount = Optional.empty();
+ if (context.needOutputCount()) {
+ joinCount = findProjectedCountStarOutput(context, outputIds);
+ if (!joinCount.isPresent()) {
+ joinCount = computeJoinCount(join, leftChildContext,
rightChildContext,
+ leftCountSlot, rightCountSlot, context);
+ }
+ }
+ Optional<Slot> projectedCountSlot = Optional.empty();
+ if (joinCount.isPresent()) {
+ appendProjectionIfAbsent(projections, outputIds, joinCount.get());
+ projectedCountSlot = Optional.of(joinCount.get().toSlot());
+ }
+ LogicalProject<Plan> project = new LogicalProject<>(projections, join);
+ if (projectedCountSlot.isPresent()) {
+ context.getBilateralState().registerCountSlot(project,
projectedCountSlot.get());
+ } else {
+ context.getBilateralState().registerNoCountSlot(project);
+ }
+ return project;
+ }
+
+ private void appendJoinSideOutputs(List<NamedExpression> projections,
Set<ExprId> outputIds, Plan originalSide,
+ Optional<PushDownAggContext> childContext, PushDownAggContext
parentContext,
+ Optional<Slot> oppositeCountSlot, boolean shouldAdjustOutput) {
+ if (childContext.isPresent()) {
+ for (AggregateFunction aggFunc :
childContext.get().getAggFunctions()) {
+ NamedExpression aggOutput = shouldAdjustOutput
+ ? adjustAggOutputUseOppositeCountOnJoin(aggFunc,
parentContext, oppositeCountSlot)
+ : buildAggOutputWithoutJoinAdjustment(aggFunc,
parentContext);
+ appendProjectionIfAbsent(projections, outputIds, aggOutput);
+ }
+ for (SlotReference groupKey : childContext.get().getGroupKeys()) {
+ appendProjectionIfAbsent(projections, outputIds, groupKey);
+ }
} else {
+ for (Slot slot : originalSide.getOutput()) {
+ appendProjectionIfAbsent(projections, outputIds, slot);
+ }
+ }
+ }
+
+ private void appendProjectionIfAbsent(List<NamedExpression> projections,
Set<ExprId> outputIds,
+ NamedExpression expression) {
+ if (outputIds.add(expression.getExprId())) {
+ projections.add(expression);
+ }
+ }
+
+ private boolean shouldUseJoinOppositeCntAdjustAggOutput(LogicalJoin<?
extends Plan, ? extends Plan> join,
+ Optional<PushDownAggContext> childContext, Optional<Slot>
oppositeCountSlot) {
+ return join.getJoinType().isInnerOrCrossJoin() &&
childContext.isPresent() && oppositeCountSlot.isPresent()
+ &&
hasAggNeedJoinMultiplicityRecovery(childContext.get().getAggFunctions());
+ }
+
+ private Optional<NamedExpression>
findProjectedCountStarOutput(PushDownAggContext context, Set<ExprId> outputIds)
{
+ BilateralState state = context.getBilateralState();
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (aggFunc instanceof Count && ((Count) aggFunc).isCountStar()) {
+ ExprId exprId = context.getAliasMap().get(aggFunc).getExprId();
+ if (state.hasAggFuncOutput(exprId)) {
+ NamedExpression countStarOutput =
state.getPushedAggFuncSlot(exprId);
+ if (outputIds.contains(countStarOutput.getExprId())) {
+ return Optional.of(countStarOutput);
+ }
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ private boolean hasAggNeedJoinMultiplicityRecovery(List<AggregateFunction>
aggFunctions) {
+ return
aggFunctions.stream().anyMatch(this::needJoinMultiplicityRecovery);
+ }
+
+ private boolean needJoinMultiplicityRecovery(AggregateFunction aggFunc) {
+ return !(aggFunc instanceof Max) && !(aggFunc instanceof Min);
+ }
+
+ private Optional<? extends NamedExpression> computeJoinCount(LogicalJoin<?
extends Plan, ? extends Plan> join,
+ Optional<PushDownAggContext> leftChildContext,
Optional<PushDownAggContext> rightChildContext,
+ Optional<Slot> leftCountSlot, Optional<Slot> rightCountSlot,
PushDownAggContext context) {
+ JoinType joinType = join.getJoinType();
+ if (joinType.isInnerOrCrossJoin()) {
+ if (leftCountSlot.isPresent() && rightCountSlot.isPresent()) {
+ Expression joinCnt = TypeCoercionUtils.processBinaryArithmetic(
+ new Multiply(leftCountSlot.get(),
rightCountSlot.get()));
+ return Optional.of(new Alias(joinCnt,
+ JOIN_CNT +
context.getCascadesContext().getStatementContext().generateColumnName()));
+ } else if (leftCountSlot.isPresent()) {
+ return leftCountSlot;
+ } else if (rightCountSlot.isPresent()) {
+ return rightCountSlot;
+ }
+ return Optional.empty();
+ }
+ if (joinType.isLeftOuterJoin()) {
+ if (leftChildContext.isPresent()) {
+ return leftCountSlot;
+ }
+ if (rightChildContext.isPresent() && rightCountSlot.isPresent()) {
+ Expression joinCnt = TypeCoercionUtils.processBoundFunction(
+ new Nvl(rightCountSlot.get(), BigIntLiteral.of(1)));
+ return Optional.of(new Alias(joinCnt,
+ JOIN_CNT +
context.getCascadesContext().getStatementContext().generateColumnName()));
+ }
+ return Optional.empty();
+ }
+ if (joinType.isRightOuterJoin()) {
+ if (leftChildContext.isPresent() && leftCountSlot.isPresent()) {
+ Expression joinCnt = TypeCoercionUtils.processBoundFunction(
+ new Nvl(leftCountSlot.get(), BigIntLiteral.of(1)));
+ return Optional.of(new Alias(joinCnt,
+ JOIN_CNT +
context.getCascadesContext().getStatementContext().generateColumnName()));
+ }
+ if (rightChildContext.isPresent()) {
+ return rightCountSlot;
+ }
+ return Optional.empty();
+ }
+ if (joinType.isLeftSemiOrAntiJoin()) {
+ return leftCountSlot;
+ }
+ if (joinType.isRightSemiOrAntiJoin()) {
+ return rightCountSlot;
+ }
+ return Optional.empty();
+ }
Review Comment:
When a full outer join is rewritten under an upper join that needs
`context.needOutputCount()`, this falls through to no count slot. That loses
multiplicity for parent bilateral recovery after one side of the full outer
join was pre-aggregated. Example: `p INNER JOIN (l FULL OUTER JOIN r ON l.k =
r.k)` with `sum(p.z)` and `count(l.v)` grouped by `l.k`; the full-outer child
may collapse `l` to one row per key, but because no count slot is registered
here, the parent cannot multiply `sum(p.z)` by the original `l` row count and
undercounts matched keys. The left/right outer cases above return
`leftCountSlot`, `rightCountSlot`, or `nvl(..., 1)` for exactly this reason;
full outer needs equivalent handling, or pushdown should be disabled when
`context.needOutputCount()` reaches a full outer join.
--
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]
