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: [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]
