srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4054082215
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala:
##########
@@ -86,6 +86,217 @@ abstract class BaseScriptTransformationSuite extends
QueryTest {
assert(uncaughtExceptionHandler.exception.isEmpty)
}
+ test("SPARK-59277: TRANSFORM output supports first-class CHAR/VARCHAR
without SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val input = Seq(("ab", "xyz")).toDF("c", "v")
+ checkAnswer(
+ input,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(
+ AttributeReference("c", CharType(4, "UTF8_LCASE"))(),
+ AttributeReference("v", VarcharType(5, "UNICODE_CI"))()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(Row("ab ", "xyz")))
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-59277: TRANSFORM CHAR overflow without SerDe raises
EXCEED_LIMIT_LENGTH") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val input = Seq("abcdef").toDF("c")
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("c", CharType(4))()),
+ child = input.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema),
+ spark.sqlContext)
+ }
+ 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 without SerDe raises
EXCEED_LIMIT_LENGTH") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val input = Seq("abcdefgh").toDF("v")
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("v", VarcharType(5))()),
+ child = input.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema),
+ spark.sqlContext)
+ }
+ 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: TRANSFORM converts nested CHAR/VARCHAR without SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq(
+ ("""["ab"]""", ArrayType(CharType(4)), Row(Seq("ab "))),
+ ("""["xy"]""", ArrayType(VarcharType(4)), Row(Seq("xy"))),
+ (
+ """{"1":"ab"}""",
+ MapType(IntegerType, CharType(4)),
+ Row(Map(1 -> "ab "))),
+ (
+ """{"1":{"2":"ab"}}""",
+ MapType(IntegerType, MapType(IntegerType, CharType(4))),
+ Row(Map(1 -> Map(2 -> "ab ")))),
+ (
+ """[{"1":"ab"}]""",
+ ArrayType(MapType(IntegerType, CharType(4))),
+ Row(Seq(Map(1 -> "ab ")))),
+ (
+ """{"m":{"1":"ab"}}""",
+ StructType(Seq(StructField("m", MapType(IntegerType, CharType(4))))),
+ Row(Row(Map(1 -> "ab ")))),
+ (
+ """{"value":"xy"}""",
+ StructType(Seq(StructField("value", CharType(5)))),
+ Row(Row("xy ")))).foreach { case (json, dataType, expected) =>
+ val input = Seq(json).toDF("value")
+ checkAnswer(
+ input,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("value", dataType)()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(expected))
+ }
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-59277: TRANSFORM nested CHAR/VARCHAR overflow without SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq(
+ (ArrayType(CharType(4)), """["abcdef"]"""),
+ (MapType(IntegerType, CharType(4)), """{"1":"abcdef"}"""),
+ (
+ MapType(IntegerType, MapType(IntegerType, CharType(4))),
+ """{"1":{"2":"abcdef"}}"""),
+ (
+ StructType(Seq(StructField("value", VarcharType(4)))),
+ """{"value":"abcdef"}""")).foreach { case (dataType, json) =>
+ val input = Seq(json).toDF("value")
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("value", dataType)()),
+ child = input.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema),
+ spark.sqlContext)
+ }
+ 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: malformed nested CHAR JSON without SerDe returns null") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val input = Seq("""{"1":""").toDF("value")
+ checkAnswer(
+ input,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(
+ AttributeReference("value", MapType(IntegerType, CharType(4)))()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(Row(null)))
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-59277: TRANSFORM validates restored JSON map keys without
SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val mapType = MapType(IntegerType, CharType(4))
+ Seq(
+ ("""{"1":"ab"}""", mapType, Row(Map(1 -> "ab "))),
+ ("""{"not-an-int":"ab"}""", mapType, Row(null)),
+ ("""{"1":"a","01":"b"}""", mapType, Row(null)),
+ (
+ """[{"not-an-int":"ab"}]""",
+ ArrayType(mapType),
+ Row(null)),
+ (
+ """{"m":{"1":"a","01":"b"}}""",
+ StructType(Seq(StructField("m", mapType))),
+ Row(null))).foreach { case (json, dataType, expected) =>
+ val input = Seq(json).toDF("value")
+ checkAnswer(
+ input,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("value", dataType)()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(expected))
+ }
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-59277: colliding map key followed by valid row without SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val mapType = MapType(IntegerType, CharType(4))
+ // Row 1 has duplicate converted keys (1 and 01 both cast to 1).
+ // Row 2 is valid. Both rows are in the same partition.
+ val input = Seq(
+ """{"1":"a","01":"b"}""",
+ """{"2":"cd"}""").toDF("value")
Review Comment:
This does not enforce the same-partition condition that the regression
needs. `Seq(...).toDF` produces a `LocalTableScanExec`, which uses
`min(rowCount, leafNodeDefaultParallelism)` partitions; under this suite's
`local[2]` session, these two rows normally run in separate tasks, so the test
would also pass with the original shared-builder bug. Please force the input
into one partition.
```suggestion
"""{"2":"cd"}""").toDF("value").coalesce(1)
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -376,6 +424,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 or
+ * duplicate key cannot leave shared state dirty for the next row.
+ */
+ private[sql] def makeJsonMapKeyRestorer(
+ jsonType: DataType,
Review Comment:
This signature accepts `jsonType` and `targetType` independently, but the
recursive implementation assumes matching container topology. In the struct
branch, `zip` can truncate the restorers and the subsequent loop through
`ts.length` can produce a cryptic index error; other shape mismatches silently
return `identity`. Since the only caller derives the JSON type with
`toJsonMapKeyType(targetType)`, could we make that invariant structural by
accepting only the target type and deriving the JSON type here, or validate
mismatches with a clear error?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -376,6 +424,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 or
+ * duplicate key cannot leave shared state dirty for the next row.
Review Comment:
Clarify that the failure is in key conversion.
```suggestion
* fresh [[ArrayBasedMapBuilder]] on every invocation, so a failed key
* conversion or duplicate key cannot leave shared state dirty for the
next row.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -253,7 +301,7 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
}
}
- // Keep consistent with Hive `LazySimpleSerde`, when there is a type case
error, return null
+ // Keep consistent with Hive `LazySimpleSerDe`, when there is a type case
error, return null
Review Comment:
`type case` should be `type cast`, and this can be phrased more directly.
```suggestion
// Match Hive `LazySimpleSerDe`: return null when a type cast fails.
```
--
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]
