Skip to content

Commit be37962

Browse files
lavanv11Yicong-Huang
authored andcommitted
[SPARK-59249][SQL][FOLLOWUP] Include SQL type in Python UDT identity
### What changes were proposed in this pull request? This is a follow-up to [SPARK-59249](https://issues.apache.org/jira/browse/SPARK-59249) and [`this commit`](ebb0bd2), which routed grouped-key ordering through `InternalRowComparableWrapper`'s shared ordering cache. This PR includes both `pyUDT` and `sqlType` in `PythonUserDefinedType.equals` and `hashCode`. `serializedPyClass` remains excluded from identity, and `acceptsType` retains its existing compatibility semantics. The PR also adds regression coverage for the shared `InternalRowComparableWrapper` ordering cache in both lookup orders, plus focused equality and hash-code tests. ### Why are the changes needed? `InternalRowComparableWrapper` caches generated orderings by data type. Python UDT equality previously considered only the Python class name, so UDTs with the same class name but different underlying SQL types could share a cache entry. For example, binary and `UTF8_LCASE` string-backed UDTs could reuse the first generated comparator, producing lookup-order-dependent collation semantics. ### How was this patch tested? Added regression tests to `InternalRowComparableWrapperSuite`, covering binary-first and UTF8_LCASE-first cache lookup orders, and to `DataTypeSuite`, covering equality, hash codes, differing SQL types, and exclusion of serialized Python class data. Ran: ``` build/sbt \ 'catalyst/testOnly org.apache.spark.sql.catalyst.util.InternalRowComparableWrapperSuite' \ 'catalyst/testOnly org.apache.spark.sql.types.DataTypeSuite' ``` `InternalRowComparableWrapperSuite`: 7 passed; `DataTypeSuite`: 352 passed. ### Was this patch authored or co-authored using generative AI tooling? Yes co-authored-by: OpenAI Codex GPT-5.6 sol. Closes #58799 from lavanv11/sc-244639-python-udt-ordering. Authored-by: Lavan Vivekanandasarma <lavan.vivek@databricks.com> Signed-off-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com>
1 parent 169da93 commit be37962

3 files changed

Lines changed: 46 additions & 2 deletions

File tree

‎sql/api/src/main/scala/org/apache/spark/sql/types/UserDefinedType.scala‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -154,9 +154,9 @@ private[sql] class PythonUserDefinedType(
154154
}
155155

156156
override def equals(other: Any): Boolean = other match {
157-
case that: PythonUserDefinedType => pyUDT == that.pyUDT
157+
case that: PythonUserDefinedType => pyUDT == that.pyUDT && sqlType == that.sqlType
158158
case _ => false
159159
}
160160

161-
override def hashCode(): Int = Objects.hashCode(pyUDT)
161+
override def hashCode(): Int = Objects.hash(pyUDT, sqlType)
162162
}

‎sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapperSuite.scala‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,40 @@
1717

1818
package org.apache.spark.sql.catalyst.util
1919

20+
import java.util.UUID
21+
2022
import org.apache.spark.SparkFunSuite
2123
import org.apache.spark.sql.catalyst.InternalRow
2224
import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
2325
import org.apache.spark.sql.types._
26+
import org.apache.spark.unsafe.types.UTF8String
2427

2528
class InternalRowComparableWrapperSuite extends SparkFunSuite {
2629

2730
private val structA = new StructType().add("a", IntegerType)
2831
private val structB = new StructType().add("b", IntegerType)
2932

33+
private def checkPythonUDTOrderingCache(binaryFirst: Boolean): Unit = {
34+
// The cache lasts for the JVM, so use a name that no earlier test run could have populated.
35+
val pyUDT = UUID.randomUUID().toString
36+
val lcaseType = StringType("UTF8_LCASE")
37+
def ordering(sqlType: DataType) = KeyedPartitioning.groupedKeyRowOrdering(
38+
Seq(new PythonUserDefinedType(sqlType, pyUDT, "serializedPyClass")))
39+
40+
val (binaryOrdering, lcaseOrdering) = if (binaryFirst) {
41+
val binaryOrdering = ordering(StringType)
42+
(binaryOrdering, ordering(lcaseType))
43+
} else {
44+
val lcaseOrdering = ordering(lcaseType)
45+
(ordering(StringType), lcaseOrdering)
46+
}
47+
val uppercase = InternalRow(UTF8String.fromString("A"))
48+
val lowercase = InternalRow(UTF8String.fromString("a"))
49+
50+
assert(binaryOrdering.compare(uppercase, lowercase) !== 0)
51+
assert(lcaseOrdering.compare(uppercase, lowercase) === 0)
52+
}
53+
3054
test("SPARK-59187: comparableTypes erases the naming and nothing else") {
3155
// The erasure has to answer exactly what `DataType.equalsStructurally` answers with
3256
// `ignoreNullability`, since that is the question two rows are asking of each other. Anything
@@ -95,6 +119,14 @@ class InternalRowComparableWrapperSuite extends SparkFunSuite {
95119
assert(KeyedPartitioning.groupedKeyRowOrdering(Seq(IntegerType, LongType)) eq wrapper.ordering)
96120
}
97121

122+
test("Python UDT ordering cache distinguishes SQL types with binary first") {
123+
checkPythonUDTOrderingCache(binaryFirst = true)
124+
}
125+
126+
test("Python UDT ordering cache distinguishes SQL types with UTF8_LCASE first") {
127+
checkPythonUDTOrderingCache(binaryFirst = false)
128+
}
129+
98130
test("SPARK-59187: a factory answers for the types it settled on") {
99131
// A caller that reports a type list beside the rows a factory built takes it from here, so the
100132
// two cannot answer differently.

‎sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeSuite.scala‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,18 @@ class DataTypeSuite extends SparkFunSuite with SQLHelper {
3838
private val UNICODE_COLLATION_ID = CollationFactory.collationNameToId("UNICODE")
3939
private val UTF8_LCASE_COLLATION_ID = CollationFactory.collationNameToId("UTF8_LCASE")
4040

41+
test("PythonUserDefinedType equality includes the SQL type") {
42+
val pyUDT = "org.apache.spark.sql.types.ExamplePythonUDT"
43+
val first = new PythonUserDefinedType(StringType, pyUDT, "firstSerializedClass")
44+
val same = new PythonUserDefinedType(StringType, pyUDT, "secondSerializedClass")
45+
val differentSqlType =
46+
new PythonUserDefinedType(StringType("UTF8_LCASE"), pyUDT, "firstSerializedClass")
47+
48+
assert(first === same)
49+
assert(first.hashCode === same.hashCode)
50+
assert(first !== differentSqlType)
51+
}
52+
4153
test("construct an ArrayType") {
4254
val array = ArrayType(StringType)
4355

0 commit comments

Comments
 (0)