github-actions[bot] commented on code in PR #67940:
URL: https://github.com/apache/doris/pull/67940#discussion_r4063718504
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -97,28 +633,2112 @@ 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));
+ }
+
+ /**
+ * 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));
Review Comment:
[P1] Expose the empty-domain marker from every intermediate aggregate. For a
three-stage chain such as `o.k NOT IN (SELECT max(m) FROM (SELECT max(c) m FROM
(SELECT count(*) c FROM i WHERE i.k=o.k GROUP BY i.g) x) y)`, the keep-empty
path rewrites the middle MAX with `exposesTheMatchMarker=true`: it groups by
the marker and guards `MAX(c)` with it, but this line outputs only the original
expressions plus the correlation keys. `LogicalAggregate.computeOutput()`
therefore omits the marker, while the rewritten top MAX reads it in
`MAX(IF(marker,m,NULL))` (and an intervening Project may also be asked to carry
it), so `CheckAfterRewrite` rejects the valid query. Append the marker to the
explicit outputs whenever it must be exposed, and cover a 3+ aggregate chain.
Please also model HAVING filters at every aggregate boundary: the current
helper only relaxes the interval immediately above the domain aggregate, so
fixing this missing slot alone exposes the higher-HAVING NULL-versus-empty e
rror.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -97,28 +633,2112 @@ 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));
+ }
+
+ /**
+ * 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));
+ 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()
+ &&
theFiltersBetweenTheAggregationOfTheDomainAndTheOneAboveIt(aggregation).isEmpty())
{
+ // the aggregation of the domain returns a row for every
correlation key, so no
+ // aggregate above it can observe an empty input (a filter between
those aggregations is
+ // the HAVING clause of the aggregation of the domain: it decides
on the row of the empty
+ // input and may reject it, which the aggregates above it observe)
+ 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).
+ if (aboveTheDomain.stream()
+
.anyMatch(UnCorrelatedApplyAggregateFilter::returnsAValueForAnEmptyInput)) {
+ return true;
+ }
+ // The nodes above the aggregation of the domain may expose a value of
their own for the empty
+ // input as well, even though the aggregates are nullable: the
projection of 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)
+ //
+ // turns the null which the max of the empty derived table returns
into the 0 which an outer
+ // row with the value 0 compares with, while the rewrite has no row to
compare it with and the
+ // semi join drops that row (see exposesAValueForAnEmptyInput).
+ if (exposesAValueForAnEmptyInput(apply, aggregation, aboveTheDomain)) {
+ return true;
+ }
+ // The missing row of a key is observable when the result of the IN is
not read as the decision
+ // on the outer row alone: the null which the subquery of the original
query compares with
+ // (the row of the aggregation of an empty derived table, for example)
makes the IN unknown,
+ // while the rewrite compares with nothing, which is false for an IN
and true for a NOT IN.
+ // The result of an IN which is used as a value is its mark, so its
null and its false are
+ // observable as well (the plan of such an IN is a mark join). For
example the subquery of
+ //
+ // select o.k from o where o.k not in (
+ // select max(c) from (select count(*) as c from i where i.k =
o.k group by i.g) x)
+ //
+ // returns one row for the outer rows whose correlated domain is empty
(the max of the empty
+ // derived table is null), so their NOT IN is unknown and those rows
are not returned, while
+ // the rewrite produces no row for those keys and their NOT IN is
true. A global aggregate
+ // above the aggregation of the domain is what makes such a row
disappear: the key is added to
+ // the group by of that aggregate (see withTheKeysInTheGroupBy), so a
key without rows below it
+ // has no group at all.
+ return (apply.isNot() || apply.getMarkJoinSlotReference().isPresent())
+ && aboveTheDomain.stream()
+ .anyMatch(aggregate ->
aggregate.getGroupByExpressions().isEmpty());
+ }
+
+ /**
+ * Whether one of the aggregates of the aggregation declares a value of
its own for an empty
+ * input (the count 0 or the empty array of an array_agg, for example). An
aggregate which
+ * declares no such value is left to the rewrite of the other cases, which
reads the value of an
+ * empty input the way the rest of the engine does (see
keepsTheValueOfAnEmptyDomain): the value
+ * of a UDAF is written in the UDAF itself, so its declaration does not
tell it.
+ */
+ private static boolean returnsAValueForAnEmptyInput(LogicalAggregate<?>
aggregate) {
+ for (NamedExpression output : aggregate.getOutputExpressions()) {
+ if (output.collect(AggregateFunction.class::isInstance).stream()
+ .anyMatch(function -> function instanceof
NotNullableAggregateFunction)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * The filters which sit between the aggregation of the domain and the
aggregation above it: they
+ * decide on the rows which the aggregation of the domain produces (they
are the HAVING clauses of
+ * the aggregation below them), so they are not evaluated for a
correlation key whose rows below
+ * that aggregation are missing (see keepsTheRowOfAnEmptyDomain).
+ */
+ private static Set<LogicalFilter>
theFiltersBetweenTheAggregationOfTheDomainAndTheOneAboveIt(
+ TheAggregation aggregation) {
+ if (aggregation.topAggregation() == aggregation.domainAggregation()) {
+ // the aggregation of the domain is the only aggregation of the
subquery, so there is no
+ // aggregation above it and no filter between such aggregations
either (the filter of the
+ // domain itself reads the rows of the domain and is not a filter
between the aggregates)
+ return ImmutableSet.of();
+ }
+ List<LogicalAggregate<?>> chain = aggregation.aggregationChain();
+ Plan below = chain.get(chain.size() - 2).child(0);
+ Set<LogicalFilter> filters = Sets.newLinkedHashSet();
+ while (below != aggregation.domainAggregation()) {
+ if (below instanceof LogicalFilter) {
+ filters.add((LogicalFilter) below);
+ }
+ below = below.child(0);
+ }
+ return filters;
+ }
+
+ /**
+ * Whether the nodes above the top aggregate expose a value of their own
for the empty input of a
+ * correlation key: the input of such a key is empty when the aggregation
of the domain has no row
+ * for it and a global aggregate above that aggregation returns no row for
that key in the
+ * rewrite, while the aggregation of the original subquery computes the
nodes above it out of the
+ * row of the empty input. The projection of 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)
+ *
+ * turns the null which the max of the empty derived table returns into
the 0 which the outer row
+ * with the value 0 compares with, while the rewrite has no row to compare
it with: the semi join
+ * would drop that outer row. The value which a node computes for the
empty input is read by
+ * replacing the outputs of the top aggregate with the values they return
for it and folding the
+ * expression.
+ */
+ private static boolean exposesAValueForAnEmptyInput(LogicalApply<?, ?>
apply, TheAggregation aggregation,
+ List<LogicalAggregate<?>> aboveTheDomain) {
+ if (aboveTheDomain.stream().noneMatch(aggregate ->
aggregate.getGroupByExpressions().isEmpty())) {
+ // every aggregate above the aggregation of the domain groups the
rows it reads, so a key
+ // without rows below that aggregation has no group in those
aggregates either
+ return false;
+ }
+ Map<Expression, Expression> emptyValues = Maps.newHashMap();
+ Set<Slot> outputsOfTheTopAggregation = Sets.newHashSet();
+ for (NamedExpression output :
aggregation.topAggregation().getOutputExpressions()) {
+ Expression expression = output instanceof Alias ? ((Alias)
output).child() : output;
+ if (!(expression instanceof AggregateFunction)) {
+ continue;
+ }
+ Expression emptyValue =
emptyValueForEmptyInput((AggregateFunction) expression);
+ if (emptyValue == null) {
+ // the aggregate declares no value for an empty input: it
returns the null of the empty
+ // input (the max of no row, for example)
+ emptyValue = new NullLiteral(output.getDataType());
+ }
+ emptyValues.put(output.toSlot(), emptyValue);
+ outputsOfTheTopAggregation.add(output.toSlot());
+ }
+ if (emptyValues.isEmpty()) {
+ return false;
+ }
+ Plan below = apply.right();
+ while (below != aggregation.topAggregation()) {
+ if (below instanceof LogicalProject) {
+ for (NamedExpression project : ((LogicalProject<?>)
below).getProjects()) {
+ Expression expression = project instanceof Alias ?
((Alias) project).child() : project;
+ if (Sets.intersection(expression.getInputSlots(),
outputsOfTheTopAggregation).isEmpty()) {
+ // the projection does not read the aggregation of the
domain
+ continue;
+ }
+ Expression folded =
FoldConstantRuleOnFE.evaluateWithoutContext(
+ ExpressionUtils.replace(expression, emptyValues));
+ if (folded instanceof Literal && !(folded instanceof
NullLiteral)) {
+ // the nodes above the aggregation expose this value
for the empty input
+ return true;
+ }
+ }
+ }
+ below = below.child(0);
+ }
+ return false;
+ }
+
+ /**
+ * Replace every aggregate of the chain with its rewritten version and
expose the keys through
+ * the projections between them, so that every aggregate above the deepest
one can group by the
+ * keys. The walk stops at the deepest aggregate: its rewritten version
already reads the rows of
+ * the filter below it (see pullUpCorrelatedFilter).
+ *
+ * The projections below the top aggregate only expose the keys, while the
projections above it
+ * expose the outputs of the top aggregate as well: the aggregate which
defines those outputs
+ * sits above the projections below it, so they cannot produce them. The
outputs to expose are
+ * therefore dropped as soon as the walk reaches the top aggregate.
+ *
+ * For example the plan of the example of TheAggregation is rewritten into
+ *
+ * Apply(correlationFilter=[t2.c1 = t1.c1])
+ * |-- t1
+ * +-- Filter(sum(x.c) > 2)
+ * +-- Aggregate(group by [key.c1], output [sum(c) as
sum(x.c), key.c1])
+ * +-- Project([c, key.c1]) [the key
exposed between the
+ * +-- Filter(c > 1)
aggregates]
+ * +-- Aggregate(group by [key.c1, t2.c2],
+ * output [t2.c2, count(*) as c,
key.c1])
+ * +-- t2
+ */
+ private static Plan rebuildTheAggregationChain(Plan plan, TheAggregation
aggregation,
+ Map<LogicalAggregate<?>, Plan> newAggregations, Set<Slot>
keysToExpose,
+ Set<Slot> outputsOfTheTopAggregation, Slot
matchMarkerOfTheEmptyDomain,
+ Set<LogicalFilter> filtersWhichKeepTheRowOfAnEmptyDomain, boolean
belowTheTopAggregate) {
+ Plan replacement = newAggregations.get(plan);
+ if (plan == aggregation.domainAggregation()) {
+ return replacement;
+ }
+ // the nodes below the top aggregate cannot produce its outputs, so
they only carry the keys
+ Plan child = rebuildTheAggregationChain(plan.child(0), aggregation,
newAggregations, keysToExpose,
+ plan == aggregation.topAggregation() ? ImmutableSet.of() :
outputsOfTheTopAggregation,
+ matchMarkerOfTheEmptyDomain,
filtersWhichKeepTheRowOfAnEmptyDomain,
+ belowTheTopAggregate || plan == aggregation.topAggregation());
+ if (replacement != null) {
+ return replacement.withChildren(child);
+ }
+ if (filtersWhichKeepTheRowOfAnEmptyDomain.contains(plan)) {
+ // The row which the rewrite keeps for an empty domain is not a
row of the rows below this
+ // filter (its marker is null), and the aggregation of the
original subquery does not
+ // evaluate the filter for the empty input of such a key either:
the row passes the filter,
+ // so that the aggregates above it return the values of an empty
input for the key (see
+ // keepsTheRowOfAnEmptyDomain). Without the relaxation the filter
would remove the row which
+ // the rewrite keeps for the key, and the aggregates above it
(which the rewrite grouped by
+ // the correlation key) would produce no row at all for the key.
+ List<Expression> conjuncts = Lists.newArrayList();
+ for (Expression conjunct : ((LogicalFilter<Plan>)
plan).getConjuncts()) {
+ conjuncts.add(ExpressionUtils.or(conjunct, new
IsNull(matchMarkerOfTheEmptyDomain)));
Review Comment:
[P2] Rewrite retained wrapper expressions through `nullableInnerSlots` too.
With a declared-NOT-NULL `i.g`, `o.k NOT IN (SELECT max(c) FROM (SELECT
count(*) c FROM i WHERE i.k=o.k GROUP BY i.g HAVING i.g + count(*) > 0) x)`
retains `Project(c) -> Filter(i.g + c > 0) -> grouped Aggregate`. The
keep-empty LEFT OUTER join makes `i.g` nullable and the aggregate now uses that
nullable slot, but this Filter is rebuilt from its original nonnullable `i.g`
(Projects below follow the same pattern). Final `AdjustNullable` finds the same
ExprId with nullable=true and throws when `fe_debug` is enabled. Pass the
nullable replacement map through chain reconstruction and apply it to retained
Filter/Project expressions; add a `fe_debug` planner/result case for a NOT NULL
group 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]
