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