This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 9dede1fa81fb [SPARK-57505][SQL] Do not wrap window functions in an
`AggregateExpression` when converting an expression-backed `Column`
9dede1fa81fb is described below
commit 9dede1fa81fb202b95764483dfe2dd118330ce87
Author: Peter Toth <[email protected]>
AuthorDate: Thu Jun 18 13:07:16 2026 +0200
[SPARK-57505][SQL] Do not wrap window functions in an `AggregateExpression`
when converting an expression-backed `Column`
### What changes were proposed in this pull request?
`ColumnNodeToExpressionConverter` (sql/core
.../classic/columnNodeSupport.scala) converts an expression-backed `Column` (an
`ExpressionColumnNode`) into a Catalyst `Expression` and, if the wrapped
expression is an `AggregateFunction`, wraps it in an `AggregateExpression` via
`toAggregateExpression()`. Because `AggregateWindowFunction extends
DeclarativeAggregate (which extends AggregateFunction) with WindowFunction`, a
window function also matched this branch and got wrapped, producin [...]
### Why are the changes needed?
Wrapping a window function in an `AggregateExpression` makes analysis fail:
`CheckAnalysis` sees an `AggregateExpression` whose child is a `WindowFunction`
but which is not itself a `WindowExpression`, and throws
`WINDOW_FUNCTION_WITHOUT_OVER_CLAUSE`. As a result, a user-defined
`AggregateWindowFunction` (or any built-in window function expression) wrapped
into a `Column` through the `ClassicConversions` / `ColumnConversions`
DeveloperApi and used with `over(...)` could not be analyze [...]
### Does this PR introduce _any_ user-facing change?
Yes. Before this change, wrapping a window function expression into a
`Column` and calling `.over(window)` failed analysis with
`WINDOW_FUNCTION_WITHOUT_OVER_CLAUSE`; now it analyzes and executes correctly,
matching the behavior of the equivalent by-name window function. This is a fix
within the unreleased master branch relative to the existing Column/Catalyst
conversion behavior.
### How was this patch tested?
Added two regression tests to `DataFrameWindowFunctionsSuite`: one wraps
the built-in `RowNumber()` expression into a `Column` and checks it matches
by-name `row_number()`, and one defines a minimal custom
`AggregateWindowFunction` (`NonNullRunningCount`) and checks it produces
correct results when used with `over(...)`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #56570 from
peter-toth/SPARK-57505-fix-custom-aggregatewindowfunction-analysis.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit 201da65b7add4be81bb466183a85a88e61edaf30)
Signed-off-by: Peter Toth <[email protected]>
---
.../spark/sql/classic/columnNodeSupport.scala | 9 ++++-
.../spark/sql/DataFrameWindowFunctionsSuite.scala | 47 +++++++++++++++++++++-
2 files changed, 53 insertions(+), 3 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/columnNodeSupport.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/columnNodeSupport.scala
index 5766535ac5da..4fda00b1c79a 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/columnNodeSupport.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/columnNodeSupport.scala
@@ -22,7 +22,7 @@ import org.apache.spark.SparkException
import org.apache.spark.sql.Column
import org.apache.spark.sql.catalyst.{analysis, expressions,
CatalystTypeConverters}
import org.apache.spark.sql.catalyst.analysis.{MultiAlias, UnresolvedAlias}
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, Generator, NamedExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, Generator, NamedExpression, Unevaluable, WindowFunction}
import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
AggregateFunction}
import org.apache.spark.sql.catalyst.parser.{ParserInterface, ParserUtils}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -185,7 +185,12 @@ private[sql] trait ColumnNodeToExpressionConverter extends
(ColumnNode => Expres
case ColumnNodeExpression(node) => apply(node)
}
transformed match {
- case f: AggregateFunction => f.toAggregateExpression()
+ // A window function (e.g. an AggregateWindowFunction) is also an
AggregateFunction, but
+ // it must not be wrapped in an AggregateExpression: it is used
directly as the child of
+ // a WindowExpression. Wrapping it would later fail analysis with
+ // WINDOW_FUNCTION_WITHOUT_OVER_CLAUSE.
+ case f: AggregateFunction if !f.isInstanceOf[WindowFunction] =>
+ f.toAggregateExpression()
case _ => transformed
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
index 01e72daead44..552e2b2e2749 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
@@ -20,10 +20,11 @@ package org.apache.spark.sql
import org.scalatest.matchers.must.Matchers.the
import org.apache.spark.TestUtils.{assertNotSpilled, assertSpilled}
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, Lag, Literal, NonFoldableLiteral}
+import org.apache.spark.sql.catalyst.expressions.{Add,
AggregateWindowFunction, AttributeReference, Expression, If, IsNotNull, Lag,
Literal, NonFoldableLiteral, RowNumber}
import org.apache.spark.sql.catalyst.optimizer.TransposeWindow
import org.apache.spark.sql.catalyst.plans.logical.{Window => LogicalWindow}
import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS, Exchange,
ShuffleExchangeExec}
import org.apache.spark.sql.execution.window.WindowExec
@@ -879,6 +880,28 @@ class DataFrameWindowFunctionsSuite extends QueryTest
)
}
+ test("SPARK-57505: a window function expression wrapped into a Column works
with over()") {
+ val df = Seq((1, "a"), (2, "a"), (3, "b")).toDF("value", "key")
+ val window = Window.partitionBy($"key").orderBy($"value")
+ // Wrapping a catalyst window function expression directly with
Column(expr) used to box the
+ // AggregateWindowFunction (RowNumber is one) in an AggregateExpression,
which then failed
+ // analysis with WINDOW_FUNCTION_WITHOUT_OVER_CLAUSE. It must now behave
like by-name
+ // row_number().
+ checkAnswer(
+ df.select($"value", Column(RowNumber()).over(window).as("rn")),
+ Seq(Row(1, 1), Row(2, 2), Row(3, 1)))
+ }
+
+ test("SPARK-57505: a custom AggregateWindowFunction wrapped into a Column
works with over()") {
+ val df = Seq((1, "a"), (2, "a"), (3, "b")).toDF("value", "key")
+ val window = Window.partitionBy($"key").orderBy($"value")
+ // Mirrors plugging in a user-defined AggregateWindowFunction through the
Column API:
+ // Column(MyWindowFunction(inputColumn.expr)).over(window)
+ checkAnswer(
+ df.select($"value",
Column(NonNullRunningCount($"value".expr)).over(window).as("cnt")),
+ Seq(Row(1, 1), Row(2, 2), Row(3, 1)))
+ }
+
test("SPARK-12989 ExtractWindowExpressions treats alias as regular
attribute") {
val src = Seq((0, 3, 5)).toDF("a", "b", "c")
.withColumn("Data", struct("a", "b"))
@@ -1620,3 +1643,25 @@ class DataFrameWindowFunctionsSuite extends QueryTest
}
}
}
+
+/**
+ * A minimal user-defined window function, it counts the non-null values of
`child` from the start
+ * of the window frame up to and including the current row.
+ */
+case class NonNullRunningCount(child: Expression)
+ extends AggregateWindowFunction with UnaryLike[Expression] {
+
+ private lazy val count = AttributeReference("count", IntegerType, nullable =
false)()
+
+ override lazy val aggBufferAttributes: Seq[AttributeReference] = count :: Nil
+ override lazy val initialValues: Seq[Expression] = Literal(0) :: Nil
+ override lazy val updateExpressions: Seq[Expression] =
+ If(IsNotNull(child), Add(count, Literal(1)), count) :: Nil
+ override lazy val evaluateExpression: Expression = count
+
+ override def nullable: Boolean = false
+ override def prettyName: String = "non_null_running_count"
+
+ override protected def withNewChildInternal(newChild: Expression):
NonNullRunningCount =
+ copy(child = newChild)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
