srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4054033006
##########
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:
Replaced the expression-tree approach (`MapFromArrays` with a shared
`ArrayBasedMapBuilder`) with a runtime function that allocates a fresh
`ArrayBasedMapBuilder` per invocation. A failed or duplicate key now cannot
leave shared state dirty for the next row. Added a focused test with a
colliding row followed by a valid row in the same partition. Fixed in
05da5e1d256.
##########
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:
Updated the remaining `LazySimpleSerde` to `LazySimpleSerDe` in the
`wrapperConvertException` comment. Fixed in 05da5e1d256.
--
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]
