cloud-fan commented on code in PR #58545:
URL: https://github.com/apache/spark/pull/58545#discussion_r3991524810
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,33 +629,58 @@ 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
+ var partialResultException: Option[Throwable] = None
+ var badMapException: Option[Throwable] = None
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
+ partialResultException =
partialResultException.orElse(Some(err.cause))
+ Some(err.partialResult)
+ case DuplicateMapKeyException(e) => throw e
case NonFatal(e) if enablePartialResults =>
- badRecordException = badRecordException.orElse(Some(e))
+ badMapException = badMapException.orElse(Some(e))
parser.skipChildren()
+ None
+ }
+ value.foreach { parsedValue =>
Review Comment:
**Blocking (P1):** `convertMap` records and normalizes the key only inside
`value.foreach`. With partial results enabled, parsing `{"a":"bad","a ":2}` as
`MAP<CHAR(2), INT>` drops the first key because its value has no partial
result; the builder sees only one normalized key, so the default EXCEPTION
policy is bypassed. Please track normalized constrained keys for duplicate
detection independently of whether their values are retained in the partial
map, while keeping returned key/value pairs atomic.
**Recommended change:** Track every successfully normalized constrained key
for duplicate-policy evaluation after consuming its value, while retaining
key/value pairs for the returned partial map only when a successful or partial
value exists.
**Why this works:** Separate the deduplication key stream from the retained
key/value stream. Consume each value exactly once, normalize the raw key after
consumption, record it for policy evaluation even when value conversion yields
no partial value, and preserve atomic pair insertion for materialized map data.
**Scope:** sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json,
sql/core/src/test/scala/org/apache/spark/sql
**Compatibility:** Entries whose values have no partial result remain absent
from the returned map, and parser tokens are consumed before key validation
errors surface.
**Risks:** Running ArrayBasedMapBuilder on placeholder values could
accidentally expose failed entries in LAST_WIN output. Normalizing before the
value is consumed would reintroduce parser-position and sibling-recovery
regressions.
**Constraints:** Invoke the value converter once per entry. Keep returned
key and value arrays equal length. Preserve ordinary STRING behavior and
non-duplicate partial-result causes. Preserve LAST_WIN ordering among retained
entries.
**Success:** The two-entry malformed-value collision raises
DUPLICATED_MAP_KEY under EXCEPTION. A no-partial value failure still
contributes no entry to a returned partial map. Sibling recovery, nested
duplicates, and ordinary STRING compatibility remain unchanged.
##########
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)
}
+ /**
Review Comment:
**Nit (P3):** `overflow raises` is too broad here: both delegated assignment
checks trim excess trailing spaces when the remaining value fits, and `This is
assignment semantics` is ungrammatical. Please describe that the helper uses
assignment semantics, pads CHAR, permits excess trailing-space trimming, and
raises only for remaining non-space overflow.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -2226,6 +2251,138 @@ 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 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>"))
+ }
+ }
+
+ test("SPARK-59274: normalized CHAR map key collisions honor the dedup
policy") {
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true",
Review Comment:
**Non-blocking (P2):** The new suite still leaves three independently
implemented paths unprotected: VarcharType map-key normalization, duplicate-key
propagation under FAILFAST, and the optimized `spark.read` XML wrapper. Please
add focused MAP<VARCHAR> collision/overflow coverage, assert a normalized
duplicate in FAILFAST mode, and exercise XML file reading with a constrained
user schema so those branches cannot regress while these tests pass.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala:
##########
@@ -374,31 +378,52 @@ class StaxXmlParser(
*/
private def convertMap(
parser: XMLEventReader,
+ keyType: DataType,
valueType: DataType,
attributes: Array[Attribute]): MapData = {
val kvPairs = ArrayBuffer.empty[(UTF8String, Any)]
+ var mapKeyException: Option[Throwable] = None
+ def mapKey(raw: String): UTF8String = {
+ CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(raw),
keyType)
+ }
+ def appendPair(rawKey: String, value: Any): Unit = {
+ try {
+ kvPairs += (mapKey(rawKey) -> value)
+ } catch {
+ case NonFatal(e) => mapKeyException = mapKeyException.orElse(Some(e))
+ }
+ }
attributes.foreach { attr =>
- kvPairs += (UTF8String.fromString(options.attributePrefix +
attr.getName.getLocalPart)
- -> convertTo(attr.getValue, valueType))
+ val value = convertTo(attr.getValue, valueType)
+ 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 = convertField(parser, valueType, rawKey)
+ appendPair(rawKey, value)
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))
+ val value = convertTo(c.getData, valueType)
+ appendPair(options.valueTag, value)
case _: EndElement | _: EndDocument =>
shouldStop = true
case _ => // do nothing
}
}
- ArrayBasedMapData(kvPairs.toMap)
+ mapKeyException.foreach(throw _)
Review Comment:
**Non-blocking (P2):** `mapKeyException.foreach(throw _)` runs before the
constrained-key builder. Consequently, an overlength key followed by `<a>` and
`<a >` records the first error and prevents the later normalized collision from
reaching EXCEPTION. Please apply duplicate policy to the accumulated valid
pairs first, then surface the remembered non-duplicate key error when no
duplicate wins; the map should still be fully consumed before either error
escapes.
**Recommended change:** For CHAR/VARCHAR maps, build and apply duplicate
policy to accumulated valid pairs before throwing a remembered non-duplicate
key conversion error; retain the early error path for ordinary STRING maps.
**Why this works:** Move the remembered-key-error check into the key-type
branches so the constrained branch invokes ArrayBasedMapBuilder first, matching
JSON's duplicate-before-ordinary-error precedence without changing event
consumption.
**Scope:** sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml,
sql/core/src/test/scala/org/apache/spark/sql
**Compatibility:** The XML event reader consumes the complete map before any
constrained-key error is surfaced.
**Risks:** Throwing the remembered error too late under LAST_WIN must not
return a map that also contains an invalid key. Changing event-loop order would
risk consuming the next record or sibling.
**Constraints:** Keep value conversion before constrained-key validation.
Keep key/value insertion atomic. Preserve ordinary STRING last-wins behavior.
**Success:** A prior overlength key cannot suppress a later normalized
duplicate under EXCEPTION. Without a duplicate, the remembered constrained-key
error still follows the configured parse-mode path after the map is consumed.
Later sibling fields remain recoverable.
--
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]
