LuciferYang commented on code in PR #53468:
URL: https://github.com/apache/spark/pull/53468#discussion_r3836300949
##########
sql/core/src/test/resources/sql-tests/inputs/array.sql:
##########
@@ -206,3 +206,17 @@ select trim_array(array(1, 2, 3), 4);
-- trim_array with NULL array or NULL n returns NULL
select trim_array(CAST(null AS ARRAY<INT>), 1);
select trim_array(array(1, 2, 3), CAST(null AS INT));
+
+-- SPARK-54698: Confirm 0.0, -0.0, and NaN are handled appropriately for
complex types.
+select array_union(
Review Comment:
The new SQL cases cannot exercise the wrapper's -0.0/NaN normalization:
NormalizeFloatingNumbers runs in the FinishAnalysis batch, before
ConstantFolding can fold anything, and it wraps both literals and column data,
so the operator never sees raw bits via SQL execution; the 0.0 in the golden
output (instead of -0.0) is the tell. To test that defensive logic, use
Scala-side values in CollectionExpressionsSuite, e.g. a struct containing -0.0
and Double.longBitsToDouble(0x7ff0000000000001L); checkEvaluation bypasses the
optimizer, so the bits reach the operator as-is.
Two more gaps: collated strings nested inside structs/arrays have no
coverage anywhere, and RTRIM collations ('a' vs 'a ' under UTF8_BINARY_RTRIM)
are untested for these functions. Direct collated-string elements are already
covered by collations-basic.sql, worth confirming those still pass.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/ArraySetLikeBenchmark.scala:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.benchmark
+
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.functions._
+
+/**
+ * Benchmark for measuring perf of array set-like operations.
+ * To run this benchmark:
+ * {{{
+ * 1. without sbt:
+ * bin/spark-submit --class <this class> --jars <spark core test jar>
<sql core test jar>
+ * 2. build/sbt "sql/Test/runMain <this class>"
+ * 3. generate result:
+ * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this
class>"
+ * Results will be written to
"benchmarks/ArraySetLikeBenchmark-results.txt".
+ * }}}
+ */
+object ArraySetLikeBenchmark extends SqlBasedBenchmark {
+ private val N = 1000L
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ val range1 = lit((1 to 20000).map(x => Array(x, x)).toArray)
+ val range2 = lit((10001 to 30000).map(x => Array(x, x)).toArray)
+
+
+ val arrayDistinctBenchmark = new Benchmark("Array Distinct", N, output =
output)
+ arrayDistinctBenchmark.addCase("array_distinct") { _ =>
Review Comment:
Each benchmark group has a single case, so every row in the results files
shows Relative 1.0X and the committed numbers cannot show the win. Since the
old implementation is gone, a baseline case inside the file is not possible;
the simplest fix is to paste the before/after numbers already measured in the
review thread (array_union on 100k elements: 49269ms to 1779ms) into the
description's "Local benchmark results" section (or attach the results of the
Run benchmarks action), which is currently empty.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -4424,8 +4424,14 @@ trait ArraySetLike {
case _ => false
}
- @transient protected lazy val ordering: Ordering[Any] =
- TypeUtils.getInterpretedOrdering(et)
+ // If the element type supports proper equals, we use the values directly
for comparison,
+ // otherwise we use the generic comparable wrapper so all types support
hash-based operations
+ @transient protected lazy val keyGenerator: (Any => Any) =
Review Comment:
The description says "all array set-like expressions", but arrays_overlap is
unchanged: it still splits on typeWithProperEquals and falls back to the O(n²)
bruteForceEval for complex types, so two 10k-element struct arrays still mean
about 10^8 comparisons. Since the thread agrees on a follow-up, could you add a
scope note to the description (this PR covers the ArraySetLike implementors)
and open a JIRA for the follow-up, so "all" does not overstate the coverage?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -4539,62 +4545,32 @@ case class ArrayDistinct(child: Expression)
}
}
- override def nullSafeEval(array: Any): Any = {
- val data = array.asInstanceOf[ArrayData]
- doEvaluation(data)
- }
-
- @transient private lazy val doEvaluation = if
(TypeUtils.typeWithProperEquals(elementType)) {
- (array: ArrayData) =>
- val arrayBuffer = new scala.collection.mutable.ArrayBuffer[Any]
- val hs = new SQLOpenHashSet[Any]()
- val withNaNCheckFunc = SQLOpenHashSet.withNaNCheckFunc(elementType, hs,
- (value: Any) =>
- if (!hs.contains(value)) {
- if (arrayBuffer.size > ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH) {
- throw
QueryExecutionErrors.arrayFunctionWithElementsExceedLimitError(
- prettyName, arrayBuffer.size)
- }
- arrayBuffer += value
- hs.add(value)
- },
- (valueNaN: Any) => arrayBuffer += valueNaN)
- val withNullCheckFunc = SQLOpenHashSet.withNullCheckFunc(elementType, hs,
- (value: Any) => withNaNCheckFunc(value),
- () => arrayBuffer += null)
- var i = 0
- while (i < array.numElements()) {
- withNullCheckFunc(array, i)
- i += 1
- }
- new GenericArrayData(arrayBuffer)
- } else {
- (data: ArrayData) => {
- val array = data.toArray[AnyRef](elementType)
- val arrayBuffer = new scala.collection.mutable.ArrayBuffer[AnyRef]
- var alreadyStoredNull = false
- for (i <- array.indices) {
- if (array(i) != null) {
- var found = false
- var j = 0
- while (!found && j < arrayBuffer.size) {
- val va = arrayBuffer(j)
- found = (va != null) && ordering.equiv(va, array(i))
- j += 1
- }
- if (!found) {
- arrayBuffer += array(i)
- }
- } else {
- // De-duplicate the null values.
- if (!alreadyStoredNull) {
- arrayBuffer += array(i)
- alreadyStoredNull = true
+ override def nullSafeEval(input: Any): Any = {
+ val array = input.asInstanceOf[ArrayData]
+ val arrayBuffer = new scala.collection.mutable.ArrayBuffer[Any]
+ val hs = new SQLOpenHashSet[Any]()
+ val withNaNCheckFunc = SQLOpenHashSet.withNaNCheckFunc(elementType, hs,
+ (value: Any) => {
+ val key = keyGenerator(value)
+ if (!hs.contains(key)) {
+ if (arrayBuffer.size > ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH) {
Review Comment:
After unifying the paths, array_distinct on complex element types now
enforces the MAX_ROUNDED_ARRAY_LENGTH limit: an array with more than ~2^31
distinct elements used to return a result from the nested loop and now throws
arrayFunctionWithElementsExceedLimitError. The boundary is unreachable in
practice and matches what array_union and the codegen path already did, so no
code change needed; just worth a clause in the description, since it currently
claims "No user-facing change".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
