This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new f4f3e79b8af9 [SPARK-37019][SQL][FOLLOWUP] Resolve nested higher-order
function arguments first
f4f3e79b8af9 is described below
commit f4f3e79b8af9349adc358305755dc41066b388c8
Author: Chao Sun <[email protected]>
AuthorDate: Fri Jun 19 16:53:34 2026 -0700
[SPARK-37019][SQL][FOLLOWUP] Resolve nested higher-order function arguments
first
### What changes were proposed in this pull request?
Resolve each higher-order function's argument expressions before checking
their data types and binding its lambda functions.
The analyzer now follows this order:
1. Resolve the argument expressions using the current outer lambda scope.
2. Rebuild the higher-order function with those resolved arguments.
3. If the arguments are ready and valid, bind the lambda functions
immediately.
4. Otherwise, resolve only the function expressions and defer binding.
This matches the established sequence in the single-pass
`HigherOrderFunctionResolver`, so both analyzer paths now resolve arguments
before binding lambdas.
This is intentionally narrow. It does not change `ArrayAggregate`
accumulator types, casts, code generation, or runtime execution.
The PR also adds a focused regression test for the nested `transform` /
`filter` / `aggregate` expression that exposed the bug.
### Why are the changes needed?
`ResolveLambdaVariables` previously bound a higher-order function only when
its arguments were already resolved at the start of the visit. If nested
argument expressions became resolved during that visit, Spark still walked and
rebuilt the remaining expression tree without binding the current lambda
functions.
For complex nested types, that ordering could inspect a field extraction
whose lambda variable was still unresolved and fail analysis with:
```
Invalid call to dataType on unresolved object
```
In short:
```
Before: check readiness -> resolve nested arguments -> wait for another
analyzer pass
After: resolve nested arguments -> check readiness -> bind lambdas in the
same pass
```
### Does this PR introduce _any_ user-facing change?
Yes. Valid queries with nested higher-order functions that previously
failed during analysis can now be analyzed and executed.
There is no public API, configuration, or intended runtime behavior change
for queries that already worked.
### How was this patch tested?
- Added `ArrayAggregate resolves nested lambda arguments before inspecting
their types` to reproduce the production-shaped failure and verify the result.
- `ResolveLambdaVariablesSuite`: 6 tests passed.
- `DataFrameComplexTypeSuite`: 18 tests passed.
- Catalyst and SQL test Scalastyle: 0 errors and 0 warnings.
- `git diff --check` passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
Closes #56507 from sunchao/dev/chao/codex/hof-gate-analyzer-behavior-oss.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
(cherry picked from commit 3fa2bea78b1e48b439efd412990798af91d9b94b)
Signed-off-by: Chao Sun <[email protected]>
---
.../catalyst/analysis/higherOrderFunctions.scala | 25 ++++++++++--
.../spark/sql/DataFrameComplexTypeSuite.scala | 45 ++++++++++++++++++++--
2 files changed, 64 insertions(+), 6 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/higherOrderFunctions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/higherOrderFunctions.scala
index 9c94d045ae86..107345dda379 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/higherOrderFunctions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/higherOrderFunctions.scala
@@ -52,9 +52,28 @@ object ResolveLambdaVariables extends Rule[LogicalPlan] {
private def resolve(e: Expression, parentLambdaMap: LambdaVariableMap):
Expression = e match {
case _ if e.resolved => e
- case h: HigherOrderFunction if h.argumentsResolved &&
h.checkArgumentDataTypes().isSuccess =>
- SubqueryExpressionInLambdaOrHigherOrderFunctionValidator(e)
- h.bind(LambdaBinder(_, _)).mapChildren(resolve(_, parentLambdaMap))
+ case h: HigherOrderFunction =>
+ // An argument can contain lambda variables from an outer scope. Resolve
those variables
+ // before inspecting the argument type, which may otherwise access an
unresolved extractor.
+ // Keep this higher-order function's own lambdas unbound until its
argument types are ready.
+ val resolvedArguments = h.arguments.map(resolve(_, parentLambdaMap))
+ // HigherOrderFunction children are ordered as arguments followed by
functions.
+ val resolvedHigherOrderFunction = h
+ .withNewChildren(resolvedArguments ++ h.functions)
+ .asInstanceOf[HigherOrderFunction]
+ if (resolvedHigherOrderFunction.argumentsResolved &&
+ resolvedHigherOrderFunction.checkArgumentDataTypes().isSuccess) {
+
SubqueryExpressionInLambdaOrHigherOrderFunctionValidator(resolvedHigherOrderFunction)
+ // Arguments are already resolved; this walk resolves the newly bound
lambda bodies.
+ resolvedHigherOrderFunction.bind(LambdaBinder(_, _))
+ .mapChildren(resolve(_, parentLambdaMap))
+ } else {
+ // The arguments were already visited above. Resolve only the
functions here so nested
+ // arguments are not traversed twice in the same analyzer pass.
+ val resolvedFunctions = resolvedHigherOrderFunction.functions
+ .map(resolve(_, parentLambdaMap))
+ resolvedHigherOrderFunction.withNewChildren(resolvedArguments ++
resolvedFunctions)
+ }
case l: LambdaFunction if !l.bound =>
SubqueryExpressionInLambdaOrHigherOrderFunctionValidator(e)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameComplexTypeSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameComplexTypeSuite.scala
index 8dda985030af..f65b5a61b6a2 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameComplexTypeSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameComplexTypeSuite.scala
@@ -30,7 +30,9 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{ArrayType, BooleanType, Decimal,
DoubleType, IntegerType, MapType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{
+ ArrayType, BooleanType, Decimal, DoubleType, IntegerType, LongType, MapType,
StringType,
+ StructField, StructType}
import org.apache.spark.unsafe.types.CalendarInterval
/**
@@ -235,6 +237,45 @@ class DataFrameComplexTypeSuite extends SharedSparkSession
{
}
}
+ test("ArrayAggregate resolves nested lambda arguments before inspecting
their types") {
+ val positionType = StructType(Seq(
+ StructField("turn_idx", LongType, nullable = false),
+ StructField("turn_kind", StringType, nullable = false),
+ StructField("start_message_index", LongType, nullable = false),
+ StructField("previous_start_message_index", LongType, nullable = true)))
+ val positionsType = ArrayType(positionType, containsNull = false)
+ val messages = array(struct(
+ lit("user").as("turn_kind"),
+ lit(0L).as("message_index")))
+ val candidates = filter(
+ transform(messages, message => struct(
+ message.getField("turn_kind").as("turn_kind"),
+ message.getField("message_index").as("message_index"))),
+ candidate => candidate.getField("turn_kind").isNotNull)
+ // The aggregate first sees `candidates` with its transform/filter lambdas
still unbound. Those
+ // argument expressions must resolve before the aggregate inspects their
complex data types.
+ val lastPosition = aggregate(
+ candidates,
+ array().cast(positionsType),
+ (positions, candidate) => {
+ val previousPosition = try_element_at(positions, lit(-1))
+ when(
+
previousPosition.getField("turn_kind").eqNullSafe(candidate.getField("turn_kind")),
+ positions).otherwise(
+ concat(positions, array(struct(
+ (size(positions).cast(LongType) + 1L).as("turn_idx"),
+ candidate.getField("turn_kind").as("turn_kind"),
+ candidate.getField("message_index").as("start_message_index"),
+ previousPosition.getField("start_message_index")
+ .as("previous_start_message_index")))))
+ },
+ positions => try_element_at(positions, lit(-1)))
+
+ checkAnswer(
+ spark.range(1).select(lastPosition),
+ Row(Row(1L, "user", 0L, null)))
+ }
+
test("SPARK-31552: array encoder with different types") {
// primitives
val booleans = Array(true, false)
@@ -394,5 +435,3 @@ extends DefinedByConstructorParams
case class S100_5(
s1: S100 = new S100(), s2: S100 = new S100(), s3: S100 = new S100(),
s4: S100 = new S100(), s5: S100 = new S100())
-
-
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
