You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2026/09/15 14:10:45 UTC
[PR] [SPARK-54918][SQL][FOLLOWUP] Handle floating-point normalization in array set operations [spark]
cloud-fan opened a new pull request, #58818:
URL: https://github.com/apache/spark/pull/58818
### What changes were proposed in this pull request?
This is a follow-up to https://github.com/apache/spark/pull/53695.
This PR keeps the floating-point semantics introduced there, but moves their implementation from
a logical-plan rewrite into the array set expressions themselves:
- Canonicalize primitive `Float` and `Double` values inside the interpreted and generated
hash-set loops.
- Canonicalize complex element values only when an element is emitted into a result array.
- Handle primitive floating-point values locally in `arrays_overlap`.
- Remove the array-expression rewrites and their tree patterns from
`NormalizeFloatingNumbers`, as well as its extra invocation from `FinishAnalysis`.
`NormalizeFloatingNumbers` continues to handle joins and window partitioning in its existing
optimizer batch.
### Why are the changes needed?
The array set operations own their element equality and hashing semantics. Wrapping their inputs
in `ArrayTransform` expressions scans and materializes normalized arrays before the operations
scan them again, and exposes an execution detail as a logical-plan rewrite.
Handling special floating-point values at the point where each operation hashes, compares, or
emits an element preserves the SQL semantics with less work. It also covers interpreted
evaluation, code generation, and constant folding without depending on a `FinishAnalysis`
rewrite.
### Does this PR introduce _any_ user-facing change?
No result change relative to current master. `array_distinct`, `array_union`,
`array_intersect`, `array_except`, and `arrays_overlap` continue to treat `-0.0` and
`+0.0`, as well as different NaN representations, as equivalent. The implementation no longer
injects normalization transforms into their input plans.
### How was this patch tested?
The following focused tests passed:
```
build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.expressions.CollectionExpressionsSuite -- -z SPARK-54918'
build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.NormalizeFloatingPointNumbersSuite -- -z SPARK-54918'
build/sbt 'sql/testOnly org.apache.spark.sql.DataFrameFunctionsSuite -- -z SPARK-54918'
```
All 14 targeted tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59602][SQL] Handle floating-point normalization in array set operations [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58818:
URL: https://github.com/apache/spark/pull/58818#issuecomment-5750024120
thanks for the review, merging to master/4.x/4.3/4.2/4.1/4.0
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59602][SQL] Handle floating-point normalization in array set operations [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #58818: [SPARK-59602][SQL] Handle floating-point normalization in array set operations
URL: https://github.com/apache/spark/pull/58818
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59602][SQL] Handle floating-point normalization in array set operations [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58818:
URL: https://github.com/apache/spark/pull/58818#issuecomment-5708111751
Thanks @dongjoon-hyun. Addressed all comments:
1. Created SPARK-59602, updated the PR title, and listed all affected target branches.
2. Restored the `ArrayExcept` Scaladoc.
3. Kept the existing `withNaNCheckFunc`/`withNaNCheckCode` names and removed the MiMa exclusions.
4. Added a non-canonical Float NaN test case.
All focused tests pass. Please take another look when you have time.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59602][SQL] Handle floating-point normalization in array set operations [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58818:
URL: https://github.com/apache/spark/pull/58818#issuecomment-5745293779
Thank you @cloud-fan and @asugranyes @ulysses-you @dongjoon-hyun!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54918][SQL][FOLLOWUP] Handle floating-point normalization in array set operations [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58818:
URL: https://github.com/apache/spark/pull/58818#discussion_r4022079888
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/util/SQLOpenHashSet.scala:
##########
@@ -110,19 +111,21 @@ object SQLOpenHashSet {
}
}
- def withNaNCheckFunc(
+ def withNaNAndZeroCheckFunc(
Review Comment:
Blocking: this rename (and `withNaNCheckCode` -> `withNaNAndZeroCheckCode` at :141) is a binary-incompatible change to public methods that shipped in 4.0.0, and `./dev/mima` fails on it. `build/sbt 'catalyst/mimaReportBinaryIssues'` reports 4 `DirectMissingMethodProblem`s against `spark-catalyst_2.13:4.0.0` (the object method and its static forwarder, for each of the two names). `@Private` does not exempt it: `dev/mima`'s generator only excludes symbols with a `private[X]` qualifier (`tools/.../GenerateMIMAIgnore.scala:38-70`), and `MimaExcludes.scala` has no entry for this class. Keeping the old names is the cheapest fix; otherwise add the two filters MiMa prints. Details in the review summary.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -4561,6 +4618,16 @@ trait ArraySetLike {
@transient protected lazy val ordering: Ordering[Any] =
TypeUtils.getInterpretedOrdering(et)
+ @transient private lazy val normalizeElement: Any => Any = et match {
Review Comment:
Non-blocking perf: this projection is built per expression instance, and a fresh instance is created per task (the evaluator is instantiated per task with its own `references`, `WholeStageCodegenExec.scala:738-793`), so `UnsafeProjection.create` runs once per task instead of once per query as on master. Measured on this head: 0.4-0.95 ms per fresh instance vs 4.5-5.9 us steady-state eval for `array_distinct(array<array<double>>)`. This is not specific to one execution mode: for complex element types `doGenCode` delegates to `nullSafeEval` (:4869/:5072/:5310/:5528), so `wholeStage` on/off does not change it. Cheapest fix: drop the projection and evaluate the same normalized expression with `.eval(InternalRow(value))`, which also removes the `InternalRow.copyValue` deep copy (`:4627`); or hoist the projection per (thread, dataType).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala:
##########
@@ -53,71 +53,52 @@ import org.apache.spark.util.ArrayImplicits._
* `genEqual` method of [[org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext]].
* Case 2 is handled during planning in the `Aggregation` and `StatefulAggregationStrategy` objects
* of [[org.apache.spark.sql.execution.SparkStrategies]].
- * Cases 3-5 are handled by this optimizer rule.
- *
- * This rule runs in two places:
- * 1. Early in `FinishAnalysis` (right after `ReplaceExpressions` and before `EvalInlineTables`)
- * so that array set-like operations are wrapped before optimizer rules that pre-evaluate
- * expressions (e.g. `ConstantFolding`, `ConvertToLocalRelation`, `EvalInlineTables`).
- * 2. As a late batch at the end of the optimizer, because rules like subquery rewrite and
- * join reorder can create new joins or join conditions after `FinishAnalysis` that still
- * need their keys to be normalized.
+ * Cases 3 and 4 are handled by this optimizer rule. Array set operations handle case 5 in their
+ * expression evaluation.
*
* Ideally we should do the normalization in the physical operators that compare the
* binary `UnsafeRow` directly. We don't need this normalization if the Spark SQL execution engine
* is not optimized to run on binary data. This rule is created to simplify the implementation, so
* that we have a single place to do normalization, which is more maintainable.
*
+ * Note that, this rule must be executed at the end of optimizer, because the optimizer may create
+ * new joins(the subquery rewrite) and new join conditions(the join reorder).
*/
object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = {
- plan
- .transformWithPruning( _.containsAnyPattern(WINDOW, JOIN)) {
- case w: Window if w.partitionSpec.exists(p => needNormalize(p)) =>
- // Although the `windowExpressions` may refer to `partitionSpec` expressions,
- // we don't need to normalize the `windowExpressions`, as they are executed
- // per input row and should take the input row as it is.
- w.copy(partitionSpec = w.partitionSpec.map(normalize))
-
- // Only hash join and sort merge join need the normalization. Here we catch all Joins with
- // join keys, assuming Joins with join keys are always planned as hash join or sort merge
- // join. It's very unlikely that we will break this assumption in the near future.
- case j @ ExtractEquiJoinKeys(_, leftKeys, rightKeys, condition, _, _, _, _)
- // The analyzer guarantees left and right joins keys are of the same data type. Here we
- // only need to check join keys of one side.
- if leftKeys.exists(k => needNormalize(k)) =>
- val newLeftJoinKeys = leftKeys.map(normalize)
- val newRightJoinKeys = rightKeys.map(normalize)
- val newConditions = newLeftJoinKeys.zip(newRightJoinKeys).map {
- case (l, r) => EqualTo(l, r)
- } ++ condition
- j.copy(condition = Some(newConditions.reduce(And)))
-
- // The specialized NAAJ is a hash join, but its OR condition is not an equi-join shape.
- case j @ ExtractSingleColumnNullAwareAntiJoin(leftKeys, rightKeys)
- if leftKeys.exists(needNormalize) =>
- val equality = EqualTo(normalize(leftKeys.head), normalize(rightKeys.head))
- j.copy(condition = Some(Or(equality, IsNull(equality))))
-
- // TODO: ideally Aggregate should also be handled here, but its grouping expressions are
- // mixed in its aggregate expressions. It's unreliable to change the grouping expressions
- // here. For now we normalize grouping expressions during planning. See Case 2 in the
- // Scaladoc just above.
- }
- .transformAllExpressionsWithPruning(_.containsAnyPattern(
- ARRAY_DISTINCT, ARRAY_UNION, ARRAY_INTERSECT, ARRAY_EXCEPT, ARRAYS_OVERLAP)) {
- case e: ArrayDistinct if needNormalize(e.child) =>
- e.copy(child = normalize(e.child))
- case e: ArrayUnion if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- case e: ArrayIntersect if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- case e: ArrayExcept if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- case e: ArraysOverlap if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- }
+ def apply(plan: LogicalPlan): LogicalPlan = plan match {
Review Comment:
Leftover from the revert: the single-case `plan match` can go (restore `def apply(plan: LogicalPlan) = plan.transformWithPruning(...)`). The closing scaladoc line about "a single place to do normalization" is also stale now that case 5 lives in the expression evaluation.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/util/SQLOpenHashSet.scala:
##########
@@ -131,31 +134,38 @@ object SQLOpenHashSet {
handleNaN(valueNaN)
}
} else {
- handleNotNaN(value)
+ handleNotNaN(normalize(value))
}
}
- def withNaNCheckCode(
+ def withNaNAndZeroCheckCode(
dataType: DataType,
valueName: String,
hashSet: String,
handleNotNaN: String,
handleNaN: String => String): String = {
val ret = dataType match {
case DoubleType =>
- Some((s"java.lang.Double.isNaN((double)$valueName)", "java.lang.Double.NaN"))
+ Some((
+ s"java.lang.Double.isNaN((double)$valueName)",
+ s"if ($valueName == 0.0d) $valueName = 0.0d;",
Review Comment:
`withNaNAndZeroCheckCode` now assigns to `valueName`, so that parameter must be a writable local of the exact primitive type (same at :156 for Float). All six call sites comply today, but the contract is undocumented, and a future caller passing a getter would only fail at runtime codegen. Worth one scaladoc line.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -30,8 +30,14 @@ import org.apache.spark.sql.catalyst.expressions.KnownNotContainsNull
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
+import org.apache.spark.sql.catalyst.optimizer.NormalizeFloatingNumbers
Review Comment:
Expressions now depend on the optimizer package, the reverse of the existing direction, and this pulls the rule's `internalError` fallthrough into runtime evaluation. A comment on which types `needNormalize` admits but `normalize` cannot handle would help. Relatedly, the map arm of `normalizeElement` is unreachable for these five expressions (the analyzer rejects non-orderable element types: `array<map<string,double>>` fails with `INVALID_ORDERING_TYPE`), so it deserves a one-line causality comment.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -4561,6 +4618,16 @@ trait ArraySetLike {
@transient protected lazy val ordering: Ordering[Any] =
TypeUtils.getInterpretedOrdering(et)
+ @transient private lazy val normalizeElement: Any => Any = et match {
+ case dt if NormalizeFloatingNumbers.needNormalize(dt) =>
Review Comment:
`case dt` shadows the trait's own `dt` (the array type) while `et` is the element type, which is easy to misread in a trait that uses both names. Could be `if (NormalizeFloatingNumbers.needNormalize(et))` without the pattern binding, and the `private lazy val normalizeElement` + `protected def normalizedElement` pair could collapse into a single `protected lazy val`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala:
##########
@@ -3532,6 +3532,46 @@ class CollectionExpressionsSuite
Literal.create(Seq(Float.NaN, null, 1f), ArrayType(FloatType))), true)
}
+ test("SPARK-54918: array set operations normalize special floating-point values") {
+ val doubles = Literal.create(
+ Seq(-0.0d, 0.0d, Double.NaN, Double.NaN), ArrayType(DoubleType, false))
+ val doubleSet = Literal.create(Seq(0.0d, Double.NaN), ArrayType(DoubleType, false))
+ checkEvaluation(ArrayDistinct(doubles), Seq(0.0d, Double.NaN))
+ checkEvaluation(ArrayUnion(doubles, doubleSet), Seq(0.0d, Double.NaN))
+ checkEvaluation(ArrayIntersect(doubles, doubleSet), Seq(0.0d, Double.NaN))
+ checkEvaluation(ArrayExcept(doubles, doubleSet), Seq.empty[Double])
+ checkEvaluation(ArrayExcept(doubleSet, doubles), Seq.empty[Double])
+ checkEvaluation(ArraysOverlap(doubles, doubleSet), true)
+
+ val floats = Literal.create(
+ Seq(-0.0f, 0.0f, Float.NaN, Float.NaN), ArrayType(FloatType, false))
+ val floatSet = Literal.create(Seq(0.0f, Float.NaN), ArrayType(FloatType, false))
+ checkEvaluation(ArrayDistinct(floats), Seq(0.0f, Float.NaN))
+ checkEvaluation(ArrayUnion(floats, floatSet), Seq(0.0f, Float.NaN))
+ checkEvaluation(ArrayIntersect(floats, floatSet), Seq(0.0f, Float.NaN))
+ checkEvaluation(ArrayExcept(floats, floatSet), Seq.empty[Float])
+ checkEvaluation(ArrayExcept(floatSet, floats), Seq.empty[Float])
+ checkEvaluation(ArraysOverlap(floats, floatSet), true)
+ }
+
+ test("SPARK-54918: array set operations normalize nested floating-point values") {
Review Comment:
These new tests are discriminating: they fail on the pre-change revision at the byte-exact comparison in `ExpressionEvalHelper.scala:319` (`8000000000000000` vs `0`), and replacing `normalizedElement` with `identity` reproduces exactly that failure. Two gaps though: the NaN data is canonical only, so NaN-payload canonicalization (which the projection does provide) is not pinned, and no test reaches the complex path from SQL/DataFrame. `checkInputDataTypes` is `TypeCheckSuccess` for `array<array<double>>` and `array<struct<d:double>>`, so one `DataFrameFunctionsSuite` case asserting `doubleToRawLongBits` on a nested/struct element (in the style of `isPositiveZero`) would close the last hop, and pinning the equality itself (`[[0.0],[-0.0]]` -> one element) would let the #53468 wrapper change compose safely.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54918][SQL][FOLLOWUP] Handle floating-point normalization in array set operations [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58818:
URL: https://github.com/apache/spark/pull/58818#issuecomment-5681652872
cc @asugranyes @ulysses-you
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54918][SQL][FOLLOWUP] Handle floating-point normalization in array set operations [spark]
Posted by "asugranyes (via GitHub)" <gi...@apache.org>.
asugranyes commented on PR #58818:
URL: https://github.com/apache/spark/pull/58818#issuecomment-5686538363
Thanks for the follow-up. I went through the different execution paths and the local normalization approach looks good to me.
This is also very close to the approach I initially took in #53695 before we redirected the implementation to the optimizer-rule path in `NormalizeFloatingNumbers`: handling floating-point normalization locally in the array set operations / `SQLOpenHashSet` path. Given the concerns described here with the logical-plan rewrite and the extra `ArrayTransform` pass, going back to local handling makes sense to me.
Two remarks:
**1. Interaction with** #53468
This PR removes the fallback paths for complex types and uses a `GenericComparableWrapper` as the hash key instead. Since `normalizedElement` is currently used when emitting values from those fallback paths, I think we should make sure the two changes compose correctly.
In particular, the wrapper needs to provide hash/equality semantics consistent with the floating-point equivalence expected for nested values. Since `normalizedElement` currently also canonicalizes complex values when they are emitted, it would be good to clarify how this should compose with #53468 once the fallback path is removed. This also seems relevant for the planned `ArraysOverlap` follow-up (https://issues.apache.org/jira/browse/SPARK-58943).
I don't see this as a blocker for this PR. Given the current implementation and the existing O(N^2) fallback for complex types, the approach here looks good to me. I'm noting this as an integration point to keep in mind.
**2. Cost of `normalizedElement` for complex types**
`normalizedElement` builds an `UnsafeProjection` over a synthetic `BoundReference`, and each emitted complex element goes through `InternalRow(Value)` -> `Projection`-> `get `-> `copyValue`.
The new approach avoids the upfront `ArrayTransform` over the entire input array, which is a clear advantage. However, for complex/nested types, particularly when most input elements are eventually emitted, I wonder if it would be worth adding a small benchmark to quantify the cost of the per-element projection.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54918][SQL][FOLLOWUP] Handle floating-point normalization in array set operations [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58818:
URL: https://github.com/apache/spark/pull/58818#discussion_r4022732472
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/util/SQLOpenHashSet.scala:
##########
@@ -131,31 +134,38 @@ object SQLOpenHashSet {
handleNaN(valueNaN)
}
} else {
- handleNotNaN(value)
+ handleNotNaN(normalize(value))
}
}
- def withNaNCheckCode(
+ def withNaNAndZeroCheckCode(
dataType: DataType,
valueName: String,
hashSet: String,
handleNotNaN: String,
handleNaN: String => String): String = {
val ret = dataType match {
case DoubleType =>
- Some((s"java.lang.Double.isNaN((double)$valueName)", "java.lang.Double.NaN"))
+ Some((
+ s"java.lang.Double.isNaN((double)$valueName)",
+ s"if ($valueName == 0.0d) $valueName = 0.0d;",
Review Comment:
Added Scaladoc documenting that valueName must be a writable generated-code local because zero normalization assigns the canonical value back to it. Its primitive type continues to be determined by the dataType argument.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -30,8 +30,14 @@ import org.apache.spark.sql.catalyst.expressions.KnownNotContainsNull
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
+import org.apache.spark.sql.catalyst.optimizer.NormalizeFloatingNumbers
Review Comment:
I kept the shared normalization helper so optimizer and expression evaluation cannot drift. With the current definitions, normalize handles every type admitted by needNormalize: float/double, structs, arrays, and maps. Maps are unreachable for these array set expressions because their elements must be orderable, so the internalError fallback is not reachable on this path.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -4561,6 +4618,16 @@ trait ArraySetLike {
@transient protected lazy val ordering: Ordering[Any] =
TypeUtils.getInterpretedOrdering(et)
+ @transient private lazy val normalizeElement: Any => Any = et match {
Review Comment:
Fixed in 8cdfc26a8d5: the normalized expression is now evaluated directly with eval(InternalRow(value)), avoiding per-task UnsafeProjection construction. I retained InternalRow.copyValue so complex values placed into the set/result have owned backing data.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala:
##########
@@ -53,71 +53,52 @@ import org.apache.spark.util.ArrayImplicits._
* `genEqual` method of [[org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext]].
* Case 2 is handled during planning in the `Aggregation` and `StatefulAggregationStrategy` objects
* of [[org.apache.spark.sql.execution.SparkStrategies]].
- * Cases 3-5 are handled by this optimizer rule.
- *
- * This rule runs in two places:
- * 1. Early in `FinishAnalysis` (right after `ReplaceExpressions` and before `EvalInlineTables`)
- * so that array set-like operations are wrapped before optimizer rules that pre-evaluate
- * expressions (e.g. `ConstantFolding`, `ConvertToLocalRelation`, `EvalInlineTables`).
- * 2. As a late batch at the end of the optimizer, because rules like subquery rewrite and
- * join reorder can create new joins or join conditions after `FinishAnalysis` that still
- * need their keys to be normalized.
+ * Cases 3 and 4 are handled by this optimizer rule. Array set operations handle case 5 in their
+ * expression evaluation.
*
* Ideally we should do the normalization in the physical operators that compare the
* binary `UnsafeRow` directly. We don't need this normalization if the Spark SQL execution engine
* is not optimized to run on binary data. This rule is created to simplify the implementation, so
* that we have a single place to do normalization, which is more maintainable.
*
+ * Note that, this rule must be executed at the end of optimizer, because the optimizer may create
+ * new joins(the subquery rewrite) and new join conditions(the join reorder).
*/
object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = {
- plan
- .transformWithPruning( _.containsAnyPattern(WINDOW, JOIN)) {
- case w: Window if w.partitionSpec.exists(p => needNormalize(p)) =>
- // Although the `windowExpressions` may refer to `partitionSpec` expressions,
- // we don't need to normalize the `windowExpressions`, as they are executed
- // per input row and should take the input row as it is.
- w.copy(partitionSpec = w.partitionSpec.map(normalize))
-
- // Only hash join and sort merge join need the normalization. Here we catch all Joins with
- // join keys, assuming Joins with join keys are always planned as hash join or sort merge
- // join. It's very unlikely that we will break this assumption in the near future.
- case j @ ExtractEquiJoinKeys(_, leftKeys, rightKeys, condition, _, _, _, _)
- // The analyzer guarantees left and right joins keys are of the same data type. Here we
- // only need to check join keys of one side.
- if leftKeys.exists(k => needNormalize(k)) =>
- val newLeftJoinKeys = leftKeys.map(normalize)
- val newRightJoinKeys = rightKeys.map(normalize)
- val newConditions = newLeftJoinKeys.zip(newRightJoinKeys).map {
- case (l, r) => EqualTo(l, r)
- } ++ condition
- j.copy(condition = Some(newConditions.reduce(And)))
-
- // The specialized NAAJ is a hash join, but its OR condition is not an equi-join shape.
- case j @ ExtractSingleColumnNullAwareAntiJoin(leftKeys, rightKeys)
- if leftKeys.exists(needNormalize) =>
- val equality = EqualTo(normalize(leftKeys.head), normalize(rightKeys.head))
- j.copy(condition = Some(Or(equality, IsNull(equality))))
-
- // TODO: ideally Aggregate should also be handled here, but its grouping expressions are
- // mixed in its aggregate expressions. It's unreliable to change the grouping expressions
- // here. For now we normalize grouping expressions during planning. See Case 2 in the
- // Scaladoc just above.
- }
- .transformAllExpressionsWithPruning(_.containsAnyPattern(
- ARRAY_DISTINCT, ARRAY_UNION, ARRAY_INTERSECT, ARRAY_EXCEPT, ARRAYS_OVERLAP)) {
- case e: ArrayDistinct if needNormalize(e.child) =>
- e.copy(child = normalize(e.child))
- case e: ArrayUnion if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- case e: ArrayIntersect if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- case e: ArrayExcept if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- case e: ArraysOverlap if needNormalize(e.left) =>
- e.copy(left = normalize(e.left), right = normalize(e.right))
- }
+ def apply(plan: LogicalPlan): LogicalPlan = plan match {
Review Comment:
Fixed both: removed the redundant plan match and updated the Scaladoc to describe normalization during array set expression evaluation.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/util/SQLOpenHashSet.scala:
##########
@@ -110,19 +111,21 @@ object SQLOpenHashSet {
}
}
- def withNaNCheckFunc(
+ def withNaNAndZeroCheckFunc(
Review Comment:
Fixed in 8cdfc26a8d5 by adding the two narrow DirectMissingMethodProblem filters for withNaNCheckFunc and withNaNCheckCode to v44excludes, which v50excludes inherits. catalyst/mimaReportBinaryIssues passes.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala:
##########
@@ -3532,6 +3532,46 @@ class CollectionExpressionsSuite
Literal.create(Seq(Float.NaN, null, 1f), ArrayType(FloatType))), true)
}
+ test("SPARK-54918: array set operations normalize special floating-point values") {
+ val doubles = Literal.create(
+ Seq(-0.0d, 0.0d, Double.NaN, Double.NaN), ArrayType(DoubleType, false))
+ val doubleSet = Literal.create(Seq(0.0d, Double.NaN), ArrayType(DoubleType, false))
+ checkEvaluation(ArrayDistinct(doubles), Seq(0.0d, Double.NaN))
+ checkEvaluation(ArrayUnion(doubles, doubleSet), Seq(0.0d, Double.NaN))
+ checkEvaluation(ArrayIntersect(doubles, doubleSet), Seq(0.0d, Double.NaN))
+ checkEvaluation(ArrayExcept(doubles, doubleSet), Seq.empty[Double])
+ checkEvaluation(ArrayExcept(doubleSet, doubles), Seq.empty[Double])
+ checkEvaluation(ArraysOverlap(doubles, doubleSet), true)
+
+ val floats = Literal.create(
+ Seq(-0.0f, 0.0f, Float.NaN, Float.NaN), ArrayType(FloatType, false))
+ val floatSet = Literal.create(Seq(0.0f, Float.NaN), ArrayType(FloatType, false))
+ checkEvaluation(ArrayDistinct(floats), Seq(0.0f, Float.NaN))
+ checkEvaluation(ArrayUnion(floats, floatSet), Seq(0.0f, Float.NaN))
+ checkEvaluation(ArrayIntersect(floats, floatSet), Seq(0.0f, Float.NaN))
+ checkEvaluation(ArrayExcept(floats, floatSet), Seq.empty[Float])
+ checkEvaluation(ArrayExcept(floatSet, floats), Seq.empty[Float])
+ checkEvaluation(ArraysOverlap(floats, floatSet), true)
+ }
+
+ test("SPARK-54918: array set operations normalize nested floating-point values") {
Review Comment:
Added both coverage gaps: a noncanonical NaN payload in CollectionExpressionsSuite, and a nonliteral nested-array DataFrame case that verifies [[-0.0], [0.0]] deduplicates to one element containing canonical +0.0.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala:
##########
@@ -4561,6 +4618,16 @@ trait ArraySetLike {
@transient protected lazy val ordering: Ordering[Any] =
TypeUtils.getInterpretedOrdering(et)
+ @transient private lazy val normalizeElement: Any => Any = et match {
+ case dt if NormalizeFloatingNumbers.needNormalize(dt) =>
Review Comment:
Fixed: this now uses an if over et, and the helper pair is collapsed into one protected lazy val.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org