github-actions[bot] commented on code in PR #67940:
URL: https://github.com/apache/doris/pull/67940#discussion_r4052475691
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/SubExprAnalyzer.java:
##########
@@ -125,7 +125,10 @@ public Expression visitInSubquery(InSubquery expr, T
context) {
AnalyzedResult analyzedResult = analyzeSubquery(expr);
checkOutputColumn(analyzedResult.getLogicalPlan());
- checkNoCorrelatedSlotsUnderAgg(analyzedResult);
+ // the correlated predicate of an IN subquery may sit below the
aggregation of the subquery:
+ // the rewrite which unnests it (UnCorrelatedApplyAggregateFilter)
computes the aggregation
+ // of the domain of every outer row, the empty correlated domain
included, so that the value
+ // which the IN compares exists for every outer row
Review Comment:
[P1] Keep unsupported outer references under IN aggregates behind a narrowed
analyzer guard. This removal also admits `o.k IN (SELECT sum(i.v + o.k) FROM
i)`: unlike the scalar path, `visitInSubquery` never runs
`CorrelatedSlotsValidator`, so the outer slot in the aggregate is accepted. The
resulting plan is `Apply(IN) -> [Scan(o), Aggregate[sum(i.v + o.k)] ->
Scan(i)]` with correlation slots but no correlation filter.
`UnCorrelatedApplyAggregateFilter` requires a correlated filter below the
aggregate and never matches, while `InApplyToJoin` emits only the IN equality.
The right aggregate still reads `o.k` from a scan that cannot produce it, so
`CheckAfterRewrite` rejects the newly admitted query. This is distinct from the
existing retained-project issue, which starts with a supported domain filter
and loses a rewrite-added key. Either restrict this removal to the filter shape
the PR handles, or extend unnesting to aggregate arguments/grouping/join
predicates and add a negative or
result oracle.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -97,28 +559,1746 @@ 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());
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));
+ }
+
+ /**
+ * 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) {
+ List<Expression> groupBy = Lists.newArrayList(keys);
+ groupBy.addAll(aggregate.getGroupByExpressions());
+ List<NamedExpression> outputs =
Lists.newArrayList(aggregate.getOutputExpressions());
+ keys.forEach(key -> outputs.add((NamedExpression) key));
+ return new LogicalAggregate<>(groupBy, outputs, aggregate.child(0));
+ }
+
+ /**
+ * 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 the
rewrite produces no row
+ * for those keys and the semi join drops the outer row. Neither the
aggregation of the inner
+ * side (a global aggregate above the aggregation of the domain would
aggregate the rows of every
+ * correlation key together) nor the aggregation of the outer side (it
groups that aggregate by
+ * the correlation key) is equivalent for such subqueries, so the caller
reports them.
+ */
+ private static boolean
observesTheEmptyInputOfAGlobalAggregate(LogicalApply<?, ?> apply,
+ TheAggregation aggregation, CorrelatedAggregatePredicates
predicates) {
+ List<LogicalAggregate<?>> chain = aggregation.aggregationChain();
+ if (chain.get(chain.size() - 1).getGroupByExpressions().isEmpty()) {
+ // the aggregation of the domain returns a row for every
correlation key, so no
+ // aggregate above it can observe an empty input
+ return false;
+ }
+ // the aggregates above the deepest one: the deepest one reads the
rows of the domain of a
+ // correlation key, and the predicates of that domain may leave them
empty
+ List<LogicalAggregate<?>> aboveTheDomain = chain.subList(0,
chain.size() - 1);
+ if (apply.isExist()) {
+ // the row which the global aggregate returns for the empty input
decides whether the
+ // EXISTS reports the outer row, unless the nodes above the
aggregation reject that row
+ // (the predicates of the HAVING clause which reference the outer
query were pulled into
+ // the apply, and they decide on the row of the empty input as
well)
+ List<Expression> havingConjuncts = predicates.havingPredicates();
+ return aboveTheDomain.stream()
+ .filter(aggregate ->
aggregate.getGroupByExpressions().isEmpty())
+ .anyMatch(aggregate ->
havingMayHoldWithEmptyInput(aggregate,
+ Sets.newLinkedHashSet(havingConjuncts)));
+ }
+ if (apply.isScalar()) {
+ // A scalar subquery exposes the output of the aggregation of its
domain: the join of
+ // the rewrite reports a null for the keys whose rows below the
aggregation are missing,
+ // and SubqueryToApply repairs that null with the nvl of the value
which the top
+ // aggregate returns for an empty input. A global aggregate below
the top aggregate
+ // changes the value which the top one computes out of the row of
the empty input.
+ boolean hasAGlobalAggregateBelowTheTop =
aboveTheDomain.stream().skip(1)
+ .anyMatch(aggregate ->
aggregate.getGroupByExpressions().isEmpty());
+ return hasAGlobalAggregateBelowTheTop
+ && (returnsAValueForAnEmptyInput(chain.get(0)) ||
aboveTheDomain.stream().skip(1)
+
.anyMatch(UnCorrelatedApplyAggregateFilter::returnsAValueForAnEmptyInput));
+ }
+ // An IN subquery compares the outer value with the value of the
aggregation of its domain:
+ // the value which a global aggregate returns for an empty input can
match the outer value,
+ // while the rewrite has no row to compare it with (a null value of
the aggregation does not
+ // match either, so an aggregation of nullable aggregates alone is
left alone).
Review Comment:
[P1] Preserve the NULL row for nested nullable aggregates used by IN/NOT IN.
For `o.k NOT IN (SELECT max(c) FROM (SELECT count(*) c FROM i WHERE i.k=o.k
GROUP BY i.g) x)`, an empty correlated domain makes the original global `MAX`
return one NULL row, so NOT IN is UNKNOWN and the outer row is rejected.
Because `MAX` is not a `NotNullableAggregateFunction`, this branch permits the
rewrite; adding the key to both aggregates then leaves no lower row carrying
that key, so the upper aggregate returns no row and `NOT IN (empty)` becomes
TRUE. A mark/value IN similarly changes from NULL to false. This differs from
the existing nested-EXISTS/HAVING report: this new detector handles that branch
but treats nullable IN results as unobservable, which is only true for positive
IN used directly as a filter. Reject or preserve nullable global stages for NOT
IN and mark/value contexts, and add those oracles.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -17,75 +17,537 @@
package org.apache.doris.nereids.rules.rewrite;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.hint.DistributeHint;
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.expression.rules.FoldConstantRuleOnFE;
+import org.apache.doris.nereids.trees.copier.DeepCopierContext;
+import org.apache.doris.nereids.trees.copier.LogicalPlanDeepCopier;
import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
+import org.apache.doris.nereids.trees.expressions.NullSafeEqual;
import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.VolatileExpression;
+import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
+import
org.apache.doris.nereids.trees.expressions.functions.NoneMovableFunction;
+import org.apache.doris.nereids.trees.expressions.functions.Udf;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.NotNullableAggregateFunction;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.NullIgnoringAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
+import org.apache.doris.nereids.trees.plans.DistributeType;
+import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalApply;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
import org.apache.doris.nereids.util.ExpressionUtils;
import org.apache.doris.nereids.util.PlanUtils;
import org.apache.doris.nereids.util.Utils;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
/**
- * Merge the correlated predicate and agg in the filter under apply.
- * And keep the unCorrelated predicate under agg.
- * <p>
- * Use the correlated column as the group by column of agg,
- * the output column is the correlated column and the input column.
- * <pre>
- * before:
- * apply
- * / \
- * Input(output:b) Filter(this node's existence depends on having
clause's existence)
- * |
- * agg(output:fn; group by:null)
- * |
- * Filter(correlated predicate(Input.e = this.f)/Unapply
predicate)
- *
- * end:
- * apply(correlated predicate(Input.e = this.f))
- * / \
- * Input(output:b) Filter(this node's existence depends on having clause's
existence)
- * |
- * agg(output:fn,this.f; group by:this.f)
- * |
- * Filter(Uncorrelated predicate)
- * </pre>
+ * Pull the correlated predicates of a subquery which aggregates out of the
subquery, so that the
+ * aggregation computes the aggregation of the correlated domain of one outer
row instead of the
+ * aggregation of all the inner rows of the subquery.
+ *
+ * The correlated predicates sit in the filter below the aggregation of the
subquery before the
+ * rewrite (its WHERE clause): the right side of the apply is the nodes above
that filter (the
+ * HAVING clause, the projections of the select list, the aggregation), the
filter itself and the
+ * inner table. This rule pulls the correlated conjuncts of that filter into
the apply (they become
+ * the conditions which unnest the apply) and rebuilds the aggregation around
them, keeping the
+ * nodes above it in place; the uncorrelated conjuncts stay in the filter
below the aggregation.
+ *
+ * The shape of the subquery selects one of two strategies for the new
aggregation:
+ *
+ * The aggregation stays on the inner side when the inner rows of one outer
row are exactly the
+ * groups of the aggregation: the inner side of the correlated predicate is
added to the group by
+ * of the aggregation and to its output (so that the conditions of the apply
can read the key), and
+ * every outer row is paired with the group of its own domain. This needs an
equality between the
+ * outer side and the inner side (eg. t2.c1 = t1.c1), because the value of the
inner side is what
+ * selects the group of the outer row. For the subquery of
+ *
+ * select t1.c1 from t1 where exists (select count(*) from t2
+ * where t2.c1 = t1.c1 group by t2.c2 having count(*) > 0)
+ *
+ * the rewritten plan is (the EXISTS reads the aggregation through a semi
join, see below):
+ *
+ * LEFT SEMI JOIN (t2.c1 = t1.c1) [the correlation
filter of the apply]
+ * |-- t1
+ * +-- Filter(count(*) > 0) [the HAVING
clause, kept in place]
+ * +-- Aggregate(group by [t2.c1, t2.c2], output [t2.c1, t2.c2,
count(*)])
+ * +-- t2
+ *
+ * The aggregation is built on the outer side in every other case: a deep copy
of the outer plan
+ * computes the distinct correlation keys, the keys are joined with the inner
side on the
+ * predicates of the domain, the result is grouped by the keys, and every
outer row is paired with
+ * the aggregation of its own key (see
pullUpCorrelatedPredicateByAggregatingOuter). For the
+ * subquery of
+ *
+ * select t1.c1 from t1 where exists (select count(*) from t2
+ * where t2.c1 < t1.c1 group by t2.c2 having count(*) = 2)
+ *
+ * the rewritten plan is:
+ *
+ * LEFT SEMI JOIN (t1.c1 <=> key.c1)
+ * |-- t1
+ * +-- Filter(count(*) = 2)
+ * +-- Aggregate(group by [key.c1, t2.c2], output [key.c1, t2.c2,
count(*)])
+ * +-- INNER JOIN (t2.c1 < key.c1)
+ * |-- Aggregate(group by [t1.c1], output [t1.c1])
[the keys of t1]
+ * | +-- t1
+ * +-- t2
+ *
+ * The two properties which select the strategy are independent: whether the
correlated predicate
+ * is an equality or not, and whether the aggregation has a group by or not (a
global aggregate).
+ * The four combinations:
+ *
+ * 1. equality + group by: inner side (the first example). The domain of an
outer row is exactly
+ * the group whose key is the value of the inner side, so the HAVING clause
of that group is the
+ * HAVING clause of the outer row, and an outer row with an empty domain
has no group at all,
+ * exactly like the grouped aggregate of the subquery.
+ *
+ * 2. non-equality + group by: outer side (the second example). The domain of
an outer row is the
+ * union of several groups, so the aggregates of the subquery (eg.
count(*)) are computed over
+ * the union, while one group of the inner side holds a part of the domain
only.
+ *
+ * 3. non-equality, no group by: outer side, for the reason of 2. For the
subquery of
+ *
+ * select t1.c1 from t1 where exists (select count(*) from t2
+ * where t2.c1 < t1.c1 having count(*) = 0)
+ *
+ * the plan is the plan of 2. without the group by of the subquery, with
the row of the empty
+ * domain kept by a left outer join (a global aggregate returns a row for
an empty correlated
+ * domain, and the count of that row has to be 0):
+ *
+ * LEFT SEMI JOIN (t1.c1 <=> key.c1)
+ * |-- t1
+ * +-- Filter(count($correlation_match_marker) = 0)
+ * +-- Aggregate(group by [key.c1], output [key.c1,
count($correlation_match_marker)])
+ * +-- LEFT OUTER JOIN (t2.c1 < key.c1)
[keeps the empty domain]
+ * |-- Aggregate(group by [t1.c1], output [t1.c1])
+ * | +-- t1
+ * +-- Project([true AS $correlation_match_marker,
t2.c1])
+ * +-- t2
+ *
+ * 4. equality, no group by: inner side when the subquery does not need the
row which the global
+ * aggregate returns for an empty correlated domain, outer side when it
does, because that row
+ * has no group on the inner side to be produced from. For the subquery of
+ *
+ * select t1.c1 from t1 where exists (select count(*) from t2
+ * where t2.c1 = t1.c1 having count(*) > 0)
+ *
+ * the HAVING clause is false for the empty input, so the row of the empty
domain disappears
+ * from the result anyway and the aggregation stays on the inner side (the
plan of 1. without
+ * the group by of the subquery); for the subquery of
+ *
+ * select t1.c1 from t1 where exists (select count(*) from t2
+ * where t2.c1 = t1.c1 having count(*) = 0)
+ *
+ * the row of the empty domain survives (its count is 0), so the
aggregation is built on the
+ * outer side (the plan of 3. with the equality of the domain):
+ *
+ * LEFT SEMI JOIN (t1.c1 <=> key.c1)
+ * |-- t1
+ * +-- Filter(count($correlation_match_marker) = 0)
+ * +-- Aggregate(group by [key.c1], output [key.c1,
count($correlation_match_marker)])
+ * +-- LEFT OUTER JOIN (t2.c1 = key.c1)
[keeps the empty domain]
+ * |-- Aggregate(group by [t1.c1], output [t1.c1])
+ * | +-- t1
+ * +-- Project([true AS $correlation_match_marker,
t2.c1])
+ * +-- t2
+ *
+ * The three subquery types differ in what they need from the aggregation (see
+ * needCorrelatedAggregationOnOuter for the exact conditions):
+ *
+ * - EXISTS/NOT EXISTS is decided by the aggregation, which this rule reads
through a LEFT SEMI
+ * JOIN (a LEFT ANTI JOIN for NOT EXISTS): it needs the row of an empty
correlated domain only
+ * when that row can decide the subquery (a HAVING clause which may hold for
it, or a node which
+ * the subquery keeps above its aggregation), see the examples of 4.
+ * - IN/NOT IN compares the outer expression with the value of the subquery,
which is the first
+ * column of the aggregation (the keys are appended after the outputs of the
subquery, so this
+ * column does not move). The row which a global aggregate returns for an
empty correlated domain
+ * is part of the values the IN compares: the subquery of
+ *
+ * select t1.c1 from t1 where t1.c1 in (select count(*) from t2 where
t2.c1 = t1.c1)
+ *
+ * compares the outer row with the count 0 of its empty domain (k in (0)
holds for the outer row
+ * 0), so its aggregation is built on the outer side (the plan of 3. with
the count of the
+ * marker) and the rewrite keeps the apply, whose correlation filter pairs
every outer row with
+ * the aggregation of its own key; the rule which converts the IN into a
join then compares the
+ * outer expression with the first column of that aggregation. That row is
needed for every
+ * aggregate, whatever value it returns for an empty input: the aggregation
of the inner side has
+ * no group for an outer row without a match, and the comparison which the
rewrite of the IN builds
+ * reads the rows of the domain of one outer row, so the null which the left
outer join keeps for it
+ * is not the value of its empty domain for the IN. The not in of
+ *
+ * select t1.c1 from t1 where t1.c1 not in (select sum(t2.c2) from t2
where t2.c1 = t1.c1)
+ *
+ * returns true for the outer rows without a match instead of the null which
the sum of the empty
+ * domain produces. A grouped aggregate has no value for an empty domain, so
the subquery of
+ *
+ * select t1.c1 from t1 where t1.c1 in (select count(*) from t2
+ * where t2.c1 = t1.c1 group by t2.c2)
+ *
+ * keeps the aggregation on the inner side (the plan of 1. with the
projection of the select list
+ * kept below the apply, see PullUpProjectUnderApply):
+ *
+ * Apply(IN, correlationFilter=[(t2.c1 = t1.c1)])
+ * |-- t1
+ * +-- Project([count(*), t2.c1]) [the select list,
the key appended]
+ * +-- Aggregate(group by [t2.c1, t2.c2], output [t2.c2,
count(*), t2.c1])
+ * +-- t2
+ * - scalar exposes the value of the aggregation as the value of the subquery.
The left outer join
+ * which pairs an outer row with the aggregation of its own key returns null
for the outer rows
+ * whose domain is empty, which is the value of those outer rows for the
aggregates whose value
+ * for an empty input the rewrite reads that way (see
keepsTheValueOfAnEmptyDomain): the subquery of
+ *
+ * select t1.c1, (select max(t2.c2) from t2 where t2.c1 = t1.c1) from t1
+ *
+ * keeps the aggregation on the inner side (max returns null for an empty
input, which is the
+ * null of the join), while the subquery of
+ *
+ * select t1.c1, (select count(*) from t2 where t2.c1 < t1.c1) from t1
+ *
+ * is built on the outer side (the correlated predicate is not an equality,
and the count of an
+ * empty domain is 0, which no null of the inner side can produce):
+ *
+ * LEFT OUTER JOIN (t1.c1 <=> key.c1)
+ * |-- t1
+ * +-- Aggregate(group by [key.c1], output [key.c1,
count($correlation_match_marker)])
+ * +-- LEFT OUTER JOIN (t2.c1 < key.c1)
[keeps the empty domain]
+ * |-- Aggregate(group by [t1.c1], output [t1.c1])
+ * | +-- t1
+ * +-- Project([true AS $correlation_match_marker, t2.c1])
+ * +-- t2
+ *
+ * To summarize, the aggregation of the subquery stays on the inner side for
the subqueries whose
+ * correlated domain of one outer row is one group of the aggregation and
whose empty domain needs
+ * no row: every equality predicate with a group by, the global aggregates
whose HAVING clause
+ * rejects the empty input (or whose value for the empty input is the null
which the left outer
+ * join keeps), the grouped IN subqueries, and the scalar subqueries of the
aggregates whose empty
+ * value the rewrite reads from that null. It is built on the outer side for
the subqueries whose
+ * domain is several groups of the aggregation (a non-equality predicate,
whatever the subquery type
+ * and the aggregate are), and for the subqueries which need the row of an
empty correlated domain (a
+ * global aggregate whose HAVING clause may hold for the empty input, an IN
subquery which compares
+ * the value of a global aggregate, an EXISTS subquery which keeps nodes above
its aggregation, a
+ * scalar subquery whose empty value is not the null of the join).
+ *
+ * Keeping the aggregation on the inner side in the cases of the outer side
returns a wrong result,
+ * which is why the rewrite of the outer side has to exist. The subquery of
+ *
+ * select cq_o.k from cq_o where exists (select count(*) from cq_i
+ * where cq_i.k = cq_o.k having count(*) = 0)
+ *
+ * returns the outer row 7 when its domain is empty (the count of an empty
domain is 0, which the
+ * HAVING clause accepts), while the aggregation of the inner side groups the
inner rows by
+ * cq_i.k: the outer row 7 has no group there, so its row would be dropped.
The subquery of
+ *
+ * select cn_o.k from cn_o where exists (select count(*) from cn_i
+ * where cn_i.k < cn_o.k group by cn_i.g having count(*) = 2)
+ *
+ * returns the outer row 3 when its domain is the two rows (1, 10) and (2, 10)
of the single group
+ * g = 10 (the count of the domain is 2), while the aggregation of the inner
side groups the inner
+ * rows by (cn_i.k, cn_i.g): those two rows are two groups of count 1 and the
HAVING clause would
+ * never hold.
*/
public class UnCorrelatedApplyAggregateFilter implements RewriteRuleFactory {
+
+ /**
+ * name of the projected column which tells whether an inner row matched
the correlated
+ * predicate: the left outer join of the aggregation of the outer side
keeps one row whose
+ * marker is null for every correlation key which has no inner row, so the
aggregations of this
+ * column see the empty correlated domain as an empty input (see
guardAggregateArguments). For
+ * example the plan of the second case of needCorrelatedAggregationOnOuter
evaluates
+ * count($correlation_match_marker) over the rows of Project([true AS
+ * $correlation_match_marker, t2.c1]), and the row which the left outer
join keeps for an empty
+ * domain has a null marker, which that count does not count.
+ */
+ private static final String CORRELATION_MATCH_MARKER =
"$correlation_match_marker";
+
@Override
public List<Rule> buildRules() {
+ // The nodes between the apply and the aggregate of the subquery are
projections and filters
+ // in any order, so the rules cannot enumerate their shapes: match
every correlated apply
+ // whose right side starts with one of them and locate the aggregate
in the rule.
return ImmutableList.of(
- logicalApply(any(), logicalAggregate(logicalFilter()))
- .when(LogicalApply::isCorrelated)
-
.then(UnCorrelatedApplyAggregateFilter::pullUpCorrelatedFilter)
- .toRule(RuleType.UN_CORRELATED_APPLY_AGGREGATE_FILTER),
- logicalApply(any(),
logicalFilter(logicalAggregate(logicalFilter())))
+ logicalApply(any(), subTree(LogicalAggregate.class,
LogicalProject.class, LogicalFilter.class))
.when(LogicalApply::isCorrelated)
+ .when(apply -> locateAggregate(apply).isPresent())
.then(UnCorrelatedApplyAggregateFilter::pullUpCorrelatedFilter)
-
.toRule(RuleType.UN_CORRELATED_APPLY_FILTER_AGGREGATE_FILTER));
+
.toRule(RuleType.UN_CORRELATED_APPLY_AGGREGATE_FILTER));
+ }
+
+ /**
+ * The aggregation of the subquery: the aggregates between the apply and
the filter which holds
+ * the predicates of its WHERE clause (from the one below the apply to the
one above that
+ * filter), the filters whose predicates decide which rows of that
aggregation survive (the
+ * HAVING clause of the subquery), and the filter of the WHERE clause
itself.
+ *
+ * The nodes of the subquery of
+ *
+ * select t1.c1 from t1 where exists (
+ * select sum(x.c) from (select count(*) as c from t2
+ * where t2.c1 = t1.c1 group by t2.c2 having count(*) > 1) x
having sum(x.c) > 2)
+ *
+ * before the rewrite, with the member which holds every node:
+ *
+ * Apply(correlationFilter=empty)
[the apply of the subquery]
+ * |-- t1
+ * +-- Filter(sum(x.c) > 2) [havingFilter,
filtersAboveTheAggregation(0)]
+ * +-- Aggregate(group by [], output [sum(c) as sum(x.c)])
[chain(0) =
+ * +-- Project([c])
topAggregation]
+ * +-- Filter(c > 1)
[filtersAboveTheAggregation(1)]
+ * +-- Aggregate(group by [t2.c2], output
[t2.c2, count(*) as c])
+ * ...
[chain(1) =
+ * +-- Project([t2.c2])
domainAggregation]
+ * +-- Filter(t2.c1 = t1.c1)
[domainFilter]
+ * +-- t2
+ *
+ * - chain: every aggregate between the apply and the filter of the WHERE
clause, from the top
+ * down (here the sum over the derived table, then the count of that
table);
+ * - topAggregation: chain(0), the aggregate below the apply, which the
whole chain is rebuilt
+ * around;
+ * - domainAggregation: the deepest aggregate of the chain, which reads
the rows of the filter
+ * of the WHERE clause;
+ * - filtersAboveTheAggregation: the filters whose predicates decide on
the rows of the
+ * aggregation below them (here both HAVING clauses: the sum(x.c) > 2 of
the subquery and the
+ * count(*) > 1 of the derived table); a filter whose predicate selects
the rows of the
+ * domain of an outer row instead is not one of them, see
selectsTheRowsOfTheDomain;
+ * - havingFilter: the deepest filter above the topAggregation, i.e. the
first filter met when
+ * walking down from the apply (here the filter of sum(x.c) > 2, which
sits directly above
+ * the aggregation of the sum);
+ * - domainFilter: the filter which holds the predicates of the WHERE
clause (here
+ * t2.c1 = t1.c1, which the walk finds when it stops below the deepest
aggregate).
+ *
+ * A subquery may wrap its aggregation with further aggregations:
SubqueryToApply adds a
+ * count(*)/any_value(*) aggregation above the aggregation of a correlated
scalar subquery whose
+ * output is used in the outer scope and which has no top level scalar
aggregation (the count is
+ * the runtime check of the scalar subquery, the any_value is its value),
and the user may
+ * aggregate the aggregation of a derived table (the example above). The
rows of the subquery
+ * for one outer row are the rows of the whole chain for the correlation
key of that row, so
+ * every aggregate of the chain has to keep the keys in its group by (see
pullUpCorrelatedFilter
+ * and withTheKeysInTheGroupBy).
+ */
+ private static final class TheAggregation {
+ /** the aggregates between the apply and the filter of the WHERE
clause, from the top down */
+ private final List<LogicalAggregate<?>> chain;
+ /** every filter above the deepest aggregate, whose predicates decide
on the aggregation rows */
+ private final List<LogicalFilter<Plan>> filtersAboveTheAggregation;
+ /** the deepest filter above the top aggregate, which sits directly
below the projections above it */
+ private final Optional<LogicalFilter<Plan>> havingFilter;
+ /** the filter which holds the predicates of the WHERE clause of the
subquery */
+ private final LogicalFilter<Plan> domainFilter;
+
+ private TheAggregation(List<LogicalAggregate<?>> chain,
+ List<LogicalFilter<Plan>> filtersAboveTheAggregation,
+ Optional<LogicalFilter<Plan>> havingFilter,
LogicalFilter<Plan> domainFilter) {
+ this.chain = chain;
+ this.filtersAboveTheAggregation = filtersAboveTheAggregation;
+ this.havingFilter = havingFilter;
+ this.domainFilter = domainFilter;
+ }
+
+ /** the aggregate which reads the rows of the filter, at the bottom of
the chain */
+ private LogicalAggregate<?> domainAggregation() {
+ return chain.get(chain.size() - 1);
+ }
+
+ /** the aggregate below the apply, which the whole chain is rebuilt
around */
+ private LogicalAggregate<?> topAggregation() {
+ return chain.get(0);
+ }
+
+ /** the aggregates between the apply and the filter, from the top down
*/
+ private List<LogicalAggregate<?>> aggregationChain() {
+ return chain;
+ }
+
+ /** every filter above the deepest aggregate, whose predicates decide
on the aggregation rows */
+ private List<LogicalFilter<Plan>> filtersAboveTheAggregation() {
+ return filtersAboveTheAggregation;
+ }
+
+ /** the filter which holds the predicates of the WHERE clause of the
subquery */
+ private LogicalFilter<Plan> domainFilter() {
+ return domainFilter;
+ }
+
+ /** whether the subquery does not wrap its aggregation with another
aggregation */
+ private boolean onlyTheAggregationOfTheDomain() {
+ return chain.size() == 1;
+ }
+ }
+
+ /**
+ * Locate the aggregation of the subquery, walking down from the right
side of the apply: the
+ * nodes above its aggregates are the projections and the filters of the
subquery, the
+ * projections between its aggregates only carry the columns which they
need, and the filter
+ * below the deepest aggregate holds the predicates of its WHERE clause.
The walk stops at a
+ * filter whose child is not an aggregate: that filter is the filter of
the WHERE clause, while
+ * a filter which sits directly above an aggregate (through projections)
is a HAVING clause of
+ * that aggregate.
+ *
+ * The steps of the walk for the subquery of
+ *
+ * select t1.c1 from t1 where exists (
+ * select sum(x.c) from (select count(*) as c from t2
+ * where t2.c1 = t1.c1 group by t2.c2 having count(*) > 1) x
having sum(x.c) > 2)
+ *
+ * whose plan is (see TheAggregation):
+ *
+ * Apply(t1, Filter(sum(x.c) > 2) - Aggregate(sum) - Project([c]) -
Filter(c > 1)
+ * - Aggregate(count) - Project([t2.c2]) - Filter(t2.c1 = t1.c1) -
t2)
+ *
+ * 1. the first loop walks down from the apply while the node is not an
aggregate: the filter
+ * sum(x.c) > 2 is remembered as the havingFilter and added to
filtersAboveTheAggregation,
+ * and the walk stops at the aggregate of the sum (the topAggregation
of the chain);
+ * 2. the second loop walks down from that aggregate: the projection of c
carries a column of
+ * the node below it alone, so it is walked through; the filter c > 1
sits above the
+ * aggregate of the count (through that projection) and its predicate
reads the output of
+ * that aggregate, so it decides on its rows (a HAVING clause of that
aggregate): it is
+ * added to filtersAboveTheAggregation and the chain continues below it;
+ * 3. the aggregate of the count is the deepest aggregate of the chain
(the domainAggregation,
+ * which reads the rows of the filter of the WHERE clause);
+ * 4. the projection of t2.c2 carries a column of the node below it alone,
so it is walked
+ * through, and the walk stops at the filter t2.c1 = t1.c1: its child
is a scan and not an
+ * aggregate, so it is the domainFilter, which holds the predicates of
the WHERE clause (the
+ * correlated predicate is one of them). The walk returns an empty
result (the rule leaves
+ * the apply alone) when a node of the subquery is neither a projection
nor a filter nor an
+ * aggregate, when a projection below an aggregate computes one of its
columns instead of
+ * passing a column of the node below it through, and when the filter
of the WHERE clause
+ * has no child.
+ */
+ private static Optional<TheAggregation> locateAggregate(LogicalApply<?, ?>
apply) {
+ Plan below = apply.right();
+ Optional<LogicalFilter<Plan>> havingFilter = Optional.empty();
+ List<LogicalFilter<Plan>> filtersAboveTheAggregation =
Lists.newArrayList();
+ while (!(below instanceof LogicalAggregate)) {
+ if (below instanceof LogicalFilter) {
+ havingFilter = Optional.of((LogicalFilter<Plan>) below);
+ filtersAboveTheAggregation.add((LogicalFilter<Plan>) below);
+ } else if (!(below instanceof LogicalProject)) {
+ return Optional.empty();
+ }
+ below = below.child(0);
+ }
+ List<LogicalAggregate<?>> chain = Lists.newArrayList();
+ chain.add((LogicalAggregate<?>) below);
+ Plan belowAggregate = below.child(0);
+ LogicalFilter<Plan> domainFilter = null;
+ while (true) {
+ if (belowAggregate instanceof LogicalProject) {
+ for (NamedExpression project : ((LogicalProject<?>)
belowAggregate).getProjects()) {
+ if (!(project instanceof Slot)) {
Review Comment:
[P1] Handle computed scalar outputs between the cardinality wrapper and
grouped aggregate. A newly admitted query such as `SELECT o.k, (SELECT count(*)
+ 1 FROM i WHERE i.k=o.k GROUP BY i.g) FROM o` normalizes to `Apply(SCALAR) ->
wrapper Aggregate[count(*), any_value(v)] -> Project[count + 1 AS v] -> grouped
Aggregate[count] -> Filter[i.k=o.k]`. This check rejects the computed project,
and the fallback matcher only handles `Aggregate -> Project -> Filter`, so no
rule pulls up `i.k=o.k`; `ScalarApplyToJoin` then takes the no-filter path and
`CheckAfterRewrite` rejects the retained outer slot. This is distinct from the
existing project-key reports: those rewrites run and then hide/reference a
newly added key, while this supported WHERE correlation never reaches the
rewrite. Rebuild or carry this normalized project chain (or reject it at the
widened analyzer boundary), and cover computed grouped-scalar outputs with
zero, one, and multiple groups.
--
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]
