You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2024/07/31 04:26:52 UTC
(spark) branch master updated: [SPARK-49003][SQL] Fix interpreted code path hashing to be collation aware
This is an automated email from the ASF dual-hosted git repository.
wenchen 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 5929412e65cd [SPARK-49003][SQL] Fix interpreted code path hashing to be collation aware
5929412e65cd is described below
commit 5929412e65cd82e72f74b21772d8c7ed04217399
Author: Marko <ma...@databricks.com>
AuthorDate: Wed Jul 31 12:26:35 2024 +0800
[SPARK-49003][SQL] Fix interpreted code path hashing to be collation aware
### What changes were proposed in this pull request?
Changed hash function to be collation aware. This change is just for interpreted code path. Codegen hashing was already collation aware.
### Why are the changes needed?
We were getting the wrong hash for collated strings because our hash function only used the binary representation of the string. It didn't take collation into account.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added tests to `HashExpressionsSuite.scala`
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #47502 from ilicmarkodb/ilicmarkodb/fix_string_hash.
Lead-authored-by: Marko <ma...@databricks.com>
Co-authored-by: Marko Ilic <ma...@databricks.com>
Co-authored-by: Marko Ilić <ma...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/expressions/hash.scala | 12 ++++++++--
.../expressions/HashExpressionsSuite.scala | 26 +++++++++++++++++++++-
.../spark/sql/CollationExpressionWalkerSuite.scala | 8 ++++---
.../spark/sql/CollationSQLExpressionsSuite.scala | 12 +++++-----
4 files changed, 46 insertions(+), 12 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/hash.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/hash.scala
index fa342f641509..3a667f370428 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/hash.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/hash.scala
@@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.expressions.Cast._
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.catalyst.util.{ArrayData, MapData}
+import org.apache.spark.sql.catalyst.util.{ArrayData, CollationFactory, MapData}
import org.apache.spark.sql.catalyst.util.DateTimeConstants._
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
@@ -565,7 +565,15 @@ abstract class InterpretedHashFunction {
case a: Array[Byte] =>
hashUnsafeBytes(a, Platform.BYTE_ARRAY_OFFSET, a.length, seed)
case s: UTF8String =>
- hashUnsafeBytes(s.getBaseObject, s.getBaseOffset, s.numBytes(), seed)
+ val st = dataType.asInstanceOf[StringType]
+ if (st.supportsBinaryEquality) {
+ hashUnsafeBytes(s.getBaseObject, s.getBaseOffset, s.numBytes(), seed)
+ } else {
+ val stringHash = CollationFactory
+ .fetchCollation(st.collationId)
+ .hashFunction.applyAsLong(s)
+ hashLong(stringHash, seed)
+ }
case array: ArrayData =>
val elementType = dataType match {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HashExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HashExpressionsSuite.scala
index 474c27c0de9a..6f3890cafd2a 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HashExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HashExpressionsSuite.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.{RandomDataGenerator, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.encoders.{ExamplePointUDT, ExpressionEncoder}
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateMutableProjection
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, DateTimeUtils, GenericArrayData, IntervalUtils}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, CollationFactory, DateTimeUtils, GenericArrayData, IntervalUtils}
import org.apache.spark.sql.types.{ArrayType, StructType, _}
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.ArrayImplicits._
@@ -620,6 +620,30 @@ class HashExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
checkHiveHashForDecimal("123456.123456789012345678901234567890", 38, 31, 1728235666)
}
+ for (collation <- Seq("UTF8_LCASE", "UNICODE_CI", "UTF8_BINARY")) {
+ test(s"hash check for collated $collation strings") {
+ val s1 = "aaa"
+ val s2 = "AAA"
+
+ val murmur3Hash1 = Murmur3Hash(Seq(Collate(Literal(s1), collation)), 42)
+ val murmur3Hash2 = Murmur3Hash(Seq(Collate(Literal(s2), collation)), 42)
+
+ // Interpreted hash values for s1 and s2
+ val interpretedHash1 = murmur3Hash1.eval()
+ val interpretedHash2 = murmur3Hash2.eval()
+
+ // Check that interpreted and codegen hashes are equal
+ checkEvaluation(murmur3Hash1, interpretedHash1)
+ checkEvaluation(murmur3Hash2, interpretedHash2)
+
+ if (CollationFactory.fetchCollation(collation).supportsBinaryEquality) {
+ assert(interpretedHash1 != interpretedHash2)
+ } else {
+ assert(interpretedHash1 == interpretedHash2)
+ }
+ }
+ }
+
test("SPARK-18207: Compute hash for a lot of expressions") {
def checkResult(schema: StructType, input: InternalRow): Unit = {
val exprs = schema.fields.zipWithIndex.map { case (f, i) =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CollationExpressionWalkerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CollationExpressionWalkerSuite.scala
index e4f4bb6e8557..b88c67e56c9b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CollationExpressionWalkerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CollationExpressionWalkerSuite.scala
@@ -618,6 +618,8 @@ class CollationExpressionWalkerSuite extends SparkFunSuite with SharedSparkSessi
"reflect",
"try_reflect",
"java_method",
+ "hash",
+ "xxhash64",
// need to skip as these are random functions
"rand",
"random",
@@ -630,11 +632,11 @@ class CollationExpressionWalkerSuite extends SparkFunSuite with SharedSparkSessi
)
for (funInfo <- funInfos.filter(f => !toSkip.contains(f.getName))) {
- for (m <- "> .*;".r.findAllIn(funInfo.getExamples)) {
+ for (query <- "> .*;".r.findAllIn(funInfo.getExamples).map(s => s.substring(2))) {
try {
- val resultUTF8 = sql(m.substring(2))
+ val resultUTF8 = sql(query)
withSQLConf(SqlApiConf.DEFAULT_COLLATION -> "UTF8_LCASE") {
- val resultUTF8Lcase = sql(m.substring(2))
+ val resultUTF8Lcase = sql(query)
assert(resultUTF8.collect() === resultUTF8Lcase.collect())
}
} catch {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CollationSQLExpressionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CollationSQLExpressionsSuite.scala
index 300f23c317ec..e31411ea212f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CollationSQLExpressionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CollationSQLExpressionsSuite.scala
@@ -171,9 +171,9 @@ class CollationSQLExpressionsSuite
val testCases = Seq(
Murmur3HashTestCase("Spark", "UTF8_BINARY", 228093765),
- Murmur3HashTestCase("Spark", "UTF8_LCASE", 228093765),
- Murmur3HashTestCase("SQL", "UNICODE", 17468742),
- Murmur3HashTestCase("SQL", "UNICODE_CI", 17468742)
+ Murmur3HashTestCase("Spark", "UTF8_LCASE", -1928694360),
+ Murmur3HashTestCase("SQL", "UNICODE", -1923567940),
+ Murmur3HashTestCase("SQL", "UNICODE_CI", 1029527950)
)
// Supported collations
@@ -199,9 +199,9 @@ class CollationSQLExpressionsSuite
val testCases = Seq(
XxHash64TestCase("Spark", "UTF8_BINARY", -4294468057691064905L),
- XxHash64TestCase("Spark", "UTF8_LCASE", -4294468057691064905L),
- XxHash64TestCase("SQL", "UNICODE", -2147923034195946097L),
- XxHash64TestCase("SQL", "UNICODE_CI", -2147923034195946097L)
+ XxHash64TestCase("Spark", "UTF8_LCASE", -3142112654825786434L),
+ XxHash64TestCase("SQL", "UNICODE", 5964849564945649886L),
+ XxHash64TestCase("SQL", "UNICODE_CI", 3732497619779520590L)
)
// Supported collations
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org