[SPARK-58503][SQL] Evaluate HAVING before window functions with generators

docs/sql-migration-guide.md

@@ -39,6 +39,7 @@ license: |
 - Since Spark 4.3, `unix_seconds`, `unix_millis`, and `unix_micros` accept `TIMESTAMP_NTZ` and the nanosecond-precision timestamp types directly, reading them with no time-zone shift. Previously these functions accepted only `TIMESTAMP_LTZ`; a `TIMESTAMP_NTZ` or nanosecond-timestamp argument was rejected with a `DATATYPE_MISMATCH` error. This is a new capability and does not change the result of any query that previously succeeded.
 - Since Spark 4.3, `hash()` and `xxhash64()` include the `days` field of `CalendarInterval` when computing the hash, so their output for interval values differs from earlier releases. Previously the codegen path dropped `days`, disagreeing with interpreted evaluation.
 - Since Spark 4.3, when a `SELECT` or `INSERT` statement references the same table more than once with different `WITH (...)` options (for example a self-join, or `INSERT INTO t WITH (...) SELECT * FROM t WITH (...)`), each reference now uses its own options instead of the second reference silently inheriting the first reference's options via the analyzer's relation cache.
+- Since Spark 4.3, `HAVING` is evaluated before window functions when the `SELECT` list also contains generator functions such as `explode`. Previously, window functions could include groups removed by `HAVING` and produce incorrect results.
 
 ## Upgrading from Spark SQL 4.1 to 4.2
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -4706,7 +4706,50 @@ object ResolveUnresolvedHaving extends Rule[LogicalPlan] {
     plan.resolveOperatorsWithPruning(_.containsPattern(UNRESOLVED_HAVING), ruleId) {
       case u @ UnresolvedHaving(havingCondition, child)
         if havingCondition.resolved && child.resolved =>
-        Filter(condition = havingCondition, child = child)
+        val filter = Filter(condition = havingCondition, child = child)
+        insertFilterBeforeWindow(filter).getOrElse(filter)
+    }
+  }
+
+  /**
+   * Searches through Project and Generate nodes for a Window chain and places HAVING below every
+   * Window in that chain. This restores SQL clause order for plans produced by queries such as:
+   *
+   * {{{
+   * SELECT explode(array(a)), count(*) OVER ()
+   * FROM VALUES (1), (2), (NULL) AS t(a)
+   * GROUP BY a
+   * HAVING a IS NOT NULL
+   * }}}
+   *
+   * 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))
+        .map(child => project.withNewChildren(Seq(child)))
+    case generate: Generate =>
+      insertFilterBeforeWindow(filter.copy(child = generate.child))
+        .map(child => generate.withNewChildren(Seq(child)))
+    case window: Window =>
+      insertFilterBeforeWindowChain(filter, window)
+    case _ =>
+      None
+  }
+
+  private def insertFilterBeforeWindowChain(
+      filter: Filter,
+      window: Window): Option[LogicalPlan] = {
+    if (!filter.condition.references.subsetOf(window.child.outputSet)) {
+      None
+    } else {
+      val child = window.child match {
+        case childWindow: Window =>
+          insertFilterBeforeWindowChain(filter, childWindow)
+        case child =>
+          Some(filter.copy(child = child))
+      }
+      child.map(child => window.withNewChildren(Seq(child)))
     }
   }
 }

sql/core/src/test/resources/sql-tests/analyzer-results/generators-resolution-edge-cases.sql.out

@@ -781,3 +781,63 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "val"
   } ]
 }
+
+
+-- !query
+SELECT explode(array(a)) AS col, count(*) OVER () AS cnt, a
+FROM VALUES (1), (2), (3), (NULL) AS t(a)
+GROUP BY a
+HAVING a IS NOT NULL
+-- !query analysis
+Project [col#x, cnt#xL, a#x]
++- Generate explode(_gen_input_0#x), false, [col#x]
+   +- Project [_gen_input_0#x, cnt#xL, a#x]
+      +- Project [_gen_input_0#x, a#x, cnt#xL, cnt#xL]
+         +- Window [count(1) windowspecdefinition(specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS cnt#xL]
+            +- Filter isnotnull(a#x)
+               +- Aggregate [a#x], [array(a#x) AS _gen_input_0#x, a#x]
+                  +- SubqueryAlias t
+                     +- LocalRelation [a#x]
+
+
+-- !query
+SELECT explode(array(a)) AS col,
+       count(*) OVER () AS group_count,
+       count(*) AS row_count
+FROM VALUES (1), (1), (2), (3), (3) AS t(a)
+GROUP BY a
+HAVING row_count > 1
+-- !query analysis
+Project [col#x, group_count#xL, row_count#xL]
++- Generate explode(_gen_input_0#x), false, [col#x]
+   +- Project [_gen_input_0#x, group_count#xL, row_count#xL]
+      +- Project [_gen_input_0#x, row_count#xL, group_count#xL, group_count#xL]
+         +- Window [count(1) windowspecdefinition(specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS group_count#xL]
+            +- Filter (row_count#xL > cast(1 as bigint))
+               +- Aggregate [a#x], [array(a#x) AS _gen_input_0#x, count(1) AS row_count#xL]
+                  +- SubqueryAlias t
+                     +- LocalRelation [a#x]
+
+
+-- !query
+SELECT explode(array(a)) AS col,
+       count(*) OVER () AS cnt,
+       count(*) OVER (
+         ORDER BY a
+         ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+       ) AS ordered_cnt,
+       a
+FROM VALUES (1), (2), (3), (NULL) AS t(a)
+GROUP BY a
+HAVING a IS NOT NULL
+-- !query analysis
+Project [col#x, cnt#xL, ordered_cnt#xL, a#x]
++- Generate explode(_gen_input_0#x), false, [col#x]
+   +- Project [_gen_input_0#x, cnt#xL, ordered_cnt#xL, a#x]
+      +- Project [_gen_input_0#x, a#x, cnt#xL, ordered_cnt#xL, cnt#xL, ordered_cnt#xL]
+         +- Window [count(1) windowspecdefinition(a#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS ordered_cnt#xL], [a#x ASC NULLS FIRST]
+            +- Window [count(1) windowspecdefinition(specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS cnt#xL]
+               +- Filter isnotnull(a#x)
+                  +- Aggregate [a#x], [array(a#x) AS _gen_input_0#x, a#x]
+                     +- SubqueryAlias t
+                        +- LocalRelation [a#x]

sql/core/src/test/resources/sql-tests/inputs/generators-resolution-edge-cases.sql

@@ -215,3 +215,29 @@ SELECT posexplode(array('x', 'y')) as (pos, val), pos, val FROM (VALUES (42)) AS
 
 -- generator's multi-alias does not shadow table column with aggregate
 SELECT posexplode(array('x', 'y')) as (pos, val), pos, val, count(*) FROM (VALUES (42)) AS t(pos) GROUP BY pos;
+
+-- HAVING on a grouping key should be evaluated before a window function with a generator
+SELECT explode(array(a)) AS col, count(*) OVER () AS cnt, a
+FROM VALUES (1), (2), (3), (NULL) AS t(a)
+GROUP BY a
+HAVING a IS NOT NULL;
+
+-- HAVING on an aggregate should be evaluated before a window function with a generator
+SELECT explode(array(a)) AS col,
+       count(*) OVER () AS group_count,
+       count(*) AS row_count
+FROM VALUES (1), (1), (2), (3), (3) AS t(a)
+GROUP BY a
+HAVING row_count > 1;
+
+-- HAVING should be evaluated before multiple window functions with a generator
+SELECT explode(array(a)) AS col,
+       count(*) OVER () AS cnt,
+       count(*) OVER (
+         ORDER BY a
+         ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+       ) AS ordered_cnt,
+       a
+FROM VALUES (1), (2), (3), (NULL) AS t(a)
+GROUP BY a
+HAVING a IS NOT NULL;

sql/core/src/test/resources/sql-tests/results/generators-resolution-edge-cases.sql.out

@@ -781,3 +781,49 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "val"
   } ]
 }
+
+
+-- !query
+SELECT explode(array(a)) AS col, count(*) OVER () AS cnt, a
+FROM VALUES (1), (2), (3), (NULL) AS t(a)
+GROUP BY a
+HAVING a IS NOT NULL
+-- !query schema
+struct<col:int,cnt:bigint,a:int>
+-- !query output
+1	3	1
+2	3	2
+3	3	3
+
+
+-- !query
+SELECT explode(array(a)) AS col,
+       count(*) OVER () AS group_count,
+       count(*) AS row_count
+FROM VALUES (1), (1), (2), (3), (3) AS t(a)
+GROUP BY a
+HAVING row_count > 1
+-- !query schema
+struct<col:int,group_count:bigint,row_count:bigint>
+-- !query output
+1	2	2
+3	2	2
+
+
+-- !query
+SELECT explode(array(a)) AS col,
+       count(*) OVER () AS cnt,
+       count(*) OVER (
+         ORDER BY a
+         ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+       ) AS ordered_cnt,
+       a
+FROM VALUES (1), (2), (3), (NULL) AS t(a)
+GROUP BY a
+HAVING a IS NOT NULL
+-- !query schema
+struct<col:int,cnt:bigint,ordered_cnt:bigint,a:int>
+-- !query output
+1	3	3	1
+2	3	3	2
+3	3	3	3