srielau commented on code in PR #58936:
URL: https://github.com/apache/spark/pull/58936#discussion_r4199704580
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -62,6 +80,12 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
override def outputPartitioning: Partitioning = child.outputPartitioning
+ // Snapshot at plan construction. CHAR_VARCHAR_STANDARD_SEMANTICS is
PERSISTED, so a
+ // plan built under one setting must keep that path even if a later task
SQLConf.get
+ // (ReadOnlySQLConf with the executing session) differs. SparkPlan.session is
+ // @transient, so executor `conf` cannot recover the planning session.
+ private val standardCharVarcharSemantics: Boolean =
conf.charVarcharStandardSemantics
Review Comment:
Addressed. `ScriptInputOutputSchema` now captures
`standardCharVarcharSemantics` at parse under the query or persisted-view
SQLConf, and `ScriptTransformationIOSchema` copies that value. The physical
operator no longer snapshots `SQLConf` at plan construction.
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala:
##########
@@ -268,6 +273,24 @@ object HiveScriptIOSchema extends HiveInspectors {
(columns, columnTypes)
}
+ /**
+ * Hive LazySimpleSerDe CHAR/VARCHAR types truncate on deserialize. Under
standard
+ * semantics, map them to STRING so Spark applies length checks. Only
LazySimpleSerDe
+ * and subclasses get this rewrite; other SerDes keep the declared schema.
+ * preserveCharVarcharTypeInfo without standard semantics is unchanged.
+ */
+ private def outputTypesForSerDe(
+ serdeClassName: String,
+ columnTypes: Seq[DataType]): Seq[DataType] = {
+ val serdeClass = Utils.classForName[AbstractSerDe](serdeClassName)
+ if (SQLConf.get.charVarcharStandardSemantics &&
Review Comment:
Addressed. Hive LazySimpleSerDe CHAR/VARCHAR rewrite now uses
`ioschema.standardCharVarcharSemantics` instead of `SQLConf.get`, so a later
session or task conf cannot change the wire schema.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -376,6 +442,106 @@ 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
+ }
+
+ /**
+ * Convert a Catalyst value from `sourceType` (JSON or unbounded STRING) to
Review Comment:
Updated the Scaladoc. `sourceType` is the parsed Catalyst type, and maps
rebuild through `ArrayBasedMapBuilder` only when the source and target map
types differ or the target contains CHAR/VARCHAR.
--
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]
