This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 3b850515ebb2 [SPARK-58341][SQL] Fix wrong results with 5+ positional
parameters in `sql`
3b850515ebb2 is described below
commit 3b850515ebb2295daae9663ad10dc715985abea2
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Sat Jul 25 12:39:30 2026 -0700
[SPARK-58341][SQL] Fix wrong results with 5+ positional parameters in `sql`
### What changes were proposed in this pull request?
This PR fixes a correctness issue where `spark.sql` binds 5 or more
positional parameters in the wrong order. The resolved parameters were
collected via `resolvedParams.values.toSeq` from a `Map` keyed by `_pos_<idx>`,
but Scala's immutable `Map` does not preserve insertion order for 5+ entries.
This PR looks them up by the positional key instead, in both affected places:
- `classic.SparkSession.sql(sqlText, args: Array[_], tracker)`
- `SparkConnectPlanner.buildParameterContext` (`pos_arguments`)
### Why are the changes needed?
Since Apache Spark 4.1.0 (SPARK-53573), queries with 5+ positional
parameters
silently return wrong results in both the classic API and Spark Connect.
- https://github.com/apache/spark/pull/52334
**BEFORE (Spark 4.2.0)**
```
$ bin/spark-shell
WARNING: Using incubator modules: jdk.incubator.vector
Using Spark's default log4j profile:
org/apache/spark/log4j2-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use
setLogLevel(newLevel).
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 4.2.0
/_/
Using Scala version 2.13.18 (OpenJDK 64-Bit Server VM, Java 21.0.12)
Type in expressions to have them evaluated.
Type :help for more information.
26/07/24 15:46:06 WARN NativeCodeLoader: Unable to load native-hadoop
library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://localhost:4040
Spark context available as 'sc' (master = local[*], app id =
local-1784933166639).
Spark session available as 'spark'.
scala> spark.sql("SELECT ?, ?, ?, ?, ?, ?", Array(1, 2, 3, 4, 5,
6)).collect()
val res0: Array[org.apache.spark.sql.Row] = Array([6,2,5,3,4,1])
```
**AFTER (this PR)**
```
$ bin/spark-shell
WARNING: Using incubator modules: jdk.incubator.vector
Using Spark's default log4j profile:
org/apache/spark/log4j2-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use
setLogLevel(newLevel).
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 5.0.0-SNAPSHOT
/_/
Using Scala version 2.13.18 (OpenJDK 64-Bit Server VM, Java 21.0.12)
Type in expressions to have them evaluated.
Type :help for more information.
26/07/24 15:48:50 WARN NativeCodeLoader: Unable to load native-hadoop
library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://localhost:4040
Spark context available as 'sc' (master = local[*], app id =
local-1784933331484).
Spark session available as 'spark'.
scala> spark.sql("SELECT ?, ?, ?, ?, ?, ?", Array(1, 2, 3, 4, 5,
6)).collect()
val res0: Array[org.apache.spark.sql.Row] = Array([1,2,3,4,5,6])
```
### Does this PR introduce _any_ user-facing change?
Yes, this is a correctness bug fix of the released versions 4.1.0, 4.1.1,
4.1.2, 4.1.3 and 4.2.0.
### How was this patch tested?
Pass the CIs with the newly added test case.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #57516 from dongjoon-hyun/SPARK-58341.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit b910d6f4101c53dd5f9b866122d92e4c41c768f3)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/connect/planner/SparkConnectPlanner.scala | 4 +++-
.../spark/sql/connect/planner/SparkConnectPlannerSuite.scala | 12 ++++++++++++
.../scala/org/apache/spark/sql/classic/SparkSession.scala | 11 +++++++----
.../test/scala/org/apache/spark/sql/ParametersSuite.scala | 10 ++++++++++
4 files changed, 32 insertions(+), 5 deletions(-)
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index 1922eb14bda7..841ce26402cf 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -391,7 +391,9 @@ class SparkConnectPlanner(
s"_pos_$idx" -> expr
}.toMap
val resolvedParams = session.resolveAndValidateParameters(paramMap)
- Some(PositionalParameterContext(resolvedParams.values.toSeq))
+ // Look up by the positional key instead of relying on
`resolvedParams.values`:
+ // the map does not preserve insertion order for 5+ entries.
+ Some(PositionalParameterContext(paramList.indices.map(idx =>
resolvedParams(s"_pos_$idx"))))
} else if (!args.isEmpty) {
// Use named arguments (literals) - already resolved
val paramMap = args.asScala.toMap.transform((_, v) =>
transformLiteral(v))
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
index 991b3f2a6c05..a8eec9ec7d56 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
@@ -656,6 +656,18 @@ class SparkConnectPlannerSuite extends SparkFunSuite with
SparkConnectPlanTest {
assert(array(2).toString == InternalRow(3, "kafka", 3, "kafka").toString)
}
+ test("SPARK-58341: transform SQL with 5 or more positional arguments binds
in order") {
+ val sql = proto.SQL
+ .newBuilder()
+ .setQuery("SELECT ?, ?, ?, ?, ?, ?")
+ (1 to 6).foreach { v =>
+
sql.addPosArguments(proto.Expression.newBuilder().setLiteral(toLiteralProto(v)))
+ }
+
+ val df = Dataset.ofRows(spark,
transform(proto.Relation.newBuilder.setSql(sql).build()))
+ assert(df.collect() === Array(Row(1, 2, 3, 4, 5, 6)))
+ }
+
test("transform UnresolvedStar with target field") {
val rows = (0 until 10).map { i =>
InternalRow(InternalRow(InternalRow(i, i + 1)))
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
index f03b4796314b..c5f01de37000 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
@@ -531,16 +531,19 @@ class SparkSession private(
val parsedPlan = {
// Always parse with parameter context to detect unbound parameter
markers.
// Even if args is empty, we need to detect and reject parameter
markers in the SQL.
- val (paramMap, resolvedParams) = if (args.nonEmpty) {
+ val resolvedParams = if (args.nonEmpty) {
val pMap = args.zipWithIndex.map { case (arg, idx) =>
s"_pos_$idx" -> lit(arg).expr
}.toMap
- (pMap, resolveAndValidateParameters(pMap))
+ resolveAndValidateParameters(pMap)
} else {
- (Map.empty[String, Expression], Map.empty[String, Expression])
+ Map.empty[String, Expression]
}
- val paramContext =
PositionalParameterContext(resolvedParams.values.toSeq)
+ // Look up by the positional key instead of relying on
`resolvedParams.values`:
+ // the map does not preserve insertion order for 5+ entries.
+ val paramContext =
+ PositionalParameterContext(args.indices.map(idx =>
resolvedParams(s"_pos_$idx")))
val parsed = sessionState.sqlParser.parsePlanWithParameters(sqlText,
paramContext)
// Check for SQL scripting with positional parameters
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
index 91a999ad663f..edca2c36f256 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
@@ -74,6 +74,16 @@ class ParametersSuite extends SharedSparkSession {
Row(true))
}
+ test("SPARK-58341: bind 5 or more positional parameters in order") {
+ checkAnswer(
+ spark.sql("SELECT ?, ?, ?, ?, ?, ?", Array(1, 2, 3, 4, 5, 6)),
+ Row(1, 2, 3, 4, 5, 6))
+
+ checkAnswer(
+ spark.sql("SELECT ? + ?, ? * ?, ? - ?", Array(1, 2, 30, 4, 500, 6)),
+ Row(3, 120, 494))
+ }
+
test("parameter binding is case sensitive") {
checkAnswer(
spark.sql("SELECT :p, :P", Map("p" -> 1, "P" -> 2)),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
