(spark) branch branch-4.1 updated: [SPARK-53635][SQL] Support Scala UDFs with input args of type Seq[Row]

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.1 by this push:
     new 36e056e0f9f0 [SPARK-53635][SQL] Support Scala UDFs with input args of type Seq[Row]
36e056e0f9f0 is described below

commit 36e056e0f9f0e4c9b2ae4cb41428b8c640d9bff5
Author: jameswillis <ja...@wherobots.com>
AuthorDate: Sat Nov 22 15:35:41 2025 -0800

    [SPARK-53635][SQL] Support Scala UDFs with input args of type Seq[Row]
    
    ### What changes were proposed in this pull request?
    The refactors of encoders in Spark 4 broke support for Scala UDFs that take in an argument of type `Seq[Row]`. This change restores the Spark 3.5 behavior of passing None for the encoder for arguments of that type into `SparkUserDefinedFunction`.
    
    ### Why are the changes needed?
    This change resolves issue SPARK-53635. This will help scala UDF users migrate more smoothly to Spark 4 and maintain backwards compatability for UDFs in Scala.
    
    ### Does this PR introduce _any_ user-facing change?
    Yes. It makes Spark 4 behave like Spark 3 when faced with a Scala UDF with an input argument of type `Seq[Row]`
    
    ### How was this patch tested?
    Existing and new unit tests.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    Generated-by: Copilot/GPT-5
    
    Closes #53027 from james-willis/james-willis/SPARK53635.
    
    Authored-by: jameswillis <ja...@wherobots.com>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit ea74e57b260d6549f4e05303f08feb01997ce93a)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../spark/sql/expressions/UserDefinedFunction.scala       | 15 ++++++++++++---
 .../org/apache/spark/sql/DataFrameComplexTypeSuite.scala  |  8 ++++++++
 2 files changed, 20 insertions(+), 3 deletions(-)

diff --git a/sql/api/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala b/sql/api/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala
index 6a22cbfaf351..d6da8ce17e83 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala
@@ -17,11 +17,11 @@
 
 package org.apache.spark.sql.expressions
 
-import scala.reflect.runtime.universe.TypeTag
+import scala.reflect.runtime.universe.{typeOf, TypeTag}
 import scala.util.Try
 
 import org.apache.spark.annotation.Stable
-import org.apache.spark.sql.{Column, Encoder}
+import org.apache.spark.sql.{Column, Encoder, Row}
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
 import org.apache.spark.sql.internal.{InvokeInlineUserDefinedFunction, UserDefinedFunctionLike}
@@ -130,8 +130,17 @@ object SparkUserDefinedFunction {
       returnTypeTag: TypeTag[_],
       inputTypeTags: TypeTag[_]*): SparkUserDefinedFunction = {
     val outputEncoder = ScalaReflection.encoderFor(returnTypeTag)
+    // Return None for any input that is a Seq[Row].
+    // This replicates the Spark 3 behavior of passing None as the encoder for such inputs.
+    val seqRowType = typeOf[Seq[_]]
+    val rowType = typeOf[Row]
     val inputEncoders = inputTypeTags.map { tag =>
-      Try(ScalaReflection.encoderFor(tag)).toOption
+      val tpe = tag.tpe
+      if (tpe <:< seqRowType && tpe.typeArgs.nonEmpty && tpe.typeArgs.head =:= rowType) {
+        None
+      } else {
+        Try(ScalaReflection.encoderFor(tag)).toOption
+      }
     }
     SparkUserDefinedFunction(
       f = function,
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 8024b579e5d0..984ff8fc51dd 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
@@ -71,6 +71,14 @@ class DataFrameComplexTypeSuite extends QueryTest with SharedSparkSession {
     df.select(map($"a", $"b").as("s")).select(f($"s".getItem("a"))).collect()
   }
 
+  test("SPARK-53635 array of structure to Seq[Row]") {
+    val f = udf((items: Seq[Row], num: Int) => items.length + num)
+    val df = spark.sql("SELECT array(struct(1 as a), struct(2 as a), struct(3 as a)) as items")
+    val result = df.select(f($"items", lit(3)).alias("countNum")).collect()
+    assert(result.length == 1)
+    assert(result(0).getAs[Int]("countNum") == 6)
+  }
+
   test("SPARK-12477 accessing null element in array field") {
     val df = sparkContext.parallelize(Seq((Seq("val1", null, "val2"),
       Seq(Some(1), None, Some(2))))).toDF("s", "i")


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org