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 c25f857a984b [SPARK-59917][SQL] Do not rtrim CHAR arguments to
from_json, from_csv, and from_xml
c25f857a984b is described below
commit c25f857a984b4abe37d2d4e4ffac61c7f2663c8c
Author: Serge Rielau <[email protected]>
AuthorDate: Wed Oct 7 12:37:21 2026 -0700
[SPARK-59917][SQL] Do not rtrim CHAR arguments to from_json, from_csv, and
from_xml
### What changes were proposed in this pull request?
Follow-up to
[SPARK-59274](https://issues.apache.org/jira/browse/SPARK-59274). `from_json`,
`from_csv`, `from_xml`, and `schema_of_*` take a STRING document. CHAR padding
is part of the value after CAST to STRING (`LENGTH('abc'::CHAR(10)::STRING)` is
10 under `spark.sql.charVarchar.standardSemantics.enabled`). This PR removes
`SupportTrimmedCharInput` so those functions no longer rtrim CHAR arguments.
Schema-driven parse of CHAR/VARCHAR *fields* from SPARK-59274 is unchanged:
pad, overflow, and FAILFAST still apply to parsed output.
JIRA: https://issues.apache.org/jira/browse/SPARK-59917
### Why are the changes needed?
These functions are STRING parsers. Special-casing CHAR by stripping
trailing blanks disagrees with CAST-to-STRING semantics and with VARCHAR/STRING
arguments, which keep trailing spaces.
In default mode, CHAR table columns are padded on read
(`spark.sql.readSideCharPadding`, on by default) and reach these functions as
padded STRINGs. The removed trait only matched first-class `CharType`, so on
master `from_csv(char_col, ...)` could disagree depending on
`spark.sql.charVarchar.standardSemantics.enabled`. This PR makes those paths
agree.
### Does this PR introduce _any_ user-facing change?
Yes, relative to current `master` (unreleased SPARK-59274). CHAR document
arguments are no longer rtrimmed before parse.
JSON still accepts trailing whitespace, so a padded CHAR JSON document can
still parse:
```sql
SELECT from_json(CAST('{"a":1}' AS CHAR(12)), 'a INT')
-- same result as
SELECT from_json('{"a":1} ', 'a INT')
```
CSV makes the padding visible with a space delimiter. Extra empty fields
are a token-count mismatch:
```sql
SELECT schema_of_csv(CAST('1' AS CHAR(3)), map('delimiter', ' '))
-- STRUCT<_c0: INT, _c1: STRING, _c2: STRING>
SELECT from_csv(CAST('1' AS CHAR(3)), '_c0 INT', map('delimiter', ' '))
-- PERMISSIVE still returns named_struct('_c0', 1)
SELECT from_csv(CAST('1' AS CHAR(3)), '_c0 INT', map('delimiter', ' ',
'mode', 'FAILFAST'))
-- MALFORMED_RECORD_IN_PARSING (previously parsed the trimmed value)
```
With the default comma delimiter, padding also lands on an unquoted last
field, because `from_csv` and `schema_of_csv` keep trailing whitespace by
default. Integral, decimal, and boolean converters reject those spaces; float
and double accept them, so inference can fall through to DOUBLE:
```sql
SELECT schema_of_csv(CAST('1' AS CHAR(3)))
-- STRUCT<_c0: DOUBLE> (was STRUCT<_c0: INT> after rtrim)
SELECT from_csv(CAST('1,2' AS CHAR(5)), 'a INT, b INT')
-- named_struct('a', 1, 'b', NULL) (was {1, 2})
SELECT from_csv(CAST('1,ab' AS CHAR(6)), 'a INT, b STRING')
-- named_struct('a', 1, 'b', 'ab ')
SELECT from_csv(
CAST('1,2' AS CHAR(5)),
'a INT, b INT',
map('ignoreTrailingWhiteSpace', 'true'))
-- named_struct('a', 1, 'b', 2)
```
`map('ignoreTrailingWhiteSpace', 'true')` restores the trimmed last-field
parse. FAILFAST on the INT last-field case raises `MALFORMED_RECORD_IN_PARSING`.
Default `spark.sql.charVarchar.standardSemantics.enabled` remains false.
Released Spark versions are unchanged.
### How was this patch tested?
- Dropped Catalyst tests that expected CHAR-document rtrim.
- `CsvExpressionsSuite`: space-delimiter CHAR padding (interpreted and
codegen, including FAILFAST); comma-delimiter last-field pad (`schema_of_csv`
DOUBLE, INT null, STRING pad, `ignoreTrailingWhiteSpace`).
- `BasicCharVarcharTestSuite`: padded CHAR `from_json` vs STRING with
trailing spaces; malformed CHAR corrupt records for JSON/XML; space-delimiter
`from_csv` FAILFAST; comma-delimiter `schema_of_csv` / `from_csv` including
`ignoreTrailingWhiteSpace` and a table-backed `CHAR(5)` column; schema-output
CHAR/VARCHAR cases from SPARK-59274 remain.
Targeted local run: `sql/testOnly
org.apache.spark.sql.BasicCharVarcharTestSuite -- -z SPARK-59274` and
`catalyst/testOnly
org.apache.spark.sql.catalyst.expressions.CsvExpressionsSuite -- -z
SPARK-59917`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #59188 from srielau/SPARK-59917.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
(cherry picked from commit 6fbec03caab5776014b3bfc23e8dfb8dcaab1dd7)
Signed-off-by: Daniel Tenedorio <[email protected]>
---
.../sql/catalyst/analysis/TypeCoercionHelper.scala | 9 +-
.../spark/sql/catalyst/expressions/ExprUtils.scala | 25 +----
.../sql/catalyst/expressions/csvExpressions.scala | 14 ++-
.../sql/catalyst/expressions/jsonExpressions.scala | 12 +--
.../sql/catalyst/expressions/xmlExpressions.scala | 12 +--
.../catalyst/expressions/CsvExpressionsSuite.scala | 39 +++++--
.../expressions/JsonExpressionsSuite.scala | 9 --
.../catalyst/expressions/XmlExpressionsSuite.scala | 9 --
.../apache/spark/sql/CharVarcharTestSuite.scala | 113 ++++++++++++++-------
9 files changed, 124 insertions(+), 118 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionHelper.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionHelper.scala
index 12de23c06fa3..f76433684c40 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionHelper.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionHelper.scala
@@ -53,7 +53,6 @@ import org.apache.spark.sql.catalyst.expressions.{
SpecialFrameBoundary,
SpecifiedWindowFrame,
SubtractTimestamps,
- SupportTrimmedCharInput,
TimeBucket,
TimestampAddInterval,
WindowSpecDefinition
@@ -747,14 +746,10 @@ abstract class TypeCoercionHelper {
case e: ExpectsInputTypes if e.inputTypes.nonEmpty =>
// Convert NullType into some specific target type for
ExpectsInputTypes that don't do
// general implicit casting. Also promote CHAR/VARCHAR to STRING here
because these
- // expressions skip ImplicitCastInputTypes. Expressions that trim CHAR
padding themselves
- // retain the original type so they can distinguish CHAR from VARCHAR
and STRING.
+ // expressions skip ImplicitCastInputTypes.
val children: Seq[Expression] = e.children.zip(e.inputTypes).map {
case (in, expected) =>
- val promotedType = e match {
- case _: SupportTrimmedCharInput => None
- case _ => charVarcharToPlainString(in.dataType, expected)
- }
+ val promotedType = charVarcharToPlainString(in.dataType, expected)
promotedType
.map(dt => if (dt == in.dataType) in else Cast(in, dt))
.getOrElse {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
index 12c8d80f2dcc..e4f79b9ced09 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
@@ -24,7 +24,6 @@ import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import
org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch,
TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
-import org.apache.spark.sql.catalyst.expressions.objects.Invoke
import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions
import org.apache.spark.sql.catalyst.plans.logical.Aggregate
import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
@@ -32,7 +31,7 @@ import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData,
CharVarcharUtils}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase,
QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.{AbstractMapType,
StringTypeWithCollation}
-import org.apache.spark.sql.types.{CharType, DataType, MapType, StringType,
StructType, VariantType}
+import org.apache.spark.sql.types.{DataType, MapType, StringType, StructType,
VariantType}
import org.apache.spark.unsafe.types.UTF8String
object ExprUtils extends EvalHelper with QueryErrorsBase {
@@ -271,25 +270,3 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
case _ => false
}
}
-
-private[sql] trait SupportTrimmedCharInput extends UnaryExpression {
-
- // Keep this type-based so the effective input does not change if SQLConf
changes after analysis.
- // A first-class CharType child already establishes that CHAR semantics
apply.
- @transient
- protected final lazy val stringInput: Expression = child.dataType match {
- case _: CharType =>
- Invoke(
- child,
- "trimRight",
- child.dataType,
- returnNullable = false,
- isDeterministic = child.deterministic)
- case _ => child
- }
-
- protected final def trimStringInput(value: UTF8String): UTF8String =
child.dataType match {
- case _: CharType => value.trimRight()
- case _ => value
- }
-}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
index bfe3e8271372..9c2c644cd675 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
@@ -66,8 +66,7 @@ case class CsvToStructs(
requiredSchema: Option[StructType] = None)
extends UnaryExpression
with TimeZoneAwareExpression
- with ExpectsInputTypes
- with SupportTrimmedCharInput {
+ with ExpectsInputTypes {
override def nullable: Boolean = child.nullable
override def nullIntolerant: Boolean = true
@@ -113,12 +112,12 @@ case class CsvToStructs(
override def stateful: Boolean = true
override def nullSafeEval(input: Any): Any = {
- evaluator.evaluate(trimStringInput(input.asInstanceOf[UTF8String]))
+ evaluator.evaluate(input.asInstanceOf[UTF8String])
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
val refEvaluator = ctx.addReferenceObj("evaluator", evaluator)
- val eval = stringInput.genCode(ctx)
+ val eval = child.genCode(ctx)
val resultType = CodeGenerator.boxedType(dataType)
val resultTerm = ctx.freshName("result")
ev.copy(code =
@@ -161,8 +160,7 @@ case class SchemaOfCsv(
extends UnaryExpression
with RuntimeReplaceable
with DefaultStringProducingExpression
- with QueryErrorsBase
- with SupportTrimmedCharInput {
+ with QueryErrorsBase {
def this(child: Expression) = this(child, Map.empty[String, String])
@@ -210,8 +208,8 @@ case class SchemaOfCsv(
Literal.create(evaluator, ObjectType(classOf[SchemaOfCsvEvaluator])),
"evaluate",
dataType,
- Seq(stringInput),
- Seq(stringInput.dataType),
+ Seq(child),
+ Seq(child.dataType),
returnNullable = false)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index b10efd91f4f7..91ddcec4473b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -1878,8 +1878,7 @@ case class JsonToStructs(
with TimeZoneAwareExpression
with CodegenFallback
with ExpectsInputTypes
- with QueryErrorsBase
- with SupportTrimmedCharInput {
+ with QueryErrorsBase {
// The JSON input data might be missing certain fields. We force the
nullability
// of the user-provided schema to avoid data corruptions. In particular, the
parquet-mr encoder
@@ -1940,7 +1939,7 @@ case class JsonToStructs(
override def stateful: Boolean = true
override def nullSafeEval(json: Any): Any = {
- evaluator.evaluate(trimStringInput(json.asInstanceOf[UTF8String]))
+ evaluator.evaluate(json.asInstanceOf[UTF8String])
}
override def inputTypes: Seq[AbstractDataType] =
@@ -2083,8 +2082,7 @@ case class SchemaOfJson(
extends UnaryExpression
with RuntimeReplaceable
with DefaultStringProducingExpression
- with QueryErrorsBase
- with SupportTrimmedCharInput {
+ with QueryErrorsBase {
def this(child: Expression) = this(child, Map.empty[String, String])
@@ -2126,8 +2124,8 @@ case class SchemaOfJson(
Literal.create(evaluator, ObjectType(classOf[SchemaOfJsonEvaluator])),
"evaluate",
dataType,
- Seq(stringInput),
- Seq(stringInput.dataType),
+ Seq(child),
+ Seq(child.dataType),
returnNullable = false)
override def prettyName: String = "schema_of_json"
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
index 7567ad39055f..0c4f4890ac58 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
@@ -66,8 +66,7 @@ case class XmlToStructs(
extends UnaryExpression
with TimeZoneAwareExpression
with ExpectsInputTypes
- with QueryErrorsBase
- with SupportTrimmedCharInput {
+ with QueryErrorsBase {
def this(child: Expression, schema: Expression, options: Map[String,
String]) =
this(
@@ -116,7 +115,7 @@ case class XmlToStructs(
}
override def nullSafeEval(xml: Any): Any = {
- evaluator.evaluate(trimStringInput(xml.asInstanceOf[UTF8String]))
+ evaluator.evaluate(xml.asInstanceOf[UTF8String])
}
override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode = {
@@ -177,8 +176,7 @@ case class SchemaOfXml(
extends UnaryExpression
with RuntimeReplaceable
with DefaultStringProducingExpression
- with QueryErrorsBase
- with SupportTrimmedCharInput {
+ with QueryErrorsBase {
def this(child: Expression) = this(child, Map.empty[String, String])
@@ -236,8 +234,8 @@ case class SchemaOfXml(
XmlExpressionEvalUtils.getClass,
dataType,
"schemaOfXml",
- Seq(Literal(xmlInferSchema, xmlInferSchemaObjectType), stringInput),
- Seq(xmlInferSchemaObjectType, stringInput.dataType),
+ Seq(Literal(xmlInferSchema, xmlInferSchemaObjectType), child),
+ Seq(xmlInferSchemaObjectType, child.dataType),
returnNullable = false)
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
index 7e9939ded587..3a36e276702b 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
@@ -22,7 +22,7 @@ import java.util.{Calendar, Locale, TimeZone}
import org.scalatest.exceptions.TestFailedException
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkException, SparkFunSuite}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util._
@@ -171,16 +171,39 @@ class CsvExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
"STRUCT<_c0: INT, _c1: STRING>")
}
- test("schema_of_csv and from_csv trim CHAR padding") {
- val input = Literal.create("1 2 ", CharType(6, "UTF8_LCASE"))
- val options = Map("delimiter" -> " ", "mode" -> FailFastMode.name)
- val schema = new StructType().add("_c0", IntegerType).add("_c1",
IntegerType)
+ test("SPARK-59917: schema_of_csv and from_csv keep CHAR padding") {
+ val padded = Literal.create("1 ", StringType)
+ val charInput = Literal.create("1 ", CharType(3))
+ val options = Map("delimiter" -> " ")
+ val schema = new StructType().add("_c0", IntegerType)
checkEvaluation(
- SchemaOfCsv(input, options),
- "STRUCT<_c0: INT, _c1: INT>")
+ SchemaOfCsv(padded, options),
+ "STRUCT<_c0: INT, _c1: STRING, _c2: STRING>")
checkEvaluation(
- CsvToStructs(schema, options, input, UTC_OPT),
+ SchemaOfCsv(charInput, options),
+ "STRUCT<_c0: INT, _c1: STRING, _c2: STRING>")
+ // Extra empty fields from CHAR padding: PERMISSIVE still parses the first
token.
+ checkEvaluation(CsvToStructs(schema, options, padded, UTC_OPT),
InternalRow(1))
+ checkEvaluation(CsvToStructs(schema, options, charInput, UTC_OPT),
InternalRow(1))
+ checkErrorInExpression[SparkException](
+ CsvToStructs(schema, options + ("mode" -> FailFastMode.name), charInput,
UTC_OPT),
+ "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ Map("badRecord" -> "[1]", "failFastMode" -> "FAILFAST"))
+
+ val commaChar = Literal.create("1,2 ", CharType(5))
+ val twoInts = new StructType().add("a", IntegerType).add("b", IntegerType)
+ val intString = new StructType().add("a", IntegerType).add("b", StringType)
+ checkEvaluation(
+ SchemaOfCsv(Literal.create("1 ", CharType(3)), Map.empty),
+ "STRUCT<_c0: DOUBLE>")
+ checkEvaluation(CsvToStructs(twoInts, Map.empty, commaChar, UTC_OPT),
InternalRow(1, null))
+ checkEvaluation(
+ CsvToStructs(intString, Map.empty, Literal.create("1,ab ",
CharType(6)), UTC_OPT),
+ InternalRow(1, UTF8String.fromString("ab ")))
+ checkEvaluation(
+ CsvToStructs(
+ twoInts, Map("ignoreTrailingWhiteSpace" -> "true"), commaChar,
UTC_OPT),
InternalRow(1, 2))
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index a5d30b417835..d8d68ef75c10 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -803,15 +803,6 @@ class JsonExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
"STRUCT<col: BIGINT>")
}
- test("schema_of_json and from_json trim CHAR padding") {
- val document = """{"col":1} """
- val input = Literal.create(document, CharType(document.length,
"UTF8_LCASE"))
- val schema = new StructType().add("col", LongType)
-
- checkEvaluation(SchemaOfJson(input, Map.empty), "STRUCT<col: BIGINT>")
- checkEvaluation(JsonToStructs(schema, Map.empty, input, UTC_OPT),
InternalRow(1L))
- }
-
test("parse date with locale") {
Seq("en-US", "ru-RU").foreach { langTag =>
val locale = Locale.forLanguageTag(langTag)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
index c1c53b720761..a0011e1dbbb9 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
@@ -371,15 +371,6 @@ class XmlExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
"STRUCT<col: BIGINT>")
}
- test("schema_of_xml and from_xml trim CHAR padding") {
- val document = "<ROW><col>1</col></ROW> "
- val input = Literal.create(document, CharType(document.length,
"UTF8_LCASE"))
- val schema = new StructType().add("col", LongType)
-
- checkEvaluation(SchemaOfXml(input, Map.empty), "STRUCT<col: BIGINT>")
- checkEvaluation(XmlToStructs(schema, Map.empty, input, UTC_OPT),
InternalRow(1L))
- }
-
test("parse date with locale") {
Seq("en-US", "ru-RU").foreach { langTag =>
val locale = Locale.forLanguageTag(langTag)
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 ab553de65de6..58397079b8d7 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
@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
import org.apache.spark.sql.catalyst.analysis.resolver.ResolverGuard
import org.apache.spark.sql.catalyst.expressions.{
Alias, ArrayJoin, Attribute, Concat, EqualTo, Expression, GreaterThan,
InSet, Literal,
- ScalarSubquery, StringRPad, StringToMap, SupportTrimmedCharInput, Upper
+ ScalarSubquery, StringRPad, StringToMap, Upper
}
import org.apache.spark.sql.catalyst.expressions.Cast.toSQLId
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
@@ -2645,44 +2645,89 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
test("SPARK-59274: from_json/csv/xml honor CHAR/VARCHAR under
standardSemantics") {
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
- def checkTextInput(query: String, inputType: DataType): DataFrame = {
- val df = sql(query)
- val inputTypes =
df.queryExecution.analyzed.expressions.flatMap(_.collect {
- case e: SupportTrimmedCharInput => e.child.dataType
- })
- assert(inputTypes === Seq(inputType), df.queryExecution.analyzed)
- df
- }
-
+ // CHAR padding is part of the STRING value after CAST. Parsers do not
rtrim it.
+ // from_json on padded CHAR must match the same document as STRING with
trailing spaces.
+ checkAnswer(
+ sql("""SELECT from_json(CAST('{"a":1}' AS CHAR(12)), 'a INT')"""),
+ sql("""SELECT from_json('{"a":1} ', 'a INT')"""))
+ // XML ignores trailing whitespace, so a padded CHAR document still
parses.
checkAnswer(
- checkTextInput(
- """SELECT from_json(CAST('{"a":1}' AS CHAR(12)), 'a INT')""",
- CharType(12)),
+ sql(
+ """SELECT from_xml(
+ | CAST('<ROW><a>1</a></ROW>' AS CHAR(30)),
+ | 'a INT')""".stripMargin),
Row(Row(1)))
+ // Malformed padded CHAR documents keep the full pad in the corrupt
record.
+ // Reinstating rtrim would drop those trailing spaces from _unparsed.
+ checkAnswer(
+ sql(
+ """SELECT from_json(
+ | CAST('{"a":' AS CHAR(12)),
+ | 'a INT, _unparsed STRING',
+ | map('columnNameOfCorruptRecord', '_unparsed'))""".stripMargin),
+ Row(Row(null, "{\"a\": ")))
+ checkAnswer(
+ sql(
+ """SELECT from_xml(
+ | CAST('<ROW>' AS CHAR(10)),
+ | 'a INT, _unparsed STRING',
+ | map('columnNameOfCorruptRecord', '_unparsed'))""".stripMargin),
+ Row(Row(null, "<ROW> ")))
+ // With a space delimiter, CHAR padding tokenizes into extra empty
fields.
checkAnswer(
- checkTextInput(
+ sql(
+ """SELECT schema_of_csv(
+ | CAST('1' AS CHAR(3)),
+ | map('delimiter', ' '))""".stripMargin),
+ Row("STRUCT<_c0: INT, _c1: STRING, _c2: STRING>"))
+ checkAnswer(
+ sql(
"""SELECT from_csv(
| CAST('1' AS CHAR(3)),
| '_c0 INT',
- | map('delimiter', ' ', 'mode', 'FAILFAST'))""".stripMargin,
- CharType(3)),
- Row(Row(1)))
- checkAnswer(
- checkTextInput(
- """SELECT from_xml(
- | CAST('<ROW><a>1</a></ROW>' AS CHAR(30)),
- | 'a INT')""".stripMargin,
- CharType(30)),
+ | map('delimiter', ' '))""".stripMargin),
Row(Row(1)))
+ checkError(
+ exception = intercept[SparkException] {
+ sql(
+ """SELECT from_csv(
+ | CAST('1' AS CHAR(3)),
+ | '_c0 INT',
+ | map('delimiter', ' ', 'mode',
'FAILFAST'))""".stripMargin).collect()
+ },
+ condition = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ parameters = Map("badRecord" -> "[1]", "failFastMode" -> "FAILFAST"))
Seq(
- ("CAST('1 ' AS VARCHAR(2))", VarcharType(2)),
- ("CAST('1 ' AS STRING)", StringType)).foreach { case (input,
inputType) =>
+ "CAST('1 ' AS VARCHAR(2))",
+ "CAST('1 ' AS STRING)").foreach { input =>
checkAnswer(
- checkTextInput(
- s"SELECT schema_of_csv($input, map('delimiter', ' '))",
- inputType),
+ sql(s"SELECT schema_of_csv($input, map('delimiter', ' '))"),
Row("STRUCT<_c0: INT, _c1: STRING>"))
}
+ // Default comma delimiter: padding lands on the unquoted last field.
+ checkAnswer(
+ sql("SELECT schema_of_csv(CAST('1' AS CHAR(3)))"),
+ Row("STRUCT<_c0: DOUBLE>"))
+ checkAnswer(
+ sql("SELECT from_csv(CAST('1,2' AS CHAR(5)), 'a INT, b INT')"),
+ Row(Row(1, null)))
+ checkAnswer(
+ sql("SELECT from_csv(CAST('1,ab' AS CHAR(6)), 'a INT, b STRING')"),
+ Row(Row(1, "ab ")))
+ checkAnswer(
+ sql(
+ """SELECT from_csv(
+ | CAST('1,2' AS CHAR(5)),
+ | 'a INT, b INT',
+ | map('ignoreTrailingWhiteSpace', 'true'))""".stripMargin),
+ Row(Row(1, 2)))
+ withTable("csv_char_doc") {
+ sql("CREATE TABLE csv_char_doc(c CHAR(5)) USING parquet")
+ sql("INSERT INTO csv_char_doc VALUES ('1,2')")
+ checkAnswer(
+ sql("SELECT from_csv(c, 'a INT, b INT') FROM csv_char_doc"),
+ Row(Row(1, null)))
+ }
val jsonChar = sql("""SELECT from_json('{"a": "str"}', 'a CHAR(5)')""")
val jsonCharType = jsonChar.schema.head.dataType.asInstanceOf[StructType]
@@ -2814,16 +2859,6 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
checkAnswer(
sql("SELECT length(CAST('1' AS CHAR(3)))"),
Row(3))
- checkAnswer(
- sql("SELECT schema_of_csv(CAST('1' AS CHAR(3)))"),
- Row("STRUCT<_c0: INT>"))
- // Without trailing-only CHAR padding removal, the space delimiter
creates empty columns.
- checkAnswer(
- sql(
- """SELECT schema_of_csv(
- | CAST('1' AS CHAR(3)),
- | map('delimiter', ' '))""".stripMargin),
- Row("STRUCT<_c0: INT>"))
checkAnswer(
sql("SELECT length(CAST('<ROW><a>1</a></ROW>' AS CHAR(30)))"),
Row(30))
@@ -2834,7 +2869,7 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
"schema_of_json(CAST('{\"a\":1}' AS CHAR(20) COLLATE SR_AI))" ->
"STRUCT<a: BIGINT>",
"schema_of_csv(CAST('1' AS CHAR(3) COLLATE SR_AI), map('delimiter', '
'))" ->
- "STRUCT<_c0: INT>",
+ "STRUCT<_c0: INT, _c1: STRING, _c2: STRING>",
"schema_of_xml(CAST('<ROW><a>1</a></ROW>' AS CHAR(30) COLLATE SR_AI))"
->
"STRUCT<a: BIGINT>").foreach { case (expression, expected) =>
checkAnswer(sql(s"SELECT $expression"), Row(expected))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
