dtenedor commented on code in PR #59302:
URL: https://github.com/apache/spark/pull/59302#discussion_r4232976437
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala:
##########
@@ -70,6 +70,28 @@ trait SparkCharVarcharUtils {
case v: VarcharType if !SqlApiConf.get.charVarcharFirstClassTypes =>
v.toStringType
case _ => dt
}
+
+ /**
+ * Always replaces CharType/VarcharType with unbounded StringType, including
when first-class
+ * CHAR/VARCHAR types are enabled. Used for Hive TRANSFORM script I/O, which
is
+ * STRING-on-the-wire regardless of the declared AS types.
+ */
+ def replaceCharVarcharWithStringAlways(dt: DataType): DataType = dt match {
Review Comment:
**Low: the new helpers duplicate existing ones.**
`replaceCharVarcharWithStringAlways` is `replaceCharVarcharWithString` without
the config check, and
`CharVarcharUtils.replaceCharVarcharWithStringInSchemaAlways` copies
`replaceCharVarcharWithStringInSchema`; this one also becomes public API on the
`sql/api` trait for a single catalyst caller. Consider having the conditional
versions delegate to these, or keeping a private helper in `CharVarcharUtils`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1856,6 +1857,21 @@ class AstBuilder extends DataTypeAstBuilder
schemaLess
)
)
+ // Under standard semantics, assign declared CHAR/VARCHAR (pad / length
check) in a
+ // Project so the query result type matches AS (...). Flag off and
preserve-only keep
+ // the historical STRING result with no pad and no EXCEED_LIMIT_LENGTH.
+ if (conf.charVarcharStandardSemantics &&
+ attributes.exists(a =>
CharVarcharUtils.getRawType(a.metadata).isDefined)) {
+ Project(
+ attributes.map { a =>
+ Alias(
+ CharVarcharUtils.stringLengthCheck(a, a),
Review Comment:
**High: collated CHAR/VARCHAR inside a STRUCT loses its declared type.** For
a nullable struct, `stringLengthCheck` wraps the rebuilt struct in
`If(IsNull(s), Literal(null, t), named_struct(...))`
(`CharVarcharUtils.scala:270`). Scans add that expression post-hoc in
`ApplyCharTypePadding`, but here it is in the parsed plan, so
`CollationTypeCoercion` rewrites the `If` (`CollationTypeCoercion.scala:96`;
`CaseWhen` gets a same-type skip at line 51, `If` does not) and widens
`STRUCT<value: VARCHAR(6) COLLATE UNICODE_CI>` to `STRUCT<value: STRING COLLATE
UNICODE_CI>`, which looks like why `COLLATE UNICODE_CI` was dropped from the
struct in `HiveScriptTransformationSuite` (L385, L389). Could you restore that
assertion (plus a no-SerDe variant) and make the null guard coercion-stable,
e.g. `CaseWhen(Seq(IsNull(s) -> Literal(null, t)), Some(struct))`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1856,6 +1857,21 @@ class AstBuilder extends DataTypeAstBuilder
schemaLess
)
)
+ // Under standard semantics, assign declared CHAR/VARCHAR (pad / length
check) in a
+ // Project so the query result type matches AS (...). Flag off and
preserve-only keep
+ // the historical STRING result with no pad and no EXCEED_LIMIT_LENGTH.
+ if (conf.charVarcharStandardSemantics &&
+ attributes.exists(a =>
CharVarcharUtils.getRawType(a.metadata).isDefined)) {
Review Comment:
**Low: re-parsing the stored type string breaks quoted struct field names.**
`getRawType` re-parses the `catalogString` that
`replaceCharVarcharWithStringInSchemaAlways` stored at L1822, and
`StructType.catalogString` does not quote field names, so ``AS (s STRUCT<`a b`:
CHAR(4)>)`` stores `struct<a b:char(4)>` and fails to parse under standard
semantics. Since `schema` is already in scope at L1821, consider zipping it
with `attributes` and calling `stringLengthCheck(a, field.dataType)` only for
fields that contain CHAR/VARCHAR, which also avoids re-aliasing the other
columns.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -224,19 +201,12 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
private lazy val outputFieldWriters: Seq[String => Any] = output.map { attr
=>
val converter =
CatalystTypeConverters.createToCatalystConverter(attr.dataType)
attr.dataType match {
+ // CHAR/VARCHAR must not reach the script reader; AstBuilder rewrites
them to
+ // unbounded STRING and applies assignment in a Project when standard
semantics
+ // are enabled. Reject any that still arrive (e.g. hand-built plans).
Review Comment:
**Low: the guard only rejects top-level CHAR/VARCHAR.** Nested CHAR/VARCHAR
from a hand-built plan still reaches the `JsonToStructs` branch, where
`JacksonParser` applies the length check inside `wrapperConvertException`, so
an overflow silently becomes NULL; the Hive SerDe path has no guard at all.
Consider matching `case dt if CharVarcharUtils.hasCharVarchar(dt)`, or
narrowing the comment to top-level types.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1814,11 +1814,12 @@ class AstBuilder extends DataTypeAstBuilder
if (transformClause.setQuantifier != null) {
throw
QueryParsingErrors.transformNotSupportQuantifierError(transformClause.setQuantifier)
}
- // Create the attributes.
+ // Create the attributes. Script I/O is always unbounded STRING (Hive wire
format), with
+ // the declared CHAR/VARCHAR kept in metadata for optional assignment
below.
val (attributes, schemaLess) = if (transformClause.colTypeList != null) {
// Typed return columns.
val schema = createSchema(transformClause.colTypeList)
- val replacedSchema =
CharVarcharUtils.replaceCharVarcharWithStringInSchema(schema)
+ val replacedSchema =
CharVarcharUtils.replaceCharVarcharWithStringInSchemaAlways(schema)
Review Comment:
**Medium: preserve-only mode changes behavior without saying so.** With only
`spark.sql.preserveCharVarcharTypeInfo=true`, `AS (c CHAR(4))` now becomes
STRING that still carries the `char(4)` metadata and gets no Project, which
contradicts that config's doc ("does not replace CHAR/VARCHAR with STRING") and
the comment at L1860-1862, and changes released results (nested
`ARRAY<CHAR(4)>` without a SerDe returned `ARRAY<CHAR(4)>` in 4.2.0). The
leftover metadata also makes consumers disagree: `WHERE c = 'ab'` pads only the
literal and drops the row, and reading through a view casts to `CHAR(4)` and
pads or raises where the direct query does not. Could you either use the
`...Always` rewrite only under standard semantics or gate the Project on
`charVarcharFirstClassTypes`, and call out the choice in the PR description?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala:
##########
@@ -42,8 +42,91 @@ class SparkScriptTransformationSuite extends
BaseScriptTransformationSuite with
)
}
- test("SPARK-59683: TRANSFORM view keeps bound CHAR/VARCHAR mode") {
+ // Nested complex TRANSFORM output is JSON without SerDe. Hive sessions
default to
+ // LazySimpleSerDe, so keep these no-SerDe JSON cases in the Spark suite.
+ test("SPARK-60090: TRANSFORM nested CHAR/VARCHAR assignment via Project") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('["ab"]') t(value)
+ |""".stripMargin),
+ Row(Seq("ab ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<VARCHAR(4)>)
+ |FROM VALUES ('["xy"]') t(value)
+ |""".stripMargin),
+ Row(Seq("xy")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value STRUCT<value: CHAR(5)>)
+ |FROM VALUES ('{"value":"xy"}') t(value)
+ |""".stripMargin),
+ Row(Row("xy ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<STRING, CHAR(4)>)
+ |FROM VALUES ('{"k":"ab"}') t(value)
+ |""".stripMargin),
+ Row(Map("k" -> "ab ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<CHAR(4), INT>)
Review Comment:
**Medium: still-supported behavior lost coverage.** This covers CHAR key
padding, but the deleted collision case (`MAP<CHAR(4), INT>` over `{"a":1,"a
":2}` raising `DUPLICATED_MAP_KEY`) has no replacement, none of the nested
cases assert `query.schema` (which would have caught the collated struct
issue), and the reverse view case (`v_disabled`: created without standard
semantics, read with it on) was dropped. Could you port those, plus a
STRING-keyed map-in-map case and malformed JSON returning null, and add schema
assertions to the nested cases?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -317,7 +259,7 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
}
}
- // Match Hive `LazySimpleSerDe`: return null when a type cast fails.
+ // Keep consistent with Hive `LazySimpleSerde`, when there is a type case
error, return null
Review Comment:
**Nit: this reverts the comment fix from SPARK-59683.** The old wording
returns, including the typo ("type case error"); consider keeping the corrected
text or leaving the line untouched.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1856,6 +1857,21 @@ class AstBuilder extends DataTypeAstBuilder
schemaLess
)
)
+ // Under standard semantics, assign declared CHAR/VARCHAR (pad / length
check) in a
+ // Project so the query result type matches AS (...). Flag off and
preserve-only keep
+ // the historical STRING result with no pad and no EXCEED_LIMIT_LENGTH.
+ if (conf.charVarcharStandardSemantics &&
+ attributes.exists(a =>
CharVarcharUtils.getRawType(a.metadata).isDefined)) {
+ Project(
Review Comment:
**Low: the length check can now be pruned or reordered.** Because assignment
lives in a `Project`, column pruning and filter pushdown can skip it, so
`SELECT count(*)` over overflowing output, or an unreferenced CHAR column, no
longer raises `EXCEED_LIMIT_LENGTH`, whereas base checked every row in the
reader. That matches scan-side checks, so it may well be intended; a test or a
line in the PR description would make it explicit.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala:
##########
@@ -42,8 +42,91 @@ class SparkScriptTransformationSuite extends
BaseScriptTransformationSuite with
)
}
- test("SPARK-59683: TRANSFORM view keeps bound CHAR/VARCHAR mode") {
+ // Nested complex TRANSFORM output is JSON without SerDe. Hive sessions
default to
+ // LazySimpleSerDe, so keep these no-SerDe JSON cases in the Spark suite.
+ test("SPARK-60090: TRANSFORM nested CHAR/VARCHAR assignment via Project") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('["ab"]') t(value)
+ |""".stripMargin),
+ Row(Seq("ab ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<VARCHAR(4)>)
+ |FROM VALUES ('["xy"]') t(value)
+ |""".stripMargin),
+ Row(Seq("xy")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value STRUCT<value: CHAR(5)>)
+ |FROM VALUES ('{"value":"xy"}') t(value)
+ |""".stripMargin),
+ Row(Row("xy ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<STRING, CHAR(4)>)
+ |FROM VALUES ('{"k":"ab"}') t(value)
+ |""".stripMargin),
+ Row(Map("k" -> "ab ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<CHAR(4), INT>)
+ |FROM VALUES ('{"ab":1}') t(value)
+ |""".stripMargin),
+ Row(Map("ab " -> 1)))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('[null]') t(value)
+ |""".stripMargin),
+ Row(Seq(null)))
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-60090: TRANSFORM nested CHAR/VARCHAR overflow via Project") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('["abcdef"]') t(value)
+ |""".stripMargin,
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value STRUCT<value: VARCHAR(4)>)
+ |FROM VALUES ('{"value":"abcdef"}') t(value)
+ |""".stripMargin,
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<STRING, CHAR(4)>)
+ |FROM VALUES ('{"k":"abcdef"}') t(value)
+ |""".stripMargin).foreach { query =>
+ checkExceedLimitLength(intercept[Exception](sql(query).collect()), "4")
+ }
+ }
+ }
+
+ test("SPARK-60090: TRANSFORM view keeps CHAR/VARCHAR Project assignment") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ // The Project with stringLengthCheck is baked into the parsed view plan,
so a later
+ // session conf change cannot drop pad / EXCEED_LIMIT_LENGTH.
Review Comment:
**Nit: the comment describes the view mechanism inaccurately.** The view
text is re-parsed on each read; the Project survives because
`spark.sql.charVarchar.standardSemantics.enabled` is `PERSISTED`, so the view
is parsed under the value saved at creation. Consider rewording along those
lines.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -277,34 +247,6 @@ 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. Restore non-string keys inside the
malformed
- // wrapper, then apply CHAR/VARCHAR (and CHAR/VARCHAR map-key dedup)
outside it.
- val jsonType =
ScriptTransformationIOSchema.toJsonMapKeyType(physicalType)
- val tz = Some(conf.sessionLocalTimeZone)
- val complexTypeFactory = JsonToStructs(
- jsonType,
- ioschema.outputSerdeProps.toMap,
- Literal(null),
- tz)
- val restoreJsonKeys = if (jsonType.sameType(physicalType)) {
- identity[Any] _
- } else {
- ScriptTransformationIOSchema.makeJsonToTargetRestorer(
- jsonType, physicalType, tz)
- }
- val toDeclaredType =
ScriptTransformationIOSchema.makeJsonToTargetRestorer(
- physicalType, dt, tz)
- val parser = wrapperConvertException(
- data => restoreJsonKeys(
- complexTypeFactory.nullSafeEval(UTF8String.fromString(data))),
- identity)
- (data: String) => {
- val parsed = parser(data)
- if (parsed == null) null else toDeclaredType(parsed)
- }
case _: ArrayType | _: MapType | _: StructType =>
Review Comment:
**Low: `MAP<INT, CHAR(n)>` without a SerDe now fails with a raw
`ClassCastException`.** `JacksonParser.convertMap` always builds `UTF8String`
keys, so for `MAP<INT, CHAR(4)>` the exec's `UnsafeProjection` reads them with
`getInt` and throws; base handled (and tested) this under standard semantics,
while `MAP<INT, STRING>` has always failed this way. Since it is out of scope
here, consider a clear parse-time error when standard semantics is on, there is
no output SerDe, and the declared type mixes CHAR/VARCHAR with a non-string map
key, plus a negative test (rejecting in `outputFieldWriters` would also break
empty or null `MAP<INT, STRING>` rows that work today).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala:
##########
@@ -87,6 +87,21 @@ abstract class BaseScriptTransformationSuite extends
QueryTest {
parameters = Map("dataType" -> sqlType))
}
+ protected def checkExceedLimitLength(exception: Throwable, limit: String):
Unit = {
+ val runtimeException = Iterator.iterate[Throwable](exception)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case s: org.apache.spark.SparkRuntimeException
Review Comment:
**Nit: fully qualified class name.** Consider importing
`SparkRuntimeException` next to `SparkException` and `SparkThrowable` instead
of spelling out `org.apache.spark.SparkRuntimeException`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1856,6 +1857,21 @@ class AstBuilder extends DataTypeAstBuilder
schemaLess
)
)
+ // Under standard semantics, assign declared CHAR/VARCHAR (pad / length
check) in a
+ // Project so the query result type matches AS (...). Flag off and
preserve-only keep
+ // the historical STRING result with no pad and no EXCEED_LIMIT_LENGTH.
+ if (conf.charVarcharStandardSemantics &&
+ attributes.exists(a =>
CharVarcharUtils.getRawType(a.metadata).isDefined)) {
+ Project(
+ attributes.map { a =>
+ Alias(
+ CharVarcharUtils.stringLengthCheck(a, a),
+ a.name)(explicitMetadata =
Some(CharVarcharUtils.cleanMetadata(a.metadata)))
Review Comment:
**Nit: nested field metadata is dropped.** `explicitMetadata` keeps the
top-level metadata, but the struct rebuilt in `stringLengthCheck` carries no
field metadata and makes fields nullable when the column is nullable, so a
nested `COMMENT` (and `NOT NULL` inside a nullable struct) from the `AS` clause
disappears from the result schema. Consider building the struct from the
declared field types, or noting the limitation.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -3265,7 +3265,8 @@ class FileSourceCharVarcharTestSuite extends
CharVarcharTestSuite with SharedSpa
test("SPARK-58794: EXTERNAL TABLE CHAR/VARCHAR pad, assignment, and
oversize") {
// External file tables use the same store assignment on write and pad +
- // length-check on scan. Hive TRANSFORM is out of scope for this epic.
+ // length-check on scan. Hive TRANSFORM uses the same assignment via
Project
+ // (SPARK-60090); covered in BaseScriptTransformationSuite.
Review Comment:
**Nit: this points to only one suite.** The nested and Hive coverage lives
in `SparkScriptTransformationSuite` and `HiveScriptTransformationSuite`;
consider naming all three.
--
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]
