srielau commented on code in PR #58584:
URL: https://github.com/apache/spark/pull/58584#discussion_r4054096910
##########
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:
Added `.coalesce(1)` to force both rows into the same partition. Fixed in
d9044026aa2.
##########
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:
Tightened the signature to accept only the target type; the JSON type is now
derived internally via `toJsonMapKeyType`. Fixed in d9044026aa2.
##########
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:
Updated the Scaladoc to say "failed key conversion or duplicate key". Fixed
in d9044026aa2.
##########
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:
Replaced with `Match Hive LazySimpleSerDe: return null when a type cast
fails.` Fixed in d9044026aa2.
--
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]
