[
https://issues.apache.org/jira/browse/SPARK-58792?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58792:
-----------------------------------
Labels: pull-request-available (was: )
> [SQL] Isolate Hive GenericUDF instances per expression copy to fix wrong
> results and ClassCastException from optimizer-duplicated UDF expressions
> -------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58792
> URL: https://issues.apache.org/jira/browse/SPARK-58792
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 3.3.1
> Reporter: James Xu
> Priority: Major
> Labels: pull-request-available
>
> h3. Problem
> Two distinct optimizer-produced shapes corrupt the same shared, mutable Hive
> UDF instance. Both reduce to one pattern: the optimizer creates two copies of
> a single Hive UDF expression, both copies share one GenericUDF instance, and
> their initialize() calls disagree with each other.
> _Bug 1: ClassCastException or silently constant results from duplicated
> filter conjuncts (SPARK-57437 trigger)._
> SPARK-57437 (merged to master, ships with 4.3.0) substitutes literal bindings
> into predicates, manufacturing a copy of a UDF conjunct with a constant
> argument while keeping the original column-argument conjunct in the same
> Filter. The two copies call initialize() on one shared instance with
> different argument inspectors; the loser evaluates against state built for
> the winner. Observed in production:
> {code:java}
> java.lang.ClassCastException:
> org.apache.hadoop.hive.serde2.io.TimestampWritable cannot be cast to
> java.sql.Timestamp
> {code}
> or, with the opposite initialization order, silently wrong results: the
> shared instance ends up with the constant argument cached, so the
> column-argument copy returns the constant for every row.
> Example shape ({{{}ts_within_days{}}} is a Hive GenericUDF; {{pt}} is a DATE
> partition column):
> {code:sql}
> SELECT ... FROM t
> WHERE pt = '2024-09-10'
> AND ts_within_days(CAST(pt AS TIMESTAMP), created_at, 31)
> {code}
> _Bug 2: Swapped constant-folding results across UNION branches (stock rules;
> Spark 3.x only)._
> No custom rules and no custom UDF are needed. One UDF expression written once
> above a UNION of literals is pushed into both branches and constant-folded on
> the driver:
> {code:sql}
> SELECT dt, n, hive_date_add(dt, n) FROM (
> SELECT '2023-12-25' AS dt, 1 AS n
> UNION ALL SELECT DATE '2024-06-30', 200) t
> {code}
> On Spark 3.x (verified on public 3.3.1) the folded literals baked into the
> plan are corrupted — both rows fold to 2023-12-26, or fully swapped
> (2025-01-16 / 2023-12-26), varying run to run — while every individual UDF
> evaluate() call returns the correct value. Silent, user-visible, no exception.
> This flavor does NOT reproduce on Spark 4.x, but only incidentally:
> SPARK-51466 (4.0.0) reworked {{HiveGenericUDFEvaluator.returnInspector}} for
> an unrelated reason (avoiding Hive FunctionRegistry static initialization),
> and as a side effect each copy's initialize+evaluate+capture now runs
> contiguously, so the shared mutable output always holds the copy's own value
> at capture time. The sharing itself is unchanged on 4.x — the shared instance
> is still created and still initialized once per copy — so the 4.x safety is
> an ordering side effect, not a designed guarantee.
> h3. Root Cause
> The sharing chain has three links, all old:
> {{HiveGenericUDF.withNewChildrenInternal}} is a plain {{{}copy(children =
> ...){}}}, so every copied expression node shares one {{HiveFunctionWrapper}}
> (present since the first Catalyst commit, SPARK-1251, 2014).
> {{HiveFunctionWrapper.createFunction}} caches the GenericUDF instance for
> non-simple UDFs (since SPARK-6909, 2015). Each copied node builds its own
> evaluator, but every evaluator's lazy {{function}} resolves to the same
> cached instance.
> {{GenericUDF.initialize()}} is stateful: it derives argument inspectors,
> converters, cached constant values, and mutable output holders from the
> arguments it is given. Two copies that legitimately differ in argument
> constness (attribute vs. literal) therefore initialize one instance with two
> different inspector sets. Last initialization wins.
> The inspector divergence is structural: {{HiveInspectors}} maps {{Literal(ts,
> TimestampType)}} to a constant writable inspector (whose deferred object
> yields a {{{}TimestampWritable{}}}) but a non-literal TimestampType to a Java
> inspector (which performs {{{}(java.sql.Timestamp) object{}}}). Whichever
> inspector set the shared instance was last initialized with, one of the two
> copies is mismatched — hence the ClassCastException, or the cached-constant
> silent wrong results.
> Note that merely checking {{HiveGenericUDF.foldable}} or {{.dataType}}
> already mutates the shared instance: both force {{{}returnInspector{}}},
> which calls {{{}initializeAndFoldConstants{}}}. No evaluation is required for
> the corruption to start.
> Bug 2 uses a different corruption channel on top of the same sharing. For
> all-constant arguments, {{initializeAndFoldConstants}} pre-evaluates into the
> UDF's mutable output field ({{{}GenericUDFDateAdd{}}} computes into a single
> {{private final DateWritable output}} and returns it). The constant return
> inspectors created for that result alias the same output object with no copy,
> and Spark's {{unwrapperFor}} for a constant inspector captures the constant
> eagerly, once per node — so the literal baked into the plan is whatever the
> shared output happened to hold at capture time. On 3.x the two copies'
> initialize/evaluate/capture sequences interleave (both copies' inspectors are
> forced before either capture), so one copy captures the other's value. On 4.x
> the SPARK-51466 rework happens to keep each copy's sequence contiguous, which
> is the only reason Bug 2 does not manifest there.
> The defect has been latent for a decade because no stock rule both duplicated
> a UDF expression and diverged its argument constness while keeping both
> copies live. In-place substitution rules ({{{}ConstantPropagation{}}},
> {{{}FoldablePropagation{}}}) are safe — the old node is discarded.
> SPARK-57437 is the first rule to create the original-plus-constant-copy
> shape; the UNION-fold flavor reaches it through
> {{PushProjectionThroughUnion}} + {{CollapseProject}} with all-constant
> branches.
> h3. Solution
> Isolate the GenericUDF instance at the evaluator level — the single
> consumption point of the cached instance — rather than at the (many,
> unknowable) copy sites:
> * In {{{}HiveGenericUDFEvaluator{}}}, override the lazy {{function}} to
> return an independent clone of the cached instance. Each expression copy then
> initializes and mutates its own instance; cross-initialization becomes
> impossible regardless of which rule created the copies.
> * The base evaluator class must hold {{funcWrapper}} as a {{{}protected
> val{}}}; referencing the constructor parameter from the subclass capture
> would not survive task serialization (executor NPE).
> The clone must be a verbatim copy of Hive's
> {{{}FunctionRegistry.cloneGenericUDF{}}}, placed in
> {{HiveFunctionRegistryUtils}} (the class SPARK-51466 created for
> FunctionRegistry-avoiding forks). It must NOT call FunctionRegistry itself:
> invoking any of its static methods triggers {{{}FunctionRegistry.{}}}, which
> registers all Hive built-in UDFs and fails with NoClassDefFoundError on the
> 4.x runtime classpath (hive-llap-common is absent since SPARK-51029) — the
> exact failure SPARK-51466 eliminated from this path. The copied method
> handles every GenericUDF flavor and has not changed since Hive 2.0.0
> (verified against 2.3.x / 3.1.x bytecode and master source):
> * plain GenericUDF -> fresh instance of the same class;
> * GenericUDFBridge -> rebuilt with the wrapped UDF's name/className;
> * GenericUDFMacro -> new macro with its body cloned (a naive
> new-instance-by-class-name would silently lose the macro body);
> * {{copyToNewInstance}} and SettableUDF typeInfo propagation preserved (the
> parameterized-cast UDFs carry char/varchar length and decimal precision/scale
> in their TypeInfo).
> Scope notes:
> * {{HiveSimpleUDF}} needs no change: createFunction never caches simple UDF
> instances.
> * {{HiveGenericUDTF}} shares the same pattern and is a lower-priority
> follow-up (Generate nodes are rarely duplicated with divergent inspector
> sets, and UDTFs are not constant-folded).
> * GenericUDAF resolvers are also cached, but per-argument inspector state
> lives in the UDAF evaluator created fresh per aggregation, not in the
> resolver.
> h3. Expected Impact
> * Bug 1: queries that currently fail with ClassCastException — or worse,
> return silently constant results — return correct rows. This is the only
> shape with a merged upstream trigger (SPARK-57437 ships in 4.3.0), so master
> is directly exposed.
> * Bug 2: on Spark 3.x, deterministically wrong folded literals become
> correct. 4.x behavior is unchanged (already incidentally correct).
> * Cost: one extra UDF instantiation per used expression node per JVM (the
> clone is lazy and @transient, recreated after deserialization on executors) —
> negligible, since per-node initialize() already happens today.
> h3. Related work
> * SPARK-57437: the trigger rule for Bug 1; semantically sound by itself.
> * SPARK-51466: performance-motivated returnInspector rework that
> incidentally suppresses Bug 2 on 4.x.
> * SPARK-53038: repeated initialize() on the shared instance via copied
> evaluators (InlineCTE/DeduplicateRelations), filed as a performance issue;
> the correctness dimension went unnoticed.
> * SPARK-4244: constant folding vs. GenericUDF initialize ordering within one
> expression (2014-era).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
