You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/20 13:53:30 UTC
(spark) branch master updated: [SPARK-58792][SQL] Clone Hive GenericUDF per copy
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 <xu...@gmail.com>
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 <xu...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org