github-actions[bot] commented on code in PR #63763:
URL: https://github.com/apache/doris/pull/63763#discussion_r3592551091
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunction.java:
##########
@@ -401,6 +787,32 @@ public Boolean visit(Expression expression, Expression
expression1) {
return isClassMatch(expression, expression1) &&
isSameChild(expression, expression1);
}
+ @Override
+ public Boolean visitBoundFunction(BoundFunction boundFunction,
Expression other) {
+ // BoundFunction equality requires matching function names
+ // (via extraEquals), not just matching class and children.
+ // Two UDF wrapper nodes (e.g. JavaUdf / PythonUdf) with the
+ // same runtime class and same children but different function
+ // names are different functions — they must not be matched.
+ return isClassMatch(boundFunction, other)
+ && Objects.equals(boundFunction.getName(),
((BoundFunction) other).getName())
Review Comment:
These new node-specific checks still leave the generic fallback semantically
unsafe for `Cast`/`TryCast`. A reduced accepted shape has `TRY_CAST(f.s AS INT)
IS NULL` in the outer filter and `TRY_CAST(f2.s AS DATE) IS NULL` in the
correlated aggregate filter. After the slot replacement, `IsNull` and `TryCast`
both delegate to the generic visitor, which sees the same classes and slot
child but never compares `Cast.targetType` (even though `Cast.equals()` does).
`checkFilter()` therefore records the INT predicate as the matched inner
filter, and the rewrite computes the window under it instead of the DATE
predicate. With values such as `s='2024-01-01'` and `s='2147483647'`, those
predicates select opposite rows, so the scalar sum and rewritten window sum can
differ. This is distinct from the earlier UDF/MATCH threads: those special
cases are now guarded, but the cast target type remains unhandled. Please use
semantic equality for the generic case while retaining the comparison-commut
ation exception, or add target-type-aware Cast/TryCast matching and a
regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunction.java:
##########
@@ -362,17 +520,245 @@ 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).
+ List<CatalogRelation> outerRels = outerPlans.stream()
+ .filter(CatalogRelation.class::isInstance)
+ .map(CatalogRelation.class::cast)
+ .collect(Collectors.toList());
+ List<CatalogRelation> innerRels = innerPlans.stream()
+ .filter(CatalogRelation.class::isInstance)
+ .map(CatalogRelation.class::cast)
+ .collect(Collectors.toList());
+ Set<Long> innerTableIds = innerRels.stream()
+ .map(r -> 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) {
+ 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);
+ }
+ }
+ }
+
+ // Extract and classify conjuncts from any LogicalFilter nodes nested
+ // inside the outer child (apply.left()). Nested shared-table filters
+ // (e.g. f.v > 6 under a CrossJoin) must be hoisted ABOVE the window;
+ // otherwise the window would aggregate over a filtered subset of rows
+ // while the original scalar subquery sees all rows for the key.
+ //
+ // Nested outer-only filters (e.g. d.tag > 0) are safe below and can
+ // stay in place after the filters are stripped.
+ List<LogicalFilter<Plan>> nestedOuterFilters = apply.left()
+ .collectToList(LogicalFilter.class::isInstance);
+ Set<ExprId> extractedConjunctExprIds = Sets.newHashSet();
+ for (LogicalFilter<Plan> nf : nestedOuterFilters) {
+ for (Expression conj : nf.getConjuncts()) {
+ // Matched inner-filter conjuncts always go below.
+ if (matchedInnerFilterConjuncts.contains(conj)) {
+ belowWindowConjuncts.add(conj);
+
extractedConjunctExprIds.addAll(conj.getInputSlotExprIds());
+ continue;
+ }
+ // Volatile / NoneMovable on a shared table would restrict
+ // the window's input — reject the rewrite.
+ if (isVolatileOrNoneMovable(conj)) {
+ if (nf.collect(CatalogRelation.class::isInstance).stream()
+ .map(r -> ((CatalogRelation) r).getTable().getId())
+ .anyMatch(innerTableIds::contains)) {
+ return filter;
+ }
+ continue;
+ }
+ if (referencesSharedTable(conj, sharedOuterExprIds)) {
+ aboveWindowConjuncts.add(conj);
+ } else {
+ belowWindowConjuncts.add(conj);
+ }
+ extractedConjunctExprIds.addAll(conj.getInputSlotExprIds());
+ }
+ }
+ // Strip all nested LogicalFilter nodes from the outer child so the
+ // window operates on the unfiltered scan/join.
+ Plan strippedOuterChild = stripOuterFilters(apply.left());
+
+ // The window function's aggregate references shared-table slots
+ // (e.g. f.v for SUM(f.v) after slot replacement). A pruning
+ // project inside apply.left() may have dropped those slots even
+ // when there are no nested filters to extract. Collect the
+ // replaced aggregate's input slot ExprIds and include them in
+ // the project-expansion set so ensureProjectOutput carries them
+ // through.
+ Set<ExprId> allNeededExprIds =
Sets.newHashSet(extractedConjunctExprIds);
+ allNeededExprIds.addAll(ExpressionUtils.replace(function,
innerOuterSlotMap)
+ .getInputSlotExprIds());
+ strippedOuterChild = ensureProjectOutput(strippedOuterChild,
+ allNeededExprIds);
+
+ LogicalFilter<Plan> newFilter = filter.withConjunctsAndChild(
+ belowWindowConjuncts, strippedOuterChild);
LogicalWindow<Plan> newWindow = new
LogicalWindow<>(ImmutableList.of(windowFunctionAlias), newFilter);
- LogicalFilter<Plan> windowFilter = new
LogicalFilter<>(ImmutableSet.of(windowFilterConjunct), newWindow);
+
+ // Combine shared-table predicates with the window comparison
predicate above the window
+ Set<Expression> topConjuncts = Sets.newHashSet(windowFilterConjunct);
+ topConjuncts.addAll(aboveWindowConjuncts);
+ LogicalFilter<Plan> windowFilter = new
LogicalFilter<>(ImmutableSet.copyOf(topConjuncts), newWindow);
return windowFilter;
}
+ /**
+ * Ensure that every LogicalProject in the plan tree outputs all slots
+ * referenced by extracted conjuncts (ExprIds in {@code neededExprIds}).
+ * If a project prunes away a needed slot, the reinserted filter
+ * predicates would have dangling references.
+ *
+ * <p>Recursively walks the tree depth-first; expands each project to
+ * include any missing slots as simple identity projections (SlotReference
→
+ * SlotReference). Recursing into the child before computing the current
+ * project's {@code childOutput} ensures that stacked pruning projects
+ * (e.g. {@code Project(k) → Project(k) → Scan(k, v)}) are expanded
+ * bottom-up — the inner project is expanded first, and then the outer
+ * project can see the expanded slots in its child output.
+ */
+ private Plan ensureProjectOutput(Plan plan, Set<ExprId> neededExprIds) {
+ if (neededExprIds.isEmpty()) {
+ return plan;
+ }
+ if (plan instanceof LogicalProject) {
+ LogicalProject<Plan> project = (LogicalProject<Plan>) plan;
+ // Recurse into the child first so stacked pruning projects
+ // are expanded bottom-up. The child's output after expansion
+ // determines which slots the current project can pull through.
+ Plan newChild = ensureProjectOutput(project.child(),
neededExprIds);
+ Set<ExprId> projectOutputExprIds = project.getOutputExprIdSet();
+ Set<Slot> childOutput = newChild.getOutputSet();
+ List<NamedExpression> newProjects =
Lists.newArrayList(project.getProjects());
+ boolean expanded = false;
+ for (ExprId id : neededExprIds) {
+ if (!projectOutputExprIds.contains(id)) {
+ for (Slot slot : childOutput) {
+ if (slot.getExprId().equals(id)) {
+ newProjects.add(slot);
+ expanded = true;
+ break;
+ }
+ }
+ }
+ }
+ if (expanded) {
+ return project.withProjectsAndChild(newProjects, newChild);
+ }
+ if (newChild != project.child()) {
+ return project.withChildren(ImmutableList.of(newChild));
+ }
+ return plan;
+ }
+ if (plan.children().isEmpty()) {
+ return plan;
+ }
+ List<Plan> newChildren = plan.children().stream()
+ .map(c -> ensureProjectOutput(c, neededExprIds))
+ .collect(Collectors.toList());
+ return plan.withChildren(newChildren);
+ }
+
private WindowExpression createWindowFunction(List<Slot> correlatedSlots,
AggregateFunction function) {
// partition by clause is set by all the correlated slots.
return new WindowExpression(function,
ImmutableList.copyOf(correlatedSlots), Collections.emptyList());
}
+ /** Recursively strip deterministic, movable LogicalFilter nodes from the
+ * plan tree. Filters containing volatile predicates (e.g. random()) or
+ * NoneMovableFunction predicates (e.g. assert_true()) are left in place —
+ * moving such predicates across a join/window changes their evaluation
+ * context and can alter query results. */
+ private Plan stripOuterFilters(Plan plan) {
+ if (plan instanceof LogicalFilter) {
+ LogicalFilter<?> filter = (LogicalFilter<?>) plan;
+ // Separate unsafe-to-move conjuncts from safe ones.
+ // Volatile and NoneMovableFunction predicates stay at their
+ // original position; deterministic movable predicates have been
+ // extracted and can be stripped.
+ Set<Expression> keepConjuncts = filter.getConjuncts().stream()
+
.filter(AggScalarSubQueryToWindowFunction::isVolatileOrNoneMovable)
+ .collect(Collectors.toSet());
+ if (keepConjuncts.isEmpty()) {
+ // All conjuncts are safe to strip.
+ return stripOuterFilters(filter.child(0));
+ }
+ // Keep only unsafe-to-move conjuncts at this position.
+ Plan strippedChild = stripOuterFilters(filter.child(0));
+ return new LogicalFilter<>(ImmutableSet.copyOf(keepConjuncts),
strippedChild);
Review Comment:
This recursive stripping can still change a retained `NoneMovableFunction`
filter's input in a production-reachable nested-correlated shape. Consider
`Filter(assert_true(x.tag > g.tag, 'bad')) -> Project(alias x) -> Filter(x.keep
= 1) -> Scan dim_unique`, where `g.tag` is a grandparent outer-scope slot.
Alias normalization creates the project, but `PushDownFilterThroughProject`
must leave the assertion above it because `g.tag` is not in the project's
output, so `MergeFilters` cannot combine it with `x.keep = 1`. The inner
WinMagic candidate can otherwise pass: the grandparent is only an external
slot, `x.k` is unique/non-null, and the unsafe filter's subtree contains only
the outer-only dim table. This recursion then removes `x.keep = 1` below the
retained assertion and reinserts it above the join, so a `keep=0` row that was
filtered first can newly reach `assert_true` and raise. This is distinct from
the earlier lone-`NoneMovableFunction` thread: the unsafe node stays in place he
re, but its input domain changes. Please treat a retained unsafe filter as a
subtree movement barrier (or reject movable descendants beneath it) and add a
nested-correlated regression.
--
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]
