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 bd2c46622a6d [SPARK-60090][SQL][HIVE] Apply CHAR/VARCHAR assignment
after TRANSFORM via Project
bd2c46622a6d is described below
commit bd2c46622a6d7eace772686028e18a71aab334f1
Author: Serge Rielau <[email protected]>
AuthorDate: Sun Oct 11 20:58:33 2026 -0700
[SPARK-60090][SQL][HIVE] Apply CHAR/VARCHAR assignment after TRANSFORM via
Project
### What changes were proposed in this pull request?
Keep Hive `TRANSFORM` / `MAP` / `REDUCE` script I/O as unbounded `STRING`
(JSON for complex types), the same as sources that have no native CHAR/VARCHAR.
After the script reader, apply Spark CHAR/VARCHAR assignment
(`CharVarcharUtils.stringLengthCheck`: pad `CHAR`, `EXCEED_LIMIT_LENGTH` on
overflow) in a `Project` so the query result type is the declared
`CHAR`/`VARCHAR` (including nested types).
This supersedes the SPARK-59683 codec inside the script/Hive path (that
codec is on master / branch-4.x only; it was never released in 4.2 or 4.3):
- Recursively replace CHAR/VARCHAR with unbounded STRING for script exec
output (including nested maps/arrays/structs).
- Parse delimited / JSON as today for STRING-shaped types.
- When `spark.sql.charVarchar.standardSemantics.enabled` or
`spark.sql.preserveCharVarcharTypeInfo` is true, wrap with a Project using
`stringLengthCheck` so the result type matches `AS (...)`.
- Both configs off: keep today's STRING result (`AS (c CHAR(5))` remains
STRING).
- Remove CHAR-specific logic from no-SerDe converters, Hive unwrappers,
SerDe schema rewrite, and the bind-at-parse I/O-schema flag.
Nullable struct assignment uses `CaseWhen` rather than `If`, so
`CollationTypeCoercion` does not widen collated CHAR/VARCHAR inside the struct
to unbounded STRING.
Assignment lives in a Project, so column pruning can skip an unreferenced
CHAR/VARCHAR column (including `SELECT count(*)` over overflowing output),
matching scan-side checks.
JSON non-string map keys (`MAP<INT, ...>`, `MAP<DATE, ...>`) remain out of
scope; `MAP<INT, CHAR(n)>` without a SerDe fails the same way as `MAP<INT,
STRING>` (`ClassCastException` on the JSON key). Nested field `COMMENT` / `NOT
NULL` inside a rebuilt struct is the same limitation as scan-side
`stringLengthCheck`.
A first-class CHAR/VARCHAR TRANSFORM column compared with a STRING literal
is analyzed as `cast(CHAR as STRING) = lit`, so a short literal does not match
the padded CHAR value. That matches CHAR vs STRING under standard semantics; it
is not scan-style metadata padding of the literal.
JIRA: https://issues.apache.org/jira/browse/SPARK-60090
Parent: SPARK-59683 / PR #58936
### Why are the changes needed?
SPARK-59683 mixed CHAR/VARCHAR assignment (pad / length, same as
CSV/JSON/Parquet) with TRANSFORM JSON / Hive wire issues that are not
CHAR-specific. That made `MAP<STRING, STRING>` vs `MAP<STRING, CHAR(2)>`
diverge. TRANSFORM is Hive-compat STRING-on-the-wire; assignment belongs in a
Spark `Project` like any other STRING source.
### Does this PR introduce _any_ user-facing change?
Yes.
On the unreleased `spark.sql.charVarchar.standardSemantics.enabled` path,
and on the **released** `spark.sql.preserveCharVarcharTypeInfo` path (since
4.0.0):
- `SELECT TRANSFORM(...) USING 'cat' AS (c CHAR(4))` returns typed
`CHAR(4)` with padding.
- Overflow raises `EXCEED_LIMIT_LENGTH`.
- Nested `ARRAY` / `STRUCT` / `MAP<CHAR, VARCHAR>` go through the same
Project assignment.
On branch-4.3 with only `spark.sql.preserveCharVarcharTypeInfo` set, scalar
CHAR/VARCHAR output without a SerDe failed with the unsupported-type error,
nested CHAR/VARCHAR output without a SerDe kept the declared type with no
padding or length check, and Hive SerDe deserialized `char(n)`/`varchar(n)`
with Hive truncation. None of the preserve-only no-SerDe path returned STRING.
When both configs are false, `AS (c CHAR(n))` remains STRING with no pad and no
length check.
Compared with the unreleased SPARK-59683 CHAR-in-the-exec codec (master /
branch-4.x only), no-SerDe nested CHAR is STRING-on-the-wire plus a Project
instead of CHAR inside the reader.
### How was this patch tested?
- `sql/testOnly
org.apache.spark.sql.execution.SparkScriptTransformationSuite -- -z SPARK-60090`
- `hive/testOnly
org.apache.spark.sql.hive.execution.HiveScriptTransformationSuite -- -z
SPARK-60090`
- `catalyst/testOnly org.apache.spark.sql.catalyst.parser.PlanParserSuite
-- -z TRANSFORM`
Coverage includes scalar pad/overflow/null token, mixed CHAR and non-CHAR
columns, CHAR vs STRING comparison (no literal padding), nested
ARRAY/STRUCT/MAP assignment and schema assertions, collated nested STRUCT,
quoted struct field names, MAP CHAR key collision, malformed JSON as null,
flag-off STRING results, preserve-only assignment, Hive SerDe through the same
Spark-side Project, views (PERSISTED vs stored-schema cast), pruning of
unreferenced overflow, and hand-built leftover CHA [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #59302 from srielau/SPARK-60090.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
(cherry picked from commit b4e88aa1f11994c941f0028cb4dc68e12ff042b6)
Signed-off-by: Daniel Tenedorio <[email protected]>
---
docs/sql-migration-guide.md | 1 +
.../sql/catalyst/util/SparkCharVarcharUtils.scala | 33 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 44 +-
.../plans/logical/ScriptTransformation.scala | 4 +-
.../spark/sql/catalyst/util/CharVarcharUtils.scala | 41 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 4 +-
.../execution/BaseScriptTransformationExec.scala | 187 +--------
.../spark/sql/execution/SparkSqlParser.scala | 3 +-
.../sql-tests/analyzer-results/transform.sql.out | 86 ++--
.../apache/spark/sql/CharVarcharTestSuite.scala | 4 +-
.../execution/BaseScriptTransformationSuite.scala | 457 +++++++--------------
.../execution/SparkScriptTransformationSuite.scala | 184 ++++++++-
.../org/apache/spark/sql/hive/HiveInspectors.scala | 18 -
.../execution/HiveScriptTransformationExec.scala | 30 +-
.../execution/HiveScriptTransformationSuite.scala | 229 ++---------
15 files changed, 506 insertions(+), 819 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 4af273dcfe68..7b49f4881a26 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -26,6 +26,7 @@ license: |
- Since Spark 4.4, when `spark.sql.preserveCharVarcharTypeInfo` is true and
`spark.sql.charVarchar.standardSemantics.enabled` is false, ORC reads that
apply a CHAR/VARCHAR schema over STRING storage return the stored values
without ORC truncation, matching Parquet. Previously the ORC reader requested
`char(n)`/`varchar(n)` and truncated STRING-stored values to `n`. Read-side
length checks (`EXCEED_LIMIT_LENGTH`) apply only when
`spark.sql.charVarchar.standardSemantics.enabled` is true.
- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is
true, XML names used as `MAP<CHAR(n), _>` or `MAP<VARCHAR(n), _>` keys in
`from_xml` and the XML datasource are length-checked without padding or
trimming. The checked names are element names, `attributePrefix` plus attribute
names (default prefix `_`), and the `valueTag` (default `_VALUE`) when mixed
text is present. A `CHAR(n)` key must already be exactly `n` characters, and a
`VARCHAR(n)` key must already be [...]
+- Since Spark 4.4, Hive `TRANSFORM` / `MAP` / `REDUCE` with CHAR/VARCHAR `AS`
types keeps script I/O as unbounded STRING and applies Spark assignment (CHAR
padding and `EXCEED_LIMIT_LENGTH` on overflow) in a `Project` when
`spark.sql.charVarchar.standardSemantics.enabled` or
`spark.sql.preserveCharVarcharTypeInfo` is true. Previously, with only
`spark.sql.preserveCharVarcharTypeInfo` set, scalar CHAR/VARCHAR output without
a SerDe failed with an unsupported-type error, nested CHAR/VARCHA [...]
- Since Spark 4.4, the options maps passed to `from_csv`, `to_csv`,
`schema_of_csv`, `from_json`, `to_json`, `schema_of_json`, `from_xml`,
`to_xml`, and `schema_of_xml` must be foldable after replacing
`RuntimeReplaceable` expressions. Previously, Spark evaluated non-foldable
options during analysis, which allowed some constant expressions but could fail
with an internal error or incorrectly evaluate row-dependent expressions. To
allow deterministic and row-independent non-foldable optio [...]
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed
after a compatible schema change, connectors can return more data columns from
the current table schema in `Scan.readSchema()` than requested by
`SupportsPushDownRequiredColumns.pruneColumns`. Previously, this partial
pruning could fail planning because the scan reported columns absent from the
analyzed relation output.
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed,
if rebinding an expanded struct used as a map key causes distinct current keys
to collide under the analyzed schema, the query fails with `DUPLICATED_MAP_KEY`
by default instead of returning duplicate keys. With
`spark.sql.mapKeyDedupPolicy=LAST_WIN`, the last value is kept.
diff --git
a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
index 70684d392042..c0b54f4c7e6a 100644
---
a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
+++
b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
@@ -55,20 +55,37 @@ trait SparkCharVarcharUtils {
}
/**
- * Replaces CharType/VarcharType with StringType recursively in the given
data type.
+ * Replaces CharType/VarcharType with StringType recursively in the given
data type. First-class
+ * CHAR/VARCHAR (standard semantics or preserveCharVarcharTypeInfo) is left
as-is.
*/
- def replaceCharVarcharWithString(dt: DataType): DataType = dt match {
+ def replaceCharVarcharWithString(dt: DataType): DataType = {
+ if (SqlApiConf.get.charVarcharFirstClassTypes) {
+ dt
+ } else {
+ replaceCharVarcharWithStringAlways(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.
+ */
+ private[sql] def replaceCharVarcharWithStringAlways(dt: DataType): DataType
= dt match {
case ArrayType(et, nullable) =>
- ArrayType(replaceCharVarcharWithString(et), nullable)
+ ArrayType(replaceCharVarcharWithStringAlways(et), nullable)
case MapType(kt, vt, nullable) =>
- MapType(replaceCharVarcharWithString(kt),
replaceCharVarcharWithString(vt), nullable)
+ MapType(
+ replaceCharVarcharWithStringAlways(kt),
+ replaceCharVarcharWithStringAlways(vt),
+ nullable)
case StructType(fields) =>
StructType(fields.map { field =>
- field.copy(dataType = replaceCharVarcharWithString(field.dataType))
+ field.copy(dataType =
replaceCharVarcharWithStringAlways(field.dataType))
})
- case c: CharType if !SqlApiConf.get.charVarcharFirstClassTypes =>
c.toStringType
- case v: VarcharType if !SqlApiConf.get.charVarcharFirstClassTypes =>
v.toStringType
- case _ => dt
+ case c: CharType => c.toStringType
+ case v: VarcharType => v.toStringType
+ case other => other
}
}
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 3f0a83c51c46..84f4bd9ef25c 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
@@ -1814,21 +1814,23 @@ class AstBuilder extends DataTypeAstBuilder
if (transformClause.setQuantifier != null) {
throw
QueryParsingErrors.transformNotSupportQuantifierError(transformClause.setQuantifier)
}
- // Create the attributes.
- val (attributes, schemaLess) = if (transformClause.colTypeList != null) {
+ // Create the attributes. Script I/O is always unbounded STRING (Hive wire
format).
+ // Declared CHAR/VARCHAR types are assigned in a Project below when
first-class types
+ // are enabled, using the original schema so quoted field names are not
re-parsed.
+ val (attributes, schema, schemaLess) = if (transformClause.colTypeList !=
null) {
// Typed return columns.
- val schema = createSchema(transformClause.colTypeList)
- val replacedSchema =
CharVarcharUtils.replaceCharVarcharWithStringInSchema(schema)
- (DataTypeUtils.toAttributes(replacedSchema), false)
+ val origSchema = createSchema(transformClause.colTypeList)
+ val replacedSchema =
CharVarcharUtils.replaceCharVarcharWithStringInSchemaAlways(origSchema)
+ (DataTypeUtils.toAttributes(replacedSchema), origSchema, false)
} else if (transformClause.identifierSeq != null) {
// Untyped return columns.
val attrs = visitIdentifierSeq(transformClause.identifierSeq).map { name
=>
AttributeReference(name, StringType, nullable = true)()
}
- (attrs, false)
+ (attrs, StructType(Nil), false)
} else {
(Seq(AttributeReference("key", StringType)(),
- AttributeReference("value", StringType)()), true)
+ AttributeReference("value", StringType)()), StructType(Nil), true)
}
val plan = visitCommonSelectQueryClausePlan(
@@ -1843,7 +1845,7 @@ class AstBuilder extends DataTypeAstBuilder
isDistinct = false,
isPipeOperatorSelect = false)
- ScriptTransformation(
+ val scriptTransform = ScriptTransformation(
string(visitStringLit(transformClause.script)),
attributes,
plan,
@@ -1856,6 +1858,29 @@ class AstBuilder extends DataTypeAstBuilder
schemaLess
)
)
+ // First-class CHAR/VARCHAR (standard semantics or
preserveCharVarcharTypeInfo):
+ // assign declared types in a Project so the result matches AS (...). Flag
off keeps
+ // the historical STRING result with no pad and no EXCEED_LIMIT_LENGTH.
The check can
+ // be pruned like scan-side assignment when the column is unreferenced.
+ if (conf.charVarcharFirstClassTypes &&
+ schema.exists(f => CharVarcharUtils.hasCharVarchar(f.dataType))) {
+ Project(
+ attributes.zip(schema).map { case (a, field) =>
+ if (CharVarcharUtils.hasCharVarchar(field.dataType)) {
+ // Keep the raw CHAR/VARCHAR metadata, matching scan-side
+ // readSidePadding. First-class CHAR vs STRING still promotes to
+ // STRING equality, so this metadata does not pad the literal.
+ Alias(
+ CharVarcharUtils.stringLengthCheck(a, field.dataType),
+ a.name)(explicitMetadata = Some(a.metadata))
+ } else {
+ a
+ }
+ },
+ scriptTransform)
+ } else {
+ scriptTransform
+ }
}
/**
@@ -2050,8 +2075,7 @@ class AstBuilder extends DataTypeAstBuilder
inSerdeClass, outSerdeClass,
inSerdeProps, outSerdeProps,
reader, writer,
- schemaLess,
- conf.charVarcharStandardSemantics)
+ schemaLess)
}
/**
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 5532803eafc5..3a2cb2cd7309 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,6 +49,4 @@ case class ScriptInputOutputSchema(
outputSerdeProps: Seq[(String, String)],
recordReaderClass: Option[String],
recordWriterClass: Option[String],
- schemaLess: Boolean,
- // Bound while the script is parsed under the direct-query or
persisted-view SQLConf.
- standardCharVarcharSemantics: Boolean = false)
+ schemaLess: Boolean)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
index dc628ac5551d..4ab621e672df 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
@@ -63,11 +63,27 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
* re-construct the original data type with CharType/VarcharType later when
needed.
*/
def replaceCharVarcharWithStringInSchema(st: StructType): StructType = {
+ replaceCharVarcharWithStringInSchema(st, replaceCharVarcharWithString)
+ }
+
+ /**
+ * Like [[replaceCharVarcharWithStringInSchema]], but always rewrites the
physical type to
+ * unbounded STRING even when first-class CHAR/VARCHAR is enabled. Hive
TRANSFORM script I/O
+ * is STRING-on-the-wire; callers apply [[stringLengthCheck]] in a Project
when first-class
+ * CHAR/VARCHAR types are enabled.
+ */
+ def replaceCharVarcharWithStringInSchemaAlways(st: StructType): StructType =
{
+ replaceCharVarcharWithStringInSchema(st,
replaceCharVarcharWithStringAlways)
+ }
+
+ private def replaceCharVarcharWithStringInSchema(
+ st: StructType,
+ replace: DataType => DataType): StructType = {
StructType(st.map { field =>
if (hasCharVarchar(field.dataType)) {
val metadata = new MetadataBuilder().withMetadata(field.metadata)
.putString(CHAR_VARCHAR_TYPE_STRING_METADATA_KEY,
field.dataType.catalogString).build()
- field.copy(dataType = replaceCharVarcharWithString(field.dataType),
metadata = metadata)
+ field.copy(dataType = replace(field.dataType), metadata = metadata)
} else {
field
}
@@ -258,10 +274,8 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
if (struct.valExprs.forall(_.isInstanceOf[GetStructField])) {
// No field needs char/varchar processing, just return the original
expression.
expr
- } else if (expr.nullable) {
- If(IsNull(expr), Literal(null, struct.dataType), struct)
} else {
- struct
+ nullSafeRebuiltStruct(expr, struct)
}
case ArrayType(et, containsNull) =>
@@ -285,6 +299,19 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
}
}
+ /**
+ * Guard a rebuilt named_struct so a NULL input stays NULL. Use CaseWhen
rather than If:
+ * CollationTypeCoercion rewrites If (no same-type skip) and can widen
VARCHAR collations
+ * inside the struct to unbounded STRING. CaseWhen skips that when the
branches already match.
+ */
+ private def nullSafeRebuiltStruct(expr: Expression, struct:
CreateNamedStruct): Expression = {
+ if (expr.nullable) {
+ CaseWhen(Seq((IsNull(expr), Literal(null, struct.dataType))),
Some(struct))
+ } else {
+ struct
+ }
+ }
+
private def processStringForCharVarcharInArray(
arr: Expression,
et: DataType,
@@ -391,11 +418,7 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
// GetStructField on a NULL struct yields NULL fields, which
CreateNamedStruct would
// turn into a non-NULL struct of NULLs. Guard it the same way the
scan-side rewrite
// in processStringForCharVarchar does.
- Some(if (expr.nullable) {
- If(IsNull(expr), Literal(null, struct.dataType), struct)
- } else {
- struct
- })
+ Some(nullSafeRebuiltStruct(expr, struct))
} else {
None
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index e05931e39143..933417a98188 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -7682,7 +7682,9 @@ object SQLConf {
"for SQL standard CHAR/VARCHAR behavior (CAST/LCT/STRING-returning
transforms). " +
"ORC reads with a CHAR/VARCHAR schema over STRING storage return the
stored values " +
"without ORC truncation, matching Parquet. Read-side length checks apply
only when " +
- "spark.sql.charVarchar.standardSemantics.enabled is true.")
+ "spark.sql.charVarchar.standardSemantics.enabled is true. Hive
TRANSFORM/MAP/REDUCE " +
+ "with CHAR/VARCHAR AS types still applies Spark assignment (pad /
EXCEED_LIMIT_LENGTH) " +
+ "when this flag is true, because script I/O is STRING-on-the-wire.")
.version("4.0.0")
.booleanConf
.createWithDefault(false)
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 535db0d82fec..98d775102043 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,28 +31,10 @@ 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,
- BoundReference,
- Cast,
- Expression,
- GenericInternalRow,
- JsonToStructs,
- Literal,
- StructsToJson,
- UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet,
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.{
- ArrayBasedMapBuilder,
- ArrayData,
- CharVarcharCodegenUtils,
- CharVarcharUtils,
- DateTimeUtils,
- GenericArrayData,
- IntervalUtils,
- MapData}
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeUtils,
IntervalUtils}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -80,11 +62,6 @@ 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())
@@ -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 {
- 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)
- }
+ // CHAR/VARCHAR must not reach the script reader; AstBuilder rewrites
them to
+ // unbounded STRING and applies assignment in a Project when first-class
types
+ // are enabled. Reject leftover CHAR/VARCHAR, including nested
(hand-built plans).
+ case dt if CharVarcharUtils.hasCharVarchar(dt) =>
+ throw
QueryExecutionErrors.scriptTransformWithoutSerdeUnsupportedTypeError(dt)
+ // Match collated STRING too (CHAR/VARCHAR COLLATE rewrite via
toStringType).
case _: StringType => wrapperConvertException(data => data, converter)
case BooleanType => wrapperConvertException(data => data.toBoolean,
converter)
case ByteType => wrapperConvertException(data => data.toByte, converter)
@@ -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 =>
val complexTypeFactory = JsonToStructs(attr.dataType,
ioschema.outputSerdeProps.toMap, Literal(null),
Some(conf.sessionLocalTimeZone))
@@ -432,8 +374,7 @@ case class ScriptTransformationIOSchema(
outputSerdeProps: Seq[(String, String)],
recordReaderClass: Option[String],
recordWriterClass: Option[String],
- schemaLess: Boolean,
- standardCharVarcharSemantics: Boolean = false) extends Serializable {
+ schemaLess: Boolean) extends Serializable {
import ScriptTransformationIOSchema._
val inputRowFormatMap = inputRowFormat.toMap.withDefault((k) =>
defaultFormat(k))
@@ -441,115 +382,13 @@ 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")
)
- def defaultIOSchema: ScriptTransformationIOSchema =
ScriptTransformationIOSchema(
+ val defaultIOSchema = ScriptTransformationIOSchema(
inputRowFormat = Seq.empty,
outputRowFormat = Seq.empty,
inputSerdeClass = None,
@@ -558,8 +397,7 @@ object ScriptTransformationIOSchema {
outputSerdeProps = Seq.empty,
recordReaderClass = None,
recordWriterClass = None,
- schemaLess = false,
- standardCharVarcharSemantics = SQLConf.get.charVarcharStandardSemantics
+ schemaLess = false
)
def apply(input: ScriptInputOutputSchema): ScriptTransformationIOSchema = {
@@ -572,7 +410,6 @@ object ScriptTransformationIOSchema {
input.outputSerdeProps,
input.recordReaderClass,
input.recordWriterClass,
- input.schemaLess,
- input.standardCharVarcharSemantics)
+ input.schemaLess)
}
}
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 87213a1619ab..7747c94b584d 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,8 +1347,7 @@ class SparkSqlAstBuilder extends AstBuilder {
inSerdeClass, outSerdeClass,
inSerdeProps, outSerdeProps,
reader, writer,
- schemaLess,
- conf.charVarcharStandardSemantics)
+ schemaLess)
}
}
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 fe7aa8b4282b..1e3ddd1c198d 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,false)
+ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,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)
+- 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,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)
+- 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,false)
+ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true)
+- 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,false)
+ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true)
+- 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,false)
+ScriptTransformation cat, [key#x, value#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,true)
+- 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,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)
+- 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,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+)),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x, c#x, d#x, e#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [one#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,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)
+- 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,false)
+ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+: +- ScriptTransformation cat, [b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ +- ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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,false)
+ScriptTransformation cat, [a#x, b#x],
ScriptInputOutputSchema(List(),List(),None,None,List(),List(),None,None,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/CharVarcharTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
index f9f5062b1dfc..8276f34a2255 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
@@ -3413,7 +3413,9 @@ 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,
+ // SparkScriptTransformationSuite, and HiveScriptTransformationSuite.
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
withTempPath { dir =>
val path = dir.getCanonicalPath
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 f7fb05526d96..e95f093c8c32 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,8 @@ import org.json4s.jackson.JsonMethods._
import org.scalatest.Assertions._
import org.scalatest.exceptions.TestFailedException
-import org.apache.spark.{SparkException, SparkThrowable, TaskContext,
TestUtils}
+import org.apache.spark.{SparkException, SparkRuntimeException,
SparkThrowable, TaskContext,
+ TestUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{QueryTest, Row}
import org.apache.spark.sql.catalyst.InternalRow
@@ -87,6 +88,20 @@ 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: SparkRuntimeException if s.getCondition ==
"EXCEED_LIMIT_LENGTH" => s
+ }.getOrElse {
+ fail(s"expected EXCEED_LIMIT_LENGTH, got $exception")
+ }
+ checkError(
+ exception = runtimeException,
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> limit))
+ }
+
test("cat without SerDe") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
@@ -103,360 +118,196 @@ abstract class BaseScriptTransformationSuite extends
QueryTest {
assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59277: TRANSFORM output supports first-class CHAR/VARCHAR
without SerDe") {
+ test("SPARK-60090: TRANSFORM CHAR/VARCHAR assignment via Project") {
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")))
+ val query = sql(
+ """
+ |SELECT TRANSFORM(c, v)
+ |USING 'cat' AS (c CHAR(4) COLLATE UTF8_LCASE, v VARCHAR(5) COLLATE
UNICODE_CI)
+ |FROM VALUES ('ab', 'xyz') t(c, v)
+ |""".stripMargin)
+ assert(query.schema.map(_.dataType) === Seq(
+ CharType(4, "UTF8_LCASE"),
+ VarcharType(5, "UNICODE_CI")))
+ checkAnswer(query, Row("ab ", "xyz"))
}
assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59277: TRANSFORM CHAR/VARCHAR null token without SerDe") {
+ test("SPARK-60090: TRANSFORM mixed CHAR and non-CHAR columns") {
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")))
+ val query = sql(
+ """
+ |SELECT TRANSFORM(id, name, note)
+ |USING 'cat' AS (id INT, name CHAR(4), note STRING)
+ |FROM VALUES (1, 'ab', 'hello') t(id, name, note)
+ |""".stripMargin)
+ assert(query.schema.map(_.dataType) === Seq(IntegerType, CharType(4),
StringType))
+ checkAnswer(query, Row(1, "ab ", "hello"))
}
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") {
+ test("SPARK-60090: TRANSFORM CHAR comparison with a STRING literal") {
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))
+ val subquery =
+ """
+ |SELECT * FROM (
+ | SELECT TRANSFORM(c) USING 'cat' AS (c CHAR(4))
+ | FROM VALUES ('ab') t(c)
+ |)
+ |""".stripMargin
+ // Either first-class-types flag yields a CharType attribute, so `=`
promotes
+ // CHAR to STRING instead of padding the literal (table preserve-only
padding
+ // applies only to StringType attributes that carry CHAR metadata).
+ Seq(
+ Map(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true"),
+ Map(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false")).foreach {
confs =>
+ withSQLConf(confs.toSeq: _*) {
+ checkAnswer(sql(subquery + " WHERE c = 'ab'"), Seq.empty)
+ checkAnswer(sql(subquery), Row("ab "))
}
}
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") {
+ test("SPARK-60090: TRANSFORM CHAR/VARCHAR null token") {
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)))
+ sql(
+ """
+ |SELECT TRANSFORM(c)
+ |USING 'cat' AS (c CHAR(4))
+ |FROM VALUES (CAST(NULL AS STRING)), ('ab') t(c)
+ |""".stripMargin),
+ Seq(Row(null), Row("ab ")))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(v)
+ |USING 'cat' AS (v VARCHAR(5))
+ |FROM VALUES (CAST(NULL AS STRING)), ('xy') t(v)
+ |""".stripMargin),
+ Seq(Row(null), Row("xy")))
}
assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59277: TRANSFORM validates restored JSON map keys without
SerDe") {
+ test("SPARK-60090: TRANSFORM CHAR overflow raises EXCEED_LIMIT_LENGTH") {
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))
+ val exception = intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM(c)
+ |USING 'cat' AS (c CHAR(4))
+ |FROM VALUES ('abcdef') t(c)
+ |""".stripMargin).collect()
}
+ checkExceedLimitLength(exception, "4")
}
- assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59277: colliding map key followed by valid row without SerDe") {
+ test("SPARK-60090: TRANSFORM VARCHAR overflow raises EXCEED_LIMIT_LENGTH") {
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 "))))
+ val exception = intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM(v)
+ |USING 'cat' AS (v VARCHAR(5))
+ |FROM VALUES ('abcdefgh') t(v)
+ |""".stripMargin).collect()
+ }
+ checkExceedLimitLength(exception, "5")
}
- assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59277: colliding CHAR map keys without SerDe follow
MAP_KEY_DEDUP_POLICY") {
+ test("SPARK-60090: TRANSFORM CHAR/VARCHAR flag off stays STRING") {
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\""))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ val query = sql(
+ """
+ |SELECT TRANSFORM(c)
+ |USING 'cat' AS (c CHAR(4))
+ |FROM VALUES ('ab'), ('abcdef') t(c)
+ |""".stripMargin)
+ assert(query.schema.map(_.dataType) === Seq(StringType))
+ checkAnswer(query, Seq(Row("ab"), Row("abcdef")))
}
assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59683: preserve-only CHAR/VARCHAR without SerDe stays
unsupported") {
+ test("SPARK-60090: preserve-only TRANSFORM CHAR/VARCHAR still assigns") {
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)\"")
+ val query = sql(
+ """
+ |SELECT TRANSFORM(c)
+ |USING 'cat' AS (c CHAR(4))
+ |FROM VALUES ('ab') t(c)
+ |""".stripMargin)
+ // First-class types keep CHAR in the result; assignment still pads /
length-checks.
+ assert(query.schema.map(_.dataType) === Seq(CharType(4)))
+ checkAnswer(query, Seq(Row("ab ")))
+ checkExceedLimitLength(
+ intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM(c)
+ |USING 'cat' AS (c CHAR(4))
+ |FROM VALUES ('abcdef') t(c)
+ |""".stripMargin).collect()
+ },
+ "4")
}
+ assert(uncaughtExceptionHandler.exception.isEmpty)
}
- test("SPARK-59683: bound CHAR/VARCHAR mode survives execution conf change") {
+ test("SPARK-60090: unreferenced CHAR overflow is pruned like scans") {
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"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // Assignment lives in a Project, so count(*) can skip the overflowing
column.
+ checkAnswer(
+ sql(
+ """
+ |SELECT count(*) FROM (
+ | SELECT TRANSFORM(c) USING 'cat' AS (c CHAR(4))
+ | FROM VALUES ('abcdef') t(c)
+ |)
+ |""".stripMargin),
+ Row(1))
}
+ }
- 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)
+ test("SPARK-60090: hand-built CHAR/VARCHAR script output stays unsupported
without SerDe") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ // SQL AS (c CHAR(...)) is rewritten to STRING before the exec. A
hand-built plan
+ // that still puts CharType/VarcharType on script output must fail.
+ val input = Seq("ab").toDF("c")
+ Seq(
+ AttributeReference("c", CharType(4))() -> "\"CHAR(4)\"",
+ AttributeReference("v", VarcharType(5))() -> "\"VARCHAR(5)\"",
+ AttributeReference("a", ArrayType(CharType(4)))() ->
"\"ARRAY<CHAR(4)>\"").foreach {
+ case (attr, sqlType) =>
+ val exception = intercept[Exception] {
+ QueryTest.executePlan(
+ createScriptTransformationExec(
+ script = "cat",
+ output = Seq(attr),
+ child = input.queryExecution.sparkPlan,
+ ioschema = defaultIOSchema),
+ spark.sqlContext)
+ }
+ checkTransformWithoutSerdeUnsupportedType(exception, sqlType)
}
- checkTransformWithoutSerdeUnsupportedType(exception, "\"CHAR(4)\"")
- }
- assert(uncaughtExceptionHandler.exception.isEmpty)
}
test("script transformation should not swallow errors from upstream
operators (no serde)") {
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 692302e3af36..329e773a99dd 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
@@ -17,12 +17,13 @@
package org.apache.spark.sql.execution
-import org.apache.spark.TestUtils
+import org.apache.spark.{SparkRuntimeException, 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
+import org.apache.spark.sql.types._
class SparkScriptTransformationSuite extends BaseScriptTransformationSuite
with SharedSparkSession {
import testImplicits._
@@ -42,8 +43,152 @@ 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") {
+ val arrayQuery = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('["ab"]') t(value)
+ |""".stripMargin)
+ assert(arrayQuery.schema.head.dataType === ArrayType(CharType(4)))
+ checkAnswer(arrayQuery, Row(Seq("ab ")))
+ val varcharArray = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<VARCHAR(4)>)
+ |FROM VALUES ('["xy"]') t(value)
+ |""".stripMargin)
+ assert(varcharArray.schema.head.dataType === ArrayType(VarcharType(4)))
+ checkAnswer(varcharArray, Row(Seq("xy")))
+ val structQuery = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value STRUCT<value: CHAR(5)>)
+ |FROM VALUES ('{"value":"xy"}') t(value)
+ |""".stripMargin)
+ assert(structQuery.schema.head.dataType ===
+ StructType(Seq(StructField("value", CharType(5)))))
+ checkAnswer(structQuery, Row(Row("xy ")))
+ val collated = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value STRUCT<value: VARCHAR(6) COLLATE UNICODE_CI>)
+ |FROM VALUES ('{"value":"xyz"}') t(value)
+ |""".stripMargin)
+ assert(collated.schema.head.dataType ===
+ StructType(Seq(StructField("value", VarcharType(6, "UNICODE_CI")))))
+ checkAnswer(collated, Row(Row("xyz")))
+ val quoted = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (s STRUCT<`a b`: CHAR(4)>)
+ |FROM VALUES ('{"a b":"ab"}') t(value)
+ |""".stripMargin)
+ assert(quoted.schema.head.dataType ===
+ StructType(Seq(StructField("a b", CharType(4)))))
+ checkAnswer(quoted, Row(Row("ab ")))
+ val mapVal = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<STRING, CHAR(4)>)
+ |FROM VALUES ('{"k":"ab"}') t(value)
+ |""".stripMargin)
+ assert(mapVal.schema.head.dataType === MapType(StringType, CharType(4)))
+ checkAnswer(mapVal, Row(Map("k" -> "ab ")))
+ val mapKey = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<CHAR(4), INT>)
+ |FROM VALUES ('{"ab":1}') t(value)
+ |""".stripMargin)
+ assert(mapKey.schema.head.dataType === MapType(CharType(4), IntegerType))
+ checkAnswer(mapKey, Row(Map("ab " -> 1)))
+ val nestedMap = sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<STRING, MAP<STRING, CHAR(4)>>)
+ |FROM VALUES ('{"outer":{"k":"ab"}}') t(value)
+ |""".stripMargin)
+ assert(nestedMap.schema.head.dataType ===
+ MapType(StringType, MapType(StringType, CharType(4))))
+ checkAnswer(nestedMap, Row(Map("outer" -> Map("k" -> "ab "))))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('[null]') t(value)
+ |""".stripMargin),
+ Row(Seq(null)))
+ checkAnswer(
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value ARRAY<CHAR(4)>)
+ |FROM VALUES ('not-json') t(value)
+ |""".stripMargin),
+ Row(null))
+ }
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-60090: TRANSFORM MAP CHAR key collision is DUPLICATED_MAP_KEY") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val error = intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM(value)
+ |USING 'cat' AS (value MAP<CHAR(4), INT>)
+ |FROM VALUES ('{"a":1,"a ":2}') t(value)
+ |""".stripMargin).collect()
+ }
+ val duplicate = Iterator.iterate[Throwable](error)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case e: SparkRuntimeException if e.getCondition ==
"DUPLICATED_MAP_KEY" => e
+ }.getOrElse(fail(s"expected DUPLICATED_MAP_KEY, got $error"))
+ checkError(
+ exception = duplicate,
+ condition = "DUPLICATED_MAP_KEY",
+ parameters = Map(
+ "key" -> "a ",
+ "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ }
+ }
+
+ 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"))
+ // View SQL is re-parsed on each read. standardSemantics is PERSISTED, so
the view
+ // is parsed under the value saved at CREATE VIEW, and the assignment
Project stays.
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
withView("v") {
sql(
@@ -62,38 +207,33 @@ class SparkScriptTransformationSuite extends
BaseScriptTransformationSuite with
|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"))
+ checkExceedLimitLength(
+ intercept[Exception](sql("SELECT * FROM v_overflow").collect()),
+ "4")
}
}
}
- withSQLConf(
- SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
- SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+ assert(uncaughtExceptionHandler.exception.isEmpty)
+ }
+
+ test("SPARK-60090: TRANSFORM view created with flag off uses stored schema
cast") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ // PERSISTED parses the view text under the stored flag (off), so there is
no
+ // assignment Project. The CHAR(4) result and padding come from the view
output
+ // cast against the stored schema, not from re-parsing under the caller
session.
+ withSQLConf(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)\"")
+ val query = sql("SELECT * FROM v_disabled")
+ assert(query.schema.map(_.dataType) === Seq(CharType(4)))
+ checkAnswer(query, Row("ab "))
}
}
}
- assert(uncaughtExceptionHandler.exception.isEmpty)
}
test("SPARK-32106: TRANSFORM with serde without hive should throw
exception") {
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 1eb4b48e8338..b8758407877f 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,24 +942,6 @@ 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 c53949889611..de2d15415837 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,7 +27,6 @@ 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
@@ -37,7 +36,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._
+import org.apache.spark.sql.types.DataType
import org.apache.spark.util.{CircularBuffer, Utils}
/**
@@ -76,9 +75,7 @@ private[hive] case class HiveScriptTransformationExec(
val mutableRow = new SpecificInternalRow(output.map(_.dataType))
@transient
- lazy val unwrappers =
outputSoi.getAllStructFieldRefs.asScala.zip(output).map {
- case (field, attr) => unwrapperFor(field, attr.dataType)
- }
+ lazy val unwrappers =
outputSoi.getAllStructFieldRefs.asScala.map(unwrapperFor)
override def hasNext: Boolean = {
if (completed) {
@@ -259,9 +256,7 @@ object HiveScriptIOSchema extends HiveInspectors {
output: Seq[Attribute]): Option[(AbstractSerDe, StructObjectInspector)]
= {
ioschema.outputSerdeClass.map { serdeClass =>
val (columns, columnTypes) = parseAttrs(output)
- val serdeTypes = outputTypesForSerDe(
- serdeClass, columnTypes, ioschema.standardCharVarcharSemantics)
- val serde = initSerDe(serdeClass, columns, serdeTypes,
ioschema.outputSerdeProps)
+ val serde = initSerDe(serdeClass, columns, columnTypes,
ioschema.outputSerdeProps)
val structObjectInspector =
serde.getObjectInspector().asInstanceOf[StructObjectInspector]
(serde, structObjectInspector)
}
@@ -273,25 +268,6 @@ 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 a4df01db29f2..804fa8ab53c2 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,15 +20,8 @@ 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}
@@ -380,14 +373,16 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
}
}
- test("SPARK-59277: TRANSFORM supports nested collated CHAR/VARCHAR with Hive
SerDe") {
+ test("SPARK-60090: TRANSFORM CHAR/VARCHAR with Hive SerDe uses Project
assignment") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // Hive sessions default to LazySimpleSerDe; script I/O stays
STRING-shaped and
+ // Project applies pad / EXCEED_LIMIT_LENGTH for the declared AS types.
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)))
+ | named_struct('value', CAST('xyz' AS VARCHAR(6))))
|USING 'cat'
|AS (
| chars ARRAY<CHAR(4) COLLATE UTF8_LCASE>,
@@ -398,60 +393,29 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
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") {
+ checkExceedLimitLength(
+ intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM('abcdef')
+ |USING 'cat'
+ |AS (c CHAR(4))
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin).collect()
+ },
+ "4")
+ checkExceedLimitLength(
+ intercept[Exception] {
+ sql(
+ """
+ |SELECT TRANSFORM('abcdefgh')
+ |USING 'cat'
+ |AS (v VARCHAR(5))
+ |FROM VALUES (1) input(dummy)
+ |""".stripMargin).collect()
+ },
+ "5")
Seq(
"""
|SELECT TRANSFORM(array('abcdef'))
@@ -464,107 +428,13 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
|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"))
+ |""".stripMargin).foreach { q =>
+ checkExceedLimitLength(intercept[Exception](sql(q).collect()), "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") {
+ test("SPARK-60090: Hive TRANSFORM view keeps CHAR/VARCHAR Project
assignment") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
withView("v") {
@@ -584,18 +454,9 @@ class HiveScriptTransformationSuite extends
BaseScriptTransformationSuite with T
|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"))
+ checkExceedLimitLength(
+ intercept[Exception](sql("SELECT * FROM v_overflow").collect()),
+ "4")
}
}
}
@@ -897,29 +758,3 @@ 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]
