(spark) branch master updated: [SPARK-59717][SQL] Fix MatchError when grouping by a UDT column backed by a primitive type

[we...@apache.org]

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

cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 664356ae0707 [SPARK-59717][SQL] Fix MatchError when grouping by a UDT column backed by a primitive type
664356ae0707 is described below

commit 664356ae070720ec3d0e2da18328c4df2f81d2b0
Author: jiwen624 <ji...@gmail.com>
AuthorDate: Mon Oct 12 14:20:04 2026 +0000

    [SPARK-59717][SQL] Fix MatchError when grouping by a UDT column backed by a primitive type
    
    ### What changes were proposed in this pull request?
    Exclude UserDefinedType grouping keys from the fast hash map in HashAggregateExec.checkIfFastHashMapSupported, so these aggregations use the regular UnsafeRow-based hash map.
    
    ### Why are the changes needed?
    Grouping by a UDT backed by a primitive type fails with an internal error:
    ```scala
    spark.conf.set("spark.sql.codegen.aggregate.map.twolevel.enabled", "true")
    Seq((Year.of(2024), 1), (Year.of(2025), 2)).toDF("y", "v").groupBy("y").sum("v").collect()
    // SparkException: [INTERNAL_ERROR] ... scala.MatchError: YearUDT
    ```
    CodeGenerator.isPrimitiveType unwraps the UDT, so the fast hash map is turned on. But HashMapGenerator.genComputeHash has no UDT case, so code generation throws before codegen fallback can apply. No query loses the fast map because of this change: every UDT key that reached it failed before.
    
    **Note**: This excludes UDT grouping keys from the fast hash map. No working path is lost: every UDT key that reached it failed code generation, and these queries now use the regular hash map. Adding a UDT case to genComputeHash instead is unsafe for UDTs backed by float/double: NormalizeFloatingNumbers skips UDT keys, and the two maps then treat 0.0/-0.0 and NaN differently, so distinct can merge 0.0 with -0.0 or return NaN twice. Fast-map support for other UDTs is a possible perform [...]
    
    ### Does this PR introduce _any_ user-facing change?
    Yes, the query above now returns the correct result instead of `INTERNAL_ERROR`.
    
    ### How was this patch tested?
    Added UT case.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    Yes.
    
    Closes #58975 from jiwen624/SPARK-udt-fast-hashmap.
    
    Authored-by: jiwen624 <ji...@gmail.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../sql/execution/aggregate/HashAggregateExec.scala      |  3 ++-
 .../org/apache/spark/sql/UserDefinedTypeSuite.scala      | 16 ++++++++++++++++
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala
index 7844ca6bced4..d33eee47a944 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.execution.vectorized.MutableColumnarRow
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{CalendarIntervalType, DecimalType, StringType}
+import org.apache.spark.sql.types.{CalendarIntervalType, DecimalType, StringType, UserDefinedType}
 import org.apache.spark.unsafe.KVIterator
 import org.apache.spark.util.ArrayImplicits._
 import org.apache.spark.util.Utils
@@ -507,6 +507,7 @@ case class HashAggregateExec(
   private def checkIfFastHashMapSupported(): Boolean = {
     val isSupported =
       groupingExpressions.forall(e => UnsafeRowUtils.isBinaryStable(e.dataType)) &&
+      groupingKeySchema.forall(!_.dataType.isInstanceOf[UserDefinedType[_]]) &&
       (groupingKeySchema ++ bufferSchema).forall(f => CodeGenerator.isPrimitiveType(f.dataType) ||
         f.dataType.isInstanceOf[DecimalType] || f.dataType.isInstanceOf[StringType] ||
         f.dataType.isInstanceOf[CalendarIntervalType])
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala
index 04be47523c58..ca0800aca712 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala
@@ -287,6 +287,22 @@ class UserDefinedTypeSuite extends SharedSparkSession with ParquetTest
     assert(result.toSet === Set(FooWithDate(year, "FooFoo", 3), FooWithDate(year, "Foo", 1)))
   }
 
+  test("SPARK-59717: group by a UserDefinedType column backed by a primitive type") {
+    UDTRegistration.register(classOf[Year].getName, classOf[YearUDT].getName)
+
+    val y1 = Year.of(2024)
+    val y2 = Year.of(2025)
+    val df = Seq((y1, 1), (y2, 2), (y1, 3)).toDF("y", "v")
+    Seq("false", "true").foreach { vectorized =>
+      withSQLConf(
+        SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> "true",
+        SQLConf.ENABLE_VECTORIZED_HASH_MAP.key -> vectorized) {
+        checkAnswer(df.groupBy("y").agg(sum("v")), Row(y1, 4L) :: Row(y2, 2L) :: Nil)
+        checkAnswer(df.select("y").distinct(), Row(y1) :: Row(y2) :: Nil)
+      }
+    }
+  }
+
   test("Test unwrap_udt function") {
     val unwrappedFeatures = pointsRDD.select(unwrap_udt(col("features")))
       .rdd.map { (row: Row) => row.getAs[Seq[Double]](0).toArray }


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