Hengrui Hu created SPARK-60178:
----------------------------------
Summary: Canonicalization re-associates floating-point
Add/Multiply, producing wrong results
Key: SPARK-60178
URL: https://issues.apache.org/jira/browse/SPARK-60178
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.2.0
Reporter: Hengrui Hu
{{Add.canonicalized}} and {{Multiply.canonicalized}} flatten and reorder
adjacent operands
({{Expression.buildCanonicalizedPlan}} -> {{orderCommutative}} ->
{{gatherCommutative}}) with no
exception for FLOAT/DOUBLE. Floating-point addition and multiplication are
commutative but *not
associative*, so {{(a + b) + c}} and {{a + (b + c)}} canonicalize to the same
expression while
evaluating to different values.
Every consumer of semantic equality is therefore allowed to substitute one for
the other. Two of
them return wrong results today: the cached-plan lookup in {{CacheManager}}
({{sameResult}}) and
subexpression elimination ({{EquivalentExpressions.ExpressionEquals}} ->
{{semanticEquals}}).
h2. Reproduction
Measured on current master ({{5.0.0-SNAPSHOT}}), {{local[4]}}, AQE disabled.
One Parquet row:
{code:sql}
-- a = 1.0E16, b = -1.0E16, c = 1.0, all DOUBLE
-- (a + b) + c = 1.0 a + (b + c) = 0.0
{code}
*1. Subexpression elimination, same query*
{code:sql}
SELECT (a + b) + c AS x, a + (b + c) AS y FROM t
{code}
{noformat}
cse=true wholeStage=true -> x=1.0 y=1.0 <-- y is wrong
cse=true wholeStage=false -> x=1.0 y=1.0 <-- y is wrong
cse=false wholeStage=true -> x=1.0 y=0.0
cse=false wholeStage=false -> x=1.0 y=0.0
{noformat}
The physical plan still contains both expressions; only the evaluation is
shared. Setting
{{spark.sql.subexpressionElimination.enabled=false}} restores the correct
answer, which isolates
the mechanism.
*2. Cached-plan reuse, across queries*
{code:scala}
val d1 = spark.sql("select (a+b)+c as v from t")
d1.cache(); d1.collect() // [1.0], correct
spark.sql("select a+(b+c) as v from t").collect() // [1.0], expected [0.0]
d1.unpersist()
spark.sql("select a+(b+c) as v from t").collect() // [0.0], correct again
{code}
The second query's plan is served from the cache entry built for the first:
{noformat}
== Physical Plan ==
InMemoryTableScan [v#49]
+- InMemoryRelation [v#49], StorageLevel(disk, memory, deserialized, 1
replicas)
+- *(1) Project [((a#3 + b#4) + c#5) AS v#20]
{noformat}
*3. Directly, at the expression level*
{code:scala}
val a = AttributeReference("a", DoubleType)()
val b = AttributeReference("b", DoubleType)()
val c = AttributeReference("c", DoubleType)()
Add(Add(a, b), c).semanticEquals(Add(a, Add(b, c))) // true
Multiply(Multiply(a, b), c).semanticEquals(Multiply(a, Multiply(b, c))) // true
{code}
h2. Proposed fix
Floating-point {{+}} and {{*}} are commutative, so {{a + b}} and {{b + a}} may
still be treated as
semantically equal; it is only the re-association across levels that is
unsound. The narrow fix is
to stop flattening when the operation's {{dataType}} is {{FloatType}} or
{{DoubleType}}, and
canonicalize just the two direct children, sorting them as today. That keeps
commutative reuse and
rejects {{(a + b) + c}} vs {{a + (b + c)}}.
A more conservative variant is to return {{withCanonicalizedChildren}} for
floating-point types,
mirroring the existing decimal fallback added in SPARK-40903 for a comparable
reason (reordering
changing the result), at the cost of also losing the commutative case.
The same {{buildCanonicalizedPlan}} path produces {{MultiCommutativeOp}} above
{{spark.sql.analyzer.canonicalization.multiCommutativeOpMemoryOptThreshold}}
operands, so the guard
belongs in {{canonicalized}} before the operands are gathered rather than in
the reduce step.
Suggested coverage: an assertion pair in the canonicalization suite
({{(a+b)+c}} not semantically
equal to {{a+(b+c)}} for DOUBLE, while {{a+b}} stays equal to {{b+a}}), plus
the two end-to-end
cases above.
h2. Notes
Integer, boolean and bitwise operands are unaffected: reordering those changes
only the order in
which ANSI errors may be raised, which Spark does not specify. For FLOAT/DOUBLE
the returned value
itself changes.
Not a regression. {{Add.canonicalized}} / {{Multiply.canonicalized}} have the
same shape on
branch-3.5 through master, so every maintained branch is affected; the numbers
above were measured
on master.
Found while reviewing https://github.com/apache/spark/pull/58424, where an
optimizer rule relies on
{{sameResult}} to prove two scans are equivalent. The gap is in the shared
canonicalization, not in
that rule.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
