This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 00d66a2e2eb5 [SPARK-59683][SQL][HIVE] Support first-class CHAR/VARCHAR
in script TRANSFORM output
00d66a2e2eb5 is described below
commit 00d66a2e2eb558343e61f4d773d4848a644ea280
Author: Serge Rielau <[email protected]>
AuthorDate: Tue Oct 6 22:44:27 2026 -0700
[SPARK-59683][SQL][HIVE] Support first-class CHAR/VARCHAR in script
TRANSFORM output
### What changes were proposed in this pull request?
Support first-class CHAR/VARCHAR output from script TRANSFORM when
first-class semantics are enabled. Split out from
[#58584](https://github.com/apache/spark/pull/58584) (Hive inspector conversion
+ UDF/UDAF/UDTF) per review feedback.
- **No-SerDe path:** CHAR/VARCHAR scalar, nested, and complex output with
padding, overflow (`EXCEED_LIMIT_LENGTH`), null-token, and malformed-field
handling.
- **No-SerDe nested maps:** Restore JSON string keys to the declared
physical key type through a per-row `ArrayBasedMapBuilder` that validates null
and duplicate converted keys.
- **SerDe path:** Rewrite `LazySimpleSerDe` CHAR/VARCHAR output types to
STRING so Spark applies first-class length checks. Non-`LazySimpleSerDe` SerDes
keep the declared CHAR/VARCHAR schema.
- **Legacy:** Legacy flag-off behavior remains unchanged.
This PR depends on [#58584](https://github.com/apache/spark/pull/58584) for
Hive inspector CHAR/VARCHAR conversion. JIRA:
https://issues.apache.org/jira/browse/SPARK-59683
### Why are the changes needed?
With `spark.sql.charVarchar.standardSemantics.enabled=true`, script
TRANSFORM output was not aware of CHAR/VARCHAR types. Bounded-string values
passed through without padding or length enforcement, and nested output
containing non-string map keys was not restored from the JSON string
representation.
### Does this PR introduce _any_ user-facing change?
Yes. When first-class CHAR/VARCHAR semantics are enabled, script TRANSFORM
now pads CHAR output, enforces length limits on both CHAR and VARCHAR, and
correctly handles nested CHAR/VARCHAR in arrays, maps, and structs. Legacy
flag-off behavior remains unchanged.
### How was this patch tested?
Added focused coverage for:
- Scalar CHAR/VARCHAR output without SerDe (padding, overflow).
- Nested CHAR/VARCHAR (arrays, maps with non-string keys, structs) without
SerDe.
- Malformed JSON and colliding map-key restoration without SerDe.
- Colliding row followed by valid row in the same partition (builder-leak
regression).
- Nested collated CHAR/VARCHAR with Hive SerDe.
- Scalar CHAR/VARCHAR overflow with Hive SerDe.
- `LazySimpleSerDe`-only CHAR/VARCHAR type rewrite verification.
Ran:
```
sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64 \
-Dsbt.override.build.repos=true \
'sql/testOnly
org.apache.spark.sql.execution.SparkScriptTransformationSuite -- -z SPARK-59277'
sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64 \
-Dsbt.override.build.repos=true \
'hive/testOnly
org.apache.spark.sql.hive.execution.HiveScriptTransformationSuite -- -z
SPARK-59277'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Auto
Closes #58936 from srielau/serge-rielau_data/SPARK-59277-transform.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
(cherry picked from commit 6d020ca9295ea5121a831e98ded48dc60061b5f5)
Signed-off-by: Daniel Tenedorio <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 10 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 3 +-
.../plans/logical/ScriptTransformation.scala | 4 +-
.../spark/sql/errors/QueryExecutionErrors.scala | 13 +-
.../execution/BaseScriptTransformationExec.scala | 187 +++++++++-
.../spark/sql/execution/SparkSqlParser.scala | 3 +-
.../sql-tests/analyzer-results/transform.sql.out | 86 ++---
.../execution/BaseScriptTransformationSuite.scala | 375 ++++++++++++++++++++-
.../execution/SparkScriptTransformationSuite.scala | 56 +++
.../org/apache/spark/sql/hive/HiveInspectors.scala | 18 +
.../execution/HiveScriptTransformationExec.scala | 30 +-
.../execution/HiveScriptTransformationSuite.scala | 257 +++++++++++++-
12 files changed, 969 insertions(+), 73 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index e1ed46a58c14..30eeaf2a2ab0 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -9443,6 +9443,11 @@
"TRANSFORM with SERDE is only supported in hive mode."
]
},
+ "TRANSFORM_WITHOUT_SERDE_TYPE" : {
+ "message" : [
+ "TRANSFORM without SERDE does not support the output data type
<dataType>."
+ ]
+ },
"TRIM_COLLATION" : {
"message" : [
"TRIM specifier in the collation."
@@ -11737,11 +11742,6 @@
"Subprocess exited with status <exitCode>. Error: <stderrBuffer>."
]
},
- "_LEGACY_ERROR_TEMP_2265" : {
- "message" : [
- "<nodeName> without serde does not support <dt> as output data type."
- ]
- },
"_LEGACY_ERROR_TEMP_2266" : {
"message" : [
"Invalid `startIndex` provided for generating iterator over the array.
Total elements: <numRows>, requested `startIndex`: <startIndex>."
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index ec894bc34916..3f0a83c51c46 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -2050,7 +2050,8 @@ class AstBuilder extends DataTypeAstBuilder
inSerdeClass, outSerdeClass,
inSerdeProps, outSerdeProps,
reader, writer,
- schemaLess)
+ schemaLess,
+ conf.charVarcharStandardSemantics)
}
/**
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ScriptTransformation.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ScriptTransformation.scala
index 3a2cb2cd7309..5532803eafc5 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ScriptTransformation.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ScriptTransformation.scala
@@ -49,4 +49,6 @@ case class ScriptInputOutputSchema(
outputSerdeProps: Seq[(String, String)],
recordReaderClass: Option[String],
recordWriterClass: Option[String],
- schemaLess: Boolean)
+ schemaLess: Boolean,
+ // Bound while the script is parsed under the direct-query or
persisted-view SQLConf.
+ standardCharVarcharSemantics: Boolean = false)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 79b979b54f32..f80cfecf7aa8 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -2438,14 +2438,11 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
cause = cause)
}
- def outputDataTypeUnsupportedByNodeWithoutSerdeError(
- nodeName: String, dt: DataType): Throwable = {
- new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_2265",
- messageParameters = Map(
- "nodeName" -> nodeName,
- "dt" -> dt.getClass.getSimpleName),
- cause = null)
+ def scriptTransformWithoutSerdeUnsupportedTypeError(
+ dt: DataType): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "UNSUPPORTED_FEATURE.TRANSFORM_WITHOUT_SERDE_TYPE",
+ messageParameters = Map("dataType" -> toSQLType(dt)))
}
def invalidStartIndexError(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
index a313e2c671be..535db0d82fec 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
@@ -31,10 +31,28 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet,
Cast, Expression, GenericInternalRow, JsonToStructs, Literal, StructsToJson,
UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{
+ Attribute,
+ AttributeSet,
+ BoundReference,
+ Cast,
+ Expression,
+ GenericInternalRow,
+ JsonToStructs,
+ Literal,
+ StructsToJson,
+ UnsafeProjection}
import org.apache.spark.sql.catalyst.plans.logical.ScriptInputOutputSchema
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
-import org.apache.spark.sql.catalyst.util.{DateTimeUtils, IntervalUtils}
+import org.apache.spark.sql.catalyst.util.{
+ ArrayBasedMapBuilder,
+ ArrayData,
+ CharVarcharCodegenUtils,
+ CharVarcharUtils,
+ DateTimeUtils,
+ GenericArrayData,
+ IntervalUtils,
+ MapData}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -62,6 +80,11 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
override def outputPartitioning: Partitioning = child.outputPartitioning
+ // Bound at parse into the I/O schema under the query or persisted-view
SQLConf.
+ // Do not re-read SQLConf: planning and task conf can differ from that
captured mode.
+ private def standardCharVarcharSemantics: Boolean =
+ ioschema.standardCharVarcharSemantics
+
override def doExecute(): RDD[InternalRow] = {
val broadcastedHadoopConf =
new SerializableConfiguration(session.sessionState.newHadoopConf())
@@ -201,7 +224,20 @@ 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) =>
+ // Without standard semantics, no-SerDe scalar CHAR/VARCHAR stays
unsupported
+ // (they extend StringType, so this case must come before `_:
StringType`).
+ if (!standardCharVarcharSemantics) {
+ throw
QueryExecutionErrors.scriptTransformWithoutSerdeUnsupportedTypeError(dt)
+ }
+ // Do not use the SerDe null-on-error wrapper; that would hide
EXCEED_LIMIT_LENGTH.
+ (data: String) =>
+ if (data == ioschema.outputRowFormatMap("TOK_TABLEROWFORMATNULL")) {
+ null
+ } else {
+ converter(data)
+ }
+ case _: StringType => wrapperConvertException(data => data, converter)
case BooleanType => wrapperConvertException(data => data.toBoolean,
converter)
case ByteType => wrapperConvertException(data => data.toByte, converter)
case BinaryType =>
@@ -241,6 +277,34 @@ 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 =>
val complexTypeFactory = JsonToStructs(attr.dataType,
ioschema.outputSerdeProps.toMap, Literal(null),
Some(conf.sessionLocalTimeZone))
@@ -249,11 +313,11 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
case udt: UserDefinedType[_] =>
wrapperConvertException(data => udt.deserialize(data), converter)
case dt =>
- throw
QueryExecutionErrors.outputDataTypeUnsupportedByNodeWithoutSerdeError(nodeName,
dt)
+ throw
QueryExecutionErrors.scriptTransformWithoutSerdeUnsupportedTypeError(dt)
}
}
- // Keep consistent with Hive `LazySimpleSerde`, when there is a type case
error, return null
+ // Match Hive `LazySimpleSerDe`: return null when a type cast fails.
private val wrapperConvertException: (String => Any, Any => Any) => String
=> Any =
(f: String => Any, converter: Any => Any) =>
(data: String) => converter {
@@ -368,7 +432,8 @@ case class ScriptTransformationIOSchema(
outputSerdeProps: Seq[(String, String)],
recordReaderClass: Option[String],
recordWriterClass: Option[String],
- schemaLess: Boolean) extends Serializable {
+ schemaLess: Boolean,
+ standardCharVarcharSemantics: Boolean = false) extends Serializable {
import ScriptTransformationIOSchema._
val inputRowFormatMap = inputRowFormat.toMap.withDefault((k) =>
defaultFormat(k))
@@ -376,13 +441,115 @@ 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
+ }
+
+ /**
+ * Convert a Catalyst value from `sourceType` to `targetType`. `sourceType`
is
+ * the parsed Catalyst type (unbounded STRING in place of CHAR/VARCHAR, and
+ * STRING map keys where JSON cannot preserve the declared key type).
+ * CHAR/VARCHAR leaves use write-side length checks. Maps rebuild through a
+ * fresh [[ArrayBasedMapBuilder]] when the source and target map types differ
+ * or the target contains CHAR/VARCHAR, so CHAR/VARCHAR key collisions follow
+ * MAP_KEY_DEDUP_POLICY. Same-typed maps without CHAR/VARCHAR are unchanged.
+ */
+ private[sql] def makeJsonToTargetRestorer(
+ sourceType: DataType,
+ targetType: DataType,
+ timeZoneId: Option[String]): Any => Any = {
+
+ def make(jt: DataType, tt: DataType): Any => Any = {
+ if (jt.sameType(tt) && !CharVarcharUtils.hasCharVarchar(tt)) {
+ identity
+ } else (jt, tt) match {
+ case (_, c: CharType) =>
+ (input: Any) =>
+ CharVarcharCodegenUtils.charTypeWriteSideCheck(
+ input.asInstanceOf[UTF8String], c.length)
+ case (_, v: VarcharType) =>
+ (input: Any) =>
+ CharVarcharCodegenUtils.varcharTypeWriteSideCheck(
+ input.asInstanceOf[UTF8String], v.length)
+ 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
+ else elem(arr.get(i, jet))
+ i += 1
+ }
+ new GenericArrayData(out)
+ }
+ case (MapType(jkt, jvt, _), MapType(tkt, tvt, _)) =>
+ val keyConvert = make(jkt, tkt)
+ val valRestore = make(jvt, tvt)
+ (input: Any) => {
+ val map = input.asInstanceOf[MapData]
+ val n = map.numElements()
+ val builder = new ArrayBasedMapBuilder(tkt, tvt)
+ var i = 0
+ while (i < n) {
+ val k = keyConvert(map.keyArray().get(i, jkt))
+ val v = if (map.valueArray().isNullAt(i)) null
+ else valRestore(map.valueArray().get(i, jvt))
+ builder.put(k, v)
+ i += 1
+ }
+ builder.build()
+ }
+ case (js: StructType, ts: StructType) =>
+ val restorers = js.fields.zip(ts.fields).map {
+ case (jf, tf) => make(jf.dataType, tf.dataType)
+ }
+ (input: Any) => {
+ val row = input.asInstanceOf[InternalRow]
+ val out = new GenericInternalRow(ts.length)
+ var i = 0
+ while (i < ts.length) {
+ if (row.isNullAt(i)) out.setNullAt(i)
+ else out.update(i,
+ restorers(i)(row.get(i, js(i).dataType)))
+ i += 1
+ }
+ out
+ }
+ case _ if !jt.sameType(tt) =>
+ val c = Cast(BoundReference(0, jt, nullable = false), tt, timeZoneId)
+ (input: Any) => c.eval(InternalRow(input))
+ case _ => identity
+ }
+ }
+
+ make(sourceType, targetType)
+ }
+
val defaultFormat = Map(
("TOK_TABLEROWFORMATFIELD", "\u0001"),
("TOK_TABLEROWFORMATLINES", "\n"),
("TOK_TABLEROWFORMATNULL" -> "\\N")
)
- val defaultIOSchema = ScriptTransformationIOSchema(
+ def defaultIOSchema: ScriptTransformationIOSchema =
ScriptTransformationIOSchema(
inputRowFormat = Seq.empty,
outputRowFormat = Seq.empty,
inputSerdeClass = None,
@@ -391,7 +558,8 @@ object ScriptTransformationIOSchema {
outputSerdeProps = Seq.empty,
recordReaderClass = None,
recordWriterClass = None,
- schemaLess = false
+ schemaLess = false,
+ standardCharVarcharSemantics = SQLConf.get.charVarcharStandardSemantics
)
def apply(input: ScriptInputOutputSchema): ScriptTransformationIOSchema = {
@@ -404,6 +572,7 @@ object ScriptTransformationIOSchema {
input.outputSerdeProps,
input.recordReaderClass,
input.recordWriterClass,
- input.schemaLess)
+ input.schemaLess,
+ input.standardCharVarcharSemantics)
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 7747c94b584d..87213a1619ab 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -1347,7 +1347,8 @@ class SparkSqlAstBuilder extends AstBuilder {
inSerdeClass, outSerdeClass,
inSerdeProps, outSerdeProps,
reader, writer,
- schemaLess)
+ schemaLess,
+ conf.charVarcharStandardSemantics)
}
}
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/transform.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/transform.sql.out
index 1e3ddd1c198d..fe7aa8b4282b 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/transform.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/transform.sql.out
@@ -67,7 +67,7 @@ SELECT TRANSFORM(a)
USING 'cat' AS (a)
FROM t
-- !query analysis
-ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -98,7 +98,7 @@ SELECT a, b, decode(c, 'UTF-8'), d, e, f, g, h, i, j, k, l
FROM (
-- !query analysis
Project [a#x, b#x, decode(c#x, UTF-8) AS decode(c, UTF-8)#x, d#x, e#x, f#x,
g#xL, h#x, i#x, j#x, k#x, l#x]
+- SubqueryAlias tmp
- +- ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x,
j#x, k#x, l#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x,
j#x, k#x, l#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x, l#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x,
k#x, l#x])
@@ -129,7 +129,7 @@ SELECT a, b, decode(c, 'UTF-8'), d, e, f, g, h, i, j, k, l
FROM (
-- !query analysis
Project [a#x, b#x, decode(c#x, UTF-8) AS decode(c, UTF-8)#x, d#x, e#x, f#x,
g#x, h#x, i#x, j#x, k#x, l#x]
+- SubqueryAlias tmp
- +- ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x, f#x, g#x, h#x, i#x,
j#x, k#x, l#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x, f#x, g#x, h#x, i#x,
j#x, k#x, l#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x, l#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x,
k#x, l#x])
@@ -144,7 +144,7 @@ SELECT TRANSFORM(a)
USING 'cat'
FROM t
-- !query analysis
-ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true)
+ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true,false)
+- Project [a#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -159,7 +159,7 @@ SELECT TRANSFORM(a, b)
USING 'cat'
FROM t
-- !query analysis
-ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true)
+ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true,false)
+- Project [a#x, b#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -174,7 +174,7 @@ SELECT TRANSFORM(a, b, c)
USING 'cat'
FROM t
-- !query analysis
-ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true)
+ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true,false)
+- Project [a#x, b#x, c#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -190,7 +190,7 @@ USING 'cat' AS (a int, b short, c long, d byte, e float, f
double, g decimal(38,
FROM VALUES
('a','','1231a','a','213.21a','213.21a','0a.21d','2000-04-01123','1997-0102
00:00:') tmp(a, b, c, d, e, f, g, h, i)
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#xL, d#x, e#x, f#x, g#x, h#x, i#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#xL, d#x, e#x, f#x, g#x, h#x, i#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, d#x, e#x, f#x, g#x, h#x, i#x]
+- SubqueryAlias tmp
+- LocalRelation [a#x, b#x, c#x, d#x, e#x, f#x, g#x, h#x, i#x]
@@ -202,7 +202,7 @@ USING 'cat' AS (a, b)
FROM t
GROUP BY b
-- !query analysis
-ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Aggregate [b#x], [b#x, max(a#x) AS max(a)#x, sum(f#x) AS sum(f)#xL]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -215,7 +215,7 @@ ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None
-- !query
MAP a, b USING 'cat' AS (a, b) FROM t
-- !query analysis
-ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -228,7 +228,7 @@ ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None
-- !query
REDUCE a, b USING 'cat' AS (a, b) FROM t
-- !query analysis
-ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -253,7 +253,7 @@ FROM t
-- !query analysis
ScriptTransformation cat, [a#x, b#x, c#x, d#x],
ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,@),
(TOK_TABLEROWFORMATNULL,NULL), (TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,@), (TOK_TABLEROWFORMATNULL,NULL),
(TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, null AS NULL#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -278,7 +278,7 @@ FROM t
-- !query analysis
ScriptTransformation cat, [d#x],
ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,@),
(TOK_TABLEROWFORMATNULL,NULL), (TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,@), (TOK_TABLEROWFORMATNULL,NULL),
(TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, null AS NULL#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -302,7 +302,7 @@ FROM t
-- !query analysis
ScriptTransformation cat, [a#x, b#x, c#x, d#x],
ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,@),
(TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,@), (TOK_TABLEROWFORMATNULL,NULL),
(TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, null AS NULL#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -325,7 +325,7 @@ FROM t
-- !query analysis
ScriptTransformation cat, [a#x, b#x, c#x, d#x],
ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,@),
(TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,@), (TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, null AS NULL#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -349,7 +349,7 @@ FROM t
-- !query analysis
ScriptTransformation cat, [a#x, b#x, c#x, d#x],
ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,@),
(TOK_TABLEROWFORMATNULL,XXXX), (TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,@), (TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, null AS NULL#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -374,7 +374,7 @@ FROM t
ScriptTransformation cat, [a#x, b#x, c#x, d#x],
ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,@),
(TOK_TABLEROWFORMATNULL,
), (TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,@), (TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, null AS NULL#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x,
l#x])
@@ -415,7 +415,7 @@ Project [a#x, b#x, decode(c#x, UTF-8) AS decode(c,
UTF-8)#x, d#x, e#x, f#x, g#xL
+- SubqueryAlias tmp
+- ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x,
j#x, k#x, l#x], ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,,),
(TOK_TABLEROWFORMATNULL,NULL), (TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,,), (TOK_TABLEROWFORMATNULL,NULL),
(TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x, l#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x,
k#x, l#x])
@@ -456,7 +456,7 @@ Project [a#x, b#xL, decode(c#x, UTF-8) AS decode(c,
UTF-8)#x, d#x, e#x, f#x, g#x
+- SubqueryAlias tmp
+- ScriptTransformation cat, [a#x, b#xL, c#x, d#x, e#x, f#x, g#xL, h#x,
i#x, j#x, k#x, l#xL], ScriptInputOutputSchema(List((TOK_TABLEROWFORMATFIELD,,),
(TOK_TABLEROWFORMATNULL,NULL), (TOK_TABLEROWFORMATLINES,
)),List((TOK_TABLEROWFORMATFIELD,,), (TOK_TABLEROWFORMATNULL,NULL),
(TOK_TABLEROWFORMATLINES,
-)),None,None,List(),List(),None,None,false)
+)),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x, k#x, l#x]
+- SubqueryAlias t
+- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x, j#x,
k#x, l#x])
@@ -516,7 +516,7 @@ SELECT TRANSFORM(b, a, CAST(c AS STRING))
FROM script_trans
WHERE a <= 4
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [b#x, a#x, cast(c#x as string) AS c#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -533,7 +533,7 @@ SELECT TRANSFORM(1, 2, 3)
FROM script_trans
WHERE a <= 4
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [1 AS 1#x, 2 AS 2#x, 3 AS 3#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -552,7 +552,7 @@ LIMIT 1
-- !query analysis
GlobalLimit 1
+- LocalLimit 1
- +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [1 AS 1#x, 2 AS 2#x]
+- SubqueryAlias script_trans
+- View (`script_trans`, [a#x, b#x, c#x])
@@ -573,7 +573,7 @@ SELECT TRANSFORM(
FROM script_trans
WHERE a <= 4
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [b#x, a#x, CASE WHEN (c#x > 100) THEN 1 WHEN (c#x < 100) THEN 2
ELSE 3 END AS CASE WHEN (c > 100) THEN 1 WHEN (c < 100) THEN 2 ELSE 3 END#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -590,7 +590,7 @@ SELECT TRANSFORM(b, a, c + 1)
FROM script_trans
WHERE a <= 4
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [b#x, a#x, (c#x + 1) AS (c + 1)#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -607,7 +607,7 @@ SELECT TRANSFORM(*)
FROM script_trans
WHERE a <= 4
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x, c#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -625,7 +625,7 @@ FROM script_trans
WHERE a <= 4
GROUP BY b
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Aggregate [b#x], [b#x, max(a#x) AS max(a)#x, cast(sum(c#x) as string) AS
CAST(sum(c) AS STRING)#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -643,7 +643,7 @@ FROM script_trans
WHERE a <= 4
GROUP BY b
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Aggregate [b#x], [b#x, max(a#x) FILTER (WHERE (a#x > 3)) AS max(a) FILTER
(WHERE (a > 3))#x, cast(sum(c#x) as string) AS CAST(sum(c) AS STRING)#x]
+- Filter (a#x <= 4)
+- SubqueryAlias script_trans
@@ -661,7 +661,7 @@ FROM script_trans
WHERE a <= 2
GROUP BY b
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Aggregate [b#x], [b#x, max(a#x) AS max(a)#x, cast(sum(c#x) as string) AS
CAST(sum(c) AS STRING)#x]
+- Filter (a#x <= 2)
+- SubqueryAlias script_trans
@@ -680,7 +680,7 @@ WHERE a <= 4
GROUP BY b
HAVING MAX(a) > 0
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Filter (max(a)#x > 0)
+- Aggregate [b#x], [b#x, max(a#x) AS max(a)#x, cast(sum(c#x) as string) AS
CAST(sum(c) AS STRING)#x]
+- Filter (a#x <= 4)
@@ -700,7 +700,7 @@ WHERE a <= 4
GROUP BY b
HAVING MAX(a) > 1
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Filter (max(a)#x > 1)
+- Aggregate [b#x], [b#x, max(a#x) AS max(a)#x, cast(sum(c#x) as string) AS
CAST(sum(c) AS STRING)#x]
+- Filter (a#x <= 4)
@@ -719,7 +719,7 @@ FROM script_trans
WHERE a <= 4
WINDOW w AS (PARTITION BY b ORDER BY a)
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [b#x, max(a) OVER (PARTITION BY b ORDER BY a ASC NULLS FIRST RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#x, CAST(sum(c) OVER (PARTITION BY
b ORDER BY a ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
AS STRING)#x]
+- Project [b#x, a#x, c#x, max(a) OVER (PARTITION BY b ORDER BY a ASC NULLS
FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#x, _we1#xL, max(a)
OVER (PARTITION BY b ORDER BY a ASC NULLS FIRST RANGE BETWEEN UNBOUNDED
PRECEDING AND CURRENT ROW)#x, cast(_we1#xL as string) AS CAST(sum(c) OVER
(PARTITION BY b ORDER BY a ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW) AS STRING)#x]
+- Window [max(a#x) windowspecdefinition(b#x, a#x ASC NULLS FIRST,
specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS
max(a) OVER (PARTITION BY b ORDER BY a ASC NULLS FIRST RANGE BETWEEN UNBOUNDED
PRECEDING AND CURRENT ROW)#x, sum(c#x) windowspecdefinition(b#x, a#x ASC NULLS
FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$()))
AS _we1#xL], [b#x], [a#x ASC NULLS FIRST]
@@ -743,7 +743,7 @@ WHERE a <= 4
GROUP BY b, myCol, myCol2
HAVING max(a) > 1
-- !query analysis
-ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Filter (max(a)#x > 1)
+- Aggregate [b#x, myCol#x, myCol2#x], [b#x, max(a#x) AS max(a)#x,
cast(sum(c#x) as string) AS CAST(sum(c) AS STRING)#x, myCol#x, myCol2#x]
+- Filter (a#x <= 4)
@@ -767,7 +767,7 @@ SELECT a + 1
-- !query analysis
Project [(a#x + 1) AS (a + 1)#x]
+- SubqueryAlias t
- +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias script_trans
+- View (`script_trans`, [a#x, b#x, c#x])
@@ -790,7 +790,7 @@ ORDER BY result
Sort [result#xL ASC NULLS FIRST], true
+- Project [(cast(b#x as bigint) + cast(1 as bigint)) AS result#xL]
+- SubqueryAlias t
- +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Aggregate [a#x], [a#x, sum(b#x) AS sum(b)#xL]
+- SubqueryAlias script_trans
+- View (`script_trans`, [a#x, b#x, c#x])
@@ -803,7 +803,7 @@ Sort [result#xL ASC NULLS FIRST], true
-- !query
MAP k / 10 USING 'cat' AS (one) FROM (SELECT 10 AS k)
-- !query analysis
-ScriptTransformation cat, [one#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [one#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [(cast(k#x as double) / cast(10 as double)) AS (k / 10)#x]
+- SubqueryAlias __auto_generated_subquery_name
+- Project [10 AS k#x]
@@ -815,7 +815,7 @@ FROM (SELECT 1 AS key, 100 AS value) src
MAP src.*, src.key, CAST(src.key / 10 AS INT), CAST(src.key % 10 AS INT),
src.value
USING 'cat' AS (k, v, tkey, ten, one, tvalue)
-- !query analysis
-ScriptTransformation cat, [k#x, v#x, tkey#x, ten#x, one#x, tvalue#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [k#x, v#x, tkey#x, ten#x, one#x, tvalue#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [key#x, value#x, key#x, cast((cast(key#x as double) / cast(10 as
double)) as int) AS CAST((key / 10) AS INT)#x, cast((key#x % 10) as int) AS
CAST((key % 10) AS INT)#x, value#x]
+- SubqueryAlias src
+- Project [1 AS key#x, 100 AS value#x]
@@ -828,7 +828,7 @@ SELECT TRANSFORM(1)
FROM script_trans
HAVING true
-- !query analysis
-ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Filter cast(true as boolean)
+- Aggregate [1 AS 1#x]
+- SubqueryAlias script_trans
@@ -851,7 +851,7 @@ SELECT TRANSFORM(1)
FROM script_trans
HAVING true
-- !query analysis
-ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Filter cast(true as boolean)
+- Project [1 AS 1#x]
+- SubqueryAlias script_trans
@@ -877,7 +877,7 @@ SELECT t1.b FROM temp t1 JOIN temp t2 ON t1.b = t2.b
WithCTE
:- CTERelationDef xxxx, false
: +- SubqueryAlias temp
-: +- ScriptTransformation cat, [b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+: +- ScriptTransformation cat, [b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
: +- Project [a#x]
: +- SubqueryAlias t
: +- View (`t`, [a#x, b#x, c#x, d#x, e#x, f#x, g#xL, h#x, i#x,
j#x, k#x, l#x])
@@ -999,12 +999,12 @@ FROM (
SELECT TRANSFORM(a, b)
USING 'cat' AS (a, b)
-- !query analysis
-ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias map_output
+- Sort [a#x ASC NULLS FIRST], false
+- RepartitionByExpression [a#x]
- +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias complex_trans
+- View (`complex_trans`, [a#x, b#x])
@@ -1024,11 +1024,11 @@ FROM (
SELECT TRANSFORM(a, b)
USING 'cat' AS (a, b)
-- !query analysis
-ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias map_output
+- Sort [a#x ASC NULLS FIRST], true
- +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias complex_trans
+- View (`complex_trans`, [a#x, b#x])
@@ -1043,7 +1043,7 @@ SELECT TRANSFORM (a, b)
USING 'cat' AS (a CHAR(10), b VARCHAR(10))
FROM VALUES('apache', 'spark') t(a, b)
-- !query analysis
-ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,false,false)
+- Project [a#x, b#x]
+- SubqueryAlias t
+- LocalRelation [a#x, b#x]
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala
index 62c5f5631776..f7fb05526d96 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala
@@ -26,7 +26,7 @@ import org.json4s.jackson.JsonMethods._
import org.scalatest.Assertions._
import org.scalatest.exceptions.TestFailedException
-import org.apache.spark.{SparkException, TaskContext, TestUtils}
+import org.apache.spark.{SparkException, SparkThrowable, TaskContext,
TestUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{QueryTest, Row}
import org.apache.spark.sql.catalyst.InternalRow
@@ -70,6 +70,23 @@ abstract class BaseScriptTransformationSuite extends
QueryTest {
child: SparkPlan,
ioschema: ScriptTransformationIOSchema): BaseScriptTransformationExec
+ protected def checkTransformWithoutSerdeUnsupportedType(
+ exception: Throwable,
+ sqlType: String): Unit = {
+ val err = Iterator.iterate[Throwable](exception)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case t: SparkThrowable
+ if t.getCondition ==
"UNSUPPORTED_FEATURE.TRANSFORM_WITHOUT_SERDE_TYPE" => t
+ }.getOrElse {
+ fail(s"expected TRANSFORM_WITHOUT_SERDE_TYPE, got $exception")
+ }
+ checkError(
+ exception = err,
+ condition = "UNSUPPORTED_FEATURE.TRANSFORM_WITHOUT_SERDE_TYPE",
+ parameters = Map("dataType" -> sqlType))
+ }
+
test("cat without SerDe") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
@@ -86,6 +103,362 @@ 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/VARCHAR null token without SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val charInput = Seq[String](null, "ab").toDF("c")
+ checkAnswer(
+ charInput,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("c", CharType(4))()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(Row(null), Row("ab ")))
+ val varcharInput = Seq[String](null, "xy").toDF("v")
+ checkAnswer(
+ varcharInput,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("v", VarcharType(5))()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(Row(null), Row("xy")))
+ }
+ 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 "))),
+ ("""[null]""", ArrayType(CharType(4)), Row(Seq(null))),
+ (
+ """{"1":null}""",
+ MapType(IntegerType, CharType(4)),
+ Row(Map(1 -> null))),
+ (
+ """{"value":null}""",
+ StructType(Seq(StructField("value", CharType(5)))),
+ Row(Row(null))),
+ (
+ """{"ab":1}""",
+ MapType(CharType(4), IntegerType),
+ Row(Map("ab " -> 1)))).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").coalesce(1)
+ checkAnswer(
+ input,
+ (child: SparkPlan) => createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("value", mapType)()),
+ child = child,
+ ioschema = defaultIOSchema),
+ Seq(Row(null), Row(Map(2 -> "cd "))))
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-59277: colliding CHAR map keys without SerDe follow
MAP_KEY_DEDUP_POLICY") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ val mapType = MapType(CharType(4), IntegerType)
+ val json = """{"a":1,"a ":2}"""
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val input = Seq(json).toDF("value")
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("value", mapType)()),
+ child = input.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema),
+ spark.sqlContext)
+ }
+ val runtimeException = Iterator.iterate[Throwable](exception)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case s: org.apache.spark.SparkRuntimeException
+ if s.getCondition == "DUPLICATED_MAP_KEY" => s
+ }.getOrElse {
+ fail(s"expected DUPLICATED_MAP_KEY, got $exception")
+ }
+ checkError(
+ exception = runtimeException,
+ condition = "DUPLICATED_MAP_KEY",
+ parameters = Map(
+ "key" -> "a ",
+ "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-59683: preserve-only CHAR/VARCHAR without SerDe stays
unsupported") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val input = Seq("ab").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)
+ }
+ checkTransformWithoutSerdeUnsupportedType(exception, "\"CHAR(4)\"")
+ }
+ }
+
+ test("SPARK-59683: bound CHAR/VARCHAR mode survives execution conf change") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ val padded = Seq("ab").toDF("c")
+ val enabledPlan = withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key
-> "true") {
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("c", CharType(4))()),
+ child = padded.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema)
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ assert(QueryTest.executePlan(enabledPlan, spark.sqlContext) ===
Seq(Row("ab ")))
+ }
+
+ val overflow = Seq("abcdef").toDF("c")
+ val overflowPlan = withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key
-> "true") {
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("c", CharType(4))()),
+ child = overflow.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema)
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(overflowPlan, 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"))
+ }
+
+ val disabledPlan = withSQLConf(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(AttributeReference("c", CharType(4))()),
+ child = padded.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema)
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(disabledPlan, spark.sqlContext)
+ }
+ checkTransformWithoutSerdeUnsupportedType(exception, "\"CHAR(4)\"")
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
test("script transformation should not swallow errors from upstream
operators (no serde)") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala
index dbb8e9697089..692302e3af36 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.execution
import org.apache.spark.TestUtils
+import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class SparkScriptTransformationSuite extends BaseScriptTransformationSuite
with SharedSparkSession {
@@ -40,6 +42,60 @@ class SparkScriptTransformationSuite extends
BaseScriptTransformationSuite with
)
}
+ test("SPARK-59683: TRANSFORM view keeps bound CHAR/VARCHAR mode") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ withView("v") {
+ sql(
+ """CREATE VIEW v AS
+ |SELECT TRANSFORM('ab') USING 'cat' AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)""".stripMargin)
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ checkAnswer(sql("SELECT * FROM v"), Row("ab "))
+ }
+ }
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ withView("v_overflow") {
+ sql(
+ """CREATE VIEW v_overflow AS
+ |SELECT TRANSFORM('abcdef') USING 'cat' AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)""".stripMargin)
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val exception = intercept[Exception] {
+ sql("SELECT * FROM v_overflow").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"))
+ }
+ }
+ }
+ withSQLConf(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ withView("v_disabled") {
+ sql(
+ """CREATE VIEW v_disabled AS
+ |SELECT TRANSFORM('ab') USING 'cat' AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)""".stripMargin)
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val exception = intercept[Exception] {
+ sql("SELECT * FROM v_disabled").collect()
+ }
+ checkTransformWithoutSerdeUnsupportedType(exception, "\"CHAR(4)\"")
+ }
+ }
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
test("SPARK-32106: TRANSFORM with serde without hive should throw
exception") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
withTempView("v") {
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
index b8758407877f..1eb4b48e8338 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
@@ -942,6 +942,24 @@ private[hive] trait HiveInspectors {
unwrapperFor(objectInspector)
}
+ /**
+ * In-place unwrapper that uses DataType-aware conversion when the target
+ * contains CHAR/VARCHAR or nanosecond timestamps. Other fields keep the
+ * specialized primitive setters from the field-only overload.
+ */
+ def unwrapperFor(
+ field: HiveStructField,
+ dataType: DataType): (Any, InternalRow, Int) => Unit = {
+ if (CharVarcharUtils.hasCharVarchar(dataType) ||
+ dataType.existsRecursively(_.isInstanceOf[AnyTimestampNanoType])) {
+ val unwrapper = unwrapperFor(field.getFieldObjectInspector, dataType)
+ (value: Any, row: InternalRow, ordinal: Int) =>
+ row.update(ordinal, unwrapper(value))
+ } else {
+ unwrapperFor(field)
+ }
+ }
+
/**
* Builds unwrappers ahead of time according to object inspector
* types to avoid pattern matching and branching costs per row.
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
index de2d15415837..c53949889611 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
@@ -27,6 +27,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hive.ql.exec.{RecordReader, RecordWriter}
import org.apache.hadoop.hive.serde.serdeConstants
import org.apache.hadoop.hive.serde2.AbstractSerDe
+import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe
import org.apache.hadoop.hive.serde2.objectinspector._
import org.apache.hadoop.io.Writable
@@ -36,7 +37,7 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.execution._
import org.apache.spark.sql.hive.HiveInspectors
import org.apache.spark.sql.hive.HiveShim._
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types._
import org.apache.spark.util.{CircularBuffer, Utils}
/**
@@ -75,7 +76,9 @@ private[hive] case class HiveScriptTransformationExec(
val mutableRow = new SpecificInternalRow(output.map(_.dataType))
@transient
- lazy val unwrappers =
outputSoi.getAllStructFieldRefs.asScala.map(unwrapperFor)
+ lazy val unwrappers =
outputSoi.getAllStructFieldRefs.asScala.zip(output).map {
+ case (field, attr) => unwrapperFor(field, attr.dataType)
+ }
override def hasNext: Boolean = {
if (completed) {
@@ -256,7 +259,9 @@ object HiveScriptIOSchema extends HiveInspectors {
output: Seq[Attribute]): Option[(AbstractSerDe, StructObjectInspector)]
= {
ioschema.outputSerdeClass.map { serdeClass =>
val (columns, columnTypes) = parseAttrs(output)
- val serde = initSerDe(serdeClass, columns, columnTypes,
ioschema.outputSerdeProps)
+ val serdeTypes = outputTypesForSerDe(
+ serdeClass, columnTypes, ioschema.standardCharVarcharSemantics)
+ val serde = initSerDe(serdeClass, columns, serdeTypes,
ioschema.outputSerdeProps)
val structObjectInspector =
serde.getObjectInspector().asInstanceOf[StructObjectInspector]
(serde, structObjectInspector)
}
@@ -268,6 +273,25 @@ object HiveScriptIOSchema extends HiveInspectors {
(columns, columnTypes)
}
+ /**
+ * Hive LazySimpleSerDe CHAR/VARCHAR types truncate on deserialize. Under
standard
+ * semantics, map them to STRING so Spark applies length checks. Only
LazySimpleSerDe
+ * and subclasses get this rewrite; other SerDes keep the declared schema.
+ * preserveCharVarcharTypeInfo without standard semantics is unchanged.
+ */
+ private def outputTypesForSerDe(
+ serdeClassName: String,
+ columnTypes: Seq[DataType],
+ standardCharVarcharSemantics: Boolean): Seq[DataType] = {
+ val serdeClass = Utils.classForName[AbstractSerDe](serdeClassName)
+ if (standardCharVarcharSemantics &&
+ classOf[LazySimpleSerDe].isAssignableFrom(serdeClass)) {
+ columnTypes.map(ScriptTransformationIOSchema.toUnboundedStringType)
+ } else {
+ columnTypes
+ }
+ }
+
def initSerDe(
serdeClassName: String,
columns: Seq[String],
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationSuite.scala
index b1ff05b8c1b0..a4df01db29f2 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationSuite.scala
@@ -20,8 +20,15 @@ package org.apache.spark.sql.hive.execution
import java.sql.Timestamp
import java.time.{Duration, Period}
import java.time.temporal.ChronoUnit
+import java.util.{Arrays, Properties}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.hive.serde.serdeConstants
+import org.apache.hadoop.hive.serde2.{AbstractSerDe, SerDeStats}
import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe
+import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector,
ObjectInspectorFactory}
+import
org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory
+import org.apache.hadoop.io.{Text, Writable}
import org.scalatest.exceptions.TestFailedException
import org.apache.spark.{SparkException, TestUtils}
@@ -31,6 +38,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeConstants
import org.apache.spark.sql.execution._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.types.DayTimeIntervalType._
import org.apache.spark.sql.types.YearMonthIntervalType._
@@ -59,7 +67,7 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
)
}
- private val hiveIOSchema: ScriptTransformationIOSchema = {
+ private def hiveIOSchema: ScriptTransformationIOSchema = {
defaultIOSchema.copy(
inputSerdeClass = Some(classOf[LazySimpleSerDe].getCanonicalName),
outputSerdeClass = Some(classOf[LazySimpleSerDe].getCanonicalName)
@@ -372,6 +380,227 @@ 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)))())
+ val (_, soi) = HiveScriptIOSchema.initOutputSerDe(hiveIOSchema,
output).get
+
assert(soi.getAllStructFieldRefs.get(0).getFieldObjectInspector.getTypeName ===
+ "char(4)")
+
assert(soi.getAllStructFieldRefs.get(1).getFieldObjectInspector.getTypeName ===
+ "varchar(5)")
+
assert(soi.getAllStructFieldRefs.get(2).getFieldObjectInspector.getTypeName ===
+ "array<char(4)>")
+ }
+ }
+
+ test("SPARK-59683: bound Hive SerDe CHAR/VARCHAR mode survives conf change")
{
+ val output = Seq(
+ AttributeReference("c", CharType(4))(),
+ AttributeReference("v", VarcharType(5))(),
+ AttributeReference("nested", ArrayType(CharType(4)))())
+ val enabledSchema =
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ hiveIOSchema
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val (_, soi) = HiveScriptIOSchema.initOutputSerDe(enabledSchema,
output).get
+
assert(soi.getAllStructFieldRefs.get(0).getFieldObjectInspector.getTypeName ===
+ "string")
+
assert(soi.getAllStructFieldRefs.get(1).getFieldObjectInspector.getTypeName ===
+ "string")
+
assert(soi.getAllStructFieldRefs.get(2).getFieldObjectInspector.getTypeName ===
+ "array<string>")
+ }
+ val disabledSchema = withSQLConf(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ hiveIOSchema
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val (_, soi) = HiveScriptIOSchema.initOutputSerDe(disabledSchema,
output).get
+
assert(soi.getAllStructFieldRefs.get(0).getFieldObjectInspector.getTypeName ===
+ "char(4)")
+
assert(soi.getAllStructFieldRefs.get(1).getFieldObjectInspector.getTypeName ===
+ "varchar(5)")
+
assert(soi.getAllStructFieldRefs.get(2).getFieldObjectInspector.getTypeName ===
+ "array<char(4)>")
+ }
+ }
+
+ test("SPARK-59683: Hive TRANSFORM view keeps bound CHAR/VARCHAR mode") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ withView("v") {
+ sql(
+ """CREATE VIEW v AS
+ |SELECT TRANSFORM('ab') USING 'cat' AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)""".stripMargin)
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ checkAnswer(sql("SELECT * FROM v"), Row("ab "))
+ }
+ }
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ withView("v_overflow") {
+ sql(
+ """CREATE VIEW v_overflow AS
+ |SELECT TRANSFORM('abcdef') USING 'cat' AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)""".stripMargin)
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val exception = intercept[Exception] {
+ sql("SELECT * FROM v_overflow").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-32400: TRANSFORM doesn't support
CalendarIntervalType/UserDefinedType (hive serde)") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
withTempView("v") {
@@ -668,3 +897,29 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
}
}
}
+
+class TestLazySimpleSerDe extends LazySimpleSerDe
+
+class SchemaCapturingSerDe extends AbstractSerDe {
+ override def initialize(conf: Configuration, tbl: Properties): Unit = {
+ SchemaCapturingSerDe.lastColumnTypes =
+ tbl.getProperty(serdeConstants.LIST_COLUMN_TYPES)
+ }
+
+ override def getObjectInspector: ObjectInspector =
+ ObjectInspectorFactory.getStandardStructObjectInspector(
+ Arrays.asList("col"),
+ Arrays.asList(PrimitiveObjectInspectorFactory.javaStringObjectInspector))
+
+ override def getSerializedClass: Class[_ <: Writable] = classOf[Text]
+
+ override def getSerDeStats: SerDeStats = null
+
+ override def serialize(obj: Any, inspector: ObjectInspector): Writable = null
+
+ override def deserialize(blob: Writable): AnyRef = null
+}
+
+object SchemaCapturingSerDe {
+ @volatile var lastColumnTypes: String = _
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
