Hengrui Hu created SPARK-60179:
----------------------------------
Summary: Cast.needsTimeZone misses complex-to-STRING casts
containing a timestamp leaf
Key: SPARK-60179
URL: https://issues.apache.org/jira/browse/SPARK-60179
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.2.0
Reporter: Hengrui Hu
h2. Summary
{{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.
sdp@a4bf018d4221:~$ cat ~/hhr/spark-core-bugs/jira2-cast-needstimezone.txt
h2. Summary
{{Cast.needsTimeZone}} classifies a cast from a complex type to STRING as
time-zone independent,
although {{Cast.castToString}} recurses into ARRAY/MAP/STRUCT and renders a
nested TIMESTAMP with
the session {{zoneId}}. The function is not self-consistent:
{noformat}
needsTimeZone(TIMESTAMP, STRING) = true
needsTimeZone(ARRAY<TIMESTAMP>, ARRAY<STRING>) = true
needsTimeZone(ARRAY<TIMESTAMP>, STRING) = false <-- renders with
zoneId
needsTimeZone(MAP<INT,TIMESTAMP>, STRING) = false <-- renders with
zoneId
needsTimeZone(STRUCT<t:TIMESTAMP>, STRING) = false <-- renders with
zoneId
needsTimeZone(ARRAY<ARRAY<TIMESTAMP>>, ARRAY<STRING>) = false
{noformat}
The last line follows from the first: the complex-to-complex branch recurses to
{{(ARRAY<TIMESTAMP>, STRING)}} and stops there.
That the conversion really is zone dependent:
{code:scala}
val lit = Literal.create(Seq(0L), ArrayType(TimestampType))
Cast(lit, StringType, Some("UTC")).eval(null) // [1970-01-01
00:00:00]
Cast(lit, StringType, Some("America/Los_Angeles")).eval(null) // [1969-12-31
16:00:00]
{code}
h2. Consequences
*1. A zone-dependent cast is considered resolved without a time zone, and
crashes on eval.*
{code:scala}
val c = Cast(Literal.create(Seq(0L), ArrayType(TimestampType)), StringType) //
no timeZoneId
c.resolved // true
c.contextIndependentFoldable // true
c.eval(null) // java.util.NoSuchElementException: None.get
{code}
{{Cast.resolved}} only requires {{timeZoneId.isDefined}} when {{needsTimeZone}}
is true.
{{ResolveTimeZone}} assigns the session zone unconditionally, so plans built by
the analyzer are
covered; a {{Cast}} constructed after that point is reported as resolved and
then fails in
{{TimeZoneAwareExpression.zoneId}}.
*2. {{Cast.canonicalized}} drops the time zone, so two casts with different
zones compare equal.*
{{canonicalized}} calls {{basic.withTimeZone(null)}} when {{!needsTimeZone}}.
With a shared Parquet
table holding {{ts TIMESTAMP}} and {{ats ARRAY<TIMESTAMP>}}, caching under one
session time zone and
querying under another returns the stale rendering, because the cache lookup
goes through
{{sameResult}}:
{code:scala}
// q = select cast(<col> as string) as v from t
spark.conf.set("spark.sql.session.timeZone", "UTC"); d1 = spark.sql(q);
d1.cache(); d1.collect()
spark.conf.set("spark.sql.session.timeZone", "America/Los_Angeles");
spark.sql(q).collect()
{code}
{noformat}
col = ts (needsTimeZone = true, control)
plan under LA: *(1) Project [cast(ts#0 as string) AS v#31] +- FileScan parquet
value under LA: [1969-12-31 16:00:00] correct, cache
not reused
col = ats (needsTimeZone = false)
plan under LA: InMemoryTableScan [v#64] +- InMemoryRelation ...
value under LA: [[1970-01-01 00:00:00]] WRONG, UTC
rendering reused
value under LA after unpersist: [[1969-12-31 16:00:00]] correct
{noformat}
Measured on current master ({{5.0.0-SNAPSHOT}}), {{local[4]}}, AQE disabled.
*3. Downstream users of the predicate inherit the misclassification*, e.g.
{{contextIndependentFoldable}} (used by {{V2ExpressionBuilder}}), the
partition-predicate pushdown
check in {{HiveShim}}, and {{ResolveDefaultColumnsUtil}}.
h2. Proposed fix
Add one case to {{Cast.needsTimeZone}}: a cast whose target is STRING needs the
session time zone if
the source type recursively contains a time-zone-sensitive timestamp leaf
(TIMESTAMP /
TIMESTAMP_LTZ(p)), e.g. via {{DataType.existsRecursively}}. The existing
complex-to-complex
recursion then covers the nested {{ARRAY<ARRAY<TIMESTAMP>> -> ARRAY<STRING>}}
shape as well, since
it bottoms out in the new case. TIMESTAMP_NTZ stays zone independent,
consistent with the scalar
rule.
Since the predicate becomes stricter, casts to STRING over complex types now
require a
{{timeZoneId}} to be resolved. Call sites that construct such a cast outside
the analyzer need a
zone, and plan-string golden files may change.
Suggested coverage: extend the {{needsTimeZone}} assertions in
{{CastSuiteBase}} with the
complex-to-STRING pairs above (with TIMESTAMP_NTZ as the negative control), and
add the
cross-time-zone cache case.
h2. Notes
Not a regression. {{Cast.needsTimeZone}} has had no complex-to-STRING case
since branch-3.5, 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 uses
{{Cast.needsTimeZone}} to decide whether a projection is repeatable across two
evaluations. The gap
is in the shared classification, 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]
