github-actions[bot] commented on code in PR #67940:
URL: https://github.com/apache/doris/pull/67940#discussion_r4005533490
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -64,28 +100,95 @@
* |
* Filter(Uncorrelated predicate)
* </pre>
+ * <p>
+ * The rewrite above keeps the aggregation of the subquery on the inner side,
which is only
+ * equivalent to the original subquery when the correlated predicate is an
equality between the
+ * outer side and the inner side: in that case the inner rows of one outer row
are exactly the
+ * groups of the aggregate whose key is the value of that inner side, so the
HAVING clause of every
+ * group is the HAVING clause of the outer row.
+ * The aggregation of an EXISTS/NOT EXISTS subquery is built on the outer side
instead when that
+ * equivalence does not hold, see
+ * {@link #pullUpCorrelatedPredicateByAggregatingOuter}.
*/
public class UnCorrelatedApplyAggregateFilter implements RewriteRuleFactory {
+
+ /** name of the projected column which tells whether an inner row matched
the correlated predicate */
+ 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 and the filter which holds the
predicates of its HAVING clause
+ * which were not pulled into the apply: the nodes between the apply and
the aggregate are the
+ * projections and the filters of the subquery, and the projections
between the aggregate and its
+ * filter only carry the columns which the aggregation needs.
+ */
+ private static Optional<Pair<LogicalAggregate<?>,
Optional<LogicalFilter<Plan>>>> locateAggregate(
+ LogicalApply<?, ?> apply) {
+ Plan below = apply.right();
+ Optional<LogicalFilter<Plan>> havingFilter = Optional.empty();
+ while (!(below instanceof LogicalAggregate)) {
+ if (below instanceof LogicalFilter) {
+ havingFilter = Optional.of((LogicalFilter<Plan>) below);
Review Comment:
[P1] Preserve every wrapper above the aggregate. A valid right side can
reach this loop as `Filter(r < .5) -> Project(c, random() AS r) -> Filter(c =
1) -> GlobalAggregate(count)`: filter pushdown intentionally leaves the
predicate over the volatile alias above its project. This assignment first
remembers the upper filter and then overwrites it with the deeper HAVING
filter; the outer-aggregation path later rebuilds only `Filter(c = 1) ->
Aggregate`, so `r < .5` is deleted and a probabilistic EXISTS becomes always
true. This is distinct from the existing volatility thread, where the
expression remains but is shared per key. Track the complete ordered wrapper
chain and include it in movement-safety/reconstruction, or reject shapes that
cannot be preserved.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -115,10 +241,683 @@ public List<Rule> buildRules() {
correlatedPredicate = ExpressionUtils.replace(correlatedPredicate,
unCorrelatedExprToSlot);
LogicalAggregate newAgg = new LogicalAggregate<>(newGroupby,
newAggOutput,
PlanUtils.filterOrSelf(ImmutableSet.copyOf(unCorrelatedPredicate),
filter.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);
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));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * Whether a projection sits between the apply and the aggregation of the
subquery: it only
+ * exposes a part of the output of the aggregation (the select list of the
subquery), so the
+ * correlation keys the original rewrite adds to the group by are not
visible above it.
+ */
+ private static boolean hasProjectionAboveAggregate(LogicalApply<?, ?>
apply, LogicalAggregate<?> agg) {
+ Plan below = apply.right();
+ while (below != agg) {
+ if (below instanceof LogicalProject) {
+ return true;
+ }
+ below = below.child(0);
+ }
+ return false;
+ }
+
+ /**
+ * Keep the nodes which wrap the aggregate (the HAVING clause, the
projection of the subquery)
+ * when the aggregate is replaced.
+ */
+ private static Plan replaceAggregate(Plan plan, Plan newAggregate) {
+ if (plan instanceof LogicalAggregate) {
+ return newAggregate;
+ }
+ return plan.withChildren(replaceAggregate(plan.child(0),
newAggregate));
+ }
+
+ /**
+ * The predicates which relate the correlated subquery to the outer query,
classified by the node
+ * on which they have to be evaluated:
+ *
+ * <ul>
+ * <li>domain predicates select the inner rows which belong to the
correlated domain of one outer
+ * row. They are the predicates of the WHERE clause of the subquery,
plus the predicates
+ * which were already pulled into the apply and do not reference the
aggregation;</li>
+ * <li>aggregate predicates reference the output of the aggregate, they
are the predicates of the
+ * HAVING clause which were pulled into the apply by {@link
UnCorrelatedApplyFilter} (that
+ * rule runs before this one). They decide which rows of the
aggregation the subquery
+ * returns for one outer row.</li>
+ * </ul>
+ */
+ private static final class CorrelatedAggregatePredicates {
+ private final List<Expression> whereConjuncts = Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<?> agg, LogicalFilter<Plan> filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // Every predicate which was pulled into the apply was pulled from
the filter which sits
+ // above the aggregate (the HAVING clause of the subquery), so it
decides which rows of
+ // the aggregation the subquery returns and has to be evaluated
above the aggregation of
+ // the rewrite. Its provenance cannot be recovered from the slots
it uses: a predicate
+ // such as `outer.flag = 1` references no aggregation output and
no inner column but it
+ // still rejects the row of the aggregation.
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(predicates.aggregatePredicates::add);
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.copyOf(whereConjuncts);
+ }
+
+ private List<Expression> havingPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(havingConjuncts)
+ .addAll(aggregatePredicates)
+ .build();
+ }
+
+ private Set<Expression> havingConjuncts() {
+ return havingConjuncts;
+ }
+
+ private boolean hasHaving() {
+ return !havingConjuncts.isEmpty() ||
!aggregatePredicates.isEmpty();
+ }
+
+ private boolean hasAggregatePredicates() {
+ return !aggregatePredicates.isEmpty();
+ }
+
+ /**
+ * The correlation slots which the keys of the aggregation have to
contain: the domains of
+ * two outer rows are the same as soon as the slots their predicates
use are equal.
+ */
+ private Set<Slot> keySlots(List<Slot> correlationSlots) {
+ Set<Slot> keys = new LinkedHashSet<>();
+ for (Expression conjunct : domainPredicates()) {
+ addCorrelationSlots(conjunct, correlationSlots, keys);
+ }
+ for (Expression conjunct : aggregatePredicates) {
+ addCorrelationSlots(conjunct, correlationSlots, keys);
+ }
+ return keys;
+ }
+
+ private static void addCorrelationSlots(Expression conjunct,
List<Slot> correlationSlots, Set<Slot> keys) {
+ for (Slot slot : conjunct.getInputSlots()) {
+ if (correlationSlots.contains(slot)) {
+ keys.add(slot);
+ }
+ }
+ }
+
+ /**
+ * Whether every predicate can be evaluated by the plan of the
aggregation, that is if it only
+ * uses the correlation keys, the inner rows and the output of the
aggregate.
+ */
+ private boolean isResolvable(LogicalApply<?, ?> apply,
LogicalAggregate<?> agg,
+ LogicalFilter<Plan> filter) {
+ Set<ExprId> allowed = Sets.newHashSet();
+ apply.getCorrelationSlot().forEach(slot ->
allowed.add(slot.getExprId()));
+ agg.getOutput().forEach(slot -> allowed.add(slot.getExprId()));
+ filter.child().getOutput().forEach(slot ->
allowed.add(slot.getExprId()));
+ return domainPredicates().stream().allMatch(conjunct ->
allowed.containsAll(conjunct.getInputSlotExprIds()))
+ && havingPredicates().stream()
+ .allMatch(conjunct ->
allowed.containsAll(conjunct.getInputSlotExprIds()));
+ }
+ }
+
+ /**
+ * Whether the aggregation of the correlated subquery has to be built on
the outer side.
+ * <p>
+ * The original rewrite puts the inner side of the correlated predicate
into the group by of the
+ * aggregate, so the HAVING clause of one group is treated as the HAVING
clause of one outer row.
+ * That is wrong for:
+ * <ul>
+ * <li>a correlated predicate which is not an equality (eg. `inner.k
< outer.k`): the inner
+ * rows of one outer row are the union of several groups, so group
wide aggregates such as
+ * count(*) are computed for a part of the domain of the outer row
only;</li>
+ * <li>a global aggregate (no group by) whose HAVING clause holds for an
empty input
+ * (eg. `having count(*) = 0`): a global aggregate returns one row
for every outer row,
+ * including the outer rows without any matching inner row, and that
row disappears when
+ * the inner side of the correlated predicate becomes the group by
key;</li>
+ * <li>a HAVING clause which references the outer query: the row kept by
that HAVING clause is
+ * the one of the domain of the outer row, so it cannot be evaluated
on a group of the
+ * inner side when the domain is empty.</li>
+ * </ul>
+ */
+ private static boolean needCorrelatedAggregationOnOuter(LogicalApply<?, ?>
apply, LogicalAggregate<?> agg,
+ boolean havingFilterPulled, List<Expression> correlatedPredicate,
+ CorrelatedAggregatePredicates predicates) {
+ if (!apply.isExist()) {
+ // scalar and IN subqueries need the aggregate output to be
exposed by a join
+ return false;
+ }
+ if (havingFilterPulled && !predicates.hasHaving()) {
+ // EXISTS/NOT EXISTS without a HAVING clause only depends on the
existence of the
+ // aggregation result, which is kept by grouping the inner side
+ return false;
+ }
+ if (predicates.hasAggregatePredicates() &&
agg.getGroupByExpressions().isEmpty()) {
+ // a predicate of the HAVING clause which references the outer
query decides whether the
+ // row of a global aggregate is kept, and that row exists for
every outer row including
+ // the rows of an empty correlated domain, so the predicate cannot
be evaluated on a
+ // group of the inner side
+ return true;
+ }
+ for (Expression conjunct : correlatedPredicate) {
+ if (!isSupportedCorrelatedComparison(conjunct,
apply.getCorrelationSlot())) {
+ // keep the behavior of the original rewrite, which reports
these predicates
+ return false;
+ }
+ }
+ if (predicates.domainPredicates().stream()
+ .anyMatch(conjunct -> breaksDomainPartition(conjunct,
apply.getCorrelationSlot()))) {
+ // the domain of one outer row is the union of several groups of
the aggregate
+ return true;
+ }
+ if (predicates.hasAggregatePredicates()) {
+ // the domain of one outer row is exactly one group and the rows
of the subquery are the
+ // rows of that group, so the original rewrite is still equivalent
+ return false;
+ }
+ if (!agg.getGroupByExpressions().isEmpty()) {
+ // an equality correlated predicate maps the domain of every outer
row onto exactly one group
+ return false;
+ }
+ return havingMayHoldWithEmptyInput(agg, predicates.havingConjuncts());
+ }
+
+ /**
+ * Whether this predicate changes the domain of an outer row in a way
which is not a group of the
+ * aggregate. Only an equality between the outer side and the inner side
partitions the inner rows
+ * of one outer row into exactly the groups of the aggregate, while a
predicate which does not
+ * reference the outer query at all just filters the inner rows.
+ */
+ private static boolean breaksDomainPartition(Expression conjunct,
List<Slot> correlationSlots) {
+ if (conjunct instanceof EqualPredicate) {
+ return false;
+ }
+ return
conjunct.getInputSlots().stream().anyMatch(correlationSlots::contains);
+ }
+
+ /**
+ * Whether the correlated predicate is a comparison whose sides do not mix
the outer query and the
+ * subquery, eg. `inner.k < outer.k` or `outer.k = inner.abs(k)`. Those
are the predicates which
+ * can be evaluated by joining the two sides, and the ones supported by
the original rewrite.
+ */
+ private static boolean isSupportedCorrelatedComparison(Expression
conjunct, List<Slot> correlationSlots) {
+ Expression predicate = conjunct;
+ if (predicate instanceof Not && predicate.child(0) instanceof
BinaryExpression) {
+ predicate = predicate.child(0);
+ }
+ if (!(predicate instanceof BinaryExpression)) {
+ return false;
+ }
+ Expression left = ((BinaryExpression) predicate).left();
+ Expression right = ((BinaryExpression) predicate).right();
+ Set<Slot> leftSlots = left.getInputSlots();
+ Set<Slot> rightSlots = right.getInputSlots();
+ boolean correlatedToLeft = !leftSlots.isEmpty() &&
leftSlots.stream().allMatch(correlationSlots::contains)
+ && rightSlots.stream().noneMatch(correlationSlots::contains);
+ boolean correlatedToRight = !rightSlots.isEmpty() &&
rightSlots.stream().allMatch(correlationSlots::contains)
+ && leftSlots.stream().noneMatch(correlationSlots::contains);
+ return correlatedToLeft || correlatedToRight;
+ }
+
+ /**
+ * Rewrite `outer [not] exists (select agg from inner where <correlated
predicate>
+ * [group by ...] having ...)` into a semi/anti join whose right side
aggregates the outer rows
+ * together with their correlated inner rows, so that the aggregation of
one outer row is the
+ * aggregation of exactly the inner rows satisfying the correlated
predicate:
+ *
+ * <pre>
+ * before:
+ * Apply(EXISTS, correlationSlot=[outer.k])
+ * / \
+ * outer Filter(having)
+ * +-- Aggregate(group by [inner.g], count(*))
+ * +-- Filter(correlated
predicate(inner.k < outer.k))
+ * +-- inner
+ *
+ * after:
+ * LEFT SEMI JOIN(otherJoinConjuncts=[outer.k <=> key.k])
+ * / \
+ * outer Filter(having: count(*) => count(marker))
+ * +-- Aggregate(group by [key.k, inner.g],
count(marker))
+ * +-- LEFT OUTER JOIN(inner.k < key.k)
// keeps the empty domain
+ * |-- Aggregate(group by [k],
output=[k]) // distinct correlated keys
+ * | +-- outer'
+ * +-- Project(marker, ...)
+ * +-- Filter(uncorrelated
predicates)
+ * +-- inner
+ * </pre>
+ * outer' is a deep copy of outer, so that the two references of the outer
plan have their own
+ * slots and relation ids. The LEFT OUTER JOIN and the marker are not
needed (a plain inner join
+ * is used) when the aggregate has a group by, because such an aggregate
returns no row at all
+ * for an empty input. The predicates of the HAVING clause which reference
the outer query (they
+ * were pulled into the apply by {@link UnCorrelatedApplyFilter}) are
evaluated above the
+ * aggregate of the outer rows, so that they decide on the aggregation of
the whole domain of an
+ * outer row, the empty domain included.
+ *
+ * @return null if this rewrite cannot be applied safely, the caller then
reports the subquery as
+ * unsupported
+ */
+ private static Plan
pullUpCorrelatedPredicateByAggregatingOuter(LogicalApply<?, ?> apply,
+ LogicalAggregate<?> agg, LogicalFilter<Plan> filter,
+ List<Expression> unCorrelatedPredicate,
CorrelatedAggregatePredicates predicates) {
+ Set<Slot> correlationSlots =
predicates.keySlots(apply.getCorrelationSlot());
+ if (containsSensitiveExpression(apply.left(), correlationSlots)
+ || hasNonDeterministicRows(apply.left())
+ || containsNoneMovableFunction(apply.right())
+ || containsSensitiveSubqueryExpression(agg, filter, predicates)
+ || referencesOuterSlot(apply.right(),
ImmutableSet.copyOf(predicates.whereConjuncts),
+ apply.getCorrelationSlot())
+ || !predicates.isResolvable(apply, agg, filter)) {
+ return null;
+ }
+
+ // the domains of two outer rows are the same as soon as their
correlation slots are equal,
+ // so the correlation slots are the only outer information the
aggregation needs
+ LogicalPlan outer = (LogicalPlan) apply.left();
+ LogicalPlan outerCopy = LogicalPlanDeepCopier.INSTANCE.deepCopy(outer,
new DeepCopierContext());
+ List<Slot> outerOutput = outer.getOutput();
+ List<Slot> outerCopyOutput = outerCopy.getOutput();
Review Comment:
[P1] Require copied outer outputs to have disjoint ExprIds before using them
as keys. In `Apply(EXISTS, number#7) -> [numbers(3), HAVING count(*)=0 ->
Aggregate -> Filter(i.k < number#7)]` with inner `k={0}`, only outer number 0
should survive. The TVF copier dispatches through `visitLogicalRelation`, but
`LogicalTVFRelation.withRelationId` reuses the original logical properties, so
`outerCopy.getOutput()` still contains `number#7`. The back condition is
therefore `number#7 <=> number#7`, which `SimplifySelfComparison` folds to
TRUE; because copied key 0 exists, the semi join returns 0, 1, and 2. This is
distinct from the existing nondeterministic-copy thread: both TVF evaluations
return identical rows, but their slots were never separated. Fix the relation
copier or reject overlapping outputs.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -115,10 +241,683 @@ public List<Rule> buildRules() {
correlatedPredicate = ExpressionUtils.replace(correlatedPredicate,
unCorrelatedExprToSlot);
LogicalAggregate newAgg = new LogicalAggregate<>(newGroupby,
newAggOutput,
PlanUtils.filterOrSelf(ImmutableSet.copyOf(unCorrelatedPredicate),
filter.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);
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));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * Whether a projection sits between the apply and the aggregation of the
subquery: it only
+ * exposes a part of the output of the aggregation (the select list of the
subquery), so the
+ * correlation keys the original rewrite adds to the group by are not
visible above it.
+ */
+ private static boolean hasProjectionAboveAggregate(LogicalApply<?, ?>
apply, LogicalAggregate<?> agg) {
+ Plan below = apply.right();
+ while (below != agg) {
+ if (below instanceof LogicalProject) {
+ return true;
+ }
+ below = below.child(0);
+ }
+ return false;
+ }
+
+ /**
+ * Keep the nodes which wrap the aggregate (the HAVING clause, the
projection of the subquery)
+ * when the aggregate is replaced.
+ */
+ private static Plan replaceAggregate(Plan plan, Plan newAggregate) {
+ if (plan instanceof LogicalAggregate) {
+ return newAggregate;
+ }
+ return plan.withChildren(replaceAggregate(plan.child(0),
newAggregate));
+ }
+
+ /**
+ * The predicates which relate the correlated subquery to the outer query,
classified by the node
+ * on which they have to be evaluated:
+ *
+ * <ul>
+ * <li>domain predicates select the inner rows which belong to the
correlated domain of one outer
+ * row. They are the predicates of the WHERE clause of the subquery,
plus the predicates
+ * which were already pulled into the apply and do not reference the
aggregation;</li>
+ * <li>aggregate predicates reference the output of the aggregate, they
are the predicates of the
+ * HAVING clause which were pulled into the apply by {@link
UnCorrelatedApplyFilter} (that
+ * rule runs before this one). They decide which rows of the
aggregation the subquery
+ * returns for one outer row.</li>
+ * </ul>
+ */
+ private static final class CorrelatedAggregatePredicates {
+ private final List<Expression> whereConjuncts = Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<?> agg, LogicalFilter<Plan> filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // Every predicate which was pulled into the apply was pulled from
the filter which sits
+ // above the aggregate (the HAVING clause of the subquery), so it
decides which rows of
+ // the aggregation the subquery returns and has to be evaluated
above the aggregation of
+ // the rewrite. Its provenance cannot be recovered from the slots
it uses: a predicate
+ // such as `outer.flag = 1` references no aggregation output and
no inner column but it
+ // still rejects the row of the aggregation.
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(predicates.aggregatePredicates::add);
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.copyOf(whereConjuncts);
+ }
+
+ private List<Expression> havingPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(havingConjuncts)
+ .addAll(aggregatePredicates)
+ .build();
+ }
+
+ private Set<Expression> havingConjuncts() {
+ return havingConjuncts;
+ }
+
+ private boolean hasHaving() {
+ return !havingConjuncts.isEmpty() ||
!aggregatePredicates.isEmpty();
+ }
+
+ private boolean hasAggregatePredicates() {
+ return !aggregatePredicates.isEmpty();
+ }
+
+ /**
+ * The correlation slots which the keys of the aggregation have to
contain: the domains of
+ * two outer rows are the same as soon as the slots their predicates
use are equal.
+ */
+ private Set<Slot> keySlots(List<Slot> correlationSlots) {
+ Set<Slot> keys = new LinkedHashSet<>();
+ for (Expression conjunct : domainPredicates()) {
+ addCorrelationSlots(conjunct, correlationSlots, keys);
+ }
+ for (Expression conjunct : aggregatePredicates) {
+ addCorrelationSlots(conjunct, correlationSlots, keys);
+ }
+ return keys;
+ }
+
+ private static void addCorrelationSlots(Expression conjunct,
List<Slot> correlationSlots, Set<Slot> keys) {
+ for (Slot slot : conjunct.getInputSlots()) {
+ if (correlationSlots.contains(slot)) {
+ keys.add(slot);
+ }
+ }
+ }
+
+ /**
+ * Whether every predicate can be evaluated by the plan of the
aggregation, that is if it only
+ * uses the correlation keys, the inner rows and the output of the
aggregate.
+ */
+ private boolean isResolvable(LogicalApply<?, ?> apply,
LogicalAggregate<?> agg,
+ LogicalFilter<Plan> filter) {
+ Set<ExprId> allowed = Sets.newHashSet();
+ apply.getCorrelationSlot().forEach(slot ->
allowed.add(slot.getExprId()));
+ agg.getOutput().forEach(slot -> allowed.add(slot.getExprId()));
+ filter.child().getOutput().forEach(slot ->
allowed.add(slot.getExprId()));
+ return domainPredicates().stream().allMatch(conjunct ->
allowed.containsAll(conjunct.getInputSlotExprIds()))
+ && havingPredicates().stream()
+ .allMatch(conjunct ->
allowed.containsAll(conjunct.getInputSlotExprIds()));
+ }
+ }
+
+ /**
+ * Whether the aggregation of the correlated subquery has to be built on
the outer side.
+ * <p>
+ * The original rewrite puts the inner side of the correlated predicate
into the group by of the
+ * aggregate, so the HAVING clause of one group is treated as the HAVING
clause of one outer row.
+ * That is wrong for:
+ * <ul>
+ * <li>a correlated predicate which is not an equality (eg. `inner.k
< outer.k`): the inner
+ * rows of one outer row are the union of several groups, so group
wide aggregates such as
+ * count(*) are computed for a part of the domain of the outer row
only;</li>
+ * <li>a global aggregate (no group by) whose HAVING clause holds for an
empty input
+ * (eg. `having count(*) = 0`): a global aggregate returns one row
for every outer row,
+ * including the outer rows without any matching inner row, and that
row disappears when
+ * the inner side of the correlated predicate becomes the group by
key;</li>
+ * <li>a HAVING clause which references the outer query: the row kept by
that HAVING clause is
+ * the one of the domain of the outer row, so it cannot be evaluated
on a group of the
+ * inner side when the domain is empty.</li>
+ * </ul>
+ */
+ private static boolean needCorrelatedAggregationOnOuter(LogicalApply<?, ?>
apply, LogicalAggregate<?> agg,
+ boolean havingFilterPulled, List<Expression> correlatedPredicate,
+ CorrelatedAggregatePredicates predicates) {
+ if (!apply.isExist()) {
+ // scalar and IN subqueries need the aggregate output to be
exposed by a join
+ return false;
+ }
+ if (havingFilterPulled && !predicates.hasHaving()) {
+ // EXISTS/NOT EXISTS without a HAVING clause only depends on the
existence of the
+ // aggregation result, which is kept by grouping the inner side
+ return false;
+ }
+ if (predicates.hasAggregatePredicates() &&
agg.getGroupByExpressions().isEmpty()) {
+ // a predicate of the HAVING clause which references the outer
query decides whether the
+ // row of a global aggregate is kept, and that row exists for
every outer row including
+ // the rows of an empty correlated domain, so the predicate cannot
be evaluated on a
+ // group of the inner side
+ return true;
+ }
+ for (Expression conjunct : correlatedPredicate) {
+ if (!isSupportedCorrelatedComparison(conjunct,
apply.getCorrelationSlot())) {
+ // keep the behavior of the original rewrite, which reports
these predicates
+ return false;
+ }
+ }
+ if (predicates.domainPredicates().stream()
+ .anyMatch(conjunct -> breaksDomainPartition(conjunct,
apply.getCorrelationSlot()))) {
+ // the domain of one outer row is the union of several groups of
the aggregate
+ return true;
+ }
+ if (predicates.hasAggregatePredicates()) {
+ // the domain of one outer row is exactly one group and the rows
of the subquery are the
+ // rows of that group, so the original rewrite is still equivalent
+ return false;
+ }
+ if (!agg.getGroupByExpressions().isEmpty()) {
+ // an equality correlated predicate maps the domain of every outer
row onto exactly one group
+ return false;
+ }
+ return havingMayHoldWithEmptyInput(agg, predicates.havingConjuncts());
+ }
+
+ /**
+ * Whether this predicate changes the domain of an outer row in a way
which is not a group of the
+ * aggregate. Only an equality between the outer side and the inner side
partitions the inner rows
+ * of one outer row into exactly the groups of the aggregate, while a
predicate which does not
+ * reference the outer query at all just filters the inner rows.
+ */
+ private static boolean breaksDomainPartition(Expression conjunct,
List<Slot> correlationSlots) {
+ if (conjunct instanceof EqualPredicate) {
+ return false;
+ }
+ return
conjunct.getInputSlots().stream().anyMatch(correlationSlots::contains);
+ }
+
+ /**
+ * Whether the correlated predicate is a comparison whose sides do not mix
the outer query and the
+ * subquery, eg. `inner.k < outer.k` or `outer.k = inner.abs(k)`. Those
are the predicates which
+ * can be evaluated by joining the two sides, and the ones supported by
the original rewrite.
+ */
+ private static boolean isSupportedCorrelatedComparison(Expression
conjunct, List<Slot> correlationSlots) {
+ Expression predicate = conjunct;
+ if (predicate instanceof Not && predicate.child(0) instanceof
BinaryExpression) {
+ predicate = predicate.child(0);
+ }
+ if (!(predicate instanceof BinaryExpression)) {
+ return false;
+ }
+ Expression left = ((BinaryExpression) predicate).left();
+ Expression right = ((BinaryExpression) predicate).right();
+ Set<Slot> leftSlots = left.getInputSlots();
+ Set<Slot> rightSlots = right.getInputSlots();
+ boolean correlatedToLeft = !leftSlots.isEmpty() &&
leftSlots.stream().allMatch(correlationSlots::contains)
+ && rightSlots.stream().noneMatch(correlationSlots::contains);
+ boolean correlatedToRight = !rightSlots.isEmpty() &&
rightSlots.stream().allMatch(correlationSlots::contains)
+ && leftSlots.stream().noneMatch(correlationSlots::contains);
+ return correlatedToLeft || correlatedToRight;
+ }
+
+ /**
+ * Rewrite `outer [not] exists (select agg from inner where <correlated
predicate>
+ * [group by ...] having ...)` into a semi/anti join whose right side
aggregates the outer rows
+ * together with their correlated inner rows, so that the aggregation of
one outer row is the
+ * aggregation of exactly the inner rows satisfying the correlated
predicate:
+ *
+ * <pre>
+ * before:
+ * Apply(EXISTS, correlationSlot=[outer.k])
+ * / \
+ * outer Filter(having)
+ * +-- Aggregate(group by [inner.g], count(*))
+ * +-- Filter(correlated
predicate(inner.k < outer.k))
+ * +-- inner
+ *
+ * after:
+ * LEFT SEMI JOIN(otherJoinConjuncts=[outer.k <=> key.k])
+ * / \
+ * outer Filter(having: count(*) => count(marker))
+ * +-- Aggregate(group by [key.k, inner.g],
count(marker))
+ * +-- LEFT OUTER JOIN(inner.k < key.k)
// keeps the empty domain
+ * |-- Aggregate(group by [k],
output=[k]) // distinct correlated keys
+ * | +-- outer'
+ * +-- Project(marker, ...)
+ * +-- Filter(uncorrelated
predicates)
+ * +-- inner
+ * </pre>
+ * outer' is a deep copy of outer, so that the two references of the outer
plan have their own
+ * slots and relation ids. The LEFT OUTER JOIN and the marker are not
needed (a plain inner join
+ * is used) when the aggregate has a group by, because such an aggregate
returns no row at all
+ * for an empty input. The predicates of the HAVING clause which reference
the outer query (they
+ * were pulled into the apply by {@link UnCorrelatedApplyFilter}) are
evaluated above the
+ * aggregate of the outer rows, so that they decide on the aggregation of
the whole domain of an
+ * outer row, the empty domain included.
+ *
+ * @return null if this rewrite cannot be applied safely, the caller then
reports the subquery as
+ * unsupported
+ */
+ private static Plan
pullUpCorrelatedPredicateByAggregatingOuter(LogicalApply<?, ?> apply,
+ LogicalAggregate<?> agg, LogicalFilter<Plan> filter,
+ List<Expression> unCorrelatedPredicate,
CorrelatedAggregatePredicates predicates) {
+ Set<Slot> correlationSlots =
predicates.keySlots(apply.getCorrelationSlot());
+ if (containsSensitiveExpression(apply.left(), correlationSlots)
+ || hasNonDeterministicRows(apply.left())
+ || containsNoneMovableFunction(apply.right())
+ || containsSensitiveSubqueryExpression(agg, filter, predicates)
+ || referencesOuterSlot(apply.right(),
ImmutableSet.copyOf(predicates.whereConjuncts),
+ apply.getCorrelationSlot())
+ || !predicates.isResolvable(apply, agg, filter)) {
+ return null;
+ }
+
+ // the domains of two outer rows are the same as soon as their
correlation slots are equal,
+ // so the correlation slots are the only outer information the
aggregation needs
+ LogicalPlan outer = (LogicalPlan) apply.left();
+ LogicalPlan outerCopy = LogicalPlanDeepCopier.INSTANCE.deepCopy(outer,
new DeepCopierContext());
+ List<Slot> outerOutput = outer.getOutput();
+ List<Slot> outerCopyOutput = outerCopy.getOutput();
+ Preconditions.checkState(outerOutput.size() == outerCopyOutput.size(),
+ "the deep copy of the outer plan changed its output size");
+ Map<Expression, Expression> slotToKey = Maps.newLinkedHashMap();
+ for (Slot slot : correlationSlots) {
+ int index = outerOutput.indexOf(slot);
+ if (index < 0) {
+ return null;
+ }
+ slotToKey.put(slot, outerCopyOutput.get(index));
+ }
+ List<NamedExpression> keyExpressions = new
ArrayList<>(slotToKey.size());
+ for (Expression key : slotToKey.values()) {
+ keyExpressions.add((NamedExpression) key);
+ }
+ LogicalAggregate<Plan> keyAggregate = new LogicalAggregate<>(
+ ImmutableList.copyOf(slotToKey.values()), keyExpressions,
outerCopy);
+
+ Plan inner =
PlanUtils.filterOrSelf(ImmutableSet.copyOf(unCorrelatedPredicate),
filter.child());
+ // An aggregate with a group by returns no row at all for an empty
correlated domain, so the
+ // inner join below reproduces the behaviour of the subquery, which
has no group to report.
+ // A global aggregate returns exactly one row for every outer row
instead, the empty domain
+ // included, so the rows of an empty correlated domain have to be
kept: the left outer join
+ // gives the aggregate one row whose inner columns are null, which is
the same input the
+ // subquery aggregates for an empty input. Only the count aggregations
tell "no row" and "one
+ // row of nulls" apart, and count(*) also counts the kept row itself,
so the counts are
+ // replaced by counts of the projected marker, which is null for the
rows which were kept for
+ // an empty correlated domain.
+ boolean keepEmptyDomain = agg.getGroupByExpressions().isEmpty();
+ Slot matchMarker = null;
+ if (keepEmptyDomain) {
+ Alias marker = new Alias(BooleanLiteral.TRUE,
CORRELATION_MATCH_MARKER);
+ matchMarker = marker.toSlot();
+ List<NamedExpression> projects = Lists.newArrayList(marker);
+ projects.addAll(inner.getOutput());
+ inner = new LogicalProject<>(projects, inner);
+ }
+
+ List<Expression> domainConjuncts =
predicates.domainPredicates().stream()
+ .map(conjunct -> ExpressionUtils.replace(conjunct, slotToKey))
+ .collect(ImmutableList.toImmutableList());
+ // a grouped aggregate reports an empty domain as "no row", a global
aggregate as "one row"
+ Plan domainJoin = new LogicalJoin<>(keepEmptyDomain ?
JoinType.LEFT_OUTER_JOIN : JoinType.INNER_JOIN,
+ ExpressionUtils.EMPTY_CONDITION, domainConjuncts, new
DistributeHint(DistributeType.NONE),
+ Optional.empty(), keyAggregate, inner, null);
+
+ List<Expression> havingPredicates = predicates.havingPredicates();
+ Set<AggregateFunction> aggregates = Sets.newLinkedHashSet();
+ if (keepEmptyDomain) {
+ // the aggregation of the rewrite is computed over the rows of the
inner side plus the
+ // row which the left outer join keeps for an empty correlated
domain, and only the
+ // aggregates which ignore null arguments see that row as an empty
input
+ for (Expression expression : agg.getOutputExpressions()) {
+
aggregates.addAll(expression.collect(AggregateFunction.class::isInstance));
+ }
+ for (Expression conjunct : havingPredicates) {
+
aggregates.addAll(conjunct.collect(AggregateFunction.class::isInstance));
+ }
+ if (aggregates.stream().anyMatch(function -> !(function instanceof
NullIgnoringAggregateFunction))) {
+ // an aggregate which keeps null arguments cannot tell the row
kept for an empty
+ // correlated domain from a row of the inner side
+ return null;
+ }
+ }
+ Map<Expression, Expression> compensated =
guardAggregateArguments(aggregates, matchMarker);
+
+ List<Expression> newGroupBy = Lists.newArrayList(slotToKey.values());
+ newGroupBy.addAll(agg.getGroupByExpressions());
+ List<NamedExpression> newOutputs = Lists.newArrayList(keyExpressions);
+ for (NamedExpression output : agg.getOutputExpressions()) {
+ newOutputs.add((NamedExpression) ExpressionUtils.replace(output,
compensated));
+ }
+ LogicalAggregate<Plan> newAggregate = new
LogicalAggregate<>(newGroupBy, newOutputs, domainJoin);
+
+ Set<Expression> newHavingConjuncts = Sets.newLinkedHashSet();
+ for (Expression conjunct : havingPredicates) {
+
newHavingConjuncts.add(ExpressionUtils.replace(ExpressionUtils.replace(conjunct,
compensated), slotToKey));
+ }
+ Plan newRight = newHavingConjuncts.isEmpty() ? newAggregate
+ : new LogicalFilter<>(newHavingConjuncts, newAggregate);
+
+ // the aggregate of the subquery is now computed for the correlation
keys of every outer row,
+ // so the outer rows which own one of those groups are the rows for
which the subquery has rows
+ List<Expression> backConjuncts = new ArrayList<>(slotToKey.size());
+ for (Map.Entry<Expression, Expression> entry : slotToKey.entrySet()) {
+ backConjuncts.add(new NullSafeEqual(entry.getKey(),
entry.getValue()));
+ }
+ return new LogicalJoin<>(apply.isNot() ? JoinType.LEFT_ANTI_JOIN :
JoinType.LEFT_SEMI_JOIN,
+ ExpressionUtils.EMPTY_CONDITION, backConjuncts, new
DistributeHint(DistributeType.NONE),
+ apply.getMarkJoinSlotReference(), outer, newRight, null);
+ }
+
+ /**
+ * Replace the arguments of the aggregates so that the row which the left
outer join keeps for an
+ * empty correlated domain does not contribute to the aggregation: the
marker of that row is
+ * null, so every argument is null for it and the aggregates which ignore
null arguments (the
+ * caller only admits those) return the value of an empty input.
`count(*)` has no argument to
+ * build that guard on, so it counts the marker instead; a distinct count
keeps its argument and
+ * its distinct flag, because the null of the kept row must not be counted
as a value.
+ */
+ private static Map<Expression, Expression> guardAggregateArguments(
+ Set<AggregateFunction> aggregates, Slot matchMarker) {
+ Map<Expression, Expression> replace = Maps.newHashMap();
+ for (AggregateFunction function : aggregates) {
+ if (function instanceof Count && ((Count) function).isCountStar()
&& !function.isDistinct()) {
+ replace.put(function, new Count(matchMarker));
+ continue;
+ }
+ List<Expression> arguments =
Lists.newArrayListWithCapacity(function.arity());
+ for (Expression argument : function.getArguments()) {
+ arguments.add(new If(matchMarker, argument, new
NullLiteral(argument.getDataType())));
+ }
+ replace.put(function, function.withChildren(arguments));
+ }
+ return replace;
+ }
+
+ /**
+ * Whether the HAVING clause of a global aggregate can hold for the row
which the aggregate
+ * returns for an empty input.
+ */
+ private static boolean havingMayHoldWithEmptyInput(LogicalAggregate<?>
agg, Set<Expression> havingConjuncts) {
+ // the having clause usually references the output slots of the
aggregate, but it may also
+ // contain the aggregate functions themselves, so both of them are
replaced by the value
+ // which the aggregate returns for an empty input
+ Map<Expression, Expression> emptyValues = Maps.newHashMap();
+ for (NamedExpression output : agg.getOutputExpressions()) {
+ Expression expression = output instanceof Alias ? ((Alias)
output).child() : output;
+ if (!(expression instanceof AggregateFunction)) {
+ continue;
+ }
+ Expression emptyValue =
emptyValueForEmptyInput((AggregateFunction) expression);
+ if (emptyValue != null) {
+ emptyValues.put(output.toSlot(), emptyValue);
+ emptyValues.put(expression, emptyValue);
+ }
+ }
+ for (Expression conjunct : havingConjuncts) {
+ Expression substituted = emptyValues.isEmpty() ? conjunct
+ : ExpressionUtils.replace(conjunct, emptyValues);
+ Expression folded =
FoldConstantRuleOnFE.evaluateWithoutContext(substituted);
+ if (!(folded instanceof Literal)) {
+ // the aggregate returns an unknown value for an empty input,
rewrite conservatively
+ return true;
+ }
+ if (!BooleanLiteral.TRUE.equals(folded)) {
+ // false or null: this conjunct filters the row of the empty
input out
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * The value which an aggregate function returns for an empty input, or
null if it cannot be
+ * decided.
+ */
+ private static Expression emptyValueForEmptyInput(AggregateFunction
function) {
+ if (function instanceof Count) {
+ return new BigIntLiteral(0);
+ }
+ // only count is known to return a non null value for an empty input
Review Comment:
[P1] Keep the legacy equality rewrite when HAVING rejects the known empty
value. `SUM`, `AVG`, `MIN`, and `MAX` return SQL NULL on empty input, but this
fallback treats them as unknown. Thus `Apply(EXISTS) -> outer Limit -> HAVING
SUM(i.v) IS NOT NULL -> GlobalAggregate -> Filter(i.k = outer.k)` is classified
as needing outer aggregation; that path rejects the Limit and turns a
previously valid query into `Unsupported correlated subquery`, even though
absence of the inner group exactly represents the rejected empty domain. This
differs from the existing unsafe-fallback thread because outer aggregation is
not required here. Model exact typed NULL empty results, and scan all HAVING
conjuncts so a false/NULL conjunct dominates unknown ones, before selecting the
outer-copy path.
--
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]
