This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 2bee7e67aa72 [SPARK-58792][SQL] Clone Hive GenericUDF per copy
2bee7e67aa72 is described below
commit 2bee7e67aa72016272c9eba7a3e42f418b165a46
Author: James Xu <[email protected]>
AuthorDate: Thu Aug 20 21:53:13 2026 +0800
[SPARK-58792][SQL] Clone Hive GenericUDF per copy
### What changes were proposed in this pull request?
Copied `HiveGenericUDF` expression nodes (e.g. via
`withNewChildrenInternal`) share one
`HiveFunctionWrapper`, whose cached `GenericUDF` instance is mutable. This
PR isolates the
instance per expression copy:
- `HiveGenericUDFEvaluator` overrides the lazy `function` to return an
independent clone of
the cached instance instead of the shared instance itself, so each copied
node initializes
and mutates its own `GenericUDF`.
- The clone is produced by a verbatim copy of Hive's
`FunctionRegistry.cloneGenericUDF`
(Hive 2.3.10), added to `HiveFunctionRegistryUtils` — the class
SPARK-51466 created for
FunctionRegistry-avoiding forks. `FunctionRegistry` itself cannot be
called: its static
initialization registers all Hive built-in UDFs and fails with
`NoClassDefFoundError` on
the Spark 4.x runtime classpath (hive-llap-common is absent since
SPARK-51029). The copied
method covers every `GenericUDF` flavor: plain UDFs, `GenericUDFBridge`,
`GenericUDFMacro`
(whose body must be cloned, not dropped), plus `copyToNewInstance` and
`SettableUDF`
type-info propagation.
- `HiveUDFEvaluatorBase.funcWrapper` becomes a `protected val`, so the
subclass override
reads a real serialized field rather than a compiler-captured constructor
parameter.
`HiveSimpleUDF` needs no change (its instances are never cached);
`HiveGenericUDTF` shows
the same pattern but is a lower-priority follow-up (Generate nodes are
rarely duplicated
with divergent inspector sets and are not constant-folded).
### Why are the changes needed?
`GenericUDF.initialize()` is stateful: it derives argument converters,
cached constant
values, and mutable output holders from the arguments it is given. When two
copies of one
UDF expression legitimately differ in argument constness (attribute vs.
literal), their
inspectors diverge — a `TimestampType` literal maps to a constant writable
inspector
(runtime object: `TimestampWritable`), an attribute to a Java inspector
(runtime object:
`java.sql.Timestamp`). Both copies call `initialize()` on the one shared
instance, last
write wins, and the losing copy then evaluates against the winner's
converter state:
```
java.lang.ClassCastException:
org.apache.hadoop.hive.serde2.io.TimestampWritable cannot be cast to
java.sql.Timestamp
```
or, with the opposite initialization order, silently constant results.
This shape is reachable on master with stock rules: the literal-binding
inference
(SPARK-57437, ships with 4.3.0) feeding `InferFiltersFromConstraints`
manufactures a
constant-argument copy of a UDF conjunct and ANDs it into the same Filter
as the original
attribute-argument conjunct. Minimal repro (a bare boolean Hive UDF
conjunct over a table
with at least two rows matching the equality):
```sql
CREATE TEMPORARY FUNCTION hive_gt AS
'org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPGreaterThan';
-- gt_table(id INT, pt DATE, created_at TIMESTAMP)
SELECT id FROM gt_table
WHERE pt = DATE '2024-09-10'
AND hive_gt(CAST(pt AS TIMESTAMP), created_at);
```
On unpatched master the second row throws the `ClassCastException` above.
On Spark 3.x the
same sharing additionally corrupts constant folding across UNION branches
(silently swapped
folded literals); master avoids that channel only incidentally, via
SPARK-51466's ordering
side effect — this PR converts that incidental immunity into a designed
guarantee.
### Does this PR introduce _any_ user-facing change?
Yes (bug fix). Queries where the optimizer duplicates a Hive `GenericUDF`
expression with
divergent argument constness previously failed with `ClassCastException`
(or could return
silently constant results); they now evaluate correctly. On master the
trigger is the
literal-binding inference (SPARK-57437, unreleased 4.3.0); the UNION
constant-fold flavor
affects released Spark 3.x.
### How was this patch tested?
Two new tests in `HiveUDFSuite`, each verified to **fail before the fix
with the exact
production `ClassCastException`** and pass after:
1. Expression-level: two `copy()`-ed `HiveGenericUDF` nodes sharing one
`HiveFunctionWrapper`
with divergent argument constness, evaluated interleaved; also asserts
the copies share
the wrapper (the test's premise).
2. SQL-level with stock rules (no flags, no extensions): the repro query
above, including a
plan assertion that the optimized Filter really holds both UDF conjuncts
(one
literal-arg, one attribute-arg), so the test cannot go silently vacuous.
`HiveUDFSuite` (44/44), `HiveUDFDynamicLoadSuite` (5/5),
`HiveInspectorSuite` (13/13) via
`build/sbt testOnly`; `build/sbt hive/scalastyle` and
`hive/Test/scalastyle` clean;
checkstyle clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Pi
Closes #58027 from xumingming/SPARK-58792-hive-udf-shared-instance.
Authored-by: James Xu <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../hive/ql/exec/HiveFunctionRegistryUtils.java | 55 +++++++++++++++
.../apache/spark/sql/hive/hiveUDFEvaluators.scala | 11 ++-
.../spark/sql/hive/execution/HiveUDFSuite.scala | 80 +++++++++++++++++++++-
3 files changed, 142 insertions(+), 4 deletions(-)
diff --git
a/sql/hive/src/main/java/org/apache/hadoop/hive/ql/exec/HiveFunctionRegistryUtils.java
b/sql/hive/src/main/java/org/apache/hadoop/hive/ql/exec/HiveFunctionRegistryUtils.java
index 333ae0151a5a..1cb527d92d05 100644
---
a/sql/hive/src/main/java/org/apache/hadoop/hive/ql/exec/HiveFunctionRegistryUtils.java
+++
b/sql/hive/src/main/java/org/apache/hadoop/hive/ql/exec/HiveFunctionRegistryUtils.java
@@ -18,6 +18,10 @@
package org.apache.hadoop.hive.ql.exec;
import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.udf.SettableUDF;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFMacro;
import
org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
import
org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils.PrimitiveGrouping;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
@@ -29,6 +33,8 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import org.apache.hadoop.util.ReflectionUtils;
+
import org.apache.spark.internal.SparkLogger;
import org.apache.spark.internal.SparkLoggerFactory;
@@ -339,4 +345,53 @@ public class HiveFunctionRegistryUtils {
}
}
}
+
+ /**
+ * Create a copy of an existing GenericUDF.
+ */
+ public static GenericUDF cloneGenericUDF(GenericUDF genericUDF) {
+ if (null == genericUDF) {
+ return null;
+ }
+
+ GenericUDF clonedUDF;
+ if (genericUDF instanceof GenericUDFBridge) {
+ GenericUDFBridge bridge = (GenericUDFBridge) genericUDF;
+ clonedUDF = new GenericUDFBridge(bridge.getUdfName(),
bridge.isOperator(),
+ bridge.getUdfClassName());
+ } else if (genericUDF instanceof GenericUDFMacro) {
+ GenericUDFMacro bridge = (GenericUDFMacro) genericUDF;
+ clonedUDF = new GenericUDFMacro(bridge.getMacroName(),
bridge.getBody().clone(),
+ bridge.getColNames(), bridge.getColTypes());
+ } else {
+ clonedUDF = ReflectionUtils.newInstance(genericUDF.getClass(), null);
+ }
+
+ if (clonedUDF != null) {
+ // Copy info that may be required in the new copy.
+ // The SettableUDF calls below could be replaced using this mechanism as
well.
+ try {
+ genericUDF.copyToNewInstance(clonedUDF);
+ } catch (UDFArgumentException err) {
+ throw new IllegalArgumentException(err);
+ }
+
+ // The original may have settable info that needs to be added to the new
copy.
+ if (genericUDF instanceof SettableUDF) {
+ try {
+ TypeInfo typeInfo = ((SettableUDF)genericUDF).getTypeInfo();
+ if (typeInfo != null) {
+ ((SettableUDF)clonedUDF).setTypeInfo(typeInfo);
+ }
+ } catch (UDFArgumentException err) {
+ // In theory this should not happen - if the original copy of the
UDF had this
+ // data, we should be able to set the UDF copy with this same
settableData.
+ LOG.error("Unable to add settable data to UDF " +
genericUDF.getClass());
+ throw new IllegalArgumentException(err);
+ }
+ }
+ }
+
+ return clonedUDF;
+ }
}
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFEvaluators.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFEvaluators.scala
index 866d88dee878..f09ee7eed93e 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFEvaluators.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFEvaluators.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.hive.HiveShim.HiveFunctionWrapper
import org.apache.spark.sql.types.DataType
abstract class HiveUDFEvaluatorBase[UDFType <: AnyRef](
- funcWrapper: HiveFunctionWrapper, children: Seq[Expression])
+ protected val funcWrapper: HiveFunctionWrapper, children: Seq[Expression])
extends HiveInspectors with Serializable {
@transient
@@ -115,6 +115,15 @@ class HiveGenericUDFEvaluator(
funcWrapper: HiveFunctionWrapper, children: Seq[Expression])
extends HiveUDFEvaluatorBase[GenericUDF](funcWrapper, children) {
+ // SPARK-58792: copied expression nodes (e.g. via withNewChildrenInternal)
share one
+ // HiveFunctionWrapper, whose cached GenericUDF instance is mutable:
initialize()
+ // rewrites its converters and output holders based on the arguments of
whichever
+ // copy initialized it last. Give every evaluator its own clone so copied
nodes
+ // cannot corrupt each other.
+ @transient
+ override lazy val function: GenericUDF =
+
HiveFunctionRegistryUtils.cloneGenericUDF(funcWrapper.createFunction[GenericUDF]())
+
@transient
private lazy val argumentInspectors = children.map(toInspector).toArray
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
index 8999ff12d3c2..942172d1411c 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.hive.execution
import java.io.{DataInput, DataOutput, File, PrintWriter}
+import java.sql.{Date, Timestamp}
import java.util.{ArrayList, Arrays, Properties}
import scala.jdk.CollectionConverters._
@@ -35,13 +36,17 @@ import org.apache.hadoop.io.{LongWritable, Writable}
import org.apache.spark.{SparkException, SparkFiles, TestUtils}
import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
-import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
-import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
BindReferences, CodegenObjectFactoryMode, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, Project}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.WholeStageCodegenExec
import org.apache.spark.sql.functions.{call_function, max}
+import org.apache.spark.sql.hive.HiveGenericUDF
+import org.apache.spark.sql.hive.HiveShim.HiveFunctionWrapper
import org.apache.spark.sql.hive.test.{TestHiveSingleton, TestUDTFJar}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.TimeType
+import org.apache.spark.sql.types.{TimestampType, TimeType}
import org.apache.spark.tags.SlowHiveTest
import org.apache.spark.util.Utils
@@ -886,6 +891,75 @@ class HiveUDFSuite extends QueryTest with
TestHiveSingleton {
}
hiveContext.reset()
}
+
+ test("SPARK-58792: copied HiveGenericUDF nodes must not share a mutable
GenericUDF") {
+ val tsAttr = AttributeReference("ts", TimestampType, nullable = false)()
+ val constTs = Literal(
+ DateTimeUtils.fromJavaTimestamp(Timestamp.valueOf("2024-09-10
01:02:03")), TimestampType)
+ val original = HiveGenericUDF(
+ "default.date_add",
+ HiveFunctionWrapper(classOf[GenericUDFDateAdd].getName),
+ Seq(tsAttr, Literal(1)))
+ // Optimizer-created copies of a HiveGenericUDF (via
withNewChildrenInternal) share
+ // one HiveFunctionWrapper, and used to share the single mutable GenericUDF
+ // instance cached inside it.
+ val constCopy = original.copy(children = Seq(constTs, Literal(1)))
+ assert(original.funcWrapper eq constCopy.funcWrapper)
+
+ val boundOriginal = BindReferences.bindReference(original, Seq(tsAttr))
+ val input = InternalRow(DateTimeUtils.fromJavaTimestamp(
+ Timestamp.valueOf("2023-12-25 10:00:00")))
+ // Interleave evaluation of the two copies the way a Project/Filter would.
With a
+ // shared instance, the second evaluation of constCopy reuses the
converters that
+ // the attribute copy's initialize() installed on the shared instance, and
throws
+ // ClassCastException: TimestampWritable cannot be cast to
java.sql.Timestamp.
+ (1 to 2).foreach { _ =>
+ assert(constCopy.eval(input) ==
DateTimeUtils.fromJavaDate(Date.valueOf("2024-09-11")))
+ assert(boundOriginal.eval(input) ==
DateTimeUtils.fromJavaDate(Date.valueOf("2023-12-26")))
+ }
+ }
+
+ test("SPARK-58792: inferred literal-binding conjunct must not corrupt a
copied Hive UDF") {
+ // InferFiltersFromConstraints substitutes the pt = literal binding into
the UDF
+ // conjunct and ANDs the substituted copy into the same Filter, so the
Filter holds
+ // two copies of one UDF expression whose argument constness differs
(literal vs.
+ // attribute). The conjunct is a bare boolean UDF call rather than a
+ // BinaryComparison, so ConstantPropagation (which substitutes into
BinaryComparisons
+ // only) never rewrites the original conjunct into the same constant form,
and the
+ // divergent pair reaches execution with stock rules - no rules excluded.
A real
+ // table is used because a LocalRelation source lets the optimizer
evaluate the
+ // Filter on the driver per-conjunct, which does not interleave the two
copies.
+ // With a shared GenericUDF instance, the second row threw
ClassCastException:
+ // TimestampWritable cannot be cast to java.sql.Timestamp.
+ withUserDefinedFunction("hive_gt" -> true) {
+ sql(s"CREATE TEMPORARY FUNCTION hive_gt AS
'${classOf[GenericUDFOPGreaterThan].getName}'")
+ withTable("gt_table") {
+ sql("CREATE TABLE gt_table (id INT, pt DATE, created_at TIMESTAMP)
STORED AS PARQUET")
+ sql("""
+ |INSERT INTO gt_table VALUES
+ | (1, DATE '2024-09-10', TIMESTAMP '2024-09-01 00:00:00'),
+ | (2, DATE '2024-09-10', TIMESTAMP '2024-09-05 00:00:00'),
+ | (3, DATE '2024-09-10', TIMESTAMP '2024-09-15 00:00:00'),
+ | (4, DATE '2024-09-11', TIMESTAMP '2024-09-01 00:00:00')
+ |""".stripMargin)
+ val df = sql("""
+ |SELECT id FROM gt_table
+ |WHERE pt = DATE '2024-09-10'
+ | AND hive_gt(CAST(pt AS TIMESTAMP), created_at)
+ |ORDER BY id
+ |""".stripMargin)
+ // Guard against the test going silently vacuous: the optimized Filter
must
+ // hold both copies, exactly one of them with a literal first argument.
+ val udfs = df.queryExecution.optimizedPlan.collect {
+ case f: Filter => f.condition.collect { case u: HiveGenericUDF => u }
+ }.flatten
+ assert(udfs.size == 2)
+ assert(udfs.count(_.children.head.isInstanceOf[Literal]) == 1)
+ checkAnswer(df, Row(1) :: Row(2) :: Nil)
+ }
+ }
+ hiveContext.reset()
+ }
}
class TestPair(x: Int, y: Int) extends Writable with Serializable {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
