github-actions[bot] commented on code in PR #67940:
URL: https://github.com/apache/doris/pull/67940#discussion_r4067993543
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/SubExprAnalyzer.java:
##########
@@ -480,11 +533,12 @@ private LogicalAggregate
validateNodeInfoList(List<PlanNodeCorrelatedInfo> nodeI
throw new AnalysisException(
"access outer query's column before
two agg nodes is not supported");
}
- if (nodeInfo.hasGroupBy) {
- // TODO support later
- throw new AnalysisException(
- "access outer query's column before
agg with group by is not supported");
- }
+ // the aggregation of the subquery may group the
inner rows and it may
Review Comment:
[P1] Reject Repeat on the widened grouped-scalar path. `SELECT T1.id,
(SELECT count(*) FROM T2 WHERE T2.id=T1.id GROUP BY GROUPING SETS ((T2.score),
())) FROM T1` is normalized during nested analysis to `Aggregate -> Repeat ->
Project -> Filter(correlation)`. In this leaf-to-root walk, `LOGICAL_REPEAT`
falls through before `checkAfterAggNode` is set, and this relaxed
grouped-Aggregate branch accepts it.
`UnCorrelatedApplyAggregateFilter.locateAggregate` later stops at Repeat, so no
`correlationFilter` reaches the Apply; `ScalarApplyToJoin` takes the
uncorrelated path while the retained Filter still reads `T1.id`, and final slot
validation rejects the query. Please reject `LOGICAL_REPEAT` here or rebuild
Repeat per correlation key, and add scalar `GROUPING SETS` coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -97,28 +633,2135 @@ public List<Rule> buildRules() {
return apply;
}
- // pull up correlated filter into apply node
- List<NamedExpression> newAggOutput = new
ArrayList<>(agg.getOutputExpressions());
- List<Expression> newGroupby =
- Utils.getUnCorrelatedExprs(correlatedPredicate,
apply.getCorrelationSlot());
- newGroupby.addAll(agg.getGroupByExpressions());
+ CorrelatedAggregatePredicates predicates =
+ CorrelatedAggregatePredicates.of(apply, correlatedPredicate,
+ aggregation.filtersAboveTheAggregation());
+ // A global aggregate above an aggregate which can return no row for a
correlation key
+ // returns a row for the empty input of that key, and neither rewrite
can reproduce it (see
+ // observesTheEmptyInputOfAGlobalAggregate): report those subqueries
instead of dropping the
+ // row and evaluating the subquery to false.
+ if (observesTheEmptyInputOfAGlobalAggregate(apply, aggregation,
predicates)) {
+ throw new AnalysisException("Unsupported correlated subquery with
grouping and/or aggregation "
+ + apply.right());
+ }
+ if (needCorrelatedAggregationOnOuter(apply, aggregation,
correlatedPredicate, predicates)) {
+ Plan aggregatedOuter = pullUpCorrelatedPredicateByAggregatingOuter(
+ apply, aggregation, unCorrelatedPredicate, predicates);
+ if (aggregatedOuter != null) {
+ return aggregatedOuter;
+ }
+ // The original rewrite is known to be not equivalent for this
subquery and the rewrite
+ // above cannot be applied safely: report the subquery as
unsupported instead of building
+ // a plan whose result is wrong.
+ throw new AnalysisException("Unsupported correlated subquery with
grouping and/or aggregation "
+ + apply.right());
+ }
+
+ // pull up correlated filter into apply node: the inner side of every
correlated predicate
+ // becomes a group by column and an output column of the aggregation
below the filter, so that
+ // the aggregation of one outer row is the aggregation of the rows of
its own key, and every
+ // aggregate above that aggregation groups the rows of its child by
the same keys (a scalar
+ // subquery keeps the rows of its aggregation through an aggregation
which SubqueryToApply adds
+ // above it, and those rows may not be mixed between two correlation
keys either)
+ List<Expression> newGroupby =
Utils.getUnCorrelatedExprs(correlatedPredicate, apply.getCorrelationSlot());
Map<Expression, Slot> unCorrelatedExprToSlot = Maps.newHashMap();
+ List<NamedExpression> newGroupbyOutputs =
Lists.newArrayListWithCapacity(newGroupby.size());
for (Expression expression : newGroupby) {
if (expression instanceof Slot) {
- newAggOutput.add((NamedExpression) expression);
+ newGroupbyOutputs.add((NamedExpression) expression);
} else {
Alias alias = new Alias(expression);
unCorrelatedExprToSlot.put(expression, alias.toSlot());
- newAggOutput.add(alias);
+ newGroupbyOutputs.add(alias);
}
}
+ // the keys which the aggregates above the deepest one group by: the
slots the keys have in
+ // the output of the aggregation below them
+ List<NamedExpression> keySlots = newGroupbyOutputs.stream()
+
.map(NamedExpression::toSlot).collect(ImmutableList.toImmutableList());
correlatedPredicate = ExpressionUtils.replace(correlatedPredicate,
unCorrelatedExprToSlot);
- LogicalAggregate newAgg = new LogicalAggregate<>(newGroupby,
newAggOutput,
-
PlanUtils.filterOrSelf(ImmutableSet.copyOf(unCorrelatedPredicate),
filter.child()));
+ Map<LogicalAggregate<?>, Plan> newAggregations = new
IdentityHashMap<>();
+ for (LogicalAggregate<?> aggregate : aggregation.aggregationChain()) {
+ boolean isTheAggregationOfTheDomain = aggregate ==
aggregation.domainAggregation();
+ List<Expression> groupBy = Lists.newArrayList(
+ isTheAggregationOfTheDomain ? newGroupby : keySlots);
+ groupBy.addAll(aggregate.getGroupByExpressions());
+ List<NamedExpression> outputs =
Lists.newArrayList(aggregate.getOutputExpressions());
+ outputs.addAll(isTheAggregationOfTheDomain ? newGroupbyOutputs :
keySlots);
+ Plan child = isTheAggregationOfTheDomain
+ // the projections below it only carry the columns which
the aggregation needs, so
+ // the new aggregation reads the rows of the filter
directly
+ ?
PlanUtils.filterOrSelf(ImmutableSet.copyOf(unCorrelatedPredicate),
+ aggregation.domainFilter().child())
+ : aggregate.child(0);
+ newAggregations.put(aggregate, new LogicalAggregate<>(groupBy,
outputs, child));
+ }
+ // the predicates which were already pulled into the apply are the
predicates of the HAVING
+ // clause of the subquery: they were evaluated on the rows of the old
aggregate and have to
+ // stay in the filter of the new apply, otherwise the subquery loses
them
+ List<Expression> newCorrelationFilter = Lists.newArrayList();
+ apply.getCorrelationFilter().map(ExpressionUtils::extractConjunction)
+ .ifPresent(newCorrelationFilter::addAll);
+ newCorrelationFilter.addAll(correlatedPredicate);
+ // the join which unnests the apply reads the inner side of the
correlation predicates from
+ // the output of the right side, so the projections which wrap the new
aggregate have to
+ // expose the keys it added: an IN subquery keeps the projections of
its select list above
+ // the aggregate (for example the outputs [c1] and [c1, c2] which wrap
an aggregate
+ // computing count(*) as c1, random() as c2), and a projection which
hides one of the keys
+ // makes the apply unresolvable
+ Set<Slot> keysToExpose = keySlots.stream()
+
.map(NamedExpression::toSlot).collect(ImmutableSet.toImmutableSet());
+ // The outputs of the top aggregate are exposed by the projections
above that aggregate
+ // alone, because the projections below it cannot produce them: the
aggregate which defines
+ // them sits above those projections. The predicates which were pulled
into the apply read
+ // the outputs of the top aggregate as well (for example the max(c) <=
t1.c1 of the HAVING
+ // clause), and the projection below the top aggregate has to carry
the keys alone. For
+ // example the subquery of
+ //
+ // select t1.c1 from t1 where t1.c1 in (select max(c) from (select
count(*) as c from t2
+ // where t2.c1 = t1.c1 group by t2.c2) x having max(c) <=
t1.c1)
+ //
+ // reaches the rewrite with the plan
+ //
+ // Apply(correlationFilter=[(max(c) <= t1.c1)])
+ // |-- t1
+ // +-- Project([max(c)]) [the
select list]
+ // +-- Aggregate(group by [], output [max(c) as max(c)])
+ // +-- Project([c]) [the
projection below the
+ // +-- Aggregate(group by [t2.c2],
aggregate which defines
+ // output [t2.c2, count(*) as c]) max(c)]
+ // +-- Filter(t2.c1 = t1.c1)
+ // +-- t2
+ //
+ // and appending max(c) to the projection of the count (the projection
below the aggregate
+ // which defines it) would make that projection read a slot which its
child cannot produce,
+ // so the plan would be rejected by the slot check of the rewrite.
+ Set<Slot> outputsOfTheTopAggregation = newCorrelationFilter.stream()
+ .flatMap(conjunct -> conjunct.getInputSlots().stream())
+ .filter(slot ->
newAggregations.get(aggregation.topAggregation()).getOutput().contains(slot))
+ .filter(slot -> !keysToExpose.contains(slot))
+ .collect(ImmutableSet.toImmutableSet());
+ // the predicates of the apply are evaluated on the nodes above the
aggregation of the
+ // subquery, which produce the outputs of that aggregation themselves,
so no output of it has
+ // to be appended to the projections below them
return new LogicalApply<>(apply.getCorrelationSlot(),
apply.getSubqueryType(), apply.isNot(),
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
- ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
+ ExpressionUtils.optionalAnd(newCorrelationFilter),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ rebuildTheAggregationChain(apply.right(), aggregation,
newAggregations, keysToExpose,
+ outputsOfTheTopAggregation, null, ImmutableSet.of(),
false, Maps.newHashMap()));
+ }
+
+ /**
+ * The aggregation of the subquery with the keys of the correlation added
to its group by and to
+ * its output: the rows of one correlation key are the rows of the
subquery for the outer rows
+ * which own that key, so an aggregation above the aggregation of the
domain may not mix them.
+ * For example the aggregate of the sum of the example of TheAggregation
is rewritten into
+ *
+ * Aggregate(group by [key.c1], output [sum(c) as sum(x.c), key.c1])
+ *
+ * around the rewritten aggregation of the domain, whose rows carry the
key as well (see
+ * rebuildTheAggregationChain).
+ */
+ private static LogicalAggregate<?>
withTheKeysInTheGroupBy(LogicalAggregate<?> aggregate,
+ List<? extends Expression> keys, Slot matchMarkerOfTheEmptyDomain,
+ boolean exposesTheMatchMarker, Map<Expression, Expression>
nullableInnerSlots,
+ boolean ignoresTheKeptRowOfAnEmptyDomain) {
+ List<Expression> groupBy = Lists.newArrayList(keys);
+ for (Expression groupByExpression : aggregate.getGroupByExpressions())
{
+ // the group by of an aggregate above the aggregation of the
domain may read the columns
+ // of the inner side as well, and the left outer join of the
domain reports them as
+ // nullable (see nullableInnerSlots)
+ groupBy.add(ExpressionUtils.replace(groupByExpression,
nullableInnerSlots));
+ }
+ if (matchMarkerOfTheEmptyDomain != null && exposesTheMatchMarker) {
+ // The marker of the row which is kept for an empty domain is read
by the guard of the
+ // aggregates above the aggregation of the domain and by the
filters between the
+ // aggregates (see rebuildTheAggregationChain), so those
aggregates expose it. The marker
+ // is null for the row which is kept for an empty domain, so the
grouping of the rows of a
+ // correlation key does not change. The top aggregate does not
expose it: the rows which it
+ // produces are the rows of the subquery, and the marker belongs
to the rows below it (the
+ // aggregates above the aggregation of the domain read it from
their own input).
+ groupBy.add(matchMarkerOfTheEmptyDomain);
+ }
+ List<NamedExpression> outputs = Lists.newArrayList();
+ if (!ignoresTheKeptRowOfAnEmptyDomain) {
+ // The row which the rewrite keeps for an empty correlated domain
is the row which the
+ // original subquery computes out of the empty input of the
aggregation of the domain (its
+ // own guard makes that aggregation return the value of an empty
input, see
+ // guardAggregateArguments), so the aggregates above it read the
value which that row
+ // carries (the count 0 of
+ // select (select count(*) from t2 where t2.c1 = t1.c1 having
count(*) = 0) from t1, for
+ // example). They read the columns of the inner side of the join
through the nullable slots.
+ for (NamedExpression output : aggregate.getOutputExpressions()) {
+ outputs.add((NamedExpression) ExpressionUtils.replace(output,
nullableInnerSlots));
+ }
+ } else {
+ // The subquery produces no row for an empty correlated domain (a
grouped aggregation of
+ // the domain reports an empty domain as "no row", and a HAVING
clause which does not hold
+ // for the row of a global aggregation of the domain removes that
row), so the aggregates
+ // above the aggregation of the domain ignore the row which the
rewrite keeps for that
+ // domain: they return the value of their empty input for it (see
guardAggregateArguments),
+ // which is the value the original subquery computes for the empty
domain as well.
+ Set<AggregateFunction> aggregates = Sets.newLinkedHashSet();
+ for (NamedExpression output : aggregate.getOutputExpressions()) {
+
aggregates.addAll(output.collect(AggregateFunction.class::isInstance));
+ }
+ Map<Expression, Expression> compensated =
guardAggregateArguments(aggregates,
+ matchMarkerOfTheEmptyDomain);
+ if (compensated == null) {
+ // an aggregate of the aggregation cannot be guarded, so the
row which is kept for an
+ // empty input cannot be told apart from a row of the rows
below the aggregation
+ return null;
+ }
+ for (NamedExpression output : aggregate.getOutputExpressions()) {
+ outputs.add((NamedExpression) ExpressionUtils.replace(
+ (NamedExpression) ExpressionUtils.replace(output,
compensated), nullableInnerSlots));
+ }
+ }
+ keys.forEach(key -> outputs.add((NamedExpression) key));
+ if (matchMarkerOfTheEmptyDomain != null && exposesTheMatchMarker
+ && !outputs.contains(matchMarkerOfTheEmptyDomain)) {
+ // The aggregates above this one and the filters between them read
the marker from their
+ // own input, and the outputs of an aggregate decide on the rows
which it produces: the
+ // marker is a group key of this aggregate (see above), so it has
to be one of its outputs
+ // as well, otherwise the nodes above it cannot read it.
+ outputs.add(matchMarkerOfTheEmptyDomain);
+ }
+ return new LogicalAggregate<>(groupBy, outputs, aggregate.child(0));
+ }
+
+ /**
+ * Whether the rewrite of the outer side has to keep one row for the
correlation keys whose rows
+ * below the aggregation of the domain are missing, although that
aggregation returns no row of its
+ * own for them: every aggregate above it is global, so the aggregation of
the original subquery
+ * produces one row for the empty input of such a key, and the aggregates
which the rewrite builds
+ * above that aggregation can be guarded with the marker of the row which
is kept for it (see
+ * guardAggregateArguments and withTheKeysInTheGroupBy). The value which
that row exposes is then
+ * the value which the original subquery exposes for the key. For example
the subquery of
+ *
+ * select o.k from o where o.k in (
+ * select coalesce(max(c), 0) from
+ * (select count(*) as c from i where i.k = o.k group by i.g)
x)
+ *
+ * returns one row whose value is 0 for the outer rows whose correlated
domain is empty (the max of
+ * the empty derived table is null and the coalesce turns that null into
the 0), so the outer row
+ * of the value 0 matches the subquery: the rewrite keeps a row for such a
key, the max above it
+ * ignores that row and returns the null of its empty input, and the
coalesce of the plan of the
+ * subquery turns that null into the 0 as well.
+ *
+ * An EXISTS subquery reads whether the row of such a key exists instead
of the value it exposes,
+ * so the row has to be kept when the HAVING clause of the subquery keeps
the row of the empty
+ * input (see the EXISTS branch below).
+ */
+ private static boolean keepsTheRowOfAnEmptyDomain(LogicalApply<?, ?>
apply, TheAggregation aggregation,
+ CorrelatedAggregatePredicates predicates) {
+ List<LogicalAggregate<?>> chain = aggregation.aggregationChain();
+ List<LogicalAggregate<?>> aboveTheDomain = chain.subList(0,
chain.size() - 1);
+ if (aboveTheDomain.isEmpty()) {
+ // the aggregation of the domain is the only aggregation of the
subquery: the rewrite of the
+ // outer side keeps a row of its own for an empty domain when that
aggregation is global,
+ // and no aggregate above it observes such a row
+ return false;
+ }
+ if (aboveTheDomain.stream().anyMatch(aggregate ->
!aggregate.getGroupByExpressions().isEmpty())) {
+ // an aggregate above the aggregation of the domain groups the
rows which it reads, so the
+ // row which is kept for an empty domain builds a group of its own
in that aggregate, while
+ // the aggregation of the original subquery produces no row at all
for such an empty input
+ return false;
+ }
+ if (!chain.stream()
+ .flatMap(aggregate ->
aggregate.getOutputExpressions().stream())
+ .flatMap(output ->
output.collect(AggregateFunction.class::isInstance).stream())
+ .allMatch(function -> function instanceof
NullIgnoringAggregateFunction)) {
+ // only the aggregates which ignore null arguments can be guarded,
so that the row which is
+ // kept for an empty domain does not contribute to them (see
guardAggregateArguments)
+ return false;
+ }
+ if (apply.isExist()) {
+ // The row which the aggregation of an empty correlated domain
produces exists for the
+ // subquery when the HAVING clause of the aggregation above the
one of the domain holds for
+ // the values of that empty input (see
havingMayHoldWithEmptyInput): the EXISTS of the
+ // subquery of
+ //
+ // select t1.c1 from t1 where exists (select max(c) from
(select count(*) as c from t2
+ // where t2.c1 = t1.c1 group by t2.c2) x having max(c) is
null)
+ //
+ // is true for the outer rows whose correlated domain is empty
(the max of the empty
+ // derived table is null and the HAVING clause keeps that row).
The rewrite keeps the row of
+ // such a key and lets the aggregates above the aggregation of the
domain return the values
+ // of an empty input for it, so that the nodes above the
aggregation decide on the row the
+ // way the original subquery does (see rebuildTheAggregationChain
and
+ // guardAggregateArguments). A HAVING clause which rejects the row
of the empty input
+ // (having max(c) > 0, for example) drops it, and the nodes above
the aggregation reject
+ // the row which the rewrite keeps for such a key as well.
+ List<Expression> havingConjuncts = predicates.havingPredicates();
+ return aboveTheDomain.stream()
+ .filter(aggregate ->
aggregate.getGroupByExpressions().isEmpty())
+ .anyMatch(aggregate ->
havingMayHoldWithEmptyInput(aggregate,
+ Sets.newLinkedHashSet(havingConjuncts)));
+ }
+ return true;
+ }
+
+ /**
+ * Whether the aggregation of the subquery holds a global aggregate above
an aggregate which can
+ * return no row for a correlation key, and the subquery observes the row
which that global
+ * aggregate returns for the empty input.
+ *
+ * The rewrite adds the correlation keys to the group by of every
aggregate of the chain (see
+ * pullUpCorrelatedFilter and withTheKeysInTheGroupBy), so a global
aggregate above the
+ * aggregation of the domain produces no row at all for a key whose rows
below it are missing,
+ * while the aggregation of the original subquery returns one row for that
empty input. The
+ * subquery of
+ *
+ * select t1.c1 from t1 where exists (select max(c) from (select
count(*) as c from t2
+ * where t2.c1 = t1.c1 group by t2.c2) x having max(c) is null)
+ *
+ * is true for the outer rows whose correlated domain is empty, because
the max of the empty
+ * derived table is null and the HAVING clause keeps that row, while a
rewrite which dropped the
+ * key would produce no row for it and the semi join would drop the outer
row. The aggregation of
+ * the inner side is not equivalent for such subqueries, and the
aggregation of the outer side is
+ * only equivalent when it keeps a row for the empty domain and lets the
aggregates above the
+ * aggregation of the domain return the values of an empty input for it
(see
+ * keepsTheRowOfAnEmptyDomain); the caller reports the subqueries which
neither of them can
+ * rewrite.
+ */
+ private static boolean
observesTheEmptyInputOfAGlobalAggregate(LogicalApply<?, ?> apply,
+ TheAggregation aggregation, CorrelatedAggregatePredicates
predicates) {
+ if (keepsTheRowOfAnEmptyDomain(apply, aggregation, predicates)) {
+ // the rewrite of the outer side keeps the row which such a key is
missing (see
+ // keepsTheRowOfAnEmptyDomain), so the subquery is not reported
+ return false;
+ }
+ return theEmptyInputOfAGlobalAggregateIsObservable(apply, aggregation,
predicates);
+ }
+
+ /**
+ * The detection of observesTheEmptyInputOfAGlobalAggregate on its own:
the subqueries
+ * which this detection reports are the subqueries whose rewrite would
drop the row which the
+ * aggregation of the original subquery returns for a correlation key
whose rows below the
+ * aggregation of the domain are missing. The rewrite of the outer side
keeps that row and lets the
+ * aggregates above the aggregation of the domain return the values of an
empty input for it when
+ * every one of them is a global aggregate which ignores null arguments
(see
+ * keepsTheRowOfAnEmptyDomain), and those subqueries are rewritten instead
of reported.
+ */
+ private static boolean
theEmptyInputOfAGlobalAggregateIsObservable(LogicalApply<?, ?> apply,
+ TheAggregation aggregation, CorrelatedAggregatePredicates
predicates) {
+ List<LogicalAggregate<?>> chain = aggregation.aggregationChain();
+ if (chain.get(chain.size() - 1).getGroupByExpressions().isEmpty()
+ && theFiltersBetweenTheAggregations(aggregation).isEmpty()) {
+ // the aggregation of the domain returns a row for every
correlation key, so no
Review Comment:
[P1] Preserve the guaranteed row of a lower global aggregate before keying
this chain. For `EXISTS (SELECT count(*) FROM (SELECT count(*) c FROM i WHERE
i.k=o.k) x GROUP BY c)`, an unmatched key still makes the original lower
`COUNT` emit `c=0`; the upper grouped aggregate forms that group and emits a
row, so `EXISTS` is true. This return selects the legacy keyed rewrite, which
adds `i.k` to the lower `COUNT`'s group-by. The unmatched key then has no lower
row or upper group, and the semi join incorrectly drops it (`NOT EXISTS` flips
conversely). This is reachable through the EXISTS analyzer and has no
HAVING/marker path, unlike the existing threads. Preserve the lower global row
through upper aggregates or reject this chain, and add unmatched-key
`EXISTS`/`NOT EXISTS` oracles.
--
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]
