[jira] [Assigned] (SPARK-58792) [SQL] Isolate Hive GenericUDF instances per expression copy to fix wrong results and ClassCastException from optimizer-duplicated UDF expressions

["Wenchen Fan (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-58792?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Wenchen Fan reassigned SPARK-58792:
-----------------------------------

    Assignee: James Xu

> [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
>            Assignee: 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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org