You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/09/21 21:46:07 UTC
(spark) branch branch-4.x updated: [SPARK-48701][SQL][PS][FOLLOWUP] Make pandas_mode/mode null-safe for collated strings nested in complex types
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon 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 cd5e723e3557 [SPARK-48701][SQL][PS][FOLLOWUP] Make pandas_mode/mode null-safe for collated strings nested in complex types
cd5e723e3557 is described below
commit cd5e723e355723907cbb4be4edb766933cccda6c
Author: Vinod KC <vi...@gmail.com>
AuthorDate: Tue Sep 22 06:45:15 2026 +0900
[SPARK-48701][SQL][PS][FOLLOWUP] Make pandas_mode/mode null-safe for collated strings nested in complex types
### What changes were proposed in this pull request?
Follow-up to SPARK-48701 (#58233). The mode / pandas_mode collation fold rewrites each key through ModeCollationAware.collationAwareTransform, which recurses into struct/array/map values. The original null guard only covered a fully-null top-level key, so a null nested inside a non-binary-collated struct field, array element, or map value reached CollationFactory.getCollationKey(null, ...) and threw an NPE.
This moves the null check to the top of collationAwareTransform so it applies at any nesting depth. It also simplifies the three recursive helpers to build their result in a single pass.
### Why are the changes needed?
mode and pandas_mode (pandas-on-Spark Series.mode() / DataFrame.mode()) throw an NPE on a non-binary-collated complex type containing a nested null:
```
CREATE TABLE t (c STRUCT<f: STRING COLLATE UTF8_LCASE>) USING parquet;
INSERT INTO t VALUES (named_struct('f', 'b')), (named_struct('f', 'B')),
(named_struct('f', CAST(null AS STRING)));
-- Series.mode() over c -> NullPointerException
```
### Does this PR introduce _any_ user-facing change?
No. It fixes an NPE
### How was this patch tested?
Added three tests to CollationAggregationSuite (nested null in struct, array, and map value)
### Was this patch authored or co-authored using generative AI tooling?
Yes. Generated-by: Claude Code.
Closes #58680 from vinodkc/spark48701-nested-null-mode.
Authored-by: Vinod KC <vi...@gmail.com>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
(cherry picked from commit 14d3ec836b8636ceef3a357dde7af864a435a554)
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
.../sql/catalyst/expressions/aggregate/Mode.scala | 54 +++++++++++++---------
.../sql/collation/CollationAggregationSuite.scala | 45 ++++++++++++++++++
2 files changed, 76 insertions(+), 23 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
index d119b8c276f8..34d5e34f9943 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
@@ -28,8 +28,9 @@ import org.apache.spark.sql.catalyst.types.PhysicalDataType
import org.apache.spark.sql.catalyst.util.{ArrayData, CollationFactory, GenericArrayData, MapData, UnsafeRowUtils}
import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, ArrayType, BooleanType, DataType, DoubleType, FloatType, MapType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, ArrayType, BooleanType, DataType, DoubleType, FloatType, MapType, StringType, StructType}
import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.collection.OpenHashMap
private[aggregate] object ModeKeyNormalizer {
@@ -70,21 +71,22 @@ private[aggregate] trait ModeCollationAware { self: Expression =>
childDataType: DataType): Option[AnyRef => _] = {
childDataType match {
case _ if UnsafeRowUtils.isBinaryStable(child.dataType) => None
- // A null key is kept as its own group: PandasMode may store one when `ignoreNA`
- // is false, and passing null to collationAwareTransform would throw. Mode never
- // stores a null key, so its behavior is unchanged.
- case _ => Some((key: AnyRef) =>
- if (key == null) null else collationAwareTransform(key, childDataType))
+ case _ => Some(collationAwareTransform(_, childDataType))
}
}
determineBufferingFunction(childDataType).map(groupAndReduceBuffer).getOrElse(buffer)
}
protected[sql] def collationAwareTransform(data: AnyRef, dataType: DataType): AnyRef = {
+ // A null carries no collation, so it forms its own group. Guarding here (rather than only
+ // at the top-level key) also covers nulls nested in complex types: a null struct field,
+ // array element, or map value must not reach the string/collation-key path below, which
+ // throws on null. PandasMode stores a top-level null key when `ignoreNA` is false; Mode
+ // never stores one, so its behavior is unchanged.
+ if (data == null) return null
dataType match {
case _ if UnsafeRowUtils.isBinaryStable(dataType) => data
- case st: StructType =>
- processStructTypeWithBuffer(data.asInstanceOf[InternalRow].toSeq(st).zip(st.fields))
+ case st: StructType => processStructTypeWithBuffer(data.asInstanceOf[InternalRow], st)
case at: ArrayType => processArrayTypeWithBuffer(at, data.asInstanceOf[ArrayData])
case mt: MapType => processMapTypeWithBuffer(mt, data.asInstanceOf[MapData])
case st: StringType =>
@@ -100,26 +102,32 @@ private[aggregate] trait ModeCollationAware { self: Expression =>
}
}
- private def processStructTypeWithBuffer(
- tuples: Seq[(Any, StructField)]): Seq[Any] = {
- tuples.map(t => collationAwareTransform(t._1.asInstanceOf[AnyRef], t._2.dataType))
+ // The results below are used only as grouping keys, so they just need consistent
+ // element-wise equals/hashCode; a Scala immutable Seq/Map provides that. Each builds its
+ // result in a single pass (one array/map allocation) rather than chaining map/zip/toMap
+ // over per-element intermediate collections.
+
+ private def processStructTypeWithBuffer(row: InternalRow, st: StructType): Seq[Any] = {
+ val fields = st.fields
+ Array.tabulate(fields.length) { i =>
+ val fieldType = fields(i).dataType
+ collationAwareTransform(row.get(i, fieldType).asInstanceOf[AnyRef], fieldType)
+ }.toImmutableArraySeq
}
- private def processArrayTypeWithBuffer(
- a: ArrayType,
- data: ArrayData): Seq[Any] = {
- (0 until data.numElements()).map(i =>
- collationAwareTransform(data.get(i, a.elementType), a.elementType))
+ private def processArrayTypeWithBuffer(a: ArrayType, data: ArrayData): Seq[Any] = {
+ Array.tabulate(data.numElements()) { i =>
+ collationAwareTransform(data.get(i, a.elementType), a.elementType)
+ }.toImmutableArraySeq
}
private def processMapTypeWithBuffer(mt: MapType, data: MapData): Map[Any, Any] = {
- val transformedKeys = (0 until data.numElements()).map { i =>
- collationAwareTransform(data.keyArray().get(i, mt.keyType), mt.keyType)
- }
- val transformedValues = (0 until data.numElements()).map { i =>
- collationAwareTransform(data.valueArray().get(i, mt.valueType), mt.valueType)
- }
- transformedKeys.zip(transformedValues).toMap
+ val keys = data.keyArray()
+ val values = data.valueArray()
+ (0 until data.numElements()).iterator.map { i =>
+ collationAwareTransform(keys.get(i, mt.keyType), mt.keyType) ->
+ collationAwareTransform(values.get(i, mt.valueType), mt.valueType)
+ }.toMap
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationAggregationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationAggregationSuite.scala
index b3e9b482f456..6f8a1a429f9a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationAggregationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationAggregationSuite.scala
@@ -616,6 +616,51 @@ class CollationAggregationSuite
}
}
+ // A null nested inside a collated complex type (struct field, array element, map value)
+ // must fold as its own group instead of reaching the collation-key path, which throws on
+ // null. These guard that null-safety while also confirming the collation fold still applies
+ // to the non-null siblings ('b'/'B' collapse to one group under UTF8_LCASE).
+
+ test("SPARK-48701: pandas_mode is null-safe for a null collated string nested in struct") {
+ withTable("t") {
+ sql("CREATE TABLE t (c STRUCT<f: STRING COLLATE UTF8_LCASE>) USING parquet")
+ sql(
+ """INSERT INTO t VALUES (named_struct('f', 'b')), (named_struct('f', 'B')),
+ | (named_struct('f', CAST(null AS STRING)))""".stripMargin)
+ val modes = pandasMode(spark.table("t").repartition(4), "c", ignoreNA = true)
+ .map(_.asInstanceOf[Row])
+ // {b, B} fold to one group of 2, outvoting the null-field struct (1).
+ assert(modes.length == 1)
+ assert(modes.head.getString(0).toLowerCase(Locale.ROOT) == "b")
+ }
+ }
+
+ test("SPARK-48701: pandas_mode is null-safe for a null collated string nested in array") {
+ withTable("t") {
+ sql("CREATE TABLE t (c ARRAY<STRING COLLATE UTF8_LCASE>) USING parquet")
+ sql(
+ """INSERT INTO t VALUES (array('b')), (array('B')),
+ | (array(CAST(null AS STRING)))""".stripMargin)
+ val modes = pandasMode(spark.table("t").repartition(4), "c", ignoreNA = true)
+ .map(_.asInstanceOf[collection.Seq[String]])
+ // [b] and [B] fold to one group of 2, outvoting [null] (1).
+ assert(modes.length == 1)
+ assert(modes.head.head.toLowerCase(Locale.ROOT) == "b")
+ }
+ }
+
+ test("SPARK-48701: pandas_mode is null-safe for a null collated string nested in map value") {
+ withTable("t") {
+ sql("CREATE TABLE t (c MAP<STRING, STRING COLLATE UTF8_LCASE>) USING parquet")
+ sql(
+ """INSERT INTO t VALUES (map('k', 'b')), (map('k', 'B')),
+ | (map('k', CAST(null AS STRING)))""".stripMargin)
+ val modes = pandasMode(spark.table("t").repartition(4), "c", ignoreNA = true)
+ // {k->b, k->B} fold to one group of 2, outvoting {k->null} (1).
+ assert(modes.length == 1)
+ }
+ }
+
// `mode` (the public aggregate) is already collation-aware; these tests guard that
// behavior, which currently has no coverage (the original tests were removed with
// CollationSQLExpressionsSuite by SPARK-51067). Unlike pandas_mode, `mode` returns a
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org