You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by dt...@apache.org on 2026/09/30 16:40:28 UTC
(spark) branch branch-4.x updated: [SPARK-59274][SQL] Enforce CHAR/VARCHAR semantics in schema-driven text parsing
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 4df616bf34ef [SPARK-59274][SQL] Enforce CHAR/VARCHAR semantics in schema-driven text parsing
4df616bf34ef is described below
commit 4df616bf34ef408373dcc830fad13e0de9df555d
Author: Serge Rielau <se...@rielau.com>
AuthorDate: Wed Sep 30 09:40:03 2026 -0700
[SPARK-59274][SQL] Enforce CHAR/VARCHAR semantics in schema-driven text parsing
### What changes were proposed in this pull request?
When `spark.sql.charVarchar.standardSemantics.enabled` is true, first-class CHAR/VARCHAR schemas are already allowed through `failIfHasCharVarchar`. Schema-driven text parsers still treated those columns as unbounded STRING (no pad/overflow), `schema_of_json/csv/xml` rejected CHAR/VARCHAR input with `child.dataType != StringType`, and XML map keys / `convertTo` used exact `StringType` matches.
This patch applies assignment semantics while parsing text into a typed schema:
- JSON (`JacksonParser`), CSV (`UnivocityParser`), and XML (`StaxXmlParser`) call `CharVarcharUtils.applyTextParseSemantics` so CHAR is padded and oversize VARCHAR raises `EXCEED_LIMIT_LENGTH`
- XML map keys with CHAR/VARCHAR key types get the same checks and honor `spark.sql.mapKeyDedupPolicy` for collisions created by that normalization
- XML wildcard columns and `convertTo` keep the declared CHAR/VARCHAR type instead of collapsing to `StringType`
- `schema_of_json` / `schema_of_csv` / `schema_of_xml` accept any `StringType` subtype as the input document, and CHAR documents trim trailing padding after analysis
JSON `MAP<CHAR/VARCHAR, ...>` key normalization is **out of scope** and tracked as [SPARK-59722](https://issues.apache.org/jira/browse/SPARK-59722). Ordinary STRING maps, including non-binary collated keys, keep their historical parser behavior.
Default (flag off) is unchanged: `from_json(..., 'a CHAR(5)')` still fails with `UNSUPPORTED_CHAR_OR_VARCHAR_AS_STRING`.
### Why are the changes needed?
`from_json` / `from_csv` / `from_xml` are the remaining schema-driven text parse surfaces that drop CHAR/VARCHAR length rules under standard semantics. Without this, enabling the flag still produces unpadded CHAR values, silently truncates or accepts oversize VARCHAR, and rejects CHAR/VARCHAR documents in `schema_of_*`.
JIRA: https://issues.apache.org/jira/browse/SPARK-59274 (subtask of SPARK-58794)
Follow-up: https://issues.apache.org/jira/browse/SPARK-59722 (JSON map keys)
### Does this PR introduce _any_ user-facing change?
Yes, when `spark.sql.charVarchar.standardSemantics.enabled` is true (still default false):
- `from_json` / `from_csv` / `from_xml` with a CHAR/VARCHAR schema keep those types and apply assignment checks (CHAR pad, VARCHAR overflow -> `EXCEED_LIMIT_LENGTH`).
- `schema_of_json` / `schema_of_csv` / `schema_of_xml` accept CHAR/VARCHAR input strings.
- XML maps with CHAR/VARCHAR keys pad or reject keys the same way, and apply `mapKeyDedupPolicy` to normalization-created collisions.
### How was this patch tested?
Added `BasicCharVarcharTestSuite` coverage for SPARK-59274: CHAR padding and VARCHAR overflow in `from_json` / `from_csv` / `from_xml`, XML `MAP<CHAR(n), INT>` keys, and `schema_of_json/csv/xml` on CHAR/VARCHAR input.
Targeted local run: `sql/testOnly *CharVarcharTestSuite -- -z SPARK-59274` and `sql/testOnly *ExpressionInfoSuite -- -z "Check whether SQL expressions should extend NullIntolerant"`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #58545 from srielau/serge-rielau_data/SPARK-59274.
Authored-by: Serge Rielau <se...@rielau.com>
Signed-off-by: Daniel Tenedorio <da...@databricks.com>
(cherry picked from commit 92d512771331d4d7fd6fd270059ac3fc7c7b2d1b)
Signed-off-by: Daniel Tenedorio <da...@databricks.com>
---
.../sql/catalyst/analysis/TypeCoercionHelper.scala | 13 +-
.../spark/sql/catalyst/csv/UnivocityParser.scala | 6 +-
.../spark/sql/catalyst/expressions/ExprUtils.scala | 25 +-
.../sql/catalyst/expressions/csvExpressions.scala | 17 +-
.../sql/catalyst/expressions/jsonExpressions.scala | 19 +-
.../sql/catalyst/expressions/xmlExpressions.scala | 37 +-
.../spark/sql/catalyst/json/JacksonParser.scala | 15 +-
.../spark/sql/catalyst/util/CharVarcharUtils.scala | 24 +
.../sql/catalyst/util/DuplicateMapKeyUtils.scala | 82 +++
.../spark/sql/catalyst/xml/StaxXmlParser.scala | 247 +++++++--
.../org/apache/spark/sql/internal/SQLConf.scala | 11 +-
.../catalyst/expressions/CsvExpressionsSuite.scala | 13 +
.../expressions/JsonExpressionsSuite.scala | 9 +
.../catalyst/expressions/XmlExpressionsSuite.scala | 9 +
.../plans/ConstraintPropagationSuite.scala | 17 +-
.../apache/spark/sql/CharVarcharTestSuite.scala | 563 ++++++++++++++++++++-
16 files changed, 1030 insertions(+), 77 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 67af8f9ee111..12de23c06fa3 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,6 +53,7 @@ import org.apache.spark.sql.catalyst.expressions.{
SpecialFrameBoundary,
SpecifiedWindowFrame,
SubtractTimestamps,
+ SupportTrimmedCharInput,
TimeBucket,
TimestampAddInterval,
WindowSpecDefinition
@@ -745,12 +746,16 @@ 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: these
- // expressions skip ImplicitCastInputTypes, so without this the length constraint would
- // remain on the child.
+ // 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.
val children: Seq[Expression] = e.children.zip(e.inputTypes).map {
case (in, expected) =>
- charVarcharToPlainString(in.dataType, expected)
+ val promotedType = e match {
+ case _: SupportTrimmedCharInput => None
+ case _ => charVarcharToPlainString(in.dataType, expected)
+ }
+ promotedType
.map(dt => if (dt == in.dataType) in else Cast(in, dt))
.getOrElse {
if (in.dataType == NullType && !expected.acceptsType(NullType)) {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
index 4583c328239e..6f0965e2aed0 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
@@ -278,8 +278,10 @@ class UnivocityParser(
timeFormatter.parse(datum)
}
- case _: StringType => (d: String) =>
- nullSafeDatum(d, name, nullable, options)(UTF8String.fromString)
+ case dt: StringType => (d: String) =>
+ nullSafeDatum(d, name, nullable, options) { s =>
+ CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(s), dt)
+ }
case _: BinaryType => (d: String) =>
nullSafeDatum(d, name, nullable, options)(_.getBytes)
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 e4f79b9ced09..12c8d80f2dcc 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,6 +24,7 @@ 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
@@ -31,7 +32,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.{DataType, MapType, StringType, StructType, VariantType}
+import org.apache.spark.sql.types.{CharType, DataType, MapType, StringType, StructType, VariantType}
import org.apache.spark.unsafe.types.UTF8String
object ExprUtils extends EvalHelper with QueryErrorsBase {
@@ -270,3 +271,25 @@ 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 25d154066b67..bfe3e8271372 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,10 +66,10 @@ case class CsvToStructs(
requiredSchema: Option[StructType] = None)
extends UnaryExpression
with TimeZoneAwareExpression
- with ExpectsInputTypes {
+ with ExpectsInputTypes
+ with SupportTrimmedCharInput {
override def nullable: Boolean = child.nullable
-
override def nullIntolerant: Boolean = true
// Used in `FunctionRegistry`
@@ -113,12 +113,12 @@ case class CsvToStructs(
override def stateful: Boolean = true
override def nullSafeEval(input: Any): Any = {
- evaluator.evaluate(input.asInstanceOf[UTF8String])
+ evaluator.evaluate(trimStringInput(input.asInstanceOf[UTF8String]))
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
val refEvaluator = ctx.addReferenceObj("evaluator", evaluator)
- val eval = child.genCode(ctx)
+ val eval = stringInput.genCode(ctx)
val resultType = CodeGenerator.boxedType(dataType)
val resultTerm = ctx.freshName("result")
ev.copy(code =
@@ -161,7 +161,8 @@ case class SchemaOfCsv(
extends UnaryExpression
with RuntimeReplaceable
with DefaultStringProducingExpression
- with QueryErrorsBase {
+ with QueryErrorsBase
+ with SupportTrimmedCharInput {
def this(child: Expression) = this(child, Map.empty[String, String])
@@ -183,7 +184,7 @@ case class SchemaOfCsv(
DataTypeMismatch(
errorSubClass = "UNEXPECTED_NULL",
messageParameters = Map("exprName" -> "csv"))
- } else if (child.dataType != StringType) {
+ } else if (!child.dataType.isInstanceOf[StringType]) {
DataTypeMismatch(
errorSubClass = "UNEXPECTED_INPUT_TYPE",
messageParameters = Map(
@@ -209,8 +210,8 @@ case class SchemaOfCsv(
Literal.create(evaluator, ObjectType(classOf[SchemaOfCsvEvaluator])),
"evaluate",
dataType,
- Seq(child),
- Seq(child.dataType),
+ Seq(stringInput),
+ Seq(stringInput.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 06728984d3da..b10efd91f4f7 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,7 +1878,8 @@ case class JsonToStructs(
with TimeZoneAwareExpression
with CodegenFallback
with ExpectsInputTypes
- with QueryErrorsBase {
+ with QueryErrorsBase
+ with SupportTrimmedCharInput {
// 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
@@ -1886,11 +1887,10 @@ case class JsonToStructs(
private val nullableSchema: DataType = schema.asNullable
override def nullable: Boolean = true
+ override def nullIntolerant: Boolean = true
final override def nodePatternsInternal(): Seq[TreePattern] = Seq(JSON_TO_STRUCT)
- override def nullIntolerant: Boolean = true
-
// Used in `FunctionRegistry`
def this(child: Expression, schema: Expression, options: Map[String, String]) =
this(
@@ -1939,7 +1939,9 @@ case class JsonToStructs(
options, nullableSchema, nameOfCorruptRecord, timeZoneId, variantAllowDuplicateKeys)
override def stateful: Boolean = true
- override def nullSafeEval(json: Any): Any = evaluator.evaluate(json.asInstanceOf[UTF8String])
+ override def nullSafeEval(json: Any): Any = {
+ evaluator.evaluate(trimStringInput(json.asInstanceOf[UTF8String]))
+ }
override def inputTypes: Seq[AbstractDataType] =
StringTypeWithCollation(supportsTrimCollation = true) :: Nil
@@ -2081,7 +2083,8 @@ case class SchemaOfJson(
extends UnaryExpression
with RuntimeReplaceable
with DefaultStringProducingExpression
- with QueryErrorsBase {
+ with QueryErrorsBase
+ with SupportTrimmedCharInput {
def this(child: Expression) = this(child, Map.empty[String, String])
@@ -2103,7 +2106,7 @@ case class SchemaOfJson(
DataTypeMismatch(
errorSubClass = "UNEXPECTED_NULL",
messageParameters = Map("exprName" -> "json"))
- } else if (child.dataType != StringType) {
+ } else if (!child.dataType.isInstanceOf[StringType]) {
DataTypeMismatch(
errorSubClass = "UNEXPECTED_INPUT_TYPE",
messageParameters = Map(
@@ -2123,8 +2126,8 @@ case class SchemaOfJson(
Literal.create(evaluator, ObjectType(classOf[SchemaOfJsonEvaluator])),
"evaluate",
dataType,
- Seq(child),
- Seq(child.dataType),
+ Seq(stringInput),
+ Seq(stringInput.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 6baee74acecb..7567ad39055f 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
@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions
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.codegen.{CodegenContext, CodeGenerator, ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.expressions.xml.{StructsToXmlEvaluator, XmlExpressionEvalUtils, XmlToStructsEvaluator}
import org.apache.spark.sql.catalyst.util.DropMalformedMode
@@ -65,7 +66,8 @@ case class XmlToStructs(
extends UnaryExpression
with TimeZoneAwareExpression
with ExpectsInputTypes
- with QueryErrorsBase {
+ with QueryErrorsBase
+ with SupportTrimmedCharInput {
def this(child: Expression, schema: Expression, options: Map[String, String]) =
this(
@@ -113,22 +115,30 @@ case class XmlToStructs(
copy(timeZoneId = Option(timeZoneId))
}
- override def nullSafeEval(xml: Any): Any = evaluator.evaluate(xml.asInstanceOf[UTF8String])
+ override def nullSafeEval(xml: Any): Any = {
+ evaluator.evaluate(trimStringInput(xml.asInstanceOf[UTF8String]))
+ }
override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
val expr = ctx.addReferenceObj("this", this)
+ val inputEval = child.genCode(ctx)
// nullSafeEval returns an InternalRow for struct output and a VariantVal for variant output.
// The variant result can be null (e.g. a malformed record rescued under PERMISSIVE mode), so
// cast to the actual output type and null-check the result.
val resultType = CodeGenerator.javaType(dataType)
val result = ctx.freshName("xmlResult")
- nullSafeCodeGen(ctx, ev, input =>
- s"""
- |$resultType $result = ($resultType) $expr.nullSafeEval($input);
- |if ($result == null) {
- | ${ev.isNull} = true;
- |} else {
- | ${ev.value} = $result;
+ ev.copy(code =
+ code"""
+ |${inputEval.code}
+ |boolean ${ev.isNull} = ${inputEval.isNull};
+ |$resultType ${ev.value} = ${CodeGenerator.defaultValue(dataType)};
+ |if (!${ev.isNull}) {
+ | $resultType $result = ($resultType) $expr.nullSafeEval(${inputEval.value});
+ | if ($result == null) {
+ | ${ev.isNull} = true;
+ | } else {
+ | ${ev.value} = $result;
+ | }
|}
""".stripMargin)
}
@@ -167,7 +177,8 @@ case class SchemaOfXml(
extends UnaryExpression
with RuntimeReplaceable
with DefaultStringProducingExpression
- with QueryErrorsBase {
+ with QueryErrorsBase
+ with SupportTrimmedCharInput {
def this(child: Expression) = this(child, Map.empty[String, String])
@@ -200,7 +211,7 @@ case class SchemaOfXml(
DataTypeMismatch(
errorSubClass = "UNEXPECTED_NULL",
messageParameters = Map("exprName" -> "xml"))
- } else if (child.dataType != StringType) {
+ } else if (!child.dataType.isInstanceOf[StringType]) {
DataTypeMismatch(
errorSubClass = "UNEXPECTED_INPUT_TYPE",
messageParameters = Map(
@@ -225,8 +236,8 @@ case class SchemaOfXml(
XmlExpressionEvalUtils.getClass,
dataType,
"schemaOfXml",
- Seq(Literal(xmlInferSchema, xmlInferSchemaObjectType), child),
- Seq(xmlInferSchemaObjectType, child.dataType),
+ Seq(Literal(xmlInferSchema, xmlInferSchemaObjectType), stringInput),
+ Seq(xmlInferSchemaObjectType, stringInput.dataType),
returnNullable = false)
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index ea8061b774c4..e1a277764d0f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -302,7 +302,7 @@ class JacksonParser(
}
}
- case _: StringType => (parser: JsonParser) => {
+ case dt: StringType => (parser: JsonParser) => {
// This must be enabled if we will retrieve the bytes directly from the raw content:
val oldFeature = parser.getFeatureMask
val featureToAdd = JsonParser.Feature.INCLUDE_SOURCE_IN_LOCATION.getMask
@@ -356,7 +356,7 @@ class JacksonParser(
// to be reset. This ensures that every feature is restored to its previous
// state as defined by `oldFeature`.
parser.overrideStdFeatures(oldFeature, ~0)
- result
+ CharVarcharUtils.applyTextParseSemantics(result, dt)
}
case TimestampType =>
@@ -636,8 +636,15 @@ class JacksonParser(
}
}
- // The JSON map will never have null or duplicated map keys, it's safe to create a
- // ArrayBasedMapData directly here.
+ // Value conversion can fail after the key is recorded. Do not build MapData from
+ // unpaired buffers: ArrayBasedMapData would throw a cardinality error and hide
+ // the original conversion failure (for example EXCEED_LIMIT_LENGTH).
+ if (keys.length != values.length) {
+ throw badRecordException.get
+ }
+
+ // Preserve every parsed JSON key/value pair, including exact duplicate names.
+ // ArrayBasedMapData is used directly to retain this historical behavior.
val mapData = ArrayBasedMapData(keys.toArray, values.toArray)
if (badRecordException.isEmpty) {
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 dd0b0d3ba577..c9fc3bfd87f1 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
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.ArrayImplicits._
object CharVarcharUtils extends Logging with SparkCharVarcharUtils {
@@ -156,6 +157,29 @@ object CharVarcharUtils extends Logging with SparkCharVarcharUtils {
StructType(fields)
}
+ /**
+ * Applies assignment semantics when parsing text into a typed schema. CHAR values are padded,
+ * and excess trailing spaces are trimmed when the remaining value fits. Non-space overflow
+ * raises EXCEED_LIMIT_LENGTH. Null stays null, and unbounded STRING is unchanged.
+ *
+ * With first-class CHAR/VARCHAR semantics enabled, for example,
+ * `from_json('{"c":"ab"}', 'c CHAR(4)')` produces `"ab "`, while parsing `{"v":"abcd"}`
+ * as `v VARCHAR(3)` in FAILFAST mode raises EXCEED_LIMIT_LENGTH.
+ */
+ def applyTextParseSemantics(value: UTF8String, dt: DataType): UTF8String = {
+ if (value == null) {
+ null
+ } else {
+ dt match {
+ case c: CharType =>
+ CharVarcharCodegenUtils.charTypeWriteSideCheck(value, c.length)
+ case v: VarcharType =>
+ CharVarcharCodegenUtils.varcharTypeWriteSideCheck(value, v.length)
+ case _ => value
+ }
+ }
+ }
+
/**
* Returns an expression to apply write-side string length check for the given expression. A
* string value can not exceed N characters if it's written into a CHAR(N)/VARCHAR(N)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala
new file mode 100644
index 000000000000..fd179bc73d05
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import scala.collection.mutable
+
+import org.apache.spark.SparkRuntimeException
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.util.SparkErrorUtils
+
+private[sql] object DuplicateMapKeyUtils {
+ def cause(exception: Throwable): Option[SparkRuntimeException] = {
+ SparkErrorUtils.getRootCause(exception) match {
+ case cause: SparkRuntimeException if cause.getCondition == "DUPLICATED_MAP_KEY" =>
+ Some(cause)
+ case _ => None
+ }
+ }
+
+ def unapply(exception: Throwable): Option[SparkRuntimeException] = cause(exception)
+
+ /**
+ * Builds an XML map with a constrained CHAR/VARCHAR key type from a raw-key last-wins
+ * accumulator. Failed values still occupy a slot so normalized collisions are visible.
+ *
+ * CHAR/VARCHAR keys: repeated raw XML keys keep the last value, then
+ * `spark.sql.mapKeyDedupPolicy` applies to normalized keys.
+ *
+ * Example: parsing `a` and `a ` as CHAR(2) keys raises
+ * DUPLICATED_MAP_KEY under EXCEPTION and keeps `a ` -> 2 under LAST_WIN.
+ */
+ def buildConstrainedMap(
+ lastEntries: mutable.LinkedHashMap[UTF8String, (UTF8String, Option[Any])],
+ keyType: DataType,
+ valueType: DataType): MapData = {
+ if (SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY) ==
+ SQLConf.MapKeyDedupPolicy.EXCEPTION) {
+ val distinctKeys = keyType match {
+ case stringType: StringType if stringType.supportsBinaryEquality =>
+ new java.util.HashSet[Any]()
+ case _ =>
+ new java.util.TreeSet[Any](TypeUtils.getInterpretedOrdering(keyType))
+ }
+ val keys = mutable.ArrayBuffer.empty[Any]
+ val values = mutable.ArrayBuffer.empty[Any]
+ lastEntries.valuesIterator.foreach { case (key, value) =>
+ if (!distinctKeys.add(key)) {
+ throw QueryExecutionErrors.duplicateMapKeyFoundError(key)
+ }
+ value.foreach { v =>
+ keys += key
+ values += v
+ }
+ }
+ ArrayBasedMapData(keys.toArray, values.toArray)
+ } else {
+ val builder = new ArrayBasedMapBuilder(keyType, valueType)
+ lastEntries.valuesIterator.foreach { case (normalizedKey, value) =>
+ value.foreach(builder.put(normalizedKey, _))
+ }
+ builder.build()
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
index d964564fbde1..f0923d405436 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
@@ -26,6 +26,7 @@ import javax.xml.stream.events._
import javax.xml.transform.stream.StreamSource
import javax.xml.validation.Schema
+import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
import scala.util.Try
@@ -41,7 +42,7 @@ import org.apache.spark.{SparkIllegalArgumentException, SparkUpgradeException}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{ExprUtils, GenericInternalRow, ToStringBase}
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, BadRecordException, DateFormatter, DropMalformedMode, FailureSafeParser, GenericArrayData, MapData, ParseMode, PartialResultArrayException, PartialResultException, PermissiveMode, TimeFormatter, TimestampFormatter}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, BadRecordException, CharVarcharUtils, DateFormatter, DropMalformedMode, DuplicateMapKeyUtils, FailureSafeParser, GenericArrayData, MapData, ParseMode, PartialResultArrayException, PartialResultException, PermissiveMode, TimeFormatter, TimestampFormatter}
import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT
import org.apache.spark.sql.catalyst.xml.StaxXmlParser.convertStream
import org.apache.spark.sql.errors.QueryExecutionErrors
@@ -85,6 +86,72 @@ class StaxXmlParser(
private val caseSensitive = SQLConf.get.caseSensitiveAnalysis
+ /**
+ * Limits a view of an event stream to one element whose start event has already been consumed.
+ * Closing or draining this view consumes the matching end event without closing the underlying
+ * reader.
+ */
+ private final class ElementBoundedEventReader(delegate: XMLEventReader)
+ extends XMLEventReader {
+ private var depth = 1
+
+ private def track(event: XMLEvent): XMLEvent = {
+ event match {
+ case _: StartElement => depth += 1
+ case _: EndElement => depth -= 1
+ case _ =>
+ }
+ event
+ }
+
+ override def hasNext: Boolean = depth > 0 && delegate.hasNext
+
+ override def nextEvent(): XMLEvent = {
+ if (!hasNext) {
+ throw new NoSuchElementException("No more events in the current element")
+ }
+ track(delegate.nextEvent())
+ }
+
+ override def peek(): XMLEvent = if (hasNext) delegate.peek() else null
+
+ override def next(): AnyRef = nextEvent()
+
+ override def getElementText: String = {
+ val text = new StringBuilder
+ var complete = false
+ while (!complete && hasNext) {
+ nextEvent() match {
+ case c: Characters => text.append(c.getData)
+ case _: EndElement => complete = true
+ case _: StartElement =>
+ throw new XMLStreamException("Element text contains a nested start element")
+ case _ =>
+ }
+ }
+ text.toString()
+ }
+
+ override def nextTag(): XMLEvent = {
+ var event = nextEvent()
+ while (event.isCharacters && event.asCharacters().isWhiteSpace) {
+ event = nextEvent()
+ }
+ if (!event.isStartElement && !event.isEndElement) {
+ throw new XMLStreamException(s"Expected a start or end element, but found $event")
+ }
+ event
+ }
+
+ override def getProperty(name: String): AnyRef = delegate.getProperty(name)
+
+ def drain(): Unit = while (hasNext) {
+ nextEvent()
+ }
+
+ override def close(): Unit = drain()
+ }
+
/**
* Parses a single XML string and turns it into either one resulting row or no row (if the
* the record is malformed).
@@ -166,6 +233,7 @@ class StaxXmlParser(
// ValidatorUtil.newValidator throws this when the JAXP implementation cannot
// disable external access; that is an environment error, not a bad record.
case e: UnsupportedOperationException => throw e
+ case DuplicateMapKeyUtils(e) => throw e
case e@(_: RuntimeException | _: XMLStreamException | _: MalformedInputException
| _: SAXException) =>
// XML parser currently doesn't support partial results for corrupted records.
@@ -262,14 +330,16 @@ class StaxXmlParser(
throw BadRecordException(xmlLiteral, () => Array.empty,
wrappedCharException)
case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = xmlLiteral,
- partialResults = () => Array(row),
- cause)
+ DuplicateMapKeyUtils.cause(cause) match {
+ case Some(e) => throw e
+ case None =>
+ throw BadRecordException(record = xmlLiteral, partialResults = () => Array(row), cause)
+ }
case PartialResultArrayException(rows, cause) =>
throw BadRecordException(record = xmlLiteral, partialResults = () => rows, cause)
case e: Throwable =>
SparkErrorUtils.getRootCause(e) match {
+ case DuplicateMapKeyUtils(duplicate) => throw duplicate
case _: FileNotFoundException if options.ignoreMissingFiles =>
logWarning("Skipped missing file", e)
parser.close()
@@ -314,45 +384,47 @@ class StaxXmlParser(
attributes: Array[Attribute]): Any = dt match {
case st: StructType => convertObject(parser, st)
case MapType(StringType, vt, _) => convertMap(parser, vt, attributes)
+ case MapType(kt @ (_: CharType | _: VarcharType), vt, _) =>
+ convertConstrainedMap(parser, kt, vt, attributes)
case ArrayType(st, _) => convertField(parser, st, startElementName)
case VariantType =>
StaxXmlParser.convertVariant(parser, attributes, options)
- case _: StringType =>
+ case dt: StringType =>
convertTo(
StaxXmlParserUtils.currentStructureAsString(
parser, startElementName, options),
- StringType)
+ dt)
}
(parser.peek, dataType) match {
case (_: StartElement, dt: DataType) =>
convertComplicatedType(dt, startElementName, attributes)
- case (_: EndElement, _: StringType) =>
+ case (_: EndElement, dt: StringType) =>
StaxXmlParserUtils.skipNextEndElement(parser, startElementName, options)
// Empty. It's null if "" is the null value
if (options.nullValue == "") {
null
} else {
- UTF8String.fromString("")
+ CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(""), dt)
}
case (_: EndElement, _: DataType) =>
StaxXmlParserUtils.skipNextEndElement(parser, startElementName, options)
null
case (c: Characters, ArrayType(st, _)) =>
- // For `ArrayType`, it needs to return the type of element. The values are merged later.
+ // Consume the element before conversion so a length check cannot leave the
+ // parser on the original text or an unmatched end tag.
parser.next
- val value = convertTo(c.getData, st)
StaxXmlParserUtils.skipNextEndElement(parser, startElementName, options)
- value
+ convertTo(c.getData, st)
case (_: Characters, st: StructType) =>
convertObject(parser, st)
case (_: Characters, VariantType) =>
StaxXmlParser.convertVariant(parser, Array.empty, options)
- case (_: Characters, _: StringType) =>
+ case (_: Characters, dt: StringType) =>
convertTo(
StaxXmlParserUtils.currentStructureAsString(
parser, startElementName, options),
- StringType)
+ dt)
case (c: Characters, _: DataType) if c.isWhiteSpace =>
// When `Characters` is found, we need to look further to decide
// if this is really data or space between other elements.
@@ -390,12 +462,12 @@ class StaxXmlParser(
case e: StartElement =>
val key = StaxXmlParserUtils.getName(e.asStartElement.getName, options)
kvPairs +=
- (UTF8String.fromString(key) -> convertField(parser, valueType, key))
+ (UTF8String.fromString(key) -> convertField(parser, valueType, key))
case c: Characters if !c.isWhiteSpace =>
// Create a value tag field for it
kvPairs +=
- // TODO: We don't support an array value tags in map yet.
- (UTF8String.fromString(options.valueTag) -> convertTo(c.getData, valueType))
+ // TODO: We don't support array value tags in maps yet.
+ (UTF8String.fromString(options.valueTag) -> convertTo(c.getData, valueType))
case _: EndElement | _: EndDocument =>
shouldStop = true
case _ => // do nothing
@@ -404,6 +476,80 @@ class StaxXmlParser(
ArrayBasedMapData(kvPairs.toMap)
}
+ private def convertConstrainedMap(
+ parser: XMLEventReader,
+ keyType: DataType,
+ valueType: DataType,
+ attributes: Array[Attribute]): MapData = {
+ val lastEntries =
+ mutable.LinkedHashMap.empty[UTF8String, (UTF8String, Option[Any])]
+ var badMapException: Option[Throwable] = None
+ def mapKey(raw: UTF8String): UTF8String = {
+ CharVarcharUtils.applyTextParseSemantics(raw, keyType)
+ }
+ def appendPair(rawKey: String, value: Option[Any]): Unit = {
+ try {
+ val rawKeyUtf8 = UTF8String.fromString(rawKey)
+ lastEntries.remove(rawKeyUtf8)
+ lastEntries.update(rawKeyUtf8, (mapKey(rawKeyUtf8), value))
+ } catch {
+ case NonFatal(e) => badMapException = badMapException.orElse(Some(e))
+ }
+ }
+ attributes.foreach { attr =>
+ val value = try {
+ Some(convertTo(attr.getValue, valueType))
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ None
+ }
+ appendPair(options.attributePrefix + attr.getName.getLocalPart, value)
+ }
+ var shouldStop = false
+ while (!shouldStop) {
+ parser.nextEvent match {
+ case e: StartElement =>
+ val rawKey = StaxXmlParserUtils.getName(e.asStartElement.getName, options)
+ val entryParser = new ElementBoundedEventReader(parser)
+ val value = {
+ try {
+ Some(convertField(entryParser, valueType, rawKey))
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case DuplicateMapKeyUtils(e) => throw e
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ None
+ } finally {
+ entryParser.drain()
+ }
+ }
+ appendPair(rawKey, value)
+ case c: Characters if !c.isWhiteSpace =>
+ // Create a value tag field for it
+ // TODO: We don't support array value tags in maps yet.
+ val value = try {
+ Some(convertTo(c.getData, valueType))
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ None
+ }
+ appendPair(options.valueTag, value)
+ case _: EndElement | _: EndDocument =>
+ shouldStop = true
+ case _ => // do nothing
+ }
+ }
+ val mapData = DuplicateMapKeyUtils.buildConstrainedMap(
+ lastEntries, keyType, valueType)
+ badMapException.foreach(throw _)
+ mapData
+ }
+
/**
* Convert XML attributes to a map with the given schema types.
*/
@@ -421,6 +567,30 @@ class StaxXmlParser(
convertedValuesMap.toMap
}
+ /**
+ * Assigns attributes onto `row` one field at a time. Returns the first conversion
+ * failure so callers can keep parse-mode recovery without dropping sibling values.
+ */
+ private def assignAttributes(
+ attributes: Array[Attribute],
+ schema: StructType,
+ row: Array[Any]): Option[Throwable] = {
+ var firstError: Option[Throwable] = None
+ val valuesMap = StaxXmlParserUtils.convertAttributesToValuesMap(attributes, options)
+ valuesMap.foreach { case (f, v) =>
+ getFieldIndex(schema, f).foreach { i =>
+ try {
+ row(i) = convertTo(v, schema(i).dataType)
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case DuplicateMapKeyUtils(e) => throw e
+ case NonFatal(e) => firstError = firstError.orElse(Some(e))
+ }
+ }
+ }
+ firstError
+ }
+
/**
* [[convertObject()]] calls this in order to convert the nested object to a row.
* [[convertObject()]] contains some logic to find out which events are the start
@@ -465,6 +635,19 @@ class StaxXmlParser(
}
}
+ private def convertNestedStruct(
+ parser: XMLEventReader,
+ schema: StructType,
+ startElementName: String,
+ attributes: Array[Attribute]): InternalRow = {
+ val elementParser = new ElementBoundedEventReader(parser)
+ try {
+ convertObjectWithAttributes(elementParser, schema, startElementName, attributes)
+ } finally {
+ elementParser.drain()
+ }
+ }
+
/**
* Parse an object from the event stream into a new InternalRow representing the schema.
* Fields in the xml that are not defined in the requested schema will be dropped.
@@ -474,17 +657,17 @@ class StaxXmlParser(
schema: StructType,
rootAttributes: Array[Attribute] = Array.empty): InternalRow = {
val row = new Array[Any](schema.length)
- // If there are attributes, then we process them first.
- convertAttributes(rootAttributes, schema).toSeq.foreach {
- case (f, v) =>
- getFieldIndex(schema, f).foreach { row(_) = v }
+ var badRecordException: Option[Throwable] = None
+ // Convert each rowTag attribute independently so a CHAR/VARCHAR overflow on one
+ // field does not drop already-converted sibling attributes, and still records the
+ // failure for parse-mode recovery instead of aborting the reader.
+ assignAttributes(rootAttributes, schema, row).foreach { e =>
+ badRecordException = badRecordException.orElse(Some(e))
}
val wildcardColName = options.wildcardColName
val hasWildcard = schema.exists(_.name == wildcardColName)
- var badRecordException: Option[Throwable] = None
-
var shouldStop = false
while (!shouldStop) {
parser.nextEvent match {
@@ -495,7 +678,7 @@ class StaxXmlParser(
getFieldIndex(schema, field) match {
case Some(index) => schema(index).dataType match {
case st: StructType =>
- row(index) = convertObjectWithAttributes(parser, st, field, attributes)
+ row(index) = convertNestedStruct(parser, st, field, attributes)
case ArrayType(dt: DataType, _) =>
val values = Option(row(index))
@@ -503,7 +686,7 @@ class StaxXmlParser(
.getOrElse(ArrayBuffer.empty[Any])
val newValue = dt match {
case st: StructType =>
- convertObjectWithAttributes(parser, st, field, attributes)
+ convertNestedStruct(parser, st, field, attributes)
case VariantType =>
StaxXmlParser.convertVariant(parser, attributes, options)
case dt: DataType =>
@@ -522,12 +705,12 @@ class StaxXmlParser(
if (hasWildcard) {
// Special case: there's an 'any' wildcard element that matches anything else
// as a string (or array of strings, to parse multiple ones)
- val newValue = convertField(parser, StringType, field)
val anyIndex = schema.fieldIndex(wildcardColName)
schema(wildcardColName).dataType match {
- case StringType =>
- row(anyIndex) = newValue
- case ArrayType(StringType, _) =>
+ case dt: StringType =>
+ row(anyIndex) = convertField(parser, dt, field)
+ case ArrayType(et: StringType, _) =>
+ val newValue = convertField(parser, et, field)
val values = Option(row(anyIndex))
.map(_.asInstanceOf[ArrayBuffer[String]])
.getOrElse(ArrayBuffer.empty[String])
@@ -539,6 +722,7 @@ class StaxXmlParser(
}
} catch {
case e: SparkUpgradeException => throw e
+ case DuplicateMapKeyUtils(e) => throw e
case NonFatal(e) =>
// TODO: we don't support partial results now
badRecordException = badRecordException.orElse(Some(e))
@@ -609,7 +793,8 @@ class StaxXmlParser(
timestampNTZFormatter.parseWithoutTimeZoneNanos(datum, t.precision, false)
case _: DateType => parseXmlDate(datum, options)
case _: TimeType => timeFormatter.parse(datum)
- case _: StringType => UTF8String.fromString(datum)
+ case dt: StringType =>
+ CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(datum), dt)
case _: BinaryType => binaryParser(UTF8String.fromString(datum))
case _ => throw new SparkIllegalArgumentException(
errorClass = "_LEGACY_ERROR_TEMP_3244",
@@ -653,7 +838,7 @@ class StaxXmlParser(
case LongType => signSafeToLong(value)
case DoubleType => signSafeToDouble(value)
case BooleanType => castTo(value, BooleanType)
- case StringType => castTo(value, StringType)
+ case dt: StringType => castTo(value, dt)
case BinaryType => castTo(value, BinaryType)
case DateType => castTo(value, DateType)
case TimestampType => castTo(value, TimestampType)
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 9d3dd349a765..1c4c93dfebe6 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
@@ -6798,10 +6798,13 @@ object SQLConf {
}
val MAP_KEY_DEDUP_POLICY = buildConf("spark.sql.mapKeyDedupPolicy")
- .doc("The policy to deduplicate map keys in builtin function: CreateMap, MapFromArrays, " +
- "MapFromEntries, StringToMap, MapConcat and TransformKeys. When EXCEPTION, the query " +
- "fails if duplicated map keys are detected. When LAST_WIN, the map key that is inserted " +
- "at last takes precedence.")
+ .doc("The policy to deduplicate map keys in built-in functions: CreateMap, MapFromArrays, " +
+ "MapFromEntries, StringToMap, MapConcat and TransformKeys. The policy also applies in " +
+ "schema-driven XML parsing, including from_xml and the XML data source, when CHAR/VARCHAR " +
+ "keys normalize to the same value. " +
+ "EXCEPTION fails the query when duplicate keys are detected. LAST_WIN makes the last " +
+ "inserted key take precedence. Repeated raw XML map keys retain their historical " +
+ "last-wins behavior. Ordinary STRING keys retain each parser's historical behavior.")
.version("3.0.0")
.enumConf(MapKeyDedupPolicy)
.createWithDefault(MapKeyDedupPolicy.EXCEPTION)
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 0cda94d7b188..7e9939ded587 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
@@ -171,6 +171,19 @@ 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)
+
+ checkEvaluation(
+ SchemaOfCsv(input, options),
+ "STRUCT<_c0: INT, _c1: INT>")
+ checkEvaluation(
+ CsvToStructs(schema, options, input, UTC_OPT),
+ InternalRow(1, 2))
+ }
+
test("to_csv - struct") {
val schema = StructType(StructField("a", IntegerType) :: Nil)
val struct = Literal.create(create_row(1), schema)
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 d8d68ef75c10..a5d30b417835 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,6 +803,15 @@ 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 a0011e1dbbb9..c1c53b720761 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,6 +371,15 @@ 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/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
index 61ee06d4aaf1..be5d5f0afd02 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.util.CollationFactory
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType, LongType, StringType}
+import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType, LongType, StringType, StructType}
class ConstraintPropagationSuite extends PlanTest {
@@ -56,6 +56,21 @@ class ConstraintPropagationSuite extends PlanTest {
Cast(expr, dataType, Option(TimeZone.getDefault().getID))
}
+ test("from_json/csv/xml propagate null constraints") {
+ val input = $"input".string
+ val relation = LocalRelation(input)
+ val schema = new StructType().add("value", StringType)
+ val expressions = Seq(
+ CsvToStructs(schema, Map.empty, input),
+ JsonToStructs(schema, Map.empty, input),
+ XmlToStructs(schema, Map.empty, input))
+
+ expressions.foreach { expression =>
+ val constraints = relation.where(IsNotNull(expression)).constraints
+ assert(constraints.contains(IsNotNull(input)), expression)
+ }
+ }
+
test("propagating constraints in filters") {
val tr = LocalRelation($"a".int, $"b".string, $"c".int)
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 40783ceaa9fd..b14f113ad036 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.{
ArrayJoin, Attribute, Concat, EqualTo, Expression, GreaterThan, InSet, Literal, ScalarSubquery,
- StringRPad, StringToMap, Upper
+ StringRPad, StringToMap, SupportTrimmedCharInput, Upper
}
import org.apache.spark.sql.catalyst.expressions.Cast.toSQLId
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
@@ -1020,6 +1020,46 @@ trait CharVarcharTestSuite extends QueryTest {
class BasicCharVarcharTestSuite extends SharedSparkSession {
import testImplicits._
+ private def assertParseExceedLimit(query: String, expectedLimit: String = "5"): Unit = {
+ assertParseExceedLimitError(sql(query).collect(), expectedLimit)
+ }
+
+ private def assertParseExceedLimitError(body: => Any, expectedLimit: String = "5"): Unit = {
+ val e = intercept[SparkException] { body }
+ val cause = e.getCause match {
+ case r: SparkRuntimeException => r
+ case other =>
+ Option(other).flatMap(t => Option(t.getCause)).getOrElse(other) match {
+ case r: SparkRuntimeException => r
+ case _ => fail(s"expected EXCEED_LIMIT_LENGTH cause, got: $e")
+ }
+ }
+ checkError(
+ exception = cause,
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> expectedLimit))
+ }
+
+ private def assertDuplicateMapKey(query: String, expectedKey: String = "a "): Unit = {
+ assertDuplicateMapKeyError(sql(query).collect(), expectedKey)
+ }
+
+ private def assertDuplicateMapKeyError(body: => Any, expectedKey: String = "a "): Unit = {
+ val error = intercept[Exception] { body }
+ val duplicateError = Iterator.iterate[Throwable](error)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case e: SparkRuntimeException if e.getCondition == "DUPLICATED_MAP_KEY" => e
+ }
+ .getOrElse(fail("expected DUPLICATED_MAP_KEY cause", error))
+ checkError(
+ exception = duplicateError,
+ condition = "DUPLICATED_MAP_KEY",
+ parameters = Map(
+ "key" -> expectedKey,
+ "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ }
+
test("user-specified schema in cast") {
def assertNoCharType(df: DataFrame): Unit = {
checkAnswer(df, Row("0"))
@@ -2401,6 +2441,527 @@ 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
+ }
+
+ checkAnswer(
+ checkTextInput(
+ """SELECT from_json(CAST('{"a":1}' AS CHAR(12)), 'a INT')""",
+ CharType(12)),
+ Row(Row(1)))
+ checkAnswer(
+ checkTextInput(
+ """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)),
+ Row(Row(1)))
+ Seq(
+ ("CAST('1 ' AS VARCHAR(2))", VarcharType(2)),
+ ("CAST('1 ' AS STRING)", StringType)).foreach { case (input, inputType) =>
+ checkAnswer(
+ checkTextInput(
+ s"SELECT schema_of_csv($input, map('delimiter', ' '))",
+ inputType),
+ Row("STRUCT<_c0: INT, _c1: STRING>"))
+ }
+
+ val jsonChar = sql("""SELECT from_json('{"a": "str"}', 'a CHAR(5)')""")
+ val jsonCharType = jsonChar.schema.head.dataType.asInstanceOf[StructType]
+ assert(jsonCharType.head.dataType === CharType(5))
+ checkAnswer(jsonChar, Row(Row("str ")))
+
+ val jsonVarchar = sql("""SELECT from_json('{"a": "ab"}', 'a VARCHAR(5)')""")
+ val jsonVarcharType = jsonVarchar.schema.head.dataType.asInstanceOf[StructType]
+ assert(jsonVarcharType.head.dataType === VarcharType(5))
+ checkAnswer(jsonVarchar, Row(Row("ab")))
+
+ // Default PERMISSIVE mode returns a parsed struct whose failed field is null.
+ Seq("CHAR(5)", "VARCHAR(5)").foreach { dataType =>
+ checkAnswer(
+ sql(s"""SELECT from_json('{"a": "abcdef"}', 'a $dataType')"""),
+ Row(Row(null)))
+ assertParseExceedLimit(
+ s"""SELECT from_json(
+ | '{"a": "abcdef"}',
+ | 'a $dataType',
+ | map('mode', 'FAILFAST'))""".stripMargin)
+ }
+
+ checkAnswer(sql("SELECT from_csv('str', 'a CHAR(5)')"), Row(Row("str ")))
+ Seq("CHAR(5)", "VARCHAR(5)").foreach { dataType =>
+ checkAnswer(sql(s"SELECT from_csv('abcdef', 'a $dataType')"), Row(Row(null)))
+ assertParseExceedLimit(
+ s"SELECT from_csv('abcdef', 'a $dataType', map('mode', 'FAILFAST'))")
+ }
+
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><a>str</a></ROW>', 'a CHAR(5)')"),
+ Row(Row("str ")))
+ checkAnswer(
+ sql(
+ """SELECT from_xml(
+ | '<ROW><a></a></ROW>',
+ | 'a CHAR(5)',
+ | map('nullValue', 'NULL'))""".stripMargin),
+ Row(Row(" ")))
+ Seq("CHAR(5)", "VARCHAR(5)").foreach { dataType =>
+ checkAnswer(
+ sql(s"SELECT from_xml('<ROW><a>abcdef</a></ROW>', 'a $dataType')"),
+ Row(Row(null)))
+ assertParseExceedLimit(
+ s"SELECT from_xml('<ROW><a>abcdef</a></ROW>', 'a $dataType', " +
+ "map('mode', 'FAILFAST'))")
+ }
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><m><ab>1</ab></m></ROW>', 'm MAP<CHAR(4), INT>')"),
+ Row(Row(Map("ab " -> 1))))
+ checkAnswer(
+ sql(
+ """SELECT from_xml(
+ | '<ROW><other>str</other></ROW>',
+ | 'xs_any CHAR(5)',
+ | map('wildcardColName', 'xs_any'))""".stripMargin),
+ Row(Row("str ")))
+ checkAnswer(
+ sql(
+ """SELECT from_xml(
+ | '<ROW><first>a</first><second>bc</second></ROW>',
+ | 'xs_any ARRAY<CHAR(5)>',
+ | map('wildcardColName', 'xs_any'))""".stripMargin),
+ Row(Row(Seq("a ", "bc "))))
+ checkAnswer(
+ sql(
+ """SELECT from_xml(
+ | '<ROW><other>abcdef</other></ROW>',
+ | 'xs_any CHAR(5)',
+ | map('wildcardColName', 'xs_any'))""".stripMargin),
+ Row(Row(null)))
+ checkAnswer(
+ sql(
+ """SELECT from_xml(
+ | '<ROW><other>abcdef</other></ROW>',
+ | 'xs_any ARRAY<CHAR(5)>',
+ | map('wildcardColName', 'xs_any'))""".stripMargin),
+ Row(Row(null)))
+ assertParseExceedLimit(
+ """SELECT from_xml(
+ | '<ROW><other>abcdef</other></ROW>',
+ | 'xs_any CHAR(5)',
+ | map('wildcardColName', 'xs_any', 'mode', 'FAILFAST'))""".stripMargin)
+ assertParseExceedLimit(
+ """SELECT from_xml(
+ | '<ROW><other>abcdef</other></ROW>',
+ | 'xs_any ARRAY<CHAR(5)>',
+ | map('wildcardColName', 'xs_any', 'mode', 'FAILFAST'))""".stripMargin)
+ withTempPath { path =>
+ Seq("<ROW><m><a>1</a></m></ROW>").toDS().write.text(path.getCanonicalPath)
+ val xmlDataFrame = spark.read
+ .option("rowTag", "ROW")
+ .schema("m MAP<CHAR(2), INT>")
+ .xml(path.getCanonicalPath)
+ assert(
+ xmlDataFrame.schema("m").dataType ===
+ MapType(CharType(2), IntegerType, valueContainsNull = true))
+ checkAnswer(xmlDataFrame, Row(Map("a " -> 1)))
+ }
+ withTempPath { path =>
+ Seq("""{"c":"ab"}""").toDS().write.text(path.getCanonicalPath)
+ val jsonDataFrame = spark.read.schema("c CHAR(4)").json(path.getCanonicalPath)
+ assert(jsonDataFrame.schema("c").dataType === CharType(4))
+ checkAnswer(jsonDataFrame, Row("ab "))
+ }
+ withTempPath { path =>
+ Seq("abcdef").toDS().write.text(path.getCanonicalPath)
+ val csvOverflow = spark.read.schema("c VARCHAR(4)").csv(path.getCanonicalPath)
+ assert(csvOverflow.schema("c").dataType === VarcharType(4))
+ checkAnswer(csvOverflow, Row(null))
+ }
+
+ checkAnswer(
+ sql("""SELECT schema_of_json(CAST('{"a":1}' AS VARCHAR(20)))"""),
+ Row("STRUCT<a: BIGINT>"))
+ checkAnswer(
+ sql("SELECT schema_of_csv(CAST('1,abc' AS VARCHAR(20)))"),
+ Row("STRUCT<_c0: INT, _c1: STRING>"))
+ checkAnswer(
+ sql("SELECT schema_of_xml(CAST('<ROW><a>1</a></ROW>' AS VARCHAR(40)))"),
+ Row("STRUCT<a: BIGINT>"))
+ checkAnswer(
+ sql("""SELECT length(CAST('{"a":1}' AS CHAR(20)))"""),
+ Row(20))
+ checkAnswer(
+ sql("""SELECT schema_of_json(CAST('{"a":1}' AS CHAR(20)))"""),
+ Row("STRUCT<a: BIGINT>"))
+ 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))
+ checkAnswer(
+ sql("SELECT schema_of_xml(CAST('<ROW><a>1</a></ROW>' AS CHAR(30)))"),
+ Row("STRUCT<a: BIGINT>"))
+ Seq(
+ "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>",
+ "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))
+ }
+ }
+ }
+
+ test("SPARK-59274: normalized XML CHAR map keys honor the dedup policy") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val xmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><a>1</a>9</m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('valueTag', 'a ')).m""".stripMargin
+ val varcharXmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><ab>1</ab>2</m></ROW>',
+ | 'm MAP<VARCHAR(2), INT>',
+ | map('valueTag', 'ab ')).m""".stripMargin
+ val exactCharXmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><a>1</a><a>2</a></m></ROW>',
+ | 'm MAP<CHAR(2), INT>').m""".stripMargin
+ val exactVarcharXmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><ab>1</ab><ab>2</ab></m></ROW>',
+ | 'm MAP<VARCHAR(2), INT>').m""".stripMargin
+ val interleavedCharXmlQuery =
+ """SELECT map_entries(from_xml(
+ | '<ROW><m><a>1</a>2<a>3</a></m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('valueTag', 'a ')).m)""".stripMargin
+ val badXmlKeyThenSiblingQuery =
+ """SELECT from_xml(
+ | '<ROW><m><abc>1</abc></m><tail>2</tail></ROW>',
+ | 'm MAP<CHAR(2), INT>, tail INT').tail""".stripMargin
+ val badXmlKeyBeforeDuplicateQuery =
+ """SELECT from_xml(
+ | '<ROW><m><abc>0</abc><a>1</a>2</m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('valueTag', 'a ')).m""".stripMargin
+ val malformedXmlValueBeforeDuplicateQuery =
+ """SELECT from_xml(
+ | '<ROW><m><a>bad</a>2</m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('valueTag', 'a ')).m""".stripMargin
+ val ignoreCorruptXmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><a>1</a>2</m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('valueTag', 'a ', 'ignoreCorruptFiles', 'true')).m""".stripMargin
+ val attributePaddingQuery =
+ """SELECT from_xml(
+ | '<ROW><m a="1"><b>2</b></m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('attributePrefix', '')).m""".stripMargin
+ val attributeOverflowQuery =
+ """SELECT from_xml(
+ | '<ROW><m abc="1"><b>2</b></m></ROW>',
+ | 'm MAP<VARCHAR(2), INT>',
+ | map('attributePrefix', '')).m""".stripMargin
+ val attributeOverflowFailfastQuery =
+ """SELECT from_xml(
+ | '<ROW><m abc="1"><b>2</b></m></ROW>',
+ | 'm MAP<VARCHAR(2), INT>',
+ | map('attributePrefix', '', 'mode', 'FAILFAST')).m""".stripMargin
+ val attributeCollisionQuery =
+ """SELECT from_xml(
+ | '<ROW><m a="1"><b>0</b>2</m></ROW>',
+ | 'm MAP<CHAR(2), INT>',
+ | map('attributePrefix', '', 'valueTag', 'a ')).m""".stripMargin
+
+ assertDuplicateMapKey(xmlQuery)
+ assertDuplicateMapKey(varcharXmlQuery, expectedKey = "ab")
+ checkAnswer(sql(exactCharXmlQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(exactVarcharXmlQuery), Row(Map("ab" -> 2)))
+ assertDuplicateMapKey(interleavedCharXmlQuery)
+ assertDuplicateMapKey(badXmlKeyBeforeDuplicateQuery)
+ assertDuplicateMapKey(malformedXmlValueBeforeDuplicateQuery)
+ assertDuplicateMapKey(ignoreCorruptXmlQuery)
+ checkAnswer(sql(badXmlKeyThenSiblingQuery), Row(2))
+ checkAnswer(sql(attributePaddingQuery), Row(Map("a " -> 1, "b " -> 2)))
+ checkAnswer(sql(attributeOverflowQuery), Row(null))
+ assertParseExceedLimit(attributeOverflowFailfastQuery, expectedLimit = "2")
+ assertDuplicateMapKey(attributeCollisionQuery)
+
+ withTempPath { path =>
+ Seq("<ROW><m><a>1</a>2</m></ROW>").toDS().write.text(path.getCanonicalPath)
+ def readXmlMap(): DataFrame = spark.read
+ .option("rowTag", "ROW")
+ .option("valueTag", "a ")
+ .schema("m MAP<CHAR(2), INT>")
+ .xml(path.getCanonicalPath)
+
+ assertDuplicateMapKeyError(readXmlMap().collect())
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ checkAnswer(readXmlMap(), Row(Map("a " -> 2)))
+ }
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ checkAnswer(sql(xmlQuery), Row(Map("a " -> 9)))
+ checkAnswer(sql(varcharXmlQuery), Row(Map("ab" -> 2)))
+ checkAnswer(sql(exactCharXmlQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(exactVarcharXmlQuery), Row(Map("ab" -> 2)))
+ checkAnswer(sql(interleavedCharXmlQuery), Row(Seq(Row("a ", 3))))
+ checkAnswer(sql(badXmlKeyBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(malformedXmlValueBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(ignoreCorruptXmlQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(attributeCollisionQuery), Row(Map("a " -> 2, "b " -> 0)))
+ }
+ }
+ }
+
+ test("SPARK-59274: pretty-printed XML map failures preserve parser position") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val overflowQueries = Seq("CHAR(5)", "VARCHAR(5)").map { valueType =>
+ s"""SELECT from_xml(
+ | '<ROW>
+ | <m>
+ | <b>abcdef</b>
+ | <a>x</a>y
+ | </m>
+ | <tail>9</tail>
+ | </ROW>',
+ | 'm MAP<CHAR(2), $valueType>, tail INT',
+ | map('valueTag', 'a '))""".stripMargin
+ }
+ val nestedFailureQuery =
+ """SELECT from_xml(
+ | '<ROW>
+ | <m>
+ | <b><x>bad</x></b>
+ | <a><x>1</x></a>ignored
+ | </m>
+ | <tail>9</tail>
+ | </ROW>',
+ | 'm MAP<CHAR(2), MAP<CHAR(2), INT>>, tail INT',
+ | map('valueTag', 'a '))""".stripMargin
+
+ (overflowQueries :+ nestedFailureQuery).foreach { query =>
+ // A duplicate, rather than an AssertionError from XML recovery, wins under EXCEPTION.
+ withClue(s"$query: ") {
+ assertDuplicateMapKey(query)
+ }
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ // The failed map remains null, but its complete element is consumed and the `tail` field
+ // is parsed.
+ (overflowQueries :+ nestedFailureQuery).foreach { query =>
+ checkAnswer(sql(query), Row(Row(null, 9)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-59274: nested XML attribute failures preserve parser position") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq("STRUCT<c CHAR(2), v INT>", "ARRAY<STRUCT<c VARCHAR(2), v INT>>")
+ .foreach { nestedType =>
+ val xml = "<ROW><nested c=\"abc\"><v>1</v></nested><tail>9</tail></ROW>"
+ val schema = s"nested $nestedType, tail INT"
+ val permissiveQuery =
+ s"""SELECT from_xml(
+ | '$xml',
+ | '$schema',
+ | map('attributePrefix', ''))""".stripMargin
+ checkAnswer(sql(permissiveQuery), Row(Row(null, 9)))
+ assertParseExceedLimit(
+ s"""SELECT from_xml(
+ | '$xml',
+ | '$schema',
+ | map('attributePrefix', '', 'mode', 'FAILFAST'))""".stripMargin,
+ expectedLimit = "2")
+ }
+ }
+ }
+
+ test("SPARK-59274: JSON map value overflow keeps EXCEED_LIMIT_LENGTH") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val json = """{"m":{"k":"abcdef"},"tail":1}"""
+ val goodJson = """{"m":{"k":"ab"},"tail":2}"""
+ Seq("CHAR(2)", "VARCHAR(2)").foreach { valueType =>
+ val schema = s"m MAP<STRING, $valueType>, tail INT"
+ val fromJson = s"SELECT from_json('$json', '$schema')"
+ Seq(true, false).foreach { partial =>
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> partial.toString) {
+ withClue(s"$valueType partial=$partial") {
+ if (partial) {
+ checkAnswer(sql(fromJson), Row(Row(null, 1)))
+ } else {
+ checkAnswer(sql(fromJson), Row(Row(null, null)))
+ }
+ assertParseExceedLimit(
+ s"SELECT from_json('$json', '$schema', map('mode', 'FAILFAST'))",
+ expectedLimit = "2")
+
+ withTempPath { path =>
+ Seq(json, goodJson).toDS()
+ .repartition(1)
+ .write.text(path.getCanonicalPath)
+ val fileSchema = schema
+ val permissive = spark.read.schema(fileSchema).json(path.getCanonicalPath)
+ if (partial) {
+ checkAnswer(permissive, Seq(Row(null, 1), Row(Map("k" -> "ab"), 2)))
+ } else {
+ checkAnswer(permissive, Seq(Row(null, null), Row(Map("k" -> "ab"), 2)))
+ }
+ val failFast = spark.read
+ .option("mode", "FAILFAST")
+ .schema(fileSchema)
+ .json(path.getCanonicalPath)
+ assertParseExceedLimitError(failFast.collect(), expectedLimit = "2")
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-59274: XML rowTag attribute overflow uses parse mode") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq("CHAR(2)", "VARCHAR(2)").foreach { attrType =>
+ withTempPath { path =>
+ Seq(
+ """<ROWS><ROW c="abcdef"><v>1</v></ROW><ROW c="ok"><v>2</v></ROW></ROWS>"""
+ ).toDS().write.text(path.getCanonicalPath)
+ val schema = s"c $attrType, v INT"
+ withClue(attrType) {
+ val permissive = spark.read
+ .option("rowTag", "ROW")
+ .option("attributePrefix", "")
+ .schema(schema)
+ .xml(path.getCanonicalPath)
+ checkAnswer(permissive, Seq(Row(null, 1), Row("ok", 2)))
+ val failFast = spark.read
+ .option("rowTag", "ROW")
+ .option("attributePrefix", "")
+ .option("mode", "FAILFAST")
+ .schema(schema)
+ .xml(path.getCanonicalPath)
+ assertParseExceedLimitError(failFast.collect(), expectedLimit = "2")
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-59274: mixed XML rowTag attributes keep valid siblings") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val xml = """<ROW good="ok" bad="abcdef"/>"""
+ val schema = "good CHAR(2), bad VARCHAR(2)"
+ val opts = "map('attributePrefix', '')"
+ checkAnswer(
+ sql(s"SELECT from_xml('$xml', '$schema', $opts)"),
+ Row(Row("ok", null)))
+ assertParseExceedLimit(
+ s"SELECT from_xml('$xml', '$schema', map('attributePrefix', '', 'mode', 'FAILFAST'))",
+ expectedLimit = "2")
+ withTempPath { path =>
+ Seq(
+ """<ROWS><ROW good="ok" bad="abcdef"/><ROW good="ab" bad="cd"/></ROWS>"""
+ ).toDS().write.text(path.getCanonicalPath)
+ val permissive = spark.read
+ .option("rowTag", "ROW")
+ .option("attributePrefix", "")
+ .schema(schema)
+ .xml(path.getCanonicalPath)
+ checkAnswer(permissive, Seq(Row("ok", null), Row("ab", "cd")))
+ val failFast = spark.read
+ .option("rowTag", "ROW")
+ .option("attributePrefix", "")
+ .option("mode", "FAILFAST")
+ .schema(schema)
+ .xml(path.getCanonicalPath)
+ assertParseExceedLimitError(failFast.collect(), expectedLimit = "2")
+ }
+ }
+ }
+
+ test("SPARK-59274: ordinary STRING map duplicate behavior is unchanged") {
+ Seq("false", "true").foreach { standardSemantics =>
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> standardSemantics) {
+ checkAnswer(
+ sql(
+ """SELECT size(m), map_entries(m)
+ |FROM (
+ | SELECT from_json('{"a":1,"a":2}', 'MAP<STRING, INT>') AS m
+ |)""".stripMargin),
+ Row(2, Seq(Row("a", 1), Row("a", 2))))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><a>1</a><a>2</a></m></ROW>',
+ | 'm MAP<STRING, INT>').m""".stripMargin),
+ Row(Map("a" -> 2)))
+ }
+ }
+ }
+
+ test("SPARK-59274: collated CHAR/VARCHAR map keys honor the dedup policy") {
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true",
+ SQLConf.ALLOW_COLLATIONS_IN_MAP_KEYS.key -> "true") {
+ val xmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><a>1</a><A>2</A></m></ROW>',
+ | 'm MAP<CHAR(2) COLLATE UTF8_LCASE, INT>').m""".stripMargin
+ val varcharXmlQuery =
+ """SELECT from_xml(
+ | '<ROW><m><ab>1</ab><AB>2</AB></m></ROW>',
+ | 'm MAP<VARCHAR(2) COLLATE UTF8_LCASE, INT>').m""".stripMargin
+
+ assertDuplicateMapKey(xmlQuery, expectedKey = "A ")
+ assertDuplicateMapKey(varcharXmlQuery, expectedKey = "AB")
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ checkAnswer(sql(xmlQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(varcharXmlQuery), Row(Map("ab" -> 2)))
+ }
+ }
+ }
}
class FileSourceCharVarcharTestSuite extends CharVarcharTestSuite with SharedSparkSession {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org