srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4063704195
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala:
##########
@@ -829,6 +869,26 @@ private[hive] trait HiveInspectors {
null
}
}
+ case (_, c: CharType) =>
+ val unwrapper = unwrapperFor(objectInspector)
+ data: Any => {
+ val value = unwrapper(data).asInstanceOf[UTF8String]
+ if (value == null) {
+ null
+ } else {
+ CharVarcharCodegenUtils.charTypeReadSideCheck(value, c.length)
Review Comment:
The map unwrapper now builds through a fresh `ArrayBasedMapBuilder` when the
key type contains CHAR/VARCHAR, so CHAR-padded key collisions follow
`MAP_KEY_DEDUP_POLICY`. The fast `ArrayBasedMapData` path is retained for
non-bounded key types. Fixed in a0fb4390ae3.
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala:
##########
@@ -554,4 +592,77 @@ private[hive] case class HiveUDAFFunction(
copy(children = newChildren)
}
+object HiveUDAFFunction extends HiveInspectors {
+ private[hive] case class InitializedEvaluators(
+ partialEvaluator: GenericUDAFEvaluator,
+ partialInspector: ObjectInspector,
+ finalEvaluator: GenericUDAFEvaluator,
+ finalInspector: ObjectInspector)
+
+ def apply(
+ name: String,
+ funcWrapper: HiveFunctionWrapper,
+ children: Seq[Expression]): HiveUDAFFunction = {
+ apply(name, funcWrapper, children, isUDAFBridgeRequired = false)
+ }
+
+ def apply(
+ name: String,
+ funcWrapper: HiveFunctionWrapper,
+ children: Seq[Expression],
+ isUDAFBridgeRequired: Boolean): HiveUDAFFunction = {
+ val (partialType, resultType) =
+ inferResolvedTypes(funcWrapper, children, isUDAFBridgeRequired)
+ HiveUDAFFunction(
+ name,
+ funcWrapper,
+ children,
+ isUDAFBridgeRequired,
+ mutableAggBufferOffset = 0,
+ inputAggBufferOffset = 0,
+ partialType,
+ resultType)
+ }
+
+ private[hive] def initializeEvaluators(
+ funcWrapper: HiveFunctionWrapper,
+ children: Seq[Expression],
+ isUDAFBridgeRequired: Boolean,
+ expectedPartialType: Option[DataType] = None,
+ expectedResultType: Option[DataType] = None): InitializedEvaluators = {
+ val inputInspectors = children.map(toInspector).toArray
+ def newEvaluator(): GenericUDAFEvaluator = {
+ val resolver = if (isUDAFBridgeRequired) {
+ new SparkGenericUDAFBridge(funcWrapper.createFunction[UDAF]())
+ } else {
+ funcWrapper.createFunction[AbstractGenericUDAFResolver]()
+ }
+ val parameterInfo = new SimpleGenericUDAFParameterInfo(
+ inputInspectors, false, false, false)
+ resolver.getEvaluator(parameterInfo)
+ }
+ val partial1 = newEvaluator()
+ val partialInspector = partial1.init(GenericUDAFEvaluator.Mode.PARTIAL1,
inputInspectors)
+ val finalEvaluator = newEvaluator()
+ val finalInspector =
+ finalEvaluator.init(GenericUDAFEvaluator.Mode.FINAL,
Array(partialInspector))
+
expectedPartialType.foreach(checkCompatibleHiveReturnType(partialInspector, _))
Review Comment:
Added two focused tests: one supplies `LongType` as expected partial
(runtime returns STRING - incompatible), the other supplies `VarcharType(5)` as
expected final (runtime returns `CHAR(5)` - different bounded kind). Both
intercept the `SparkException` from `checkCompatibleHiveReturnType`. Removing
either check at the call site fails the corresponding test. Fixed in
a0fb4390ae3.
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFEvaluators.scala:
##########
@@ -155,16 +157,42 @@ class HiveGenericUDFEvaluator(
oi
}
}
+}
+
+private[hive] class HiveGenericUDFEvaluator(
+ funcWrapper: HiveFunctionWrapper,
+ children: Seq[Expression],
+ catalystReturnType: DataType)
+ 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
+
+ @transient
+ lazy val returnInspector = {
+ val inspector = HiveGenericUDFEvaluator.initialize(function,
argumentInspectors)
+ checkCompatibleHiveReturnType(inspector, catalystReturnType)
Review Comment:
Added two focused tests: one constructs a `HiveGenericUDF` with
`catalystReturnType = VarcharType(5)` but the UDF returns CHAR(5) at runtime;
the other constructs a `HiveGenericUDTF` with an element schema claiming
`VarcharType(5)` but the UDTF returns CHAR(5). Both intercept the
`SparkException`. Removing either `checkCompatibleHiveReturnType` call fails
its test. Fixed in a0fb4390ae3.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
