srielau commented on code in PR #58936:
URL: https://github.com/apache/spark/pull/58936#discussion_r4168980201
##########
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:
Updated. The snapshot is for PERSISTED `CHAR_VARCHAR_STANDARD_SEMANTICS` at
plan construction, not because the executor falls back to a default SQLConf.
Task `SQLConf.get` is the executing session's ReadOnlySQLConf;
`SparkPlan.session` is `@transient`, so executor `conf` cannot recover the
planning session.
##########
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:
Added fixtures for null array elements, null map values, and null struct
fields in the nested CHAR/VARCHAR conversion test.
##########
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:
Reworded. The comment now only states that no-SerDe scalar CHAR/VARCHAR
stays unsupported without standard semantics, and why that case must precede
`StringType`.
##########
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:
Removed the catch-all Failure. The test now asserts LazySimpleSerDe
inspectors stay `char(4)` / `varchar(5)` / `array<char(4)>` under preserve-only.
##########
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:
Added a scalar CHAR/VARCHAR null-token test (null input / configured null
token) alongside the padding cases.
--
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]
