cloud-fan commented on code in PR #58545:
URL: https://github.com/apache/spark/pull/58545#discussion_r4037332193
##########
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>")
Review Comment:
**Non-blocking (P2):** These direct-expression checks bypass
ImplicitTypeCoercion. In an analyzed SQL expression the CHAR child is first
cast to plain STRING, so SupportTrimmedCharInput no longer sees a CHAR type and
does not trim its padding. The tests can therefore pass while the public
schema_of_* path still parses the padded document; please exercise the analyzed
SQL/DataFrame path here.
**Recommended change:** Preserve or apply trailing-padding removal at the
analyzer boundary for first-class CHAR document inputs, and replace the
direct-only checks with regression-sensitive analyzed SQL/DataFrame coverage
across CSV, JSON, and XML.
**Why this works:** Carry the original first-class CHAR decision through
coercion or insert StringTrimRight before the child becomes plain STRING, so
evaluation no longer depends on inspecting an already-coerced child type.
**Scope:** Make the public analyzed expression retain the same CHAR-input
trimming semantics exercised by unit construction.
**Compatibility:** VARCHAR and ordinary STRING document inputs remain
untrimmed, and existing post-analysis configuration stability is retained.
**Risks:** Applying trimming after coercion without retaining provenance
could trim unrelated STRING casts.
**Constraints:** Do not trim VARCHAR or ordinary STRING inputs. Keep the
effective semantics stable if SQLConf changes after analysis. Do not rely on
direct case-class construction as the only public-behavior signal.
**Success:** Analyzed from_csv, from_json, from_xml, and schema_of_*
expressions ignore CHAR right padding while preserving non-padding content.
Direct and analyzed evaluation no longer disagree for the same first-class CHAR
input.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,33 +619,53 @@ class JacksonParser(
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
- val keys = ArrayBuffer.empty[UTF8String]
- val values = ArrayBuffer.empty[Any]
- var badRecordException: Option[Throwable] = None
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType): MapData = {
+ val entries = ArrayBuffer.empty[(UTF8String, UTF8String, Option[Any])]
Review Comment:
**Non-blocking (P2):** This routes every ordinary STRING-key map through the
constrained-key Tuple3/Option representation and later collection passes. That
adds multiple per-entry allocations and O(n) traversals to the unchanged common
JSON/XML path; the extra normalization bookkeeping should be confined to key
types that require it.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6729,10 +6729,12 @@ 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, and in
from_json and from_xml " +
+ "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. Exact repeated field names and ordinary
STRING keys retain " +
Review Comment:
**Nit (P3):** Ordinary STRING-key JSON does not have last-wins behavior for
all SQL observables: buildParsedMap preserves both exact duplicate pairs, so
size(from_json('{"a":1,"a":2}', 'MAP<STRING, INT>')) is 2 and map_entries
exposes both values. Please describe that historical behavior rather than
calling it last-wins.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala:
##########
@@ -374,31 +375,68 @@ class StaxXmlParser(
*/
private def convertMap(
parser: XMLEventReader,
+ keyType: DataType,
valueType: DataType,
attributes: Array[Attribute]): MapData = {
- val kvPairs = ArrayBuffer.empty[(UTF8String, Any)]
+ val kvPairs = ArrayBuffer.empty[(UTF8String, UTF8String, Option[Any])]
+ var badMapException: Option[Throwable] = None
+ def mapKey(raw: String): UTF8String = {
+ CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(raw),
keyType)
+ }
+ def appendPair(rawKey: String, value: Option[Any]): Unit = {
+ try {
+ kvPairs += ((UTF8String.fromString(rawKey), mapKey(rawKey), value))
+ } catch {
+ case NonFatal(e) => badMapException = badMapException.orElse(Some(e))
+ }
+ }
attributes.foreach { attr =>
- kvPairs += (UTF8String.fromString(options.attributePrefix +
attr.getName.getLocalPart)
- -> convertTo(attr.getValue, valueType))
+ 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 key = StaxXmlParserUtils.getName(e.asStartElement.getName,
options)
- kvPairs +=
- (UTF8String.fromString(key) -> convertField(parser, valueType, key))
+ val rawKey = StaxXmlParserUtils.getName(e.asStartElement.getName,
options)
+ val value = try {
+ Some(convertField(parser, valueType, rawKey))
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case DuplicateMapKeyUtils(e) => throw e
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
Review Comment:
**Blocking (P1):** A nested value failure is caught while the XML reader is
still inside that value. The loop can then reinterpret inner fields as map keys
and stop on the inner end tag; the parent subsequently stops on the map end tag
and loses later row fields. For example, a malformed nested value in m can
cause a valid tail field after m to disappear.
**Recommended change:** On a recoverable XML map-value failure, consume the
remainder of the current entry through its matching end before recording the
failure and continuing the map; keep key/value insertion atomic.
**Why this works:** Retain the entry's start name/depth, and when conversion
fails, advance or skip to that exact matching end element before appending no
value and resuming the enclosing map loop.
**Scope:** Restore the XML event reader to the failed map entry's boundary
before permissive parsing continues.
**Compatibility:** Successful entries, invalid constrained-key recovery,
duplicate handling, and optimized file-source parsing retain their current
semantics.
**Risks:** An imprecise skip can cross the map or row boundary and discard
valid siblings.
**Constraints:** Do not consume the next map entry or enclosing row sibling.
Invoke the value converter at most once per entry. Preserve duplicate-policy
precedence and ordinary STRING behavior.
**Success:** A malformed nested XML map value does not expose its inner
elements as peer map keys. A valid row field after the map is still parsed in
permissive mode. The remembered map failure continues through the configured
bad-record path.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.{CharType, DataType, StringType, VarcharType}
+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 a parsed JSON/XML object as a map.
+ *
+ * CHAR/VARCHAR keys: exact serialized names keep the last value, then
+ * `spark.sql.mapKeyDedupPolicy` applies to normalized keys. Failed values
still
+ * occupy a slot so collisions are visible.
+ *
+ * Example: `from_json('{"a":1,"a ":2}', 'MAP<CHAR(2), INT>')` raises
+ * DUPLICATED_MAP_KEY under EXCEPTION and keeps `a ` -> 2 under LAST_WIN.
+ * Exact `{"a":1,"a":2}` is last-wins regardless of policy.
+ *
+ * Ordinary STRING keys keep historical last-wins. When
+ * `collapseOrdinaryStringKeys` is true (XML), duplicates collapse via `Map`.
+ * When false (JSON), retained pairs are stored as parallel arrays.
+ */
+ def buildParsedMap(
+ entries: Seq[(UTF8String, UTF8String, Option[Any])],
+ keyType: DataType,
+ valueType: DataType,
+ collapseOrdinaryStringKeys: Boolean): MapData = {
+ keyType match {
+ case _: CharType | _: VarcharType =>
+ buildMapWithLastRawKeyWins(entries, keyType, valueType)
+ case _ if collapseOrdinaryStringKeys =>
+ ArrayBasedMapData(
+ entries.flatMap { case (_, key, value) => value.map(key -> _)
}.toMap)
+ case _ =>
+ val retained = entries.flatMap { case (_, key, value) => value.map(key
-> _) }
+ ArrayBasedMapData(retained.map(_._1).toArray,
retained.map(_._2).toArray)
+ }
+ }
+
+ private def buildMapWithLastRawKeyWins(
+ entries: Seq[(UTF8String, UTF8String, Option[Any])],
+ keyType: DataType,
+ valueType: DataType): MapData = {
+ val indices = lastOccurrenceIndices(entries.map(_._1).toArray)
Review Comment:
**Nit (P3):** The existing entry sequence already contains the raw keys, but
this copies them into another array just to compute indices. The EXCEPTION path
then builds a normalized-key distinct index before ArrayBasedMapBuilder indexes
the same successful keys again, adding redundant O(n) allocation and traversal
for every constrained-key map.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,33 +619,53 @@ class JacksonParser(
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
- val keys = ArrayBuffer.empty[UTF8String]
- val values = ArrayBuffer.empty[Any]
- var badRecordException: Option[Throwable] = None
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType): MapData = {
+ val entries = ArrayBuffer.empty[(UTF8String, UTF8String, Option[Any])]
+ var partialResultException: Option[Throwable] = None
+ var badMapException: Option[Throwable] = None
+
+ val constrainedKeys =
+ keyType.isInstanceOf[CharType] || keyType.isInstanceOf[VarcharType]
+ // CHAR/VARCHAR maps always drain to END_OBJECT so mapKeyDedupPolicy is
applied
+ // before a stored length error. Partial results still control STRING maps.
+ val drainErrors = constrainedKeys || enablePartialResults
while (nextUntil(parser, JsonToken.END_OBJECT)) {
- keys += UTF8String.fromString(parser.currentName)
- try {
- values += fieldConverter.apply(parser)
+ val rawKey = UTF8String.fromString(parser.currentName)
+ val value = try {
+ Some(fieldConverter.apply(parser))
} catch {
case err: PartialValueException if enablePartialResults =>
- badRecordException = badRecordException.orElse(Some(err.cause))
- values += err.partialResult
- case NonFatal(e) if enablePartialResults =>
- badRecordException = badRecordException.orElse(Some(e))
+ partialResultException =
partialResultException.orElse(Some(err.cause))
+ Some(err.partialResult)
+ case DuplicateMapKeyUtils(e) => throw e
Review Comment:
**Blocking (P1):** With partial results disabled, this new unconditional
catch can resume at a failed scalar inside a nested struct. skipChildren is
then a no-op, so remaining fields owned by that struct are read as keys of the
enclosing map. A single outer key can consequently raise DUPLICATED_MAP_KEY or
materialize nested fields at the wrong level.
**Recommended change:** Track the structural boundary of each JSON map value
and, on a caught failure, drain only the remainder of that value before
resuming outer key iteration; align this behavior for partial-results enabled
and disabled modes.
**Why this works:** Capture the value's starting parsing context/depth
before conversion, then use that ownership marker to advance to the value's
matching boundary after failure rather than applying skipChildren to the
converter's final scalar token.
**Scope:** Prevent nested JSON value tokens from being reinterpreted as
enclosing map entries after conversion failure.
**Compatibility:** Existing partial maps omit values with no partial result,
failed raw keys still affect duplicate detection, and ordinary STRING parsing
remains unchanged.
**Risks:** Incorrect depth accounting can skip the next outer map entry or
stop before the failed nested value is fully drained.
**Constraints:** Do not invoke the field converter more than once. Do not
cross the failed value's owning END_OBJECT or END_ARRAY. Failed-value raw keys
must still participate in normalized duplicate detection where required.
Returned key/value pairs remain atomic.
**Success:** Inner struct field names are never recorded as outer
constrained-map keys. A one-key outer map cannot raise a duplicate solely
because its nested value failed. A valid outer row sibling after the map
remains owned by the row parser.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.{CharType, DataType, StringType, VarcharType}
+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 a parsed JSON/XML object as a map.
+ *
+ * CHAR/VARCHAR keys: exact serialized names keep the last value, then
+ * `spark.sql.mapKeyDedupPolicy` applies to normalized keys. Failed values
still
+ * occupy a slot so collisions are visible.
+ *
+ * Example: `from_json('{"a":1,"a ":2}', 'MAP<CHAR(2), INT>')` raises
+ * DUPLICATED_MAP_KEY under EXCEPTION and keeps `a ` -> 2 under LAST_WIN.
+ * Exact `{"a":1,"a":2}` is last-wins regardless of policy.
+ *
+ * Ordinary STRING keys keep historical last-wins. When
+ * `collapseOrdinaryStringKeys` is true (XML), duplicates collapse via `Map`.
+ * When false (JSON), retained pairs are stored as parallel arrays.
+ */
+ def buildParsedMap(
+ entries: Seq[(UTF8String, UTF8String, Option[Any])],
+ keyType: DataType,
+ valueType: DataType,
+ collapseOrdinaryStringKeys: Boolean): MapData = {
+ keyType match {
+ case _: CharType | _: VarcharType =>
+ buildMapWithLastRawKeyWins(entries, keyType, valueType)
+ case _ if collapseOrdinaryStringKeys =>
+ ArrayBasedMapData(
+ entries.flatMap { case (_, key, value) => value.map(key -> _)
}.toMap)
+ case _ =>
+ val retained = entries.flatMap { case (_, key, value) => value.map(key
-> _) }
+ ArrayBasedMapData(retained.map(_._1).toArray,
retained.map(_._2).toArray)
+ }
+ }
+
+ private def buildMapWithLastRawKeyWins(
+ entries: Seq[(UTF8String, UTF8String, Option[Any])],
+ keyType: DataType,
+ valueType: DataType): MapData = {
+ val indices = lastOccurrenceIndices(entries.map(_._1).toArray)
+ 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))
+ }
+ indices.foreach { index =>
+ val key = entries(index)._2
+ if (!distinctKeys.add(key)) {
+ throw QueryExecutionErrors.duplicateMapKeyFoundError(key)
+ }
+ }
+ }
+ val builder = new ArrayBasedMapBuilder(keyType, valueType)
+ indices.foreach { index =>
+ val (_, normalizedKey, value) = entries(index)
+ value.foreach(builder.put(normalizedKey, _))
+ }
+ builder.build()
+ }
+
+ private def lastOccurrenceIndices(rawKeys: Array[UTF8String]): Seq[Int] = {
+ val lastIndices = mutable.LinkedHashMap.empty[UTF8String, Int]
+ rawKeys.indices.foreach(index => lastIndices.update(rawKeys(index), index))
Review Comment:
**Blocking (P1):** Updating an existing LinkedHashMap value does not move
that key to its last serialized position. For raw entries a -> 1, a-space -> 2,
a -> 3, this returns indices [2, 1]; once CHAR normalization makes the two
names equal, LAST_WIN processes value 3 and then value 2, incorrectly retaining
2 rather than the final serialized value 3.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -2226,6 +2251,429 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
}
}
}
+
+ test("SPARK-59274: from_json/csv/xml honor CHAR/VARCHAR under
standardSemantics") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ 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 turns length failures into a null record.
+ 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_json('{"ab": 1}', 'MAP<CHAR(4), INT>')"""),
+ Row(Map("ab " -> 1)))
+
+ 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>"))
+ }
+ }
+
+ test("SPARK-59274: normalized CHAR map key collisions honor the dedup
policy") {
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true",
+ SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ val jsonQuery =
+ """SELECT from_json('{"a":1,"a ":2}', 'MAP<CHAR(2), INT>')"""
+ val jsonFailfastQuery =
+ """SELECT from_json(
+ | '{"a":1,"a ":2}',
+ | 'MAP<CHAR(2), INT>',
+ | map('mode', 'FAILFAST'))""".stripMargin
+ val varcharJsonQuery =
+ """SELECT from_json('{"ab":1,"ab ":2}', 'MAP<VARCHAR(2), INT>')"""
+ val exactCharJsonQuery =
+ """SELECT from_json('{"a":1,"a":2}', 'MAP<CHAR(2), INT>')"""
+ val exactVarcharJsonQuery =
+ """SELECT from_json('{"ab":1,"ab":2}', 'MAP<VARCHAR(2), INT>')"""
+ val interleavedCharJsonQuery =
+ """SELECT map_entries(from_json(
+ | '{"a":1,"b":2,"a":3}',
+ | 'MAP<CHAR(2), INT>'))""".stripMargin
+ val varcharOverflowQuery =
+ """SELECT from_json('{"abc":1}', 'MAP<VARCHAR(2), INT>')"""
+ val varcharOverflowFailfastQuery =
+ """SELECT from_json(
+ | '{"abc":1}',
+ | 'MAP<VARCHAR(2), INT>',
+ | map('mode', 'FAILFAST'))""".stripMargin
+ val overflowAfterCollisionQuery =
+ """SELECT from_json(
+ | '{"a":1,"a ":2,"abc":0}',
+ | 'MAP<CHAR(2), INT>')""".stripMargin
+ val nestedJsonQuery =
+ """SELECT from_json(
+ | '{"outer":{"a":1,"a ":2}}',
+ | 'MAP<STRING, MAP<CHAR(2), INT>>')""".stripMargin
+ val badFieldBeforeDuplicateQuery =
+ """SELECT from_json(
+ | '{"bad":"not-an-int","m":{"a":1,"a ":2}}',
+ | 'bad INT, m MAP<CHAR(2), INT>')""".stripMargin
+ val badKeyThenSiblingQuery =
+ """SELECT from_json(
+ | '{"m":{"abc":1},"tail":2}',
+ | 'm MAP<CHAR(2), INT>, tail INT').tail""".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 badValueBeforeDuplicateQuery =
+ """SELECT from_json('{"bad":"not-an-int","a":1,"a ":2}', 'MAP<CHAR(2),
INT>')"""
+ val malformedValueBeforeDuplicateQuery =
+ """SELECT from_json('{"a":"bad","a ":2}', 'MAP<CHAR(2), INT>')"""
+ 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><b>2</b><a>3</a></m></ROW>',
+ | 'm MAP<CHAR(2), INT>').m)""".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 badJsonKeyBeforeDuplicateQuery =
+ """SELECT from_json(
+ | '{"abc":0,"a":1,"a ":2}',
+ | 'MAP<CHAR(2), INT>')""".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
+
+ assertDuplicateMapKey(jsonQuery)
+ assertDuplicateMapKey(jsonFailfastQuery)
+ assertDuplicateMapKey(varcharJsonQuery, expectedKey = "ab")
+ checkAnswer(sql(varcharOverflowQuery), Row(null))
+ assertParseExceedLimit(varcharOverflowFailfastQuery, expectedLimit = "2")
+ assertDuplicateMapKey(overflowAfterCollisionQuery)
+ assertDuplicateMapKey(nestedJsonQuery)
+ assertDuplicateMapKey(badFieldBeforeDuplicateQuery)
+ assertDuplicateMapKey(badValueBeforeDuplicateQuery)
+ assertDuplicateMapKey(malformedValueBeforeDuplicateQuery)
+ assertDuplicateMapKey(xmlQuery)
+ assertDuplicateMapKey(varcharXmlQuery, expectedKey = "ab")
+ checkAnswer(sql(exactCharJsonQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(exactVarcharJsonQuery), Row(Map("ab" -> 2)))
+ checkAnswer(sql(interleavedCharJsonQuery), Row(Seq(Row("a ", 3), Row("b
", 2))))
+ checkAnswer(sql(exactCharXmlQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(exactVarcharXmlQuery), Row(Map("ab" -> 2)))
+ checkAnswer(sql(interleavedCharXmlQuery), Row(Seq(Row("a ", 3), Row("b
", 2))))
+ assertDuplicateMapKey(badXmlKeyBeforeDuplicateQuery)
+ assertDuplicateMapKey(badJsonKeyBeforeDuplicateQuery)
+ assertDuplicateMapKey(malformedXmlValueBeforeDuplicateQuery)
+ assertDuplicateMapKey(ignoreCorruptXmlQuery)
+ checkAnswer(sql(badKeyThenSiblingQuery), Row(2))
+ checkAnswer(sql(badXmlKeyThenSiblingQuery), Row(2))
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ assertDuplicateMapKey(jsonQuery)
+ assertDuplicateMapKey(overflowAfterCollisionQuery)
+ assertDuplicateMapKey(badJsonKeyBeforeDuplicateQuery)
+ checkAnswer(sql(varcharOverflowQuery), Row(null))
+ assertParseExceedLimit(varcharOverflowFailfastQuery, expectedLimit =
"2")
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key ->
SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ checkAnswer(sql(jsonQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(jsonFailfastQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(varcharJsonQuery), Row(Map("ab" -> 2)))
+ checkAnswer(sql(varcharOverflowQuery), Row(null))
+ assertParseExceedLimit(varcharOverflowFailfastQuery, expectedLimit =
"2")
+ checkAnswer(sql(overflowAfterCollisionQuery), Row(null))
+ checkAnswer(sql(exactCharJsonQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(exactVarcharJsonQuery), Row(Map("ab" -> 2)))
+ checkAnswer(sql(nestedJsonQuery), Row(Map("outer" -> Map("a " -> 2))))
+ checkAnswer(sql(badValueBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(malformedValueBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(interleavedCharJsonQuery), Row(Seq(Row("a ", 3),
Row("b ", 2))))
+ 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), Row("b
", 2))))
+ checkAnswer(sql(badXmlKeyBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(badJsonKeyBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(malformedXmlValueBeforeDuplicateQuery), Row(null))
+ checkAnswer(sql(ignoreCorruptXmlQuery), Row(Map("a " -> 2)))
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ checkAnswer(sql(jsonQuery), Row(Map("a " -> 2)))
+ checkAnswer(sql(overflowAfterCollisionQuery), Row(null))
+ checkAnswer(sql(badJsonKeyBeforeDuplicateQuery), Row(null))
+ }
+ }
+ }
+ }
+
+ 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 was consumed
and tail is parsed.
+ (overflowQueries :+ nestedFailureQuery).foreach { query =>
+ checkAnswer(sql(query), Row(Row(null, 9)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-59274: ordinary STRING map duplicate behavior is unchanged") {
+ withSQLConf(SQLConf.ALLOW_COLLATIONS_IN_MAP_KEYS.key -> "true") {
+ Seq("false", "true").foreach { standardSemantics =>
+ Seq(SQLConf.MapKeyDedupPolicy.EXCEPTION,
SQLConf.MapKeyDedupPolicy.LAST_WIN)
+ .foreach { dedupPolicy =>
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key ->
standardSemantics,
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> dedupPolicy.toString) {
+ checkAnswer(
+ sql("""SELECT from_json('{"a":1,"a":2}', 'MAP<STRING,
INT>')"""),
Review Comment:
**Non-blocking (P2):** Converting the result to a Scala Map erases the
compatibility property this test needs to protect: both MapData(a -> 1, a -> 2)
and a prematurely collapsed MapData(a -> 2) compare as Map(a -> 2). Please
assert the SQL-visible cardinality and ordered entries before any Scala Map
conversion.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala:
##########
@@ -156,6 +157,28 @@ 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.
+ *
+ * For example, `from_json('{"c":"ab"}', 'c CHAR(4)')` produces `"ab "`,
while parsing
Review Comment:
**Nit (P3):** This example only produces the padded CHAR result when
first-class CHAR/VARCHAR types are enabled, for example with
spark.sql.charVarchar.standardSemantics.enabled=true. Under the default
analyzer configuration the schema is rejected or handled as legacy STRING, so
the prerequisite should be explicit.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -636,9 +640,8 @@ class JacksonParser(
}
}
- // The JSON map will never have null or duplicated map keys, it's safe to
create a
- // ArrayBasedMapData directly here.
- val mapData = ArrayBasedMapData(keys.toArray, values.toArray)
+ val mapData = new ArrayBasedMapBuilder(keyType, valueType).from(
Review Comment:
The ordinary STRING construction branches are restored. One coverage gap
remains: the exact-duplicate JSON assertion collects to a Scala Map, which
collapses both the historical two-entry MapData and an incorrectly collapsed
one-entry result to the same value. Please assert SQL `size` or `map_entries`
for that case.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3944096856","thread_id":"inline:3944096856","verdict_sha256":"1d67a83cffd94f727711a4c377b782e30aa3cf967fd5d486998fc278c510ae37"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -636,9 +640,8 @@ class JacksonParser(
}
}
- // The JSON map will never have null or duplicated map keys, it's safe to
create a
- // ArrayBasedMapData directly here.
- val mapData = ArrayBasedMapData(keys.toArray, values.toArray)
+ val mapData = new ArrayBasedMapBuilder(keyType, valueType).from(
Review Comment:
Confirmed: DUPLICATED_MAP_KEY is now rethrown through the nested JSON
partial-result boundaries, including after an earlier unrelated field error.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3944096858","thread_id":"inline:3944096858","verdict_sha256":"1d67a83cffd94f727711a4c377b782e30aa3cf967fd5d486998fc278c510ae37"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,16 +629,20 @@ class JacksonParser(
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType): MapData = {
val keys = ArrayBuffer.empty[UTF8String]
val values = ArrayBuffer.empty[Any]
var badRecordException: Option[Throwable] = None
while (nextUntil(parser, JsonToken.END_OBJECT)) {
- keys += UTF8String.fromString(parser.currentName)
+ keys += CharVarcharUtils.applyTextParseSemantics(
Review Comment:
The original failed-value collision now reaches duplicate validation
correctly. A related recovery case remains with
`spark.sql.json.enablePartialResults=false`: if a nested struct value fails,
`skipChildren()` runs on the invalid scalar and the remaining inner fields are
then read as outer map keys. Please drain the failed value to its owning
map-entry boundary before continuing.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3960161099","thread_id":"inline:3960161099","verdict_sha256":"1d67a83cffd94f727711a4c377b782e30aa3cf967fd5d486998fc278c510ae37"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala:
##########
@@ -156,6 +157,25 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
StructType(fields)
}
+ /**
+ * Applies CHAR padding and VARCHAR length checks when parsing text into a
typed schema.
+ * Null stays null. Unbounded STRING is unchanged. This is assignment
semantics
+ * (overflow raises EXCEED_LIMIT_LENGTH), not explicit CAST truncation.
+ */
+ def applyTextParseSemantics(value: UTF8String, dt: DataType): UTF8String = {
+ if (value == null) {
+ null
+ } else {
+ dt match {
+ case c: CharType =>
+ CharVarcharCodegenUtils.charTypeWriteSideCheck(value, c.length)
Review Comment:
Confirmed: the parser matrix now covers non-space overflow for both CHAR and
VARCHAR in permissive and fail-fast modes.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3960161108","thread_id":"inline:3960161108","verdict_sha256":"1d67a83cffd94f727711a4c377b782e30aa3cf967fd5d486998fc278c510ae37"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala:
##########
@@ -374,31 +378,41 @@ class StaxXmlParser(
*/
private def convertMap(
parser: XMLEventReader,
+ keyType: DataType,
valueType: DataType,
attributes: Array[Attribute]): MapData = {
val kvPairs = ArrayBuffer.empty[(UTF8String, Any)]
+ def mapKey(raw: String): UTF8String = {
+ CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(raw),
keyType)
+ }
attributes.foreach { attr =>
- kvPairs += (UTF8String.fromString(options.attributePrefix +
attr.getName.getLocalPart)
- -> convertTo(attr.getValue, valueType))
+ kvPairs += (mapKey(options.attributePrefix + attr.getName.getLocalPart)
->
+ convertTo(attr.getValue, valueType))
}
var shouldStop = false
while (!shouldStop) {
parser.nextEvent match {
case e: StartElement =>
val key = StaxXmlParserUtils.getName(e.asStartElement.getName,
options)
- kvPairs +=
- (UTF8String.fromString(key) -> convertField(parser, valueType, key))
+ kvPairs += (mapKey(key) -> convertField(parser, valueType, key))
Review Comment:
The overlength-key path now consumes the entry before validation and
preserves the following field. A related value-error path remains: a failed
nested map value can leave the XML reader inside that entry, so inner elements
are treated as map keys and a later row field is lost. Please drain the failed
value through its matching entry end before continuing.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962178690","thread_id":"inline:3962178690","verdict_sha256":"1d67a83cffd94f727711a4c377b782e30aa3cf967fd5d486998fc278c510ae37"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala:
##########
@@ -310,27 +314,27 @@ class StaxXmlParser(
startElementName: String,
attributes: Array[Attribute]): Any = dt match {
case st: StructType => convertObject(parser, st)
- case MapType(StringType, vt, _) => convertMap(parser, vt, attributes)
+ case MapType(kt: StringType, vt, _) => convertMap(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)
Review Comment:
Confirmed: the empty XML element case now asserts the exact five-space
CHAR(5) result.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962178699","thread_id":"inline:3962178699","verdict_sha256":"1d67a83cffd94f727711a4c377b782e30aa3cf967fd5d486998fc278c510ae37"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
