srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4053991970
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -376,6 +430,83 @@ case class ScriptTransformationIOSchema(
}
object ScriptTransformationIOSchema {
+ private[sql] def toUnboundedStringType(dataType: DataType): DataType = {
+ dataType.transformRecursively {
+ case c: CharType => c.toStringType
+ case v: VarcharType => v.toStringType
+ }
+ }
+
+ // JSON object keys are always strings. Rewrite every map key, including
nested maps.
+ // `transformRecursively` would stop at the first matching MapType and skip
children.
+ private[sql] def toJsonMapKeyType(dataType: DataType): DataType = dataType
match {
+ case ArrayType(et, n) => ArrayType(toJsonMapKeyType(et), n)
+ case MapType(kt, vt, n) =>
+ val jsonKey = if (kt.isInstanceOf[StringType]) kt else StringType
+ MapType(jsonKey, toJsonMapKeyType(vt), n)
+ case StructType(fields) =>
+ StructType(fields.map(f => f.copy(dataType =
toJsonMapKeyType(f.dataType))))
+ case other => other
+ }
+
+ private[sql] def restoreJsonMapKeys(
+ expression: Expression,
+ targetType: DataType,
+ timeZoneId: Option[String]): Expression = {
+ def restore(
+ expression: Expression,
+ jsonType: DataType,
+ targetType: DataType): Expression = (jsonType, targetType) match {
+ case (ArrayType(jsonElementType, containsNull),
ArrayType(targetElementType, _)) =>
+ val element = NamedLambdaVariable("element", jsonElementType,
containsNull)
+ val restoredElement = restore(element, jsonElementType,
targetElementType)
+ if (restoredElement.fastEquals(element)) {
+ expression
+ } else {
+ ArrayTransform(expression, LambdaFunction(restoredElement,
Seq(element)))
+ }
+ case (
+ MapType(jsonKeyType, jsonValueType, valueContainsNull),
+ MapType(targetKeyType, targetValueType, _)) =>
+ val keys = MapKeys(expression)
+ val key = NamedLambdaVariable("key", jsonKeyType, nullable = false)
+ val restoredKey = if (jsonKeyType.sameType(targetKeyType)) {
+ key
+ } else {
+ Cast(key, targetKeyType, timeZoneId)
+ }
+ val restoredKeys = ArrayTransform(keys, LambdaFunction(restoredKey,
Seq(key)))
+
+ val values = MapValues(expression)
+ val value = NamedLambdaVariable("value", jsonValueType,
valueContainsNull)
+ val restoredValue = restore(value, jsonValueType, targetValueType)
+ val restoredValues = if (restoredValue.fastEquals(value)) {
+ values
+ } else {
+ ArrayTransform(values, LambdaFunction(restoredValue, Seq(value)))
+ }
+ MapFromArrays(restoredKeys, restoredValues)
Review Comment:
`MapFromArrays` reuses an `ArrayBasedMapBuilder`, and `from` resets it only
after a successful build. Here a duplicate converted key throws after the first
key has been inserted, but `wrapperConvertException` catches that failure and
leaves the builder dirty; the next otherwise-valid row then hits `"'from' can
only be called with a fresh ArrayBasedMapBuilder."` and is also converted to
null. Please make this restoration state reset-safe (or fresh per input) and
cover a colliding row followed by a valid row in the same partition.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -201,7 +222,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:
`LazySimpleSerDe` is spelled correctly here, but the existing comment above
`wrapperConvertException` in this method still says `LazySimpleSerde`. Could we
update that remaining occurrence as well?
--
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]
