James Xu created SPARK-58792:
--------------------------------
Summary: [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
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.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:
h1. {{HiveGenericUDF.withNewChildrenInternal}} is a plain {{copy\(children =
...\)}}, so every copied expression node shares one {{HiveFunctionWrapper}}
\(present since the first Catalyst commit, SPARK\-1251, 2014\).
h1. {{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.
h1. {{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]
