srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4052038261
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala:
##########
@@ -1122,6 +1244,51 @@ private[hive] trait HiveInspectors {
case _: JavaVoidObjectInspector => NullType
}
+ /**
+ * Analysis snapshots the Catalyst return type, but runtime inspectors are
rebuilt from the
+ * current children (including foldability and session CHAR/VARCHAR
settings). STRING may drift
+ * to or from a bounded string type across that boundary. Two bounded types
must match exactly:
+ * accepting a different kind or length would apply the snapshotted
conversion to an incompatible
+ * runtime value.
+ */
+ def checkCompatibleHiveReturnType(
+ inspector: ObjectInspector,
+ expectedType: DataType): Unit = {
+ checkCompatibleHiveReturnType(
+ inspectorToDataType(inspector, preserveCharVarchar = true),
+ expectedType)
+ }
+
+ def checkCompatibleHiveReturnType(
+ runtimeType: DataType,
+ expectedType: DataType): Unit = {
+ if (!compatibleHiveReturnType(runtimeType, expectedType)) {
+ throw SparkException.internalError(
+ s"Hive function runtime type ${runtimeType.catalogString} is
incompatible " +
+ s"with analysis type ${expectedType.catalogString}.")
+ }
+ }
+
+ private def compatibleHiveReturnType(
+ runtimeType: DataType,
+ expectedType: DataType): Boolean = {
+ (runtimeType, expectedType) match {
+ case (rt: CharType, et: CharType) => rt == et
+ case (rt: VarcharType, et: VarcharType) => rt == et
+ case (_: CharType | _: VarcharType, _: CharType | _: VarcharType) =>
false
+ case (_: StringType, _: CharType | _: VarcharType) => true
+ case (_: CharType | _: VarcharType, _: StringType) => true
+ case (ArrayType(rt, _), ArrayType(et, _)) =>
compatibleHiveReturnType(rt, et)
+ case (MapType(rk, rv, _), MapType(ek, ev, _)) =>
Review Comment:
Added positive and negative map and struct compatibility cases in
`HiveInspectorSuite`, covering key/value recursion, struct field recursion,
arity mismatch, and bounded-length mismatch. Fixed in 0cc7cdb2bc9.
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFEvaluators.scala:
##########
@@ -111,34 +111,36 @@ class HiveSimpleUDFEvaluator(
}
}
-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
+private[hive] object HiveGenericUDFEvaluator extends HiveInspectors {
+
+ /**
+ * Driver-side Hive initialize for `SELECT hive_udf(...)`. Returns the
Catalyst type
Review Comment:
Updated the Scaladoc to say `Driver-side Hive initialization`. Fixed in
0cc7cdb2bc9.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -201,7 +211,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.
Review Comment:
Corrected the class name to `LazySimpleSerDe`. Fixed in 0cc7cdb2bc9.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -241,6 +259,31 @@ 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 cast = Cast(
Review Comment:
Replaced the whole-map cast with recursive key restoration through
`MapFromArrays`, which validates null and duplicate converted keys. The
restoration recurses through arrays, structs, and map values inside the
malformed-field wrapper, while CHAR/VARCHAR conversion remains outside so
overflow still raises `EXCEED_LIMIT_LENGTH`. Added valid, invalid,
duplicate-producing, and nested-key coverage. Fixed in 0cc7cdb2bc9.
##########
sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveInspectorSuite.scala:
##########
@@ -292,6 +301,240 @@ class HiveInspectorSuite extends SparkFunSuite with
HiveInspectors {
assert(typeInfo2.scale() === 10)
}
+ test("SPARK-59277: Hive object inspectors preserve CHAR/VARCHAR type
information") {
+ withFirstClassCharVarchar(enabled = true) {
+ Seq[DataType](CharType(5), VarcharType(7)).foreach { dataType =>
+ val inspector =
toInspector(dataType).asInstanceOf[PrimitiveObjectInspector]
+ assert(inspectorToDataType(inspector) === dataType)
+ dataType match {
+ case c: CharType =>
+ assert(inspector.getTypeInfo.asInstanceOf[CharTypeInfo].getLength
=== c.length)
+ case v: VarcharType =>
+
assert(inspector.getTypeInfo.asInstanceOf[VarcharTypeInfo].getLength ===
v.length)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59277: Hive object inspectors accept collated CHAR/VARCHAR
values") {
+ withFirstClassCharVarchar(enabled = true) {
+ Seq[DataType](
+ CharType(5, "UTF8_LCASE"),
+ VarcharType(7, "UNICODE_CI")).foreach { dataType =>
+ val inspector = toInspector(dataType)
+ val value = UTF8String.fromString(dataType match {
+ case _: CharType => "ab"
+ case _: VarcharType => "abc"
+ })
+ val expectedValue = dataType match {
+ case _: CharType => UTF8String.fromString("ab ")
+ case _: VarcharType => value
+ }
+ val expectedType = dataType match {
+ case c: CharType => CharType(c.length)
+ case v: VarcharType => VarcharType(v.length)
+ }
+ assert(inspectorToDataType(inspector) === expectedType)
+ assert(unwrap(wrap(value, inspector, dataType), inspector) ===
expectedValue)
+ }
+ }
+ }
+
+ test("SPARK-59277: writable CHAR/VARCHAR inspectors round-trip values and
nulls") {
+ withFirstClassCharVarchar(enabled = true) {
+ val charType = CharType(5)
+ val charInspector =
PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(
+ new CharTypeInfo(charType.length))
+ val charValue = UTF8String.fromString("abc")
+ val wrappedChar = wrap(charValue, charInspector, charType)
+ assert(wrappedChar.isInstanceOf[HiveCharWritable])
+
assert(wrappedChar.asInstanceOf[HiveCharWritable].getHiveChar.getPaddedValue
=== "abc ")
+ assert(unwrapperFor(charInspector, charType)(wrappedChar) ===
+ UTF8String.fromString("abc "))
+ assert(wrap(null, charInspector, charType) === null)
+ assert(unwrapperFor(charInspector, charType)(null) === null)
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ wrap(UTF8String.fromString("abcdef"), charInspector, charType)
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "5"))
+
+ val varcharType = VarcharType(7)
+ val varcharInspector =
PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(
+ new VarcharTypeInfo(varcharType.length))
+ val varcharValue = UTF8String.fromString("abc")
+ val wrappedVarchar = wrap(varcharValue, varcharInspector, varcharType)
+ assert(wrappedVarchar.isInstanceOf[HiveVarcharWritable])
+
assert(wrappedVarchar.asInstanceOf[HiveVarcharWritable].getHiveVarchar.getValue
=== "abc")
+ assert(unwrapperFor(varcharInspector, varcharType)(wrappedVarchar) ===
varcharValue)
+ assert(wrap(null, varcharInspector, varcharType) === null)
+ assert(unwrapperFor(varcharInspector, varcharType)(null) === null)
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ wrap(UTF8String.fromString("abcdefgh"), varcharInspector,
varcharType)
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "7"))
+ }
+ }
+
+ test("SPARK-59277: Hive object inspectors support nested CHAR/VARCHAR") {
+ withFirstClassCharVarchar(enabled = true) {
+ val dataType = StructType(Seq(
+ StructField("chars", ArrayType(CharType(4))),
+ StructField("varchars", MapType(IntegerType, VarcharType(8)))))
+ val inspector = toInspector(dataType)
+ assert(inspectorToDataType(inspector) === dataType)
+
+ val input = InternalRow(
+ new GenericArrayData(Array[Any](UTF8String.fromString("a"))),
+ ArrayBasedMapData(
+ Array[Any](1),
+ Array[Any](UTF8String.fromString("value"))))
+ val result = unwrapperFor(inspector, dataType)(
+ wrap(input, inspector, dataType)).asInstanceOf[InternalRow]
+ assert(result.getArray(0).getUTF8String(0) === UTF8String.fromString("a
"))
+ assert(result.getMap(1).valueArray().getUTF8String(0) ===
UTF8String.fromString("value"))
+
+ val outerType = StructType(Seq(StructField("nested", dataType)))
+ val outerInspector =
toInspector(outerType).asInstanceOf[StructObjectInspector]
+ val field = outerInspector.getAllStructFieldRefs.get(0)
+ val targetRow = new SpecificInternalRow(Seq(dataType))
+ unwrapperFor(field, dataType)(wrap(input, inspector, dataType),
targetRow, 0)
+ val nestedResult = targetRow.getStruct(0, dataType.length)
+ assert(nestedResult.getArray(0).getUTF8String(0) ===
UTF8String.fromString("a "))
+ assert(
+ nestedResult.getMap(1).valueArray().getUTF8String(0) ===
UTF8String.fromString("value"))
+ }
+ }
+
+ test("SPARK-59277: typed field unwrappers preserve nanosecond timestamps") {
+ val value = Timestamp.valueOf("2026-09-16 12:34:56.123456789")
+ Seq(
+ TimestampNTZNanosType(9) ->
+ DateTimeUtils.localDateTimeToTimestampNanos(value.toLocalDateTime, 9),
+ TimestampLTZNanosType(9) ->
+ DateTimeUtils.instantToTimestampNanos(value.toInstant, 9)).foreach {
+ case (dataType, expected) =>
+ val inspector =
ObjectInspectorFactory.getStandardStructObjectInspector(
+ util.Arrays.asList("value"),
+
util.Arrays.asList(PrimitiveObjectInspectorFactory.javaTimestampObjectInspector))
+ val field = inspector.getAllStructFieldRefs.get(0)
+ val targetRow = new SpecificInternalRow(Seq(dataType))
+ unwrapperFor(field, dataType)(value, targetRow, 0)
+ assert(targetRow.get(0, dataType) === expected)
+ }
+ }
+
+ test("SPARK-59277: Hive constant inspectors preserve CHAR/VARCHAR type
information") {
+ withFirstClassCharVarchar(enabled = true) {
+ Seq[DataType](CharType(5), VarcharType(7)).foreach { dataType =>
+ val value = UTF8String.fromString("abc")
+ val inspector = toInspector(Literal.create(value, dataType))
+ assert(inspector.isInstanceOf[ConstantObjectInspector])
+ assert(inspectorToDataType(inspector) === dataType)
+ val expected = dataType match {
+ case _: CharType => UTF8String.fromString("abc ")
+ case _: VarcharType => value
+ }
+ assert(unwrapperFor(inspector, dataType)(
+
inspector.asInstanceOf[ConstantObjectInspector].getWritableConstantValue) ===
expected)
+
+ val nullInspector = toInspector(Literal.create(null, dataType))
+ assert(nullInspector.isInstanceOf[ConstantObjectInspector])
+ assert(inspectorToDataType(nullInspector) === dataType)
+ assert(unwrapperFor(nullInspector, dataType)(
+
nullInspector.asInstanceOf[ConstantObjectInspector].getWritableConstantValue)
=== null)
+ }
+ }
+ }
+
+ test("SPARK-59277: Hive CHAR/VARCHAR inspectors remain STRING under legacy
semantics") {
+ withFirstClassCharVarchar(enabled = false) {
+ Seq[DataType](CharType(5), VarcharType(7)).foreach { dataType =>
+ val inspector = toInspector(dataType)
+ assert(inspectorToDataType(inspector) === StringType)
+ assert(inspectorToDataType(inspector, preserveCharVarchar = true) ===
dataType)
+ }
+ }
+ }
+
+ test("SPARK-59277: Hive return type compatibility allows only STRING
boundary drift") {
+ checkCompatibleHiveReturnType(StringType, CharType(5))
+ checkCompatibleHiveReturnType(CharType(5), StringType)
+ checkCompatibleHiveReturnType(ArrayType(StringType),
ArrayType(VarcharType(7)))
+ checkCompatibleHiveReturnType(CharType(5), CharType(5))
+ checkCompatibleHiveReturnType(VarcharType(3), VarcharType(3))
+
+ Seq[(DataType, DataType)](
+ VarcharType(3) -> CharType(5),
+ CharType(4) -> CharType(5),
+ VarcharType(4) -> VarcharType(5),
+ IntegerType -> CharType(5)).foreach { case (runtimeType, expectedType) =>
+ intercept[SparkException] {
+ checkCompatibleHiveReturnType(runtimeType, expectedType)
+ }
+ }
+ }
+
+ test("SPARK-59277: Hive CHAR/VARCHAR boundaries enforce Spark length
semantics") {
+ withFirstClassCharVarchar(enabled = true) {
+ val varchar = VarcharType(3)
+ val varcharInspector = toInspector(varchar)
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ wrap(UTF8String.fromString("abcd"), varcharInspector, varchar)
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "3"))
+
+ val char = CharType(5)
+ val charInspector = toInspector(char)
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ wrap(UTF8String.fromString("abcdef"), charInspector, char)
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "5"))
+
+ val varcharUnwrapper =
+
unwrapperFor(PrimitiveObjectInspectorFactory.javaStringObjectInspector, varchar)
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ varcharUnwrapper("abcd")
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "3"))
+
+ val charUnwrapper =
+
unwrapperFor(PrimitiveObjectInspectorFactory.javaStringObjectInspector, char)
+ assert(charUnwrapper("ab") === UTF8String.fromString("ab "))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ charUnwrapper("abcdef")
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "5"))
+ }
+ }
+
+ test("SPARK-59277: Hive object inspectors reject unsupported CHAR/VARCHAR
lengths") {
+ withFirstClassCharVarchar(enabled = true) {
+ Seq[DataType](CharType(0), CharType(256), VarcharType(65536)).foreach {
dataType =>
Review Comment:
Added `VarcharType(0)` to both the object-inspector and type-info
invalid-length checks. Fixed in 0cc7cdb2bc9.
--
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]
