srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4059430586
##########
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:
Acknowledged. Will address the Hive map-key collision validation in this PR.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -241,6 +266,29 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
data => IntervalUtils.microsToDuration(
IntervalUtils.castStringToDTInterval(UTF8String.fromString(data),
start, end)),
converter)
+ case dt @ (_: ArrayType | _: MapType | _: StructType)
+ if CharVarcharUtils.hasCharVarchar(dt) =>
+ val physicalType =
ScriptTransformationIOSchema.toUnboundedStringType(dt)
+ // JSON object keys are strings. Cast them to the declared map key
type after parsing.
+ val jsonType =
ScriptTransformationIOSchema.toJsonMapKeyType(physicalType)
+ val complexTypeFactory = JsonToStructs(
+ jsonType,
+ ioschema.outputSerdeProps.toMap,
+ Literal(null),
+ Some(conf.sessionLocalTimeZone))
+ val parsedToPhysical = if (jsonType.sameType(physicalType)) {
+ identity[Any] _
+ } else {
+ val restoreMapKeys =
ScriptTransformationIOSchema.makeJsonMapKeyRestorer(
+ physicalType, Some(conf.sessionLocalTimeZone))
+ value: Any => restoreMapKeys(value)
+ }
+ val toScala =
CatalystTypeConverters.createToScalaConverter(physicalType)
+ val parser = wrapperConvertException(
+ data => parsedToPhysical(
+ complexTypeFactory.nullSafeEval(UTF8String.fromString(data))),
+ identity)
+ data => converter(toScala(parser(data)))
Review Comment:
Moved to follow-up: script TRANSFORM CHAR/VARCHAR support has been split out
per the review suggestion. The follow-up branch is at
`serge-rielau_data/SPARK-59277-transform`.
##########
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:
Acknowledged. Will add incompatible-runtime-inspector fixtures for UDAF
partial and final mismatches.
##########
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:
Acknowledged. Will add custom GenericUDF and GenericUDTF fixtures whose
runtime inspector is incompatible with the analysis snapshot.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -201,7 +218,15 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
private lazy val outputFieldWriters: Seq[String => Any] = output.map { attr
=>
val converter =
CatalystTypeConverters.createToCatalystConverter(attr.dataType)
attr.dataType match {
- case StringType => wrapperConvertException(data => data, converter)
+ case _: CharType | _: VarcharType =>
+ // First-class CHAR/VARCHAR must not use Hive LazySimpleSerDe's
null-on-error path.
+ (data: String) =>
+ if (data == ioschema.outputRowFormatMap("TOK_TABLEROWFORMATNULL")) {
Review Comment:
Moved to follow-up: script TRANSFORM CHAR/VARCHAR support has been split out
per the review suggestion.
--
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]
