github-actions[bot] commented on code in PR #67940:
URL: https://github.com/apache/doris/pull/67940#discussion_r4072908212
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -17,75 +17,611 @@
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.IsNull;
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));
}
- private static LogicalApply<?, ?> pullUpCorrelatedFilter(LogicalApply<?,
?> apply) {
- boolean isRightChildAgg = apply.right() instanceof LogicalAggregate;
- // locate agg node
- LogicalAggregate<LogicalFilter<Plan>> agg =
- isRightChildAgg ? (LogicalAggregate<LogicalFilter<Plan>>)
(apply.right())
- : (LogicalAggregate<LogicalFilter<Plan>>)
(apply.right().child(0));
- LogicalFilter<Plan> filter = agg.child();
+ /**
+ * 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) {
+ // the projections between the aggregates of the chain are
kept by the rewrite, which
+ // exposes the keys through them (see
rebuildTheAggregationChain), so a projection
+ // which computes its columns is accepted when it reads the
rows of an aggregate of
+ // the chain: the aggregation below it and the aggregation
above it are rebuilt
+ // around it. The projection which carries the columns of the
rows below the
+ // aggregation of the domain into that aggregation, on the
other hand, is dropped
+ // together with the filter of the WHERE clause (the rewritten
aggregation reads the
+ // child of that filter), so it may only pass those columns
through. For example the
+ // projection of the subquery of
+ //
+ // select o.k, (select count(*) + 1 from i where i.k = o.k
group by i.g) from o
+ //
+ // computes the value which the wrapper aggregation of the
scalar subquery reads
+ // (count(*) + 1), and it sits between that aggregation and
the aggregation of the
+ // count: it is kept, while the projection of the example of
locateAggregate may only
+ // carry the column below it (the sum of the subquery
aggregates that column).
+ Plan belowTheProjections = belowAggregate.child(0);
+ while (belowTheProjections instanceof LogicalProject) {
+ belowTheProjections = belowTheProjections.child(0);
+ }
+ if (!(belowTheProjections instanceof LogicalAggregate)) {
+ for (NamedExpression project : ((LogicalProject<?>)
belowAggregate).getProjects()) {
+ if (!(project instanceof Slot)) {
+ // the projection computes the columns of the
nodes above it itself, so it
+ // cannot be replaced together with the
aggregation below it
+ return Optional.empty();
+ }
Review Comment:
[P1] Reject or uncorrelate computed projects below the deepest aggregate. A
valid shape such as `o.k NOT IN (SELECT max(c) FROM (SELECT count(z) c FROM
(SELECT i.g, i.v + 1 z FROM i WHERE i.k = o.k) p GROUP BY p.g HAVING count(z) >
0) x)` reaches this loop as `Aggregate(max) -> Project -> Filter ->
Aggregate(count) -> Project(computed z) -> Filter(i.k=o.k)`. This return leaves
the Apply unchanged, while `PullUpCorrelatedFilterUnderApplyAggregateProject`
only reaches a computed Project directly under the Apply-level aggregate and
cannot cross the upper aggregate/HAVING chain. `InApplyToJoin` then has no
correlation filter to add and final slot validation rejects the `o.k` left
inside the right subtree. Please extend the pull-up through the admitted unary
chain or reject this placement during IN analysis, and add a planning/result
case with an unmatched outer key.
--
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]
