You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/09/29 07:35:46 UTC
(spark) branch branch-4.x updated: [SPARK-59801][SQL] Fix internal error for non-foldable options maps
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new a5dbf9560ccf [SPARK-59801][SQL] Fix internal error for non-foldable options maps
a5dbf9560ccf is described below
commit a5dbf9560ccf4d0cc1fb59d1fc20d42b122f8237
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Tue Sep 29 07:34:40 2026 +0000
[SPARK-59801][SQL] Fix internal error for non-foldable options maps
### What changes were proposed in this pull request?
Validate option maps before evaluating them in `ExprUtils.convertToMapData`.
The validation recursively replaces `RuntimeReplaceable` expressions before checking
foldability, so constant SQL compatibility functions such as `nvl` and nested `decode` remain
supported. By default, other non-foldable option maps raise `NON_FOLDABLE_ARGUMENT`.
The session-bound legacy config `spark.sql.legacy.allowNonFoldableOptions` permits deterministic,
row-independent, evaluable non-foldable expressions. References, nondeterministic expressions,
and unevaluable expressions are rejected even in legacy mode.
This also documents the migration behavior and adds coverage for all nine CSV, JSON, and XML
functions that accept option maps.
### Why are the changes needed?
These SQL functions evaluate their option maps during analysis. Previously, a map containing an
input reference could fail with an internal error, and some row-dependent expressions could be
evaluated incorrectly without an input row. A direct foldability check would also reject constant
`RuntimeReplaceable` expressions even though their replacements are safely evaluable.
### Does this PR introduce _any_ user-facing change?
Yes. The affected CSV, JSON, and XML functions now report `NON_FOLDABLE_ARGUMENT` for unsafe
non-foldable option maps instead of failing internally or evaluating them without an input row.
Constant `RuntimeReplaceable` expressions remain supported. The legacy config can temporarily
allow other deterministic, row-independent, evaluable expressions.
### How was this patch tested?
Added regression tests covering all nine affected functions, constant `RuntimeReplaceable`
expressions, safe legacy-mode expressions, and rejection of columns, lambda variables, outer
references, unevaluable expressions, and nondeterministic expressions.
Ran:
```
build/sbt 'sql/testOnly org.apache.spark.sql.CsvFunctionsSuite org.apache.spark.sql.JsonFunctionsSuite org.apache.spark.sql.XmlFunctionsSuite'
build/sbt 'hive/testOnly org.apache.spark.sql.configaudit.SparkConfigBindingPolicySuite'
```
All 216 function-suite tests and all 3 config binding-policy tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #59072 from cloud-fan/fix-non-foldable-options-map.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit b03b53dcacb42c9c31c0cd2b622d495d60c3f59f)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
AGENTS.md | 2 +-
docs/sql-migration-guide.md | 1 +
.../spark/sql/catalyst/expressions/ExprUtils.scala | 15 ++-
.../sql/catalyst/expressions/csvExpressions.scala | 6 +-
.../sql/catalyst/expressions/jsonExpressions.scala | 6 +-
.../sql/catalyst/expressions/xmlExpressions.scala | 6 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 12 +++
.../org/apache/spark/sql/CsvFunctionsSuite.scala | 111 +++++++++++++++++++++
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 27 +++++
.../org/apache/spark/sql/XmlFunctionsSuite.scala | 27 +++++
10 files changed, 201 insertions(+), 12 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index aebb2de64b39..7c33a4e64722 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -101,7 +101,7 @@ These are combined with a base above rather than used on their own:
## Build and Test
-Build and tests can take a long time. If the user explicitly asked to run tests, run them. Otherwise (you are running tests on your own to verify a change), first ask the user if they have more changes to make.
+Build and tests can take a long time, so prefer the narrowest relevant build or test command.
For build and test setup, including how to run tests and troubleshoot common
local failures, see `docs/building-spark.md`.
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 88ad6d029d83..5dd87d7cacb6 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -24,6 +24,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, the options maps passed to `from_csv`, `to_csv`, `schema_of_csv`, `from_json`, `to_json`, `schema_of_json`, `from_xml`, `to_xml`, and `schema_of_xml` must be foldable after replacing `RuntimeReplaceable` expressions. Previously, Spark evaluated non-foldable options during analysis, which allowed some constant expressions but could fail with an internal error or incorrectly evaluate row-dependent expressions. To allow deterministic and row-independent non-foldable optio [...]
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed after a compatible schema change, connectors can return more data columns from the current table schema in `Scan.readSchema()` than requested by `SupportsPushDownRequiredColumns.pruneColumns`. Previously, this partial pruning could fail planning because the scan reported columns absent from the analyzed relation output.
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed, if rebinding an expanded struct used as a map key causes distinct current keys to collide under the analyzed schema, the query fails with `DUPLICATED_MAP_KEY` by default instead of returning duplicate keys. With `spark.sql.mapKeyDedupPolicy=LAST_WIN`, the last value is kept.
- Since Spark 4.4, the Oracle JDBC datasource maps bare `NUMBER` columns (declared without explicit precision or scale) to `DecimalType(38, 18)` instead of `DecimalType(38, 10)`. This uses `DecimalType.DEFAULT_SCALE` and preserves up to 18 fractional digits instead of 10, reducing silent precision loss. However, the integer range shrinks from 28 to 20 digits; bare `NUMBER` values with more than 20 integer digits that previously read correctly will raise `NUMERIC_VALUE_OUT_OF_RANGE`. The [...]
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
index f3f2988abddc..e4f79b9ced09 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
@@ -24,10 +24,12 @@ import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions
import org.apache.spark.sql.catalyst.plans.logical.Aggregate
import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, CharVarcharUtils}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase, QueryExecutionErrors}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.{AbstractMapType, StringTypeWithCollation}
import org.apache.spark.sql.types.{DataType, MapType, StringType, StructType, VariantType}
import org.apache.spark.unsafe.types.UTF8String
@@ -59,13 +61,22 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
dataType.asInstanceOf[StructType]
}
- def convertToMapData(exp: Expression): Map[String, String] = exp match {
+ def convertToMapData(exp: Expression, functionName: String): Map[String, String] = exp match {
case m: CreateMap
if AbstractMapType(
StringTypeWithCollation(supportsTrimCollation = true),
StringTypeWithCollation(supportsTrimCollation = true))
.acceptsType(m.dataType) =>
- val arrayMap = m.eval().asInstanceOf[ArrayBasedMapData]
+ val preparedMap = ReplaceExpressions.replace(m)
+ val unsafeToEvaluate = preparedMap.references.nonEmpty ||
+ !preparedMap.deterministic || preparedMap.exists(_.isInstanceOf[Unevaluable])
+ val allowNonFoldable =
+ SQLConf.get.getConf(SQLConf.LEGACY_ALLOW_NON_FOLDABLE_OPTIONS)
+ if (unsafeToEvaluate || (!preparedMap.foldable && !allowNonFoldable)) {
+ throw QueryCompilationErrors.nonFoldableArgumentError(
+ functionName, "options", m.dataType)
+ }
+ val arrayMap = preparedMap.eval().asInstanceOf[ArrayBasedMapData]
ArrayBasedMapData.toScalaMap(arrayMap).map { case (key, value) =>
if (key == null) {
throw QueryExecutionErrors.nullAsMapKeyNotAllowedError()
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
index 15d4c15dcbde..25d154066b67 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
@@ -85,7 +85,7 @@ case class CsvToStructs(
def this(child: Expression, schema: Expression, options: Expression) =
this(
schema = ExprUtils.evalSchemaExpr(schema),
- options = ExprUtils.convertToMapData(options),
+ options = ExprUtils.convertToMapData(options, "from_csv"),
child = child,
timeZoneId = None)
@@ -167,7 +167,7 @@ case class SchemaOfCsv(
def this(child: Expression, options: Expression) = this(
child = child,
- options = ExprUtils.convertToMapData(options))
+ options = ExprUtils.convertToMapData(options, "schema_of_csv"))
override def nullable: Boolean = false
@@ -255,7 +255,7 @@ case class StructsToCsv(
def this(child: Expression, options: Expression) =
this(
- options = ExprUtils.convertToMapData(options),
+ options = ExprUtils.convertToMapData(options, "to_csv"),
child = child,
timeZoneId = None)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 8ff780c7488d..06728984d3da 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -1904,7 +1904,7 @@ case class JsonToStructs(
def this(child: Expression, schema: Expression, options: Expression) =
this(
schema = ExprUtils.evalTypeExpr(schema),
- options = ExprUtils.convertToMapData(options),
+ options = ExprUtils.convertToMapData(options, "from_json"),
child = child,
timeZoneId = None)
@@ -2017,7 +2017,7 @@ case class StructsToJson(
def this(child: Expression) = this(Map.empty, child, None)
def this(child: Expression, options: Expression) =
this(
- options = ExprUtils.convertToMapData(options),
+ options = ExprUtils.convertToMapData(options, "to_json"),
child = child,
timeZoneId = None)
@@ -2087,7 +2087,7 @@ case class SchemaOfJson(
def this(child: Expression, options: Expression) = this(
child = child,
- options = ExprUtils.convertToMapData(options))
+ options = ExprUtils.convertToMapData(options, "schema_of_json"))
override def nullable: Boolean = false
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
index 31aea4910543..6baee74acecb 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
@@ -86,7 +86,7 @@ case class XmlToStructs(
def this(child: Expression, schema: Expression, options: Expression) =
this(
schema = ExprUtils.evalTypeExpr(schema),
- options = ExprUtils.convertToMapData(options),
+ options = ExprUtils.convertToMapData(options, "from_xml"),
child = child,
timeZoneId = None)
@@ -173,7 +173,7 @@ case class SchemaOfXml(
def this(child: Expression, options: Expression) = this(
child = child,
- options = ExprUtils.convertToMapData(options))
+ options = ExprUtils.convertToMapData(options, "schema_of_xml"))
override def nullable: Boolean = false
@@ -274,7 +274,7 @@ case class StructsToXml(
def this(child: Expression, options: Expression) =
this(
- options = ExprUtils.convertToMapData(options),
+ options = ExprUtils.convertToMapData(options, "to_xml"),
child = child,
timeZoneId = None)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 320eea9ec712..9d3dd349a765 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -7525,6 +7525,18 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val LEGACY_ALLOW_NON_FOLDABLE_OPTIONS =
+ buildConf("spark.sql.legacy.allowNonFoldableOptions")
+ .internal()
+ .doc("When true, allow deterministic and row-independent non-foldable option maps in " +
+ "CSV, JSON, and XML SQL functions and evaluate them during analysis, which is the " +
+ "behavior in Spark 4.3 and earlier. Row-dependent, unevaluable, and nondeterministic " +
+ "option maps are always rejected.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
+
val LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT =
buildConf("spark.sql.legacy.createHiveTableByDefault")
.internal()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala
index f79a7535cc14..854860221da3 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala
@@ -439,6 +439,117 @@ class CsvFunctionsSuite extends SharedSparkSession {
checkAnswer(df1.selectExpr("to_csv(a)"), Row("1") :: Nil)
}
+ test("SPARK-59801: CSV functions with non-foldable options") {
+ val df = Seq(("1,2", (1, 2), ",")).toDF("csv", "a", "delimiter")
+
+ def errorParameters(functionName: String): Map[String, String] = Map(
+ "funcName" -> s"`$functionName`",
+ "paramName" -> "`options`",
+ "paramType" -> "\"MAP<STRING, STRING>\"")
+
+ val nonFoldableQueries = Seq(
+ "from_csv" -> "from_csv(csv, 'a INT, b INT', map('delimiter', delimiter))",
+ "to_csv" -> "to_csv(a, map('delimiter', delimiter))",
+ "schema_of_csv" -> "schema_of_csv('1,2', map('delimiter', delimiter))")
+ nonFoldableQueries.foreach { case (functionName, query) =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr(query)
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = errorParameters(functionName),
+ context = ExpectedContext(
+ fragment = query,
+ start = 0,
+ stop = query.length - 1))
+ }
+
+ val runtimeReplaceableOptions = Seq(
+ "nvl(NULL, '|')",
+ "decode(encode('|', 'UTF-8'), 'UTF-8')")
+ runtimeReplaceableOptions.foreach { option =>
+ checkAnswer(
+ df.selectExpr(s"to_csv(a, map('delimiter', $option))"),
+ Row("1|2") :: Nil)
+ }
+
+ val evaluableNonFoldableOptions = Seq(
+ "elt(1, '|', ',')",
+ "transform(array('|'), x -> x)[0]",
+ "CASE WHEN length('|') = 1 THEN '|' ELSE raise_error('bad') END")
+ evaluableNonFoldableOptions.foreach { option =>
+ val query = s"to_csv(a, map('delimiter', $option))"
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr(query)
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = errorParameters("to_csv"),
+ context = ExpectedContext(
+ fragment = query,
+ start = 0,
+ stop = query.length - 1))
+ }
+
+ val lambdaQuery =
+ "to_csv(named_struct('a', 1, 'b', 2), map('delimiter', x))"
+ val outerReferenceQuery =
+ """SELECT (SELECT to_csv(named_struct('a', 1, 'b', 2),
+ | map('delimiter', delimiter)))
+ |FROM VALUES ('|') t(delimiter)
+ |""".stripMargin
+
+ def checkUnsafeOptions(): Unit = {
+ Seq(
+ "to_csv(a, map('delimiter', delimiter))",
+ "to_csv(a, map('delimiter', uuid()))").foreach { query =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr(query)
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = errorParameters("to_csv"),
+ context = ExpectedContext(
+ fragment = query,
+ start = 0,
+ stop = query.length - 1))
+ }
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(
+ s"""SELECT transform(array('|'),
+ | x -> $lambdaQuery)
+ |""".stripMargin).collect()
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = errorParameters("to_csv"),
+ context = ExpectedContext(fragment = lambdaQuery))
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(outerReferenceQuery).collect()
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = errorParameters("to_csv"),
+ context = ExpectedContext(
+ fragment = """to_csv(named_struct('a', 1, 'b', 2),
+ | map('delimiter', delimiter))""".stripMargin,
+ start = 15,
+ stop = 81))
+ }
+
+ checkUnsafeOptions()
+ withSQLConf(SQLConf.LEGACY_ALLOW_NON_FOLDABLE_OPTIONS.key -> "true") {
+ evaluableNonFoldableOptions.foreach { option =>
+ checkAnswer(
+ df.selectExpr(s"to_csv(a, map('delimiter', $option))"),
+ Row("1|2") :: Nil)
+ }
+ checkUnsafeOptions()
+ }
+ }
+
test("parse timestamps with locale") {
Seq("en-US", "ko-KR", "zh-CN", "ru-RU").foreach { langTag =>
val locale = Locale.forLanguageTag(langTag)
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 c0ae3fc157f7..90577945c3a3 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
@@ -1022,6 +1022,33 @@ class JsonFunctionsSuite extends SharedSparkSession {
stop = 22
)
)
+
+ }
+
+ test("SPARK-59801: JSON functions with non-foldable options") {
+ val df = Seq((Tuple1(1), """{"a":1}""", "true")).toDF("a", "json", "option")
+ val nonFoldableQueries = Seq(
+ "from_json" ->
+ "from_json(json, 'a INT', map('primitivesAsString', option))",
+ "to_json" -> "to_json(a, map('pretty', option))",
+ "schema_of_json" ->
+ """schema_of_json('{"a":1}', map('primitivesAsString', option))""")
+
+ nonFoldableQueries.foreach { case (functionName, query) =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr(query)
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = Map(
+ "funcName" -> s"`$functionName`",
+ "paramName" -> "`options`",
+ "paramType" -> "\"MAP<STRING, STRING>\""),
+ context = ExpectedContext(
+ fragment = query,
+ start = 0,
+ stop = query.length - 1))
+ }
}
test("SPARK-19967 Support from_json in SQL") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
index bfe46e681261..7f874d261b7c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
@@ -474,6 +474,33 @@ class XmlFunctionsSuite extends SharedSparkSession {
stop = 21
)
)
+
+ }
+
+ test("SPARK-59801: XML functions with non-foldable options") {
+ val df = Seq((Tuple1(1), "<ROW><a>1</a></ROW>", "ROW"))
+ .toDF("a", "xml", "rowTag")
+ val nonFoldableQueries = Seq(
+ "from_xml" -> "from_xml(xml, 'a INT', map('rowTag', rowTag))",
+ "to_xml" -> "to_xml(a, map('rowTag', rowTag))",
+ "schema_of_xml" ->
+ "schema_of_xml('<ROW><a>1</a></ROW>', map('rowTag', rowTag))")
+
+ nonFoldableQueries.foreach { case (functionName, query) =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr(query)
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = Map(
+ "funcName" -> s"`$functionName`",
+ "paramName" -> "`options`",
+ "paramType" -> "\"MAP<STRING, STRING>\""),
+ context = ExpectedContext(
+ fragment = query,
+ start = 0,
+ stop = query.length - 1))
+ }
}
test("Support from_xml in SQL") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org