You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2026/10/14 16:21:37 UTC
(spark) branch master updated: [SPARK-60229][SQL] Keep CHAR/VARCHAR map key checks on from_json(to_json)
This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 119b60993990 [SPARK-60229][SQL] Keep CHAR/VARCHAR map key checks on from_json(to_json)
119b60993990 is described below
commit 119b60993990f4d7105400e3b8ee726e67812797
Author: Serge Rielau <se...@rielau.com>
AuthorDate: Wed Oct 14 09:21:25 2026 -0700
[SPARK-60229][SQL] Keep CHAR/VARCHAR map key checks on from_json(to_json)
### What changes were proposed in this pull request?
Follow-up to SPARK-60108.
1. `OptimizeCsvJsonExprs` no longer folds `from_json(to_json(m))` to `m` when `JsonToStructs.charVarcharStandardSemantics` is on and the schema has a CHAR/VARCHAR map key at any nesting. That fold skipped the JSON map-key length check when `m` already carried an off-width key.
2. Nested `JacksonParser` converters that fail with `spark.sql.json.enablePartialResults=false` now drain their object or array through `END_OBJECT` / `END_ARRAY` before rethrowing, so inner `FIELD_NAME`s cannot be taken as parent siblings. This is a parser-position fix for any nested struct, map, or array `NonFatal` failure, not only CHAR/VARCHAR map keys. Skip failures are suppressed on the original cause.
STRING-key round-trips and the flag-off path still fold.
### Why are the changes needed?
SPARK-60108 length-checks JSON CHAR/VARCHAR map keys without pad or trim. The optimizer identity fold treated `from_json(to_json(m))` as a no-op, so a map that already had an off-width key (for example from a view created with the flag off) never ran that check.
Separately, a nested converter that failed non-partially left Jackson mid-container. The parent field loop could then assign inner names to outer fields (or overwrite fields already parsed). CHAR/VARCHAR map-key rejection was the motivating case; any nested struct/map/array failure had the same hole.
JIRA: https://issues.apache.org/jira/browse/SPARK-60229
### Does this PR introduce _any_ user-facing change?
Yes, on unreleased master / branch-4.x (SPARK-60108 is not in a released Spark version).
- `from_json(to_json(m))` with CHAR/VARCHAR map keys under standard semantics now applies the key length check instead of returning `m` unchanged.
- Nested JSON parse failures with `spark.sql.json.enablePartialResults=false` (internal, default true) now keep already-parsed siblings and continue after the failed nested value instead of stealing inner field names. Default `true` already contained most nested failures; this closes the non-partial nested path, including FAILFAST still reporting the original error.
Example (standard semantics on): `from_json(to_json(m), 'MAP<CHAR(3), INT>')` for `m` with key `a` is null in PERMISSIVE and `UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY` in FAILFAST, rather than returning `m`.
### How was this patch tested?
- `OptimizeJsonExprsSuite`: fold skipped for CHAR/VARCHAR/nested/array map keys with the flag on; STRING keys and flag-off still fold.
- `BasicCharVarcharTestSuite`: `from_json(to_json)` for CHAR and VARCHAR; nested struct steal both field orders; array element failure; FAILFAST still the map-key error.
- Existing SPARK-40646 / SPARK-44940 / SPARK-59274 expectations updated so siblings after a nested failure are parsed when partial results are off.
```
build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.OptimizeJsonExprsSuite -- -z SPARK-60229'
build/sbt 'sql/testOnly org.apache.spark.sql.BasicCharVarcharTestSuite -- -z SPARK-60229'
build/sbt 'sql/testOnly org.apache.spark.sql.JsonFunctionsSuite -- -z SPARK-40646'
build/sbt 'sql/testOnly org.apache.spark.sql.execution.datasources.json.JsonV1Suite -- -z SPARK-40646 -z SPARK-44940'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #59453 from srielau/SPARK-60229.
Authored-by: Serge Rielau <se...@rielau.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../spark/sql/catalyst/json/JacksonParser.scala | 59 +++++++++++++-
.../catalyst/optimizer/OptimizeCsvJsonExprs.scala | 21 ++++-
.../optimizer/OptimizeJsonExprsSuite.scala | 67 ++++++++++++++++
.../apache/spark/sql/CharVarcharTestSuite.scala | 89 +++++++++++++++++++---
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 51 ++++++++++++-
.../sql/execution/datasources/json/JsonSuite.scala | 15 ++--
6 files changed, 280 insertions(+), 22 deletions(-)
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 0b8da0c589b6..9f418d0da100 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
@@ -594,6 +594,7 @@ class JacksonParser(
fieldConverters: Array[ValueConverter],
structFilters: StructFilters = new NoopFilters(),
isRoot: Boolean = false): Option[InternalRow] = {
+ val containerDepth = parser.getParsingContext.getNestingDepth
val row = new GenericInternalRow(schema.length)
var badRecordException: Option[Throwable] = None
var skipRow = false
@@ -618,6 +619,9 @@ class JacksonParser(
case NonFatal(e) if isRoot || enablePartialResults =>
badRecordException = badRecordException.orElse(Some(e))
parser.skipChildren()
+ case NonFatal(e) =>
+ // Nested non-partial: drain before rethrow so parent fields stay aligned.
+ skipRestAndRethrow(parser, JsonToken.END_OBJECT, containerDepth, e)
}
case None =>
parser.skipChildren()
@@ -653,6 +657,7 @@ class JacksonParser(
parser: JsonParser,
fieldConverter: ValueConverter,
keyChecker: Option[UTF8String => Option[Throwable]]): MapData = {
+ val containerDepth = parser.getParsingContext.getNestingDepth
val keys = ArrayBuffer.empty[UTF8String]
val values = ArrayBuffer.empty[Any]
var badRecordException: Option[Throwable] = None
@@ -661,9 +666,9 @@ class JacksonParser(
val key = UTF8String.fromString(parser.currentName)
keys += key
// Avoid allocating a closure per entry on the common no-checker path. Only run the check
- // while the parser is on a FIELD_NAME: if a nested value failed and escaped non-partially
- // (an array value only contains PartialValueException), the loop can resume mid-value on a
- // non-field token whose `currentName` is null, which would otherwise NPE in the check.
+ // while the parser is on a FIELD_NAME: if a nested value failed and escaped non-partially,
+ // the loop can resume mid-value on a non-field token whose `currentName` is null, which
+ // would otherwise NPE in the check.
val keyRejection = keyChecker match {
case Some(check) if parser.currentToken == JsonToken.FIELD_NAME => check(key)
case _ => None
@@ -683,6 +688,9 @@ class JacksonParser(
case NonFatal(e) if enablePartialResults =>
badRecordException = badRecordException.orElse(Some(e))
parser.skipChildren()
+ case NonFatal(e) =>
+ // Nested non-partial: drain before rethrow so parent fields stay aligned.
+ skipRestAndRethrow(parser, JsonToken.END_OBJECT, containerDepth, e)
}
}
}
@@ -753,6 +761,7 @@ class JacksonParser(
fieldConverter: ValueConverter,
isRoot: Boolean = false,
arrayAsStructs: Boolean = false): ArrayData = {
+ val containerDepth = parser.getParsingContext.getNestingDepth
val values = ArrayBuffer.empty[Any]
var badRecordException: Option[Throwable] = None
@@ -765,6 +774,9 @@ class JacksonParser(
case err: PartialValueException if enablePartialResults =>
badRecordException = badRecordException.orElse(Some(err.cause))
values += err.partialResult
+ case NonFatal(e) if !isRoot && !enablePartialResults =>
+ // Nested non-partial: drain before rethrow so parent fields stay aligned.
+ skipRestAndRethrow(parser, JsonToken.END_ARRAY, containerDepth, e)
}
}
@@ -780,6 +792,47 @@ class JacksonParser(
}
}
+ /**
+ * Drain this object or array through its own END token and rethrow `cause`. A child can fail
+ * below this container (VARIANT_DUPLICATE_KEY at an inner END_OBJECT), so stopping at the
+ * first matching END would leave remaining siblings unconsumed. Unwind to `containerDepth`
+ * captured at this container's START, then skip remaining siblings. Compare nesting depth,
+ * not context identity: Jackson recycles child context objects. Skip failures are suppressed
+ * on `cause` so the original error is preserved.
+ */
+ private def skipRestAndRethrow(
+ parser: JsonParser,
+ stopOn: JsonToken,
+ containerDepth: Int,
+ cause: Throwable): Nothing = {
+ try {
+ var eof = false
+ while (!eof && {
+ val ctx = parser.getParsingContext
+ ctx != null && ctx.getNestingDepth > containerDepth
+ }) {
+ parser.skipChildren()
+ eof = parser.nextToken() == null
+ }
+ // An inner END token pops to this container, so the current token can already be
+ // END_OBJECT/END_ARRAY of a child. Only skip remaining siblings when we are still
+ // inside this container; our own END pops to the parent (depth < containerDepth).
+ if (!eof) {
+ val ctx = parser.getParsingContext
+ if (ctx != null && ctx.getNestingDepth == containerDepth) {
+ parser.skipChildren()
+ while (nextUntil(parser, stopOn)) {
+ parser.skipChildren()
+ }
+ }
+ }
+ } catch {
+ case NonFatal(skipErr) =>
+ cause.addSuppressed(skipErr)
+ }
+ throw cause
+ }
+
/**
* Converts the non-stacktrace exceptions to user-friendly QueryExecutionErrors.
*/
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
index f04ff990251d..a605eab66357 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
@@ -25,7 +25,8 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.{CREATE_NAMED_STRUCT, EXTRACT_VALUE,
GET_JSON_OBJECT, JSON_TO_STRUCT}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{ArrayType, StringType, StructType}
+import org.apache.spark.sql.types.{ArrayType, CharType, DataType, MapType, StringType, StructType,
+ VarcharType}
import org.apache.spark.unsafe.types.UTF8String
/**
@@ -443,7 +444,12 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
case jsonToStructs @ JsonToStructs(_, options1,
StructsToJson(options2, child, timeZoneId2), timeZoneId1)
if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
- jsonToStructs.dataType == child.dataType =>
+ jsonToStructs.dataType == child.dataType &&
+ // SPARK-60108 length-checks CHAR/VARCHAR map keys in from_json. Folding
+ // from_json(to_json(m)) => m would skip that check when m already carries
+ // an off-width key (SPARK-60229).
+ !(jsonToStructs.charVarcharStandardSemantics &&
+ hasCharVarcharMapKey(jsonToStructs.dataType)) =>
// `StructsToJson` only fails when `JacksonGenerator` encounters data types it
// cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
// verifies its child's data types is convertible to JSON. But in
@@ -472,6 +478,17 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
}
+ /**
+ * True when `dt` has a MAP whose key type is CHAR or VARCHAR at any nesting.
+ * Only keys: this preserves the SPARK-60108 map-key check, which runs only under the captured
+ * standard semantics and never pads or trims. CHAR/VARCHAR values get pad/length semantics from
+ * applyTextParseSemantics regardless of that flag and are outside this guard.
+ */
+ private def hasCharVarcharMapKey(dt: DataType): Boolean = dt.existsRecursively {
+ case MapType(_: CharType | _: VarcharType, _, _) => true
+ case _ => false
+ }
+
private val csvOptimization: PartialFunction[Expression, Expression] = {
case g @ GetStructField(c @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
if schema.length > 1 && c.options.isEmpty && schema(ordinal).name != nameOfCorruptRecord =>
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
index 801404ba55a0..832b6e018ea3 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
@@ -94,6 +94,73 @@ class OptimizeJsonExprsSuite extends PlanTest with ExpressionEvalHelper {
comparePlans(optimized2, expected)
}
+ test("SPARK-60229: not optimize from_json + to_json with CHAR/VARCHAR map keys") {
+ // CHAR/VARCHAR attributes are rejected at analysis unless one of these flags is on.
+ withSQLConf(SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+ val options = Map.empty[String, String]
+
+ def roundTrip(
+ schema: DataType, child: Expression, standardSemantics: Boolean): Expression =
+ JsonToStructs(
+ schema,
+ options,
+ StructsToJson(options, child),
+ charVarcharStandardSemantics = standardSemantics)
+
+ val charMap = MapType(CharType(3), IntegerType)
+ val varcharMap = MapType(VarcharType(3), IntegerType)
+ val stringMap = MapType(StringType, IntegerType)
+ val nestedCharMap = StructType(Seq(
+ StructField("m", charMap),
+ StructField("tail", IntegerType)))
+ val arrayCharMap = ArrayType(charMap)
+
+ val mapRel = LocalRelation(Symbol("m").map(CharType(3), IntegerType).notNull)
+ val charRoundTrip = roundTrip(charMap, Symbol("m").attr, standardSemantics = true)
+ val query = mapRel.select(charRoundTrip.as("m"))
+ val optimized = Optimizer.execute(query.analyze)
+ val expected = mapRel.select(charRoundTrip.as("m")).analyze
+ comparePlans(optimized, expected)
+
+ val varcharRel = LocalRelation(Symbol("m").map(VarcharType(3), IntegerType).notNull)
+ val varcharQuery = varcharRel.select(
+ roundTrip(varcharMap, Symbol("m").attr, standardSemantics = true).as("m"))
+ comparePlans(
+ Optimizer.execute(varcharQuery.analyze),
+ varcharRel.select(
+ roundTrip(varcharMap, Symbol("m").attr, standardSemantics = true).as("m")).analyze)
+
+ val nestedRel = LocalRelation($"s".struct(nestedCharMap).notNull)
+ val nestedQuery =
+ nestedRel.select(roundTrip(nestedCharMap, $"s", standardSemantics = true).as("s"))
+ comparePlans(
+ Optimizer.execute(nestedQuery.analyze),
+ nestedRel.select(roundTrip(nestedCharMap, $"s", standardSemantics = true).as("s")).analyze)
+
+ val arrayRel = LocalRelation($"a".array(arrayCharMap).notNull)
+ val arrayQuery =
+ arrayRel.select(roundTrip(arrayCharMap, $"a", standardSemantics = true).as("a"))
+ comparePlans(
+ Optimizer.execute(arrayQuery.analyze),
+ arrayRel.select(roundTrip(arrayCharMap, $"a", standardSemantics = true).as("a")).analyze)
+
+ // STRING keys have no length check, so the round-trip is still folded.
+ val stringRel = LocalRelation(Symbol("m").map(StringType, IntegerType).notNull)
+ val stringQuery = stringRel.select(
+ roundTrip(stringMap, Symbol("m").attr, standardSemantics = true).as("m"))
+ comparePlans(
+ Optimizer.execute(stringQuery.analyze),
+ stringRel.select($"m".as("m")).analyze)
+
+ // Flag off: the check is not applied, so folding remains valid.
+ val offQuery = mapRel.select(
+ roundTrip(charMap, Symbol("m").attr, standardSemantics = false).as("m"))
+ comparePlans(
+ Optimizer.execute(offQuery.analyze),
+ mapRel.select($"m".as("m")).analyze)
+ }
+ }
+
test("SPARK-32948: not optimize from_json + to_json if schema is different") {
val options = Map.empty[String, String]
val schema = StructType.fromDDL("a 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 65a621115070..fa091a8d4b32 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
@@ -3543,6 +3543,81 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
}
}
+ test("SPARK-60229: from_json(to_json) keeps CHAR/VARCHAR map key check") {
+ Seq(
+ ("CHAR(3)", "a", "t_spark_60229_char"),
+ ("VARCHAR(3)", "abcd", "t_spark_60229_varchar")
+ ).foreach { case (keyType, badKey, viewName) =>
+ withView(viewName) {
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+ sql(s"""CREATE TEMPORARY VIEW $viewName AS
+ |SELECT from_json('{"$badKey":1}', 'MAP<$keyType, INT>') AS m""".stripMargin)
+ }
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ checkAnswer(
+ sql(s"SELECT from_json(to_json(m), 'MAP<$keyType, INT>') FROM $viewName"),
+ Row(null))
+ assertUnsupportedJsonMapKey(
+ s"""SELECT from_json(to_json(m), 'MAP<$keyType, INT>', map('mode', 'FAILFAST'))
+ |FROM $viewName""".stripMargin,
+ key = badKey,
+ dataTypeSql = keyType)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60229: nested struct map key rejection does not steal sibling fields") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val schema = "s STRUCT<m MAP<CHAR(3), INT>, tail INT>, tail INT"
+ // Inner tail after the map: without draining s, the parent can take inner tail=5
+ // and then hit s's END_OBJECT, never seeing outer tail=1.
+ val jsonAfter = """{"s":{"m":{"a":1},"tail":5},"tail":1}"""
+ // Inner tail after an already-parsed outer tail: without draining s, inner
+ // tail=5 overwrites the sibling that was already 1.
+ val jsonBefore = """{"tail":1,"s":{"m":{"a":1},"tail":5}}"""
+ Seq(true, false).foreach { partial =>
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> partial.toString) {
+ val expected = if (partial) {
+ Row(Row(Row(null, 5), 1))
+ } else {
+ Row(Row(null, 1))
+ }
+ Seq(jsonAfter, jsonBefore).foreach { json =>
+ checkAnswer(sql(s"SELECT from_json('$json', '$schema')"), expected)
+ assertUnsupportedJsonMapKey(
+ s"""SELECT from_json('$json', '$schema', map('mode', 'FAILFAST'))""",
+ key = "a",
+ dataTypeSql = "CHAR(3)")
+ }
+ }
+ }
+
+ // A failing array element must not leak inner field names to the parent, and
+ // later siblings must still be parsed.
+ val arrayJson = """{"tail":1,"arr":[{"m":{"a":1},"tail":5}],"after":2}"""
+ val arraySchema =
+ "tail INT, arr ARRAY<STRUCT<m MAP<CHAR(3), INT>, tail INT>>, after INT"
+ Seq(true, false).foreach { partial =>
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> partial.toString) {
+ val expected = if (partial) {
+ Row(Row(1, Seq(Row(null, 5)), 2))
+ } else {
+ Row(Row(1, null, 2))
+ }
+ checkAnswer(sql(s"SELECT from_json('$arrayJson', '$arraySchema')"), expected)
+ assertUnsupportedJsonMapKey(
+ s"""SELECT from_json('$arrayJson', '$arraySchema',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ key = "a",
+ dataTypeSql = "CHAR(3)")
+ }
+ }
+ }
+ }
+
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}"""
@@ -3553,11 +3628,9 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
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)))
- }
+ // SPARK-60229: a nested map value failure consumes the map object, so
+ // sibling `tail` is parsed in both partial-results modes.
+ checkAnswer(sql(fromJson), Row(Row(null, 1)))
assertParseExceedLimit(
s"SELECT from_json('$json', '$schema', map('mode', 'FAILFAST'))",
expectedLimit = "2")
@@ -3568,11 +3641,7 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
.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)))
- }
+ checkAnswer(permissive, Seq(Row(null, 1), Row(Map("k" -> "ab"), 2)))
val failFast = spark.read
.option("mode", "FAILFAST")
.schema(fileSchema)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 90577945c3a3..77ae12e61650 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -1621,6 +1621,7 @@ class JsonFunctionsSuite extends SharedSparkSession {
// Map "c2" has "k2" key that is a string, not an integer.
val df = Seq("""{"c1": {"k1": 1, "k2": "A", "k3": 3}, "c2": "abc"}""").toDF("c0")
+ // SPARK-60229: draining the failed map keeps sibling c2 in both partial-results modes.
withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
checkAnswer(
df.select(from_json($"c0", st)),
@@ -1631,11 +1632,59 @@ class JsonFunctionsSuite extends SharedSparkSession {
withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
checkAnswer(
df.select(from_json($"c0", st)),
- Row(Row(null, null))
+ Row(Row(null, "abc"))
)
}
}
+ test("SPARK-60229: nested VARIANT failure does not steal sibling fields") {
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ def assertDuplicateKey(query: String): Unit = {
+ val exception = intercept[SparkException] {
+ sql(query).collect()
+ }
+ val cause = Iterator.iterate[Throwable](exception)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case e: SparkRuntimeException if e.getCondition == "VARIANT_DUPLICATE_KEY" => e
+ }
+ .getOrElse(fail("expected VARIANT_DUPLICATE_KEY cause", exception))
+ checkError(
+ exception = cause,
+ condition = "VARIANT_DUPLICATE_KEY",
+ parameters = Map("key" -> "k"))
+ }
+
+ val expected = Row(Row(null, 5))
+ // VARIANT struct field: duplicate key is inside nested object `a`, below `v`.
+ val structJson = """{"s":{"v":{"a":{"k":1,"k":2}},"b":7},"b":5}"""
+ val structSchema = "s STRUCT<v: VARIANT, b: INT>, b INT"
+ checkAnswer(sql(s"SELECT from_json('$structJson', '$structSchema')"), expected)
+ assertDuplicateKey(
+ s"SELECT from_json('$structJson', '$structSchema', map('mode', 'FAILFAST'))")
+
+ // VARIANT map value: same nested failure, drained through the map's END_OBJECT.
+ val mapJson = """{"s":{"x":{"a":{"k":1,"k":2}},"y":1},"b":5}"""
+ val mapSchema = "s MAP<STRING, VARIANT>, b INT"
+ checkAnswer(sql(s"SELECT from_json('$mapJson', '$mapSchema')"), expected)
+ assertDuplicateKey(
+ s"SELECT from_json('$mapJson', '$mapSchema', map('mode', 'FAILFAST'))")
+
+ // VARIANT array value: duplicate key is inside an object nested in that array.
+ val variantArrJson = """{"s":{"v":[{"a":{"k":1,"k":2}}],"b":7},"b":5}"""
+ checkAnswer(sql(s"SELECT from_json('$variantArrJson', '$structSchema')"), expected)
+ assertDuplicateKey(
+ s"SELECT from_json('$variantArrJson', '$structSchema', map('mode', 'FAILFAST'))")
+
+ // ARRAY<VARIANT> element: nested object failure must not steal sibling `b`.
+ val arrJson = """{"arr":[{"a":{"k":1,"k":2}}],"b":5}"""
+ val arrSchema = "arr ARRAY<VARIANT>, b INT"
+ checkAnswer(sql(s"SELECT from_json('$arrJson', '$arrSchema')"), expected)
+ assertDuplicateKey(
+ s"SELECT from_json('$arrJson', '$arrSchema', map('mode', 'FAILFAST'))")
+ }
+ }
+
test("SPARK-40646: return partial results for JSON arrays") {
val st = new StructType()
.add("c", ArrayType(IntegerType))
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
index 538e424a535d..a561ec255cf9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -4213,9 +4213,12 @@ abstract class JsonSuite
Seq(Row(Row(1, null), Row(1)), Row(Row(2, null), Row(2)))
)
} else {
+ // SPARK-60229: any nested object/map failure (not only CHAR map keys)
+ // still nulls `a`, but the rest of that object is consumed so sibling
+ // `b` is parsed.
checkAnswer(
df,
- Seq(Row(null, null), Row(Row(2, null), Row(2)))
+ Seq(Row(null, Row(1)), Row(Row(2, null), Row(2)))
)
}
}
@@ -4247,7 +4250,7 @@ abstract class JsonSuite
checkAnswer(
df,
Seq(
- Row("AAA", null, null, null),
+ Row("AAA", null, "id1", "XXX"),
Row("BBB", Seq(Row(12, "")), "id2", "YYY")
)
)
@@ -4283,8 +4286,8 @@ abstract class JsonSuite
checkAnswer(
df,
Seq(
- Row("AAA", null, null),
- Row("BBB", null, null)
+ Row("AAA", null, "id1"),
+ Row("BBB", null, "id2")
)
)
}
@@ -4317,7 +4320,7 @@ abstract class JsonSuite
checkAnswer(
df,
Seq(
- Row("AAA", null, null),
+ Row("AAA", null, "id1"),
Row("BBB", Map("key" -> Row(12, "")), "id2")
)
)
@@ -4349,7 +4352,7 @@ abstract class JsonSuite
checkAnswer(
df,
Seq(
- Row("AAA", null, null, null),
+ Row("AAA", null, "id1", "XXX"),
Row("BBB", Row(Seq(12)), "id2", "YYY")
)
)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org