spark-test commented on code in PR #60090:
URL: https://github.com/apache/spark/pull/60090#discussion_r4483274253
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3860,6 +3860,25 @@ class Analyzer(
Project(windowOps.output ++ newExpressionsWithWindowFunctions, windowOps)
} // end of addWindow
+ /**
+ * Extracts all nested window expressions and places the HAVING filter
below every Window.
+ */
+ private def addWindowWithHaving(
Review Comment:
**Non-blocking (P2):** This placement runs only when HAVING reaches
`ExtractWindowExpressions` still attached to its Aggregate, which grouping
analytics does only for conditions that need type coercion.
`BaseGroupingSets.resolved` is always false, so the `UnresolvedHaving` case
that calls this method cannot match while the Aggregate still has ROLLUP, CUBE
or GROUPING SETS. `ResolveGroupingAnalytics`, which runs earlier in the batch,
calls the unchanged `tryResolveHavingCondition` instead. When the condition
resolves without type coercion, that method returns `Filter(cond,
constructAggregate(...))` while the window expressions are still in the
aggregate list. Every window layer is then extracted below that Filter,
whatever `spark.sql.legacy.evaluateHavingAfterInnerWindow` is set to. For
example:
```sql
SELECT k, SUM(RANK() OVER (ORDER BY SUM(v) DESC)) OVER () AS rank_sum
FROM VALUES ('drop', 2), ('keep', 1) AS input(k, v)
GROUP BY ROLLUP(k)
HAVING k = 'keep'
```
The windows run over the groups `(NULL, 3)`, `('drop', 2)` and `('keep',
1)`, so the query returns `(keep, 6)`. Evaluating HAVING first gives `(keep,
1)`, which is what the same query with `GROUP BY k` now returns. The existing
`GROUPING_ID() = 1` golden lands below its window only because that condition
needs coercion and is deferred to this method. The migration note and the
config doc state the fix unconditionally, and no test covers this form. A
HAVING condition that reads a SELECT-list window alias also resolves on this
route and is evaluated after every window; that case is outside this gap.
**Recommended change:** In ResolveGroupingAnalytics, when
spark.sql.legacy.evaluateHavingAfterInnerWindow is false, the aggregate list
still contains window expressions, and the resolved HAVING condition reads no
window output, keep the resolved HAVING attached to the constructed Aggregate
as an unresolved HAVING, instead of turning it into a Filter. The nested-window
placement then extracts every window layer first and puts the Filter directly
on the Aggregate. A condition that reads a SELECT-list window alias keeps the
current Filter above every window. Add window-resolution SQL golden tests for
ROLLUP or GROUPING SETS with nested and single-level windows and a HAVING that
needs no coercion, plus one whose HAVING reads a window alias, under both flag
values, and update the migration note and the config doc to cover grouping
analytics.
**Why this works:** The condition is already resolved and remapped to the
constructed Aggregate's output, and that Aggregate's own expressions are
resolved. ExtractWindowExpressions' HAVING case therefore matches it in the
same analyzer batch, and the nested-window placement builds the window layers
over a Filter on the Aggregate. The existing grouping-function and aggregate
rewrites still see a Filter directly over an Aggregate. When extra aggregates
were pushed for the condition, the existing projection that drops them stays on
top. Because this rule resolves HAVING against the Aggregate output, which
includes the SELECT-list window aliases, the remapped condition's references
are checked against the outputs of the window-bearing aggregate expressions: a
condition that reads one keeps the current Filter, since the nested-window
placement would put it below the Window that produces that value and analysis
would fail with a missing attribute.
**Scope:** Route grouping-analytics HAVING that reads no window output
through the change's nested-window placement, add goldens for that route, and
align the documentation.
**Compatibility:** Grouping-analytics HAVING without window functions in the
SELECT list keeps its current plans. A grouping-analytics HAVING that reads a
window alias keeps its Filter above every window, and grouping-analytics
queries with a generator in the SELECT list never reach this rule, so both keep
their current plans.
**Risks:** Single-level window queries over ROLLUP, CUBE or GROUPING SETS
with a HAVING that needs no coercion also change plans and results. HAVING now
runs before their window, which corrects the base-branch behavior. The legacy
flag rolls this back, and the migration note should mention it.
**Constraints:** Gate the new routing on
spark.sql.legacy.evaluateHavingAfterInnerWindow, so that true keeps the current
placement on this route. Keep the HAVING condition resolved against the
constructed Aggregate, including the existing attribute remapping and any
aggregate expressions pushed for the condition. Do not defer a condition that
reads a window output; keep it as a Filter above every window, as today.
**Exclusions:**
- The single-pass resolver already declines HAVING with grouping analytics
and falls back to the fixed-point analyzer, and it stays unchanged.
- Whether a HAVING condition that reads a window alias should be rejected on
this route, as it already is for plain GROUP BY, is a separate decision that
this repair does not make.
**Success:** With the legacy flag false, ROLLUP, CUBE and GROUPING SETS
queries with window functions and a HAVING condition that resolves without
coercion and reads no window output evaluate HAVING below every window layer.
With the legacy flag true, those queries keep the base-branch plans and
results. A grouping-analytics HAVING condition that reads a SELECT-list window
alias keeps its current plan and result under both flag values. The migration
note and the config doc accurately describe grouping-analytics queries.
**Verification:**
- **Regression:** A ROLLUP or GROUPING SETS query with nested window
functions and a HAVING on a grouping key that needs no coercion places the
HAVING filter directly above the Aggregate and returns rank_sum 1 by default.
- **Compatibility:** The same query with
spark.sql.legacy.evaluateHavingAfterInnerWindow set to true keeps the
base-branch plan and result.
- **Behavior:** Grouping-analytics HAVING conditions that use grouping
functions, aggregates absent from the SELECT list, or a subquery still resolve
and keep the right groups when window functions are present.
- **Behavior:** A single-level window over ROLLUP, CUBE or GROUPING SETS
with a HAVING that needs no coercion evaluates HAVING before the window by
default.
- **Behavior:** A ROLLUP or GROUPING SETS query with window functions whose
HAVING reads a SELECT-list window alias without needing coercion keeps the
HAVING filter above every window and returns the same result as before the
change, under both flag values.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3860,6 +3860,25 @@ class Analyzer(
Project(windowOps.output ++ newExpressionsWithWindowFunctions, windowOps)
} // end of addWindow
+ /**
+ * Extracts all nested window expressions and places the HAVING filter
below every Window.
Review Comment:
**Nit (P3):** With `LEGACY_EVALUATE_HAVING_AFTER_INNER_WINDOW` set to true,
the recursion guard below fails. The method then extracts only the outermost
layer and returns `addWindow` over `Filter(condition, Aggregate(...))`, whose
aggregate list still holds the inner window expressions. Those are later
extracted below the filter, as the new legacy golden shows (`Window [sum] +-
Filter (k = keep) +- Project +- Project +- Window [rank] +- Aggregate`). The
doc states the default-branch guarantee unconditionally; it could cover both
branches:
```suggestion
* Extracts all nested window expressions and places the HAVING filter
below every Window,
* unless LEGACY_EVALUATE_HAVING_AFTER_INNER_WINDOW is true: then only
the outermost layer is
* extracted here, and the inner windows are later extracted below the
HAVING filter.
```
**Verification:**
- **Inspection:** The addWindowWithHaving Scaladoc states that HAVING goes
below every nested window layer only when
spark.sql.legacy.evaluateHavingAfterInnerWindow is false, and that with true
only the outermost layer is extracted and HAVING can end up above the inner
windows.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4935,19 +4949,47 @@ object ResolveUnresolvedHaving extends
Rule[LogicalPlan] {
*
* Returns None unless the condition can be evaluated below every Window in
the chain.
*/
- private def insertFilterBeforeWindow(filter: Filter): Option[LogicalPlan] =
filter.child match {
- case project: Project =>
- insertFilterBeforeWindow(filter.copy(child = project.child))
+ private def insertFilterBeforeWindow(filter: Filter): Option[LogicalPlan] = {
+ if
(SQLConf.get.getConf(SQLConf.LEGACY_EVALUATE_HAVING_AFTER_INNER_WINDOW)) {
+ insertFilterBeforeOuterWindow(filter)
+ } else {
+ insertFilterBeforeAllWindows(filter, foundWindow = false)
+ }
+ }
+
+ private def insertFilterBeforeAllWindows(
+ filter: Filter,
+ foundWindow: Boolean): Option[LogicalPlan] = filter.child match {
+ case project: Project if
filter.condition.references.subsetOf(project.child.outputSet) =>
+ insertFilterBeforeAllWindows(filter.copy(child = project.child),
foundWindow)
.map(child => project.withNewChildren(Seq(child)))
- case generate: Generate =>
- insertFilterBeforeWindow(filter.copy(child = generate.child))
+ case generate: Generate if
filter.condition.references.subsetOf(generate.child.outputSet) =>
+ insertFilterBeforeAllWindows(filter.copy(child = generate.child),
foundWindow)
.map(child => generate.withNewChildren(Seq(child)))
- case window: Window =>
- insertFilterBeforeWindowChain(filter, window)
+ case window: Window if
filter.condition.references.subsetOf(window.child.outputSet) =>
+ insertFilterBeforeAllWindows(filter.copy(child = window.child),
foundWindow = true)
+ .map(child => window.withNewChildren(Seq(child)))
+ case _ if foundWindow &&
filter.condition.references.subsetOf(filter.child.outputSet) =>
Review Comment:
**Blocking (P1):** Once `foundWindow` is true, this fallback accepts any
stop point, including a lower `Window` of the same chain whose child lacks the
condition's references. That breaks the all-or-none contract in the method doc
above ("Returns None unless the condition can be evaluated below every Window
in the chain").
It is reachable from SQL. In a generator query, HAVING is resolved against
the post-generator `Project`, so it can read a SELECT-list window alias:
```sql
SELECT explode(array(k)) AS g,
RANK() OVER (ORDER BY k) AS r,
COUNT(*) OVER () AS c
FROM VALUES ('a', 1), ('b', 2) AS t(k, v)
GROUP BY k
HAVING r = 1
```
`addWindow` stacks `Window[c]` over `Window[r]`. The walk passes
`Window[c]`, fails the guard at `Window[r]` (`r` is not in the Aggregate
output), and this case returns `Filter(r = 1, Window[r])`. `COUNT(*) OVER ()`
therefore counts only the surviving row, and the query returns `(a, 1, 1)`.
Before this change, and with the legacy flag set to true, the filter stays
above both windows and the query returns `(a, 1, 2)`. Listing `c` before `r`
brings back `c = 2`, so the value now depends on SELECT-list order. Neither the
migration note nor the config doc describes this, and no test covers it.
Suggested fix: return `None` whenever the walk stops at a `Window`, or at a
`Project` or `Generate` that still has a `Window` below it, so the filter is
inserted only once every Window of the chain is above the stop point.
This extends an [existing
discussion](https://github.com/apache/spark/pull/60090#pullrequestreview-5770227938).
**Verification:**
- **Regression:** A generator query whose HAVING references the alias of a
window computed below another window spec keeps the HAVING filter above both
windows and returns the base-branch values, and the same query with the two
window expressions listed in the opposite order returns the same values.
- **Behavior:** Generator queries with nested windows and HAVING on a
grouping key still place the HAVING filter directly above the Aggregate and
return rank_sum 1.
- **Compatibility:** With spark.sql.legacy.evaluateHavingAfterInnerWindow
set to true, generator-query plans and results are unchanged.
- **Inspection:** The default-path walk returns no placement whenever a
Window would remain below the inserted filter, matching the documented contract
that HAVING goes below every Window in the chain or not at all.
--
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]
