cloud-fan commented on code in PR #58936:
URL: https://github.com/apache/spark/pull/58936#discussion_r4164780396
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -241,6 +276,29 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
data => IntervalUtils.microsToDuration(
IntervalUtils.castStringToDTInterval(UTF8String.fromString(data),
start, end)),
converter)
+ case dt @ (_: ArrayType | _: MapType | _: StructType)
+ if standardCharVarcharSemantics &&
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:
**Non-blocking (P2):** For standard-semantics no-SerDe output, CHAR/VARCHAR
map keys have already been erased to STRING here, so distinct JSON keys such as
`a` and `a ` can both normalize to the same padded key during the final
conversion. `CatalystTypeConverters` then constructs `ArrayBasedMapData`
without duplicate validation, which can emit an invalid map instead of raising
`DUPLICATED_MAP_KEY` or applying `LAST_WIN`.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -201,7 +222,21 @@ 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 dt @ (_: CharType | _: VarcharType) =>
+ // Preserve-only keeps CHAR/VARCHAR in the schema but does not apply
SQL pad/overflow.
Review Comment:
**Nit (P3):** Preserve-only mode is not defined by omitting CHAR/VARCHAR
padding or overflow checks: the Catalyst converters apply the bounded-string
write checks for these types. This branch keeps no-SerDe scalar output
unsupported when standard semantics are disabled, so the comment should state
that compatibility boundary without claiming different converter semantics.
**Verification:**
- **Inspection:** Compare the final wording with
CharConverter/VarcharConverter and preserve-only encoder behavior and confirm
it states no broader semantic claim.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -62,6 +79,10 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
override def outputPartitioning: Partitioning = child.outputPartitioning
+ // Snapshot on the driver. SparkPlan.session is @transient, so executor
`conf` is the
Review Comment:
**Nit (P3):** This rationale is inaccurate: when the transient session is
unavailable in a running task, `SQLConf.get` resolves to the task's
`ReadOnlySQLConf`, which reads the driver-propagated query properties rather
than a default SQLConf. Please describe the actual reason this value needs to
be snapshotted, if it still does.
**Verification:**
- **Inspection:** Trace the final wording through SparkPlan.conf,
SQLConfHelper.conf, and running-task SQLConf.get behavior and confirm every
factual claim matches those paths.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -376,6 +434,98 @@ 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
+ }
+
+ /**
+ * Build a per-call map-key restorer that converts parsed JSON string keys
+ * back to the declared physical key type and validates the result through a
+ * fresh [[ArrayBasedMapBuilder]] on every invocation, so a failed key
+ * conversion or duplicate key cannot leave shared state dirty for the next
row.
+ */
+ private[sql] def makeJsonMapKeyRestorer(
+ targetType: DataType,
+ timeZoneId: Option[String]): Any => Any = {
+ val jsonType = toJsonMapKeyType(targetType)
+
+ def make(jt: DataType, tt: DataType): Any => Any = (jt, tt) match {
+ case (ArrayType(jet, _), ArrayType(tet, _)) =>
+ val elem = make(jet, tet)
+ (input: Any) => {
+ val arr = input.asInstanceOf[ArrayData]
+ val n = arr.numElements()
+ val out = new Array[Any](n)
+ var i = 0
+ while (i < n) {
+ out(i) = if (arr.isNullAt(i)) null
Review Comment:
**Non-blocking (P2):** The recursive no-SerDe restorer has separate
null-preservation branches for array elements, map values, and struct fields,
but the new valid nested fixtures exercise only non-null members. A regression
here could be swallowed by `wrapperConvertException` and turn the entire output
field into null, so the tests need valid nested-null cases that assert the null
remains at its original nesting level.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -201,7 +222,21 @@ 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 dt @ (_: CharType | _: VarcharType) =>
+ // Preserve-only keeps CHAR/VARCHAR in the schema but does not apply
SQL pad/overflow.
+ // Match CHAR/VARCHAR before `_: StringType` (they extend StringType).
+ if (!standardCharVarcharSemantics) {
+ throw
QueryExecutionErrors.outputDataTypeUnsupportedByNodeWithoutSerdeError(
+ nodeName, dt)
+ }
+ // Do not use the SerDe null-on-error wrapper; that would hide
EXCEED_LIMIT_LENGTH.
+ (data: String) =>
+ if (data == ioschema.outputRowFormatMap("TOK_TABLEROWFORMATNULL")) {
Review Comment:
**Non-blocking (P2):** This is a new null-token path specific to
standard-semantics scalar CHAR/VARCHAR output, while the added scalar tests
cover only padding and overflow. Please add a configured null-token case that
asserts a SQL null result; otherwise a regression can turn the marker into
padded literal data or an overflow without affecting the existing STRING null
tests.
**Verification:**
- **Regression:** Exercise scalar CHAR/VARCHAR output with the configured
null token and assert a null Catalyst/SQL result under standard semantics.
##########
sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationSuite.scala:
##########
@@ -372,6 +380,163 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
}
}
+ test("SPARK-59277: TRANSFORM supports nested collated CHAR/VARCHAR with Hive
SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val query = sql(
+ """
+ |SELECT TRANSFORM(
+ | array(CAST('ab' AS CHAR(4) COLLATE UTF8_LCASE)),
+ | named_struct('value', CAST('xyz' AS VARCHAR(6) COLLATE
UNICODE_CI)))
+ |USING 'cat'
+ |AS (
+ | chars ARRAY<CHAR(4) COLLATE UTF8_LCASE>,
+ | nested STRUCT<value: VARCHAR(6) COLLATE UNICODE_CI>)
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin)
+ assert(query.schema.map(_.dataType) === Seq(
+ ArrayType(CharType(4, "UTF8_LCASE")),
+ StructType(Seq(StructField("value", VarcharType(6, "UNICODE_CI"))))))
+ checkAnswer(query, Row(Seq("ab "), Row("xyz")))
+ }
+ }
+
+ test("SPARK-59277: TRANSFORM CHAR overflow with Hive SerDe raises
EXCEED_LIMIT_LENGTH") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val exception = intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM('abcdef')
+ |USING 'cat'
+ |AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin).collect()
+ }
+ val runtimeException = exception match {
+ case s: org.apache.spark.SparkRuntimeException => s
+ case other =>
+ other.getCause.asInstanceOf[org.apache.spark.SparkRuntimeException]
+ }
+ checkError(
+ exception = runtimeException,
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "4"))
+ }
+ }
+
+ test("SPARK-59277: TRANSFORM VARCHAR overflow with Hive SerDe raises
EXCEED_LIMIT_LENGTH") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val exception = intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM('abcdefgh')
+ |USING 'cat'
+ |AS (v VARCHAR(5))
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin).collect()
+ }
+ val runtimeException = exception match {
+ case s: org.apache.spark.SparkRuntimeException => s
+ case other =>
+ other.getCause.asInstanceOf[org.apache.spark.SparkRuntimeException]
+ }
+ checkError(
+ exception = runtimeException,
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "5"))
+ }
+ }
+
+ test("SPARK-59277: nested CHAR/VARCHAR overflow with Hive SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq(
+ """
+ |SELECT TRANSFORM(array('abcdef'))
+ |USING 'cat'
+ |AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin,
+ """
+ |SELECT TRANSFORM(named_struct('value', 'abcdef'))
+ |USING 'cat'
+ |AS (value STRUCT<value: VARCHAR(4)>)
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin).foreach { query =>
+ val exception = intercept[Exception] {
+ sql(query).collect()
+ }
+ val runtimeException = exception match {
+ case s: org.apache.spark.SparkRuntimeException => s
+ case other =>
+ other.getCause.asInstanceOf[org.apache.spark.SparkRuntimeException]
+ }
+ checkError(
+ exception = runtimeException,
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "4"))
+ }
+ }
+ }
+
+ test("SPARK-59277: output SerDe CHAR/VARCHAR rewrite is
LazySimpleSerDe-only") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val output = Seq(
+ AttributeReference("c", CharType(4))(),
+ AttributeReference("v", VarcharType(5))(),
+ AttributeReference("nested", ArrayType(CharType(4)))())
+
+ val (_, lazySoi) = HiveScriptIOSchema.initOutputSerDe(hiveIOSchema,
output).get
+
assert(lazySoi.getAllStructFieldRefs.get(0).getFieldObjectInspector.getTypeName
===
+ "string")
+
assert(lazySoi.getAllStructFieldRefs.get(1).getFieldObjectInspector.getTypeName
===
+ "string")
+
assert(lazySoi.getAllStructFieldRefs.get(2).getFieldObjectInspector.getTypeName
===
+ "array<string>")
+
+ val subclassSchema = hiveIOSchema.copy(
+ outputSerdeClass = Some(classOf[TestLazySimpleSerDe].getCanonicalName))
+ val (_, subclassSoi) =
HiveScriptIOSchema.initOutputSerDe(subclassSchema, output).get
+
assert(subclassSoi.getAllStructFieldRefs.get(0).getFieldObjectInspector.getTypeName
===
+ "string")
+
+ SchemaCapturingSerDe.lastColumnTypes = null
+ val customSchema = defaultIOSchema.copy(
+ outputSerdeClass =
Some(classOf[SchemaCapturingSerDe].getCanonicalName))
+ HiveScriptIOSchema.initOutputSerDe(customSchema, output)
+ val captured = SchemaCapturingSerDe.lastColumnTypes
+ assert(captured.contains("char(4)"))
+ assert(captured.contains("varchar(5)"))
+ assert(captured.contains("array<char(4)>"))
+ }
+ }
+
+ test("SPARK-59683: preserve-only does not rewrite LazySimpleSerDe
CHAR/VARCHAR") {
+ withSQLConf(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val output = Seq(
+ AttributeReference("c", CharType(4))(),
+ AttributeReference("v", VarcharType(5))(),
+ AttributeReference("nested", ArrayType(CharType(4)))())
+ scala.util.Try(HiveScriptIOSchema.initOutputSerDe(hiveIOSchema,
output).get) match {
+ case scala.util.Success((_, soi)) =>
+ val typeName =
soi.getAllStructFieldRefs.get(0).getFieldObjectInspector.getTypeName
+ assert(typeName === "char(4)",
+ s"preserve-only must not rewrite CHAR to STRING, found $typeName")
+
assert(soi.getAllStructFieldRefs.get(1).getFieldObjectInspector.getTypeName ===
+ "varchar(5)")
+
assert(soi.getAllStructFieldRefs.get(2).getFieldObjectInspector.getTypeName ===
+ "array<char(4)>")
+ case scala.util.Failure(_) =>
Review Comment:
**Non-blocking (P2):** This catch-all makes any `initOutputSerDe` failure
count as success, including unrelated regressions in the schema-selection or
initialization code changed by this PR. The test should prove that
preserve-only mode supplied the declared CHAR/VARCHAR types, and only tolerate
a specifically identified Hive rejection if that outcome is genuinely expected
on this lineage.
**Verification:**
- **Compatibility:** Exercise a LazySimpleSerDe subtype under preserve-only
mode and assert the exact column-type property supplied to it, with any
tolerated failure matched to a specific documented rejection.
--
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]
