github-actions[bot] commented on code in PR #63763:
URL: https://github.com/apache/doris/pull/63763#discussion_r3673945433
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunction.java:
##########
@@ -362,53 +611,325 @@ private Plan rewrite(LogicalFilter<? extends Plan>
filter, LogicalApply<Plan, Pl
windowFilterConjunct = ExpressionUtils.replace(windowFilterConjunct,
ImmutableMap.of(aggOut.toSlot(), aggOutExpr));
- LogicalFilter<Plan> newFilter =
filter.withConjunctsAndChild(conjuncts.get(true), apply.left());
+ // Split uncorrelated conjuncts: predicates that reference ONLY shared
+ // relation slots (tables appearing in both outer and inner plans) must
+ // stay ABOVE the window. Otherwise the window function would see a
+ // different set of rows than the original scalar subquery.
+ //
+ // For example, with fact(f) as shared table and dim(d) as outer-only:
+ // f.v > 6 → shared-only → must stay above the window
+ // d.tag > 0 → outer-only → safe below the window
+ // f.k = d.k → join cond → needed below the window
+ //
+ // We find shared tables by comparing table IDs that appear in both
+ // outer and inner plans, then collect ALL output slots of those
+ // tables (not just columns referenced in the inner query).
+ // outerOnlyTable is already identified by checkRelation().
+ List<CatalogRelation> outerRels = outerPlans.stream()
+ .filter(CatalogRelation.class::isInstance)
+ .map(CatalogRelation.class::cast)
+ .collect(Collectors.toList());
+ Set<Long> innerTableIds = innerPlans.stream()
+ .filter(CatalogRelation.class::isInstance)
+ .map(r -> ((CatalogRelation) r).getTable().getId())
+ .collect(Collectors.toSet());
+ Set<ExprId> sharedOuterExprIds = outerRels.stream()
+ .filter(r -> innerTableIds.contains(r.getTable().getId()))
+ .flatMap(r -> r.getOutput().stream())
+ .map(Slot::getExprId)
+ .collect(Collectors.toSet());
+ Set<Expression> uncorrelatedConjuncts = conjuncts.get(true);
+ Set<Expression> belowWindowConjuncts = Sets.newHashSet();
+ Set<Expression> aboveWindowConjuncts = Sets.newHashSet();
+ if (uncorrelatedConjuncts != null) {
+ // If the outer filter (the filter on top of the Apply) contains
+ // both a volatile/NoneMovableFunction conjunct AND another
+ // deterministic conjunct that would go to a different side of
+ // the window, reject the rewrite. Splitting conjuncts from the
+ // same filter operator changes which rows reach the
+ // side-effecting predicate and can suppress expected errors.
+ //
+ // Example: a filter with both
+ // d.tag > 0 (outer-only → below window)
+ // assert_true(d.tag > 1, 'bad') (NoneMovable → above
window)
+ // splits the pair: rows filtered by d.tag > 0 never reach
+ // assert_true above the window, whereas the original operator
+ // evaluates assert_true on every row of the input block.
+ boolean hasVolatileInOuterFilter = false;
+ boolean needsBelowFromOuterFilter = false;
+ for (Expression conj : uncorrelatedConjuncts) {
+ if (matchedInnerFilterConjuncts.contains(conj)) {
+ continue;
+ }
+ if (isVolatileOrNoneMovable(conj)) {
+ hasVolatileInOuterFilter = true;
+ } else if (!referencesSharedTable(conj, sharedOuterExprIds)) {
+ needsBelowFromOuterFilter = true;
+ }
+ }
+ if (hasVolatileInOuterFilter && needsBelowFromOuterFilter) {
+ return filter;
+ }
+ for (Expression conj : uncorrelatedConjuncts) {
+ // Conjuncts that were matched against inner subquery filter
+ // conjuncts (tracked by checkFilter) must stay BELOW the
+ // window. They are semantically part of the inner aggregate's
+ // filter, not extra outer-only predicates. Placing them above
+ // the window would let the window see more rows than the
+ // original scalar subquery, producing wrong aggregate results.
+ if (matchedInnerFilterConjuncts.contains(conj)) {
+ belowWindowConjuncts.add(conj);
+ continue;
+ }
+ // Volatile and NoneMovableFunction predicates must stay
+ // ABOVE the window — pushing them below would change
+ // evaluation frequency or move a side-effecting call.
+ if (isVolatileOrNoneMovable(conj)) {
+ aboveWindowConjuncts.add(conj);
+ continue;
+ }
+ // Predicates referencing shared-table columns must stay
+ // ABOVE the window; otherwise the window sees fewer rows
+ // than the original scalar subquery.
+ if (referencesSharedTable(conj, sharedOuterExprIds)) {
+ aboveWindowConjuncts.add(conj);
+ } else {
+ belowWindowConjuncts.add(conj);
Review Comment:
[P1] Reject unsafe aggregate arguments before filtering below the window
This branch can suppress an error from a `NoneMovableFunction` inside the
aggregate argument.
A newly eligible reduced plan is:
```text
Filter(f.k = d.k, d.tag > 0, f.v > scalar_count)
Apply(correlation = d.k)
CrossJoin(Filter(f.keep > 0, Scan fact f), Scan unique_dim d)
Aggregate(COUNT(assert_true(f2.v > 0, 'bad')))
Filter(f2.k = d.k, Scan fact f2)
```
With a failing fact row for key 1 and `d(key=1, tag=0)`, the normal Apply
fallback evaluates
the right aggregate by `f2.k` before joining/filtering and raises. Here
`d.tag > 0` and the
matched join predicate are put below
`COUNT(assert_true(f.v > 0, 'bad')) OVER (PARTITION BY d.k)`, so key 1 is
removed before
`assert_true` runs and the error disappears. The nested outer filter is what
makes this exact
shape newly accepted by the PR.
This is separate from the existing same-filter-sibling thread: every filter
conjunct here is
deterministic; the unsafe expression is inside the aggregate. Please reject
aggregates that
contain volatile or `NoneMovableFunction` expressions (or prove no predicate
is placed below
their window), and add a negative rule test for this shape.
--
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]
