github-actions[bot] commented on code in PR #67940:
URL: https://github.com/apache/doris/pull/67940#discussion_r4003853966
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -119,6 +190,487 @@ public List<Rule> buildRules() {
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * 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> pulledDomainPredicates =
Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // the slots which exist above the aggregate tell whether a
predicate was evaluated on the
+ // aggregation (HAVING clause) instead of on the inner rows (WHERE
clause)
+ Set<ExprId> belowAggregate = filter.child().getOutput().stream()
+ .map(Slot::getExprId)
+ .collect(ImmutableSet.toImmutableSet());
+ Set<ExprId> aggregateOutput = agg.getOutput().stream()
+ .map(Slot::getExprId)
+ .filter(exprId -> !belowAggregate.contains(exprId))
+ .collect(ImmutableSet.toImmutableSet());
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(conjunct -> {
+ if
(conjunct.getInputSlotExprIds().stream().anyMatch(aggregateOutput::contains)) {
+ predicates.aggregatePredicates.add(conjunct);
+ } else {
+ predicates.pulledDomainPredicates.add(conjunct);
+ }
+ });
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(whereConjuncts)
+ .addAll(pulledDomainPredicates)
+ .build();
+ }
+
+ 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()) {
Review Comment:
[P1] Preserve the pulled HAVING predicate on this fallback.
`UnCorrelatedApplyFilter` has already moved a correlated HAVING such as `c =
e.k - 1` into `apply.correlationFilter`. For a grouped aggregate with an
equality domain predicate this method returns false, and the legacy path later
rebuilds the Apply with only the pulled WHERE predicate, so the HAVING
disappears before `ExistsApplyToJoin`. The new `eq_grouped_having_refs_outer`
case should admit only key 2, but this plan admits keys 1 and 2. Please
retain/conjoin the existing correlation filter at its post-aggregate evaluation
point.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -119,6 +190,487 @@ public List<Rule> buildRules() {
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * 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> pulledDomainPredicates =
Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // the slots which exist above the aggregate tell whether a
predicate was evaluated on the
+ // aggregation (HAVING clause) instead of on the inner rows (WHERE
clause)
+ Set<ExprId> belowAggregate = filter.child().getOutput().stream()
+ .map(Slot::getExprId)
+ .collect(ImmutableSet.toImmutableSet());
+ Set<ExprId> aggregateOutput = agg.getOutput().stream()
+ .map(Slot::getExprId)
+ .filter(exprId -> !belowAggregate.contains(exprId))
+ .collect(ImmutableSet.toImmutableSet());
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(conjunct -> {
+ if
(conjunct.getInputSlotExprIds().stream().anyMatch(aggregateOutput::contains)) {
+ predicates.aggregatePredicates.add(conjunct);
+ } else {
+ predicates.pulledDomainPredicates.add(conjunct);
+ }
+ });
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(whereConjuncts)
+ .addAll(pulledDomainPredicates)
+ .build();
+ }
+
+ 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
keeps the original rewrite
+ */
+ private static Plan
pullUpCorrelatedPredicateByAggregatingOuter(LogicalApply<?, ?> apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ List<Expression> unCorrelatedPredicate,
CorrelatedAggregatePredicates predicates) {
+ if (containsSensitiveExpression(apply.left())
+ || referencesOuterSlot(apply.right(),
ImmutableSet.copyOf(predicates.whereConjuncts),
+ apply.getCorrelationSlot())
+ || !predicates.isResolvable(apply, agg, filter)) {
+ return null;
+ }
+ Set<Slot> correlationSlots =
predicates.keySlots(apply.getCorrelationSlot());
+
+ // 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<Expression> countExpressions = Sets.newLinkedHashSet();
+ if (keepEmptyDomain) {
+ // the count aggregations are the only ones which count the row
kept for an empty domain
+ for (Expression expression : agg.getOutputExpressions()) {
+
countExpressions.addAll(expression.collect(Count.class::isInstance));
+ }
+ for (Expression conjunct : havingPredicates) {
+
countExpressions.addAll(conjunct.collect(Count.class::isInstance));
+ }
+ }
+ Map<Expression, Expression> compensated =
compensateCounts(countExpressions, 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 count aggregations with a form which does not count the row
that is kept for an
+ * empty correlated domain: the argument of every count is null for that
row, while count(*)
+ * counts the row itself.
+ */
+ private static Map<Expression, Expression>
compensateCounts(Set<Expression> countExpressions, Slot matchMarker) {
+ Map<Expression, Expression> replace = Maps.newHashMap();
+ for (Expression expression : countExpressions) {
+ Count count = (Count) expression;
+ if (count.isCountStar()) {
Review Comment:
[P1] Preserve DISTINCT for literal counts. `Count.isCountStar()` is also
true for `COUNT(DISTINCT 1)`, but this branch constructs a plain
`COUNT(marker)`. With two matching inner rows the original value is 1 and the
rewritten value is 2, so `HAVING COUNT(DISTINCT 1)=1` flips. Limit this
shortcut to non-distinct counts and marker-guard the literal while retaining
the aggregate attributes for the distinct case.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -97,6 +158,16 @@ public List<Rule> buildRules() {
return apply;
}
+ CorrelatedAggregatePredicates predicates =
+ CorrelatedAggregatePredicates.of(apply, agg, filter,
havingFilter, correlatedPredicate);
+ if (needCorrelatedAggregationOnOuter(apply, agg,
havingFilter.isEmpty(), correlatedPredicate, predicates)) {
+ Plan aggregatedOuter = pullUpCorrelatedPredicateByAggregatingOuter(
Review Comment:
[P1] Do not fall back after this required rewrite rejects the outer plan.
For an outer projection containing `random()`,
`pullUpCorrelatedPredicateByAggregatingOuter` returns null; execution then
continues into the legacy rewrite even though
`needCorrelatedAggregationOnOuter` already established that it is semantically
insufficient. With an empty inner input and `HAVING COUNT(*)=0` under a
non-equality correlation, the original EXISTS is true for every outer row while
the fallback returns none. Keep the Apply unchanged or use another
semantics-preserving path when this call returns null.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -119,6 +190,487 @@ public List<Rule> buildRules() {
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * 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> pulledDomainPredicates =
Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // the slots which exist above the aggregate tell whether a
predicate was evaluated on the
+ // aggregation (HAVING clause) instead of on the inner rows (WHERE
clause)
+ Set<ExprId> belowAggregate = filter.child().getOutput().stream()
+ .map(Slot::getExprId)
+ .collect(ImmutableSet.toImmutableSet());
+ Set<ExprId> aggregateOutput = agg.getOutput().stream()
+ .map(Slot::getExprId)
+ .filter(exprId -> !belowAggregate.contains(exprId))
+ .collect(ImmutableSet.toImmutableSet());
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(conjunct -> {
+ if
(conjunct.getInputSlotExprIds().stream().anyMatch(aggregateOutput::contains)) {
+ predicates.aggregatePredicates.add(conjunct);
+ } else {
+ predicates.pulledDomainPredicates.add(conjunct);
+ }
+ });
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(whereConjuncts)
+ .addAll(pulledDomainPredicates)
+ .build();
+ }
+
+ 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
keeps the original rewrite
+ */
+ private static Plan
pullUpCorrelatedPredicateByAggregatingOuter(LogicalApply<?, ?> apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ List<Expression> unCorrelatedPredicate,
CorrelatedAggregatePredicates predicates) {
+ if (containsSensitiveExpression(apply.left())
Review Comment:
[P1] Check movement safety on the right-side expressions too. With duplicate
outer keys, a volatile correlated predicate such as `random() < 0.5 + o.k * 0`
is evaluated once per outer-row subquery today; this rewrite deduplicates by
`k`, evaluates it once for that key, and joins the one result back to every
duplicate, so it can return only both rows or neither.
`containsSensitiveExpression` inspects only `apply.left()`, while the
correlated WHERE expression is explicitly excluded from the later scan. Reject
volatile/NoneMovable domain, HAVING, and aggregate expressions without entering
the unsafe fallback.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -119,6 +190,487 @@ public List<Rule> buildRules() {
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * 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> pulledDomainPredicates =
Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // the slots which exist above the aggregate tell whether a
predicate was evaluated on the
+ // aggregation (HAVING clause) instead of on the inner rows (WHERE
clause)
+ Set<ExprId> belowAggregate = filter.child().getOutput().stream()
+ .map(Slot::getExprId)
+ .collect(ImmutableSet.toImmutableSet());
+ Set<ExprId> aggregateOutput = agg.getOutput().stream()
+ .map(Slot::getExprId)
+ .filter(exprId -> !belowAggregate.contains(exprId))
+ .collect(ImmutableSet.toImmutableSet());
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(conjunct -> {
+ if
(conjunct.getInputSlotExprIds().stream().anyMatch(aggregateOutput::contains)) {
Review Comment:
[P1] Preserve HAVING provenance for outer-only conjuncts. A conjunct such as
`o.flag = 1` is pulled from HAVING into the Apply, but because it references no
aggregate-output slot this code classifies it as a domain predicate. The new
global-aggregate path then puts it in a LEFT JOIN condition; when `flag=0`, the
join still emits its padded row, `COUNT(marker)>=0` passes, and EXISTS returns
a row that the original HAVING rejects. Keep pulled HAVING conjuncts that can
reject the aggregate row above the reconstructed aggregate instead of inferring
provenance only from slot usage.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -119,6 +190,487 @@ public List<Rule> buildRules() {
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * 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> pulledDomainPredicates =
Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // the slots which exist above the aggregate tell whether a
predicate was evaluated on the
+ // aggregation (HAVING clause) instead of on the inner rows (WHERE
clause)
+ Set<ExprId> belowAggregate = filter.child().getOutput().stream()
+ .map(Slot::getExprId)
+ .collect(ImmutableSet.toImmutableSet());
+ Set<ExprId> aggregateOutput = agg.getOutput().stream()
+ .map(Slot::getExprId)
+ .filter(exprId -> !belowAggregate.contains(exprId))
+ .collect(ImmutableSet.toImmutableSet());
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(conjunct -> {
+ if
(conjunct.getInputSlotExprIds().stream().anyMatch(aggregateOutput::contains)) {
+ predicates.aggregatePredicates.add(conjunct);
+ } else {
+ predicates.pulledDomainPredicates.add(conjunct);
+ }
+ });
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(whereConjuncts)
+ .addAll(pulledDomainPredicates)
+ .build();
+ }
+
+ 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
keeps the original rewrite
+ */
+ private static Plan
pullUpCorrelatedPredicateByAggregatingOuter(LogicalApply<?, ?> apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ List<Expression> unCorrelatedPredicate,
CorrelatedAggregatePredicates predicates) {
+ if (containsSensitiveExpression(apply.left())
+ || referencesOuterSlot(apply.right(),
ImmutableSet.copyOf(predicates.whereConjuncts),
+ apply.getCorrelationSlot())
+ || !predicates.isResolvable(apply, agg, filter)) {
+ return null;
+ }
+ Set<Slot> correlationSlots =
predicates.keySlots(apply.getCorrelationSlot());
+
+ // 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<Expression> countExpressions = Sets.newLinkedHashSet();
+ if (keepEmptyDomain) {
+ // the count aggregations are the only ones which count the row
kept for an empty domain
Review Comment:
[P1] Keep the padded row out of every aggregate. This LEFT JOIN represents
an empty correlated input with one synthetic row, but only COUNT is compensated
below. For `EXISTS (SELECT SUM(1) ... HAVING SUM(1) IS NULL)`, a true empty
input yields NULL and passes; the synthetic row evaluates `1`, yields SUM=1,
and fails. `SUM0(1)` and null-transforming arguments have the same issue.
Please model an empty group without feeding its placeholder to aggregates, or
restrict/rewrite every aggregate using its exact empty-input contract.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/UnCorrelatedApplyAggregateFilter.java:
##########
@@ -119,6 +190,487 @@ public List<Rule> buildRules() {
apply.getCompareExpr(), apply.getTypeCoercionExpr(),
ExpressionUtils.optionalAnd(correlatedPredicate),
apply.getMarkJoinSlotReference(),
apply.isNeedAddSubOutputToProjects(),
apply.isMarkJoinSlotNotNull(), apply.left(),
- isRightChildAgg ? newAgg : apply.right().withChildren(newAgg));
+ replaceAggregate(apply.right(), newAgg));
+ }
+
+ /**
+ * 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> pulledDomainPredicates =
Lists.newArrayList();
+ private final List<Expression> aggregatePredicates =
Lists.newArrayList();
+ private final Set<Expression> havingConjuncts =
Sets.newLinkedHashSet();
+
+ private static CorrelatedAggregatePredicates of(LogicalApply<?, ?>
apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ Optional<LogicalFilter<Plan>> havingFilter, List<Expression>
whereConjuncts) {
+ CorrelatedAggregatePredicates predicates = new
CorrelatedAggregatePredicates();
+ predicates.whereConjuncts.addAll(whereConjuncts);
+ // the slots which exist above the aggregate tell whether a
predicate was evaluated on the
+ // aggregation (HAVING clause) instead of on the inner rows (WHERE
clause)
+ Set<ExprId> belowAggregate = filter.child().getOutput().stream()
+ .map(Slot::getExprId)
+ .collect(ImmutableSet.toImmutableSet());
+ Set<ExprId> aggregateOutput = agg.getOutput().stream()
+ .map(Slot::getExprId)
+ .filter(exprId -> !belowAggregate.contains(exprId))
+ .collect(ImmutableSet.toImmutableSet());
+ apply.getCorrelationFilter()
+ .map(ExpressionUtils::extractConjunction)
+ .orElse(ImmutableList.of())
+ .forEach(conjunct -> {
+ if
(conjunct.getInputSlotExprIds().stream().anyMatch(aggregateOutput::contains)) {
+ predicates.aggregatePredicates.add(conjunct);
+ } else {
+ predicates.pulledDomainPredicates.add(conjunct);
+ }
+ });
+ havingFilter.ifPresent(remaining ->
predicates.havingConjuncts.addAll(remaining.getConjuncts()));
+ return predicates;
+ }
+
+ private List<Expression> domainPredicates() {
+ return ImmutableList.<Expression>builder()
+ .addAll(whereConjuncts)
+ .addAll(pulledDomainPredicates)
+ .build();
+ }
+
+ 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
keeps the original rewrite
+ */
+ private static Plan
pullUpCorrelatedPredicateByAggregatingOuter(LogicalApply<?, ?> apply,
+ LogicalAggregate<LogicalFilter<Plan>> agg, LogicalFilter<Plan>
filter,
+ List<Expression> unCorrelatedPredicate,
CorrelatedAggregatePredicates predicates) {
+ if (containsSensitiveExpression(apply.left())
+ || referencesOuterSlot(apply.right(),
ImmutableSet.copyOf(predicates.whereConjuncts),
+ apply.getCorrelationSlot())
+ || !predicates.isResolvable(apply, agg, filter)) {
+ return null;
+ }
+ Set<Slot> correlationSlots =
predicates.keySlots(apply.getCorrelationSlot());
+
+ // 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());
Review Comment:
[P1] Reuse the same evaluated outer input for the key domain. This deep copy
executes independently: an unseeded TABLESAMPLE retains its random seek state
in both scans, so the original branch and copied key branch can select disjoint
keys; the final semi join can then return zero rows even when the original
EXISTS is always true. Unordered LIMIT and tied TopN have the same plan-level
risk. Please share/materialize the outer input or conservatively reject outer
plans whose key set is not deterministic.
--
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]
