cloud-fan commented on code in PR #59167: URL: https://github.com/apache/spark/pull/59167#discussion_r4232761047
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JsonParseAssignment.scala: ########## @@ -0,0 +1,152 @@ +/* + * 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.json + +import scala.util.control.NonFatal + +import org.apache.spark.SparkUpgradeException +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.BoundReference +import org.apache.spark.sql.catalyst.util._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.Filter +import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String + +/** + * CHAR/VARCHAR assignment for JSON, outside [[JacksonParser]]. + * + * Same split as Hive TRANSFORM (SPARK-60090): the parser sees unbounded STRING, then + * write-side assignment runs on the parsed value. [[JacksonParser]] stays CHAR-unaware. + */ +private[sql] object JsonParseAssignment { + + def parserSchema(schema: DataType): DataType = { + if (needsAssignment(schema)) { + CharVarcharUtils.replaceCharVarcharWithStringAlways(schema) + } else { + schema + } + } + + def pushedFilters(declared: StructType, filters: Seq[Filter]): Seq[Filter] = { + if (!needsAssignment(declared)) { + filters + } else { + filters.filterNot { f => + f.references.exists { name => + declared.getFieldIndex(name).exists { i => + CharVarcharUtils.hasCharVarchar(declared.fields(i).dataType) + } + } + } + } + } + + def parse( + rows: => Iterable[InternalRow], + declared: DataType, + recordLiteral: () => UTF8String): Iterable[InternalRow] = { + rowAssigner(declared) match { + case None => rows + case Some(assign) => + try { + rows.map(row => assignRow(assign, row, recordLiteral, recoverable = false)) Review Comment: **Non-blocking (P2):** Assignment is applied to this eager array as one all-or-nothing map. If a later element overflows a CHAR/VARCHAR field, the exception carries no already assigned rows, so PERMISSIVE input such as a valid first element followed by an overlength second element collapses to one null result instead of preserving the valid sibling. Please retain successful elements while applying the parse-mode policy to the failing element. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JsonParseAssignment.scala: ########## @@ -0,0 +1,152 @@ +/* + * 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.json + +import scala.util.control.NonFatal + +import org.apache.spark.SparkUpgradeException +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.BoundReference +import org.apache.spark.sql.catalyst.util._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.Filter +import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String + +/** + * CHAR/VARCHAR assignment for JSON, outside [[JacksonParser]]. + * + * Same split as Hive TRANSFORM (SPARK-60090): the parser sees unbounded STRING, then + * write-side assignment runs on the parsed value. [[JacksonParser]] stays CHAR-unaware. + */ +private[sql] object JsonParseAssignment { + + def parserSchema(schema: DataType): DataType = { + if (needsAssignment(schema)) { + CharVarcharUtils.replaceCharVarcharWithStringAlways(schema) + } else { + schema + } + } + + def pushedFilters(declared: StructType, filters: Seq[Filter]): Seq[Filter] = { + if (!needsAssignment(declared)) { + filters + } else { + filters.filterNot { f => + f.references.exists { name => + declared.getFieldIndex(name).exists { i => + CharVarcharUtils.hasCharVarchar(declared.fields(i).dataType) + } + } + } + } + } + + def parse( + rows: => Iterable[InternalRow], + declared: DataType, + recordLiteral: () => UTF8String): Iterable[InternalRow] = { + rowAssigner(declared) match { Review Comment: **Non-blocking (P2):** rowAssigner recursively builds the BoundReference/stringLengthCheck expression on every invocation of this per-record callback. The schema and mode are fixed when the evaluator or reader is created, so this schema-sized planning and allocation should be prepared once and safely reused across records. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala: ########## @@ -641,15 +642,8 @@ class JacksonParser( } } - // 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. + // The JSON map will never have null or duplicated map keys, it's safe to create a Review Comment: **Non-blocking (P2):** A recoverable value-conversion failure can leave keys one element longer than values here. Constructing ArrayBasedMapData then throws its cardinality error before badRecordException can report the real conversion cause. Please restore the paired-buffer guard before building the map. See **Shared repair plan 2** in the review body. ########## sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala: ########## @@ -188,9 +188,15 @@ class DataFrameReader private[sql](sparkSession: SparkSession) val createParser = CreateJacksonParser.string _ val parsed = jsonDataset.rdd.mapPartitions { iter => - val rawParser = new JacksonParser(actualSchema, parsedOptions, allowArrayAsStructs = true) + val rawParser = new JacksonParser( Review Comment: **Nit (P3):** This is a separately wired public JSON entry point, yet the new coverage exercises only from_json and file readers. Please add a Dataset[String] regression that observes CHAR padding and an assignment failure or constrained-key collision so this wrapper cannot be removed or miswired unnoticed. See **Shared repair plan 1** in the review body. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala: ########## @@ -73,6 +73,11 @@ abstract class JsonDataSource extends Serializable with Logging with SupportsArc parser: JacksonParser, schema: StructType): Iterator[InternalRow] + // Scan schema may still be CHAR/VARCHAR. The parser is given STRING; assignment is after parse. + protected def jsonDeclaredSchema(parser: JacksonParser, schema: StructType): DataType = { Review Comment: **Nit (P3):** This new schema boundary is also used when the requested schema contains the configured corrupt-record column, but the added tests never exercise that case with active CHAR/VARCHAR assignment and malformed input. Please add a regression assertion that checks assigned field placement and preservation of the corrupt payload. See **Shared repair plan 1** in the review body. ########## sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala: ########## @@ -1044,6 +1045,20 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { parameters = Map("limit" -> expectedLimit)) } + private def assertMalformedJsonRecord(query: String): Unit = { Review Comment: **Nit (P3):** This helper is never invoked, and the generic malformed-JSON tests do not activate CHAR/VARCHAR assignment. Please call it with a malformed input and an assignment-sensitive schema so a wrapper regression that replaces MALFORMED_RECORD_IN_PARSING with an assignment error is detected. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JsonParseAssignment.scala: ########## @@ -0,0 +1,152 @@ +/* + * 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.json + +import scala.util.control.NonFatal + +import org.apache.spark.SparkUpgradeException +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.BoundReference +import org.apache.spark.sql.catalyst.util._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.Filter +import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String + +/** + * CHAR/VARCHAR assignment for JSON, outside [[JacksonParser]]. + * + * Same split as Hive TRANSFORM (SPARK-60090): the parser sees unbounded STRING, then + * write-side assignment runs on the parsed value. [[JacksonParser]] stays CHAR-unaware. + */ +private[sql] object JsonParseAssignment { + + def parserSchema(schema: DataType): DataType = { + if (needsAssignment(schema)) { + CharVarcharUtils.replaceCharVarcharWithStringAlways(schema) + } else { + schema + } + } + + def pushedFilters(declared: StructType, filters: Seq[Filter]): Seq[Filter] = { + if (!needsAssignment(declared)) { + filters + } else { + filters.filterNot { f => + f.references.exists { name => + declared.getFieldIndex(name).exists { i => + CharVarcharUtils.hasCharVarchar(declared.fields(i).dataType) + } + } + } + } + } + + def parse( + rows: => Iterable[InternalRow], + declared: DataType, + recordLiteral: () => UTF8String): Iterable[InternalRow] = { + rowAssigner(declared) match { + case None => rows + case Some(assign) => + try { + rows.map(row => assignRow(assign, row, recordLiteral, recoverable = false)) + } catch { + case e: SparkUpgradeException => throw e + case DuplicateMapKeyUtils(e) => throw e + case e: BadRecordException => + throw e.copy(partialResults = () => assignPartials(assign, e.partialResults())) + } + } + } + + def parseIterator( + rows: Iterator[InternalRow], + declared: DataType, + recordLiteral: () => UTF8String): Iterator[InternalRow] = { + rowAssigner(declared) match { + case None => rows + case Some(assign) => + new Iterator[InternalRow] { + override def hasNext: Boolean = withAssignedPartials(assign)(rows.hasNext) + override def next(): InternalRow = withAssignedPartials(assign) { + assignRow(assign, rows.next(), recordLiteral, recoverable = true) + } + } + } + } + + private def needsAssignment(schema: DataType): Boolean = { + SQLConf.get.charVarcharStandardSemantics && CharVarcharUtils.hasCharVarchar(schema) Review Comment: **Non-blocking (P2):** This execution-time SQLConf read can disagree with the setting captured in a persisted view's analyzed CHAR/VARCHAR schema. A view created with standard semantics enabled but queried from a session where it is disabled skips assignment and exposes raw STRING values. Please capture the assignment mode together with the schema when the expression or reader is constructed. See **Shared repair plan 1** in the review body. ########## sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala: ########## @@ -2987,6 +3006,216 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { } } + test("SPARK-59722: JSON CHAR/VARCHAR assignment runs after STRING parse") { + withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { + // Scalar pad still applies, via stringLengthCheck on the parsed STRING value. + checkAnswer( + sql("""SELECT from_json('{"a": "str"}', 'a CHAR(5)')"""), + Row(Row("str "))) + checkAnswer( + sql("""SELECT from_json('{"ab": 1}', 'MAP<CHAR(4), INT>')"""), + Row(Map("ab " -> 1))) + + val collision = """SELECT from_json('{"a":1,"a ":2}', 'MAP<CHAR(2), INT>')""" + val exactRepeat = """SELECT from_json('{"a":1,"a":2}', 'MAP<CHAR(2), INT>')""" + val varcharCollision = + """SELECT from_json('{"ab":1,"ab ":2}', 'MAP<VARCHAR(2), INT>')""" + val nestedCollision = + """SELECT from_json( + | '{"outer":{"a":1,"a ":2}}', + | 'MAP<STRING, MAP<CHAR(2), INT>>')""".stripMargin + val overflow = """SELECT from_json('{"abc":1}', 'MAP<VARCHAR(2), INT>')""" + val overflowFailfast = + """SELECT from_json( + | '{"abc":1}', + | 'MAP<VARCHAR(2), INT>', + | map('mode', 'FAILFAST'))""".stripMargin + val overflowWithSibling = + """SELECT from_json( + | '{"m":{"k":"abcdef"},"tail":1}', + | 'm MAP<STRING, CHAR(2)>, tail INT')""".stripMargin + + // Pad/trim collisions and exact JSON member repeats both use mapKeyDedupPolicy. + assertDuplicateMapKey(collision) + assertDuplicateMapKey(exactRepeat) + assertDuplicateMapKey(varcharCollision, expectedKey = "ab") + assertDuplicateMapKey(nestedCollision) + checkAnswer(sql(overflow), Row(null)) + assertParseExceedLimit(overflowFailfast, expectedLimit = "2") + // Assignment failure is whole-value: PERMISSIVE does not keep `tail`. + checkAnswer(sql(overflowWithSibling), Row(Row(null, null))) + assertParseExceedLimit( + """SELECT from_json( + | '{"m":{"k":"abcdef"},"tail":1}', + | 'm MAP<STRING, CHAR(2)>, tail INT', + | map('mode', 'FAILFAST'))""".stripMargin, + expectedLimit = "2") + + withTempPath { path => + Seq("""{"m":{"a":1,"a ":2}}""").toDS().write.text(path.getCanonicalPath) + def readJsonMap(): DataFrame = spark.read + .schema("m MAP<CHAR(2), INT>") + .json(path.getCanonicalPath) + assertDuplicateMapKeyError(readJsonMap().collect()) + withSQLConf( + SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { + checkAnswer(readJsonMap(), Row(Map("a " -> 2))) + } + } + + withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") { + withTempPath { path => + Seq("""[{"m":{"a":1,"a ":2}},{"m":{"b":3}}]""") + .toDS().write.text(path.getCanonicalPath) + def readMultilineJsonMap(): DataFrame = spark.read + .option("multiLine", true) + .schema("m MAP<CHAR(2), INT>") + .json(path.getCanonicalPath) + assertDuplicateMapKeyError(readMultilineJsonMap().collect()) + withSQLConf( + SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { + checkAnswer( + readMultilineJsonMap(), + Seq(Row(Map("a " -> 2)), Row(Map("b " -> 3)))) + } + } + } + + withSQLConf( + SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { + checkAnswer(sql(collision), Row(Map("a " -> 2))) + checkAnswer(sql(exactRepeat), Row(Map("a " -> 2))) + checkAnswer(sql(varcharCollision), Row(Map("ab" -> 2))) + checkAnswer(sql(nestedCollision), Row(Map("outer" -> Map("a " -> 2)))) + checkAnswer(sql(overflow), Row(null)) + } + } + } + + test("SPARK-59722: JSON constrained maps preserve SparkUpgradeException") { + val timestamp = "2020-01-27T20:06:11.847-08000" + val timestampFormat = "yyyy-MM-dd'T'HH:mm:ss.SSSz" + def assertUpgradeException(body: => Unit): Unit = { + checkError( + exception = intercept[SparkUpgradeException](body), + condition = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.PARSE_DATETIME_BY_NEW_PARSER", + parameters = Map( + "datetime" -> s"'$timestamp'", + "config" -> "\"spark.sql.legacy.timeParserPolicy\"")) + } + + withSQLConf( + SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true", + SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "exception") { + val json = Seq(s"""{"a":"$timestamp"}""").toDF("json") + val parsed = json.select(functions.from_json( + $"json", + "MAP<CHAR(2), TIMESTAMP>", + Map("timestampFormat" -> timestampFormat))) + assertUpgradeException(parsed.collect()) + + withTempPath { path => + Seq(s"""{"m":{"a":"$timestamp"}}""").toDS().write.text(path.getCanonicalPath) + val parsedFile = spark.read + .option("timestampFormat", timestampFormat) + .schema("m MAP<CHAR(2), TIMESTAMP>") + .json(path.getCanonicalPath) + assertUpgradeException(parsedFile.collect()) + } + } + } + + test("SPARK-59722: JSON assignment failures do not expose physical STRING") { + withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { + // Parser mismatch plus CHAR overflow: do not return the raw overlength STRING. + checkAnswer( + sql("""SELECT from_json('{"a":"x","b":"abcdef"}', 'a INT, b CHAR(2)')"""), + Row(Row(null, null))) + + // Map and array roots assign after parse; overflow is whole-value null. + checkAnswer( + sql("""SELECT from_json('{"a":"ab"}', 'MAP<CHAR(4), CHAR(4)>')"""), + Row(Map("a " -> "ab "))) + checkAnswer( + sql("""SELECT from_json('{"a":"abcdef"}', 'MAP<STRING, CHAR(2)>')"""), + Row(null)) + checkAnswer( + sql("""SELECT from_json('["ab","c"]', 'ARRAY<CHAR(4)>')"""), + Row(Seq("ab ", "c "))) + checkAnswer( + sql("""SELECT from_json('["ok","abcdef"]', 'ARRAY<CHAR(2)>')"""), + Row(null)) + } + } + + test("SPARK-59722: streamed JSON array resumes after assignment overflow") { + withSQLConf( + SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true", + SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") { + withTempPath { path => + Seq("""[{"c":"abcdef","id":1},{"c":"ok","id":2}]""") + .toDS().write.text(path.getCanonicalPath) + val schema = "c CHAR(2), id INT" + Seq("json", "").foreach { useV1 => + withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1) { + withClue(s"useV1=$useV1") { + val permissive = spark.read + .option("multiLine", true) + .schema(schema) + .json(path.getCanonicalPath) + checkAnswer(permissive, Seq(Row(null, null), Row("ok", 2))) + val dropMalformed = spark.read + .option("multiLine", true) + .option("mode", "DROPMALFORMED") + .schema(schema) + .json(path.getCanonicalPath) + checkAnswer(dropMalformed, Seq(Row("ok", 2))) + val failFast = spark.read + .option("multiLine", true) + .option("mode", "FAILFAST") + .schema(schema) + .json(path.getCanonicalPath) + assertParseExceedLimitError(failFast.collect(), expectedLimit = "2") + } + } + } + } + } + } + + test("SPARK-59722: JSON datasource CHAR filters stay residual") { + withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { + withTempPath { path => + Seq( + """{"id":1,"c":"ab"}""", + """{"id":2,"c":"xy"}""", + """{"id":3,"c":"ab"}""") + .toDS().repartition(1).write.text(path.getCanonicalPath) + val schema = "id INT, c CHAR(4)" + Seq("json", "").foreach { useV1 => + Seq(true, false).foreach { pushdown => + withSQLConf( + SQLConf.USE_V1_SOURCE_LIST.key -> useV1, + SQLConf.JSON_FILTER_PUSHDOWN_ENABLED.key -> pushdown.toString) { + withClue(s"useV1=$useV1 pushdown=$pushdown") { + val df = spark.read.schema(schema).json(path.getCanonicalPath) + checkAnswer(df, Seq(Row(1, "ab "), Row(2, "xy "), Row(3, "ab "))) + // Independent INT predicate. + checkAnswer(df.where($"id" === 1), Seq(Row(1, "ab "))) Review Comment: **Nit (P3):** This result is identical when the id predicate is pushed into the JSON parser and when it is evaluated only as a residual filter. As written, the test still passes if independent predicate pushdown is removed entirely. Please assert a parser-side pushdown signal for both V1 and V2. **Verification:** - **Regression:** Under enabled and disabled JSON filter pushdown for both V1 and V2, distinguish scan-pushed independent predicates from residual assignment-sensitive predicates. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala: ########## @@ -60,11 +60,50 @@ class JacksonParser( // to a value in a field for `InternalRow`. private type ValueConverter = JsonParser => AnyRef + // CHAR/VARCHAR is STRING-on-the-wire in JSON. Parse as unbounded STRING, then assign. + private val physicalSchema: DataType = + CharVarcharUtils.replaceCharVarcharWithStringAlways(schema) + // `ValueConverter`s for the root schema for all fields in the schema - private val rootConverter = makeRootConverter(schema) + private val rootConverter = makeRootConverter(physicalSchema) // Lazy: a schema that is not a struct never reaches either top-level-array path. - private lazy val arrayElementConverter = makeConverter(schema) + private lazy val arrayElementConverter = makeConverter(physicalSchema) + + private val assignExpr: Expression = + CharVarcharUtils.assignAfterParse( + BoundReference(0, physicalSchema, nullable = true), + schema) + + private val needsAssignment = + SQLConf.get.charVarcharStandardSemantics && Review Comment: The current merge base does apply JSON parse semantics in preserve-only mode: JacksonParser calls applyTextParseSemantics for the retained CHAR/VARCHAR type. This patch removes that call while needsAssignment remains false, so the flag-off compatibility claim does not hold. Please preserve that mode or explicitly document and test the compatibility change. <!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4226583756","thread_id":"inline:4226583756","verdict_sha256":"b99d61dc014e93932bba0655be66f5d486897769f4fa3a258c0487c8da7566b5"} --> ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala: ########## @@ -361,7 +362,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) - CharVarcharUtils.applyTextParseSemantics(result, dt) + result Review Comment: **Non-blocking (P2):** Removing in-parser text semantics leaves preserve-only sessions without any CHAR/VARCHAR enforcement: JsonParseAssignment is disabled when standard semantics is false. With preserveCharVarcharTypeInfo=true, CHAR values are therefore no longer padded and overlength values no longer fail even though the PR promises flag-off behavior is unchanged. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala: ########## @@ -208,6 +227,18 @@ object CharVarcharUtils extends Logging with SparkCharVarcharUtils { varcharFuncName = Some("varcharTypeWriteSideCheck")) } + /** + * Applies write-side CHAR/VARCHAR assignment to a value parsed as unbounded STRING. + * No-op when standard semantics are off or `targetType` has no CHAR/VARCHAR. + */ + def assignAfterParse(expr: Expression, targetType: DataType): Expression = { + if (SQLConf.get.charVarcharStandardSemantics && hasCharVarchar(targetType)) { + stringLengthCheck(expr, targetType) Review Comment: **Non-blocking (P2):** This rebuild also reprocesses an unchanged STRING key array when only the value type contains CHAR/VARCHAR. Exact repeated JSON names that were historically preserved then raise DUPLICATED_MAP_KEY under EXCEPTION or lose entries under LAST_WIN. Please preserve ordinary STRING keys when assignment only changes the value subtree. **Recommended change:** Split key-changing and value-only map assignment. Use MapFromArrays only when the declared key subtree actually needs CHAR/VARCHAR normalization; for value-only assignment, transform values while preserving the parsed key array, ordering, and duplicate multiplicity. Keep the implementation JSON-specific if changing generic write-side assignment would affect unrelated callers. **Why this works:** Compute keyNeedsAssignment and valueNeedsAssignment separately. A constrained key follows the existing normalization/deduplication expression. An unchanged key reuses the input MapData key array and pairs it with the assigned value array through a duplicate-preserving map representation rather than MapFromArrays. **Scope:** Make recursive map assignment sensitive to which subtree changes and protect ordinary JSON key representation. **Compatibility:** Constrained JSON keys remain normalized with write-side semantics and governed by mapKeyDedupPolicy. **Risks:** A duplicate-preserving rebuild must keep key/value cardinality and ordering aligned after nested value failures. Changing generic stringLengthCheck may affect non-JSON write paths. **Constraints:** Constrained key normalization must still honor EXCEPTION and LAST_WIN. Exact repeated constrained keys remain subject to mapKeyDedupPolicy. **Success:** MAP<STRING, CHAR/VARCHAR> and outer ordinary STRING maps retain their parsed duplicate keys while values receive assignment. CHAR/VARCHAR-keyed maps continue to normalize and deduplicate under mapKeyDedupPolicy. Nested key-changing and value-only map cases choose the correct branch independently. **Verification:** - **Compatibility:** Compare ordinary STRING-key maps with exact duplicates under EXCEPTION and LAST_WIN when only values or nested values contain CHAR/VARCHAR. - **Behavior:** Verify constrained CHAR/VARCHAR keys still pad or trim, raise duplicates under EXCEPTION, and keep the last value under LAST_WIN. - **Regression:** Exercise nested ordinary-key maps around constrained values so an outer unchanged key array is never rebuilt through deduplication. ########## docs/sql-migration-guide.md: ########## @@ -29,6 +29,7 @@ license: | ## Upgrading from Spark SQL 4.3 to 4.4 - Since Spark 4.4, when `spark.sql.preserveCharVarcharTypeInfo` is true and `spark.sql.charVarchar.standardSemantics.enabled` is false, ORC reads that apply a CHAR/VARCHAR schema over STRING storage return the stored values without ORC truncation, matching Parquet. Previously the ORC reader requested `char(n)`/`varchar(n)` and truncated STRING-stored values to `n`. Read-side length checks (`EXCEED_LIMIT_LENGTH`) apply only when `spark.sql.charVarchar.standardSemantics.enabled` is true. +- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is true, `from_json` and the JSON datasource parse CHAR/VARCHAR fields as unbounded STRING and then apply write-side assignment (`CHAR` pad, `EXCEED_LIMIT_LENGTH` on overflow), including nested values and map keys. Distinct object names that become the same key after CHAR padding or VARCHAR trailing-space trimming honor `spark.sql.mapKeyDedupPolicy`, as do exact repeated member names in a CHAR/VARCHAR-keyed map. An assignment failure is a parse-mode error for the whole parsed value. PERMISSIVE `from_json` of a STRUCT returns a struct whose fields are null (`IS NULL` is false); PERMISSIVE non-struct roots and datasource rows return null. FAILFAST raises. JSON type-conversion partial results (`spark.sql.json.enablePartialResults`) are unchanged for non-CHAR mismatches. With the flag off, JSON CHAR/VARCHAR handling is unchanged from Spark 4.3. Review Comment: **Nit (P3):** PERMISSIVE data-source parsing does not return a null InternalRow here; FailureSafeParser returns a row of the requested arity with null data fields, plus the corrupt payload when configured. Please distinguish that row shape from a null non-struct from_json result. **Verification:** - **Inspection:** Compare each documented root/data-source shape with its existing result assertion and FailureSafeParser construction. ########## sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala: ########## @@ -2987,6 +3006,125 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { } } + test("SPARK-59722: JSON CHAR/VARCHAR assignment runs after STRING parse") { + withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { + // Scalar pad still applies, via stringLengthCheck on the parsed STRING value. + checkAnswer( + sql("""SELECT from_json('{"a": "str"}', 'a CHAR(5)')"""), + Row(Row("str "))) + checkAnswer( + sql("""SELECT from_json('{"ab": 1}', 'MAP<CHAR(4), INT>')"""), + Row(Map("ab " -> 1))) + + val collision = """SELECT from_json('{"a":1,"a ":2}', 'MAP<CHAR(2), INT>')""" + val exactRepeat = """SELECT from_json('{"a":1,"a":2}', 'MAP<CHAR(2), INT>')""" + val varcharCollision = + """SELECT from_json('{"ab":1,"ab ":2}', 'MAP<VARCHAR(2), INT>')""" + val nestedCollision = + """SELECT from_json( + | '{"outer":{"a":1,"a ":2}}', + | 'MAP<STRING, MAP<CHAR(2), INT>>')""".stripMargin + val overflow = """SELECT from_json('{"abc":1}', 'MAP<VARCHAR(2), INT>')""" + val overflowFailfast = + """SELECT from_json( + | '{"abc":1}', + | 'MAP<VARCHAR(2), INT>', + | map('mode', 'FAILFAST'))""".stripMargin + val overflowWithSibling = + """SELECT from_json( + | '{"m":{"k":"abcdef"},"tail":1}', + | 'm MAP<STRING, CHAR(2)>, tail INT')""".stripMargin + + // Pad/trim collisions and exact JSON member repeats both use mapKeyDedupPolicy. + assertDuplicateMapKey(collision) + assertDuplicateMapKey(exactRepeat) + assertDuplicateMapKey(varcharCollision, expectedKey = "ab") + assertDuplicateMapKey(nestedCollision) + checkAnswer(sql(overflow), Row(null)) + assertParseExceedLimit(overflowFailfast, expectedLimit = "2") + // Assignment failure is whole-value: PERMISSIVE does not keep `tail`. + checkAnswer(sql(overflowWithSibling), Row(Row(null, null))) + assertParseExceedLimit( + """SELECT from_json( + | '{"m":{"k":"abcdef"},"tail":1}', + | 'm MAP<STRING, CHAR(2)>, tail INT', + | map('mode', 'FAILFAST'))""".stripMargin, + expectedLimit = "2") + + withTempPath { path => + Seq("""{"m":{"a":1,"a ":2}}""").toDS().write.text(path.getCanonicalPath) + def readJsonMap(): DataFrame = spark.read Review Comment: The new result assertions prove that CHAR filters stay residual, but the id predicate returns the same rows whether it is pushed or evaluated above the scan. Please assert the V1/V2 pushed-filter state or another parser-side signal for the independent predicate. <!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4226583769","thread_id":"inline:4226583769","verdict_sha256":"b99d61dc014e93932bba0655be66f5d486897769f4fa3a258c0487c8da7566b5"} --> ########## docs/sql-migration-guide.md: ########## @@ -29,6 +29,7 @@ license: | ## Upgrading from Spark SQL 4.3 to 4.4 - Since Spark 4.4, when `spark.sql.preserveCharVarcharTypeInfo` is true and `spark.sql.charVarchar.standardSemantics.enabled` is false, ORC reads that apply a CHAR/VARCHAR schema over STRING storage return the stored values without ORC truncation, matching Parquet. Previously the ORC reader requested `char(n)`/`varchar(n)` and truncated STRING-stored values to `n`. Read-side length checks (`EXCEED_LIMIT_LENGTH`) apply only when `spark.sql.charVarchar.standardSemantics.enabled` is true. +- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is true, `from_json` and the JSON datasource parse CHAR/VARCHAR fields as unbounded STRING and then apply write-side assignment (`CHAR` pad, `EXCEED_LIMIT_LENGTH` on overflow), including nested values and map keys. Distinct object names that become the same key after CHAR padding or VARCHAR trailing-space trimming honor `spark.sql.mapKeyDedupPolicy`, as do exact repeated member names in a CHAR/VARCHAR-keyed map. An assignment failure is a parse-mode error for the whole parsed value (PERMISSIVE returns null; FAILFAST raises). JSON type-conversion partial results (`spark.sql.json.enablePartialResults`) are unchanged for non-CHAR mismatches. Review Comment: The from_json STRUCT wording is fixed, but a permissive JSON data-source failure returns a non-null InternalRow whose fields are null (and may populate the corrupt-record field), not a null row. Please adjust the data-source wording too. <!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4226583772","thread_id":"inline:4226583772","verdict_sha256":"b99d61dc014e93932bba0655be66f5d486897769f4fa3a258c0487c8da7566b5"} --> -- 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]
