From 96939fc1e6ff21bd4cf7eb694925f64a3ae2757f Mon Sep 17 00:00:00 2001 From: franciscoabsampaio Date: Sun, 20 Sep 2026 10:09:46 +0100 Subject: [PATCH] [SPARK-59672][SQL] Bind SQL parameters when a custom parser is injected ### What changes were proposed in this pull request? `ParserInterface.parsePlanWithParameters` had a default implementation that discarded the `ParameterContext` and called `parsePlan(sqlText)`. Only `SparkSqlParser` overrides it, so any parser injected through `SparkSessionExtensions.injectParser` silently dropped the parameter values. The default now marks the parsed plan for parameter binding during analysis, using the same `ParameterizedQuery` nodes the analyzer already applies in legacy mode, so implementations that only override `parsePlan` keep working with parameterized SQL. ### Why are the changes needed? Since SPARK-53573 moved parameter handling into a parser pre-processor, `spark.sql(query, args)` fails with `UNBOUND_SQL_PARAMETER` on any session that sets `spark.sql.extensions`, for both positional and named parameters, in classic Spark and through Spark Connect. Delta Lake and Apache Iceberg both inject a parser, so parameterized SQL - the recommended defence against SQL injection - is unusable for their users on 4.1.x. ### Does this PR introduce any user-facing change? Yes. Parameterized SQL works again on sessions with an injected parser. Parameter markers are bound wherever the parser produces a parameter expression; parsers that want markers in every position a literal is allowed should override `parsePlanWithParameters`, as `SparkSqlParser` does. ### How was this patch tested? New test in `SparkSessionExtensionSuite` covering positional and named parameters through an injected parser, unbound markers still being rejected, and legacy mode. It fails without the change with `UNBOUND_SQL_PARAMETER`. Also ran `ParametersSuite`, `LegacyParameterSubstitutionSuite`, `SparkSessionExtensionSuite`, `SparkConnectWithSessionExtensionSuite` and `SparkConnectPlannerSuite`, plus scalastyle. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5 Co-Authored-By: Claude Opus 5 --- .../sql/catalyst/parser/ParserInterface.scala | 29 +++++++++++++++++-- .../sql/SparkSessionExtensionSuite.scala | 23 +++++++++++++++ 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala index 4ca6fe3cbd91f..c9d5ffcda7989 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala @@ -19,8 +19,10 @@ package org.apache.spark.sql.catalyst.parser import org.apache.spark.annotation.DeveloperApi import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier} +import org.apache.spark.sql.catalyst.analysis.{GeneralParameterizedQuery, NameParameterizedQuery, PosParameterizedQuery} import org.apache.spark.sql.catalyst.expressions.Expression import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.StructType /** @@ -37,12 +39,33 @@ trait ParserInterface extends DataTypeParserInterface { /** * Parse a string to a [[LogicalPlan]] with explicit parameter context. * This method avoids thread-local usage for better API design. + * + * The default implementation parses `sqlText` with [[parsePlan]] and marks the result for + * parameter binding during analysis, so that implementations which only override + * [[parsePlan]] - such as the parsers injected by `SparkSessionExtensions.injectParser` - + * still honour parameterized SQL. This binds markers wherever the parser produces a + * parameter expression; implementations that want markers supported in every position a + * literal is allowed, and error positions reported against the substituted text, should + * override this method and substitute the parameters as part of their own parse, as + * `SparkSqlParser` does. */ @throws[ParseException]("Text cannot be parsed to a LogicalPlan") def parsePlanWithParameters(sqlText: String, parameterContext: ParameterContext): LogicalPlan = { - // Default implementation falls back to regular parsePlan - // Concrete implementations can override this for parameter support - parsePlan(sqlText) + val plan = parsePlan(sqlText) + // In legacy mode the callers of this method wrap the plan themselves. + if (SQLConf.get.legacyParameterSubstitutionConstantsOnly) { + plan + } else { + parameterContext match { + case NamedParameterContext(params) if params.nonEmpty => + NameParameterizedQuery(plan, params) + case PositionalParameterContext(params) if params.nonEmpty => + PosParameterizedQuery(plan, params) + case HybridParameterContext(args, paramNames) if args.nonEmpty => + GeneralParameterizedQuery(plan, args, paramNames) + case _ => plan + } + } } /** diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionExtensionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionExtensionSuite.scala index ee3bdfba4c54e..b47e0cb1e35fe 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionExtensionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionExtensionSuite.scala @@ -173,6 +173,29 @@ class SparkSessionExtensionSuite extends PlanTest with AdaptiveSparkPlanHelper { } } + test("SPARK-59672: injected parser binds SQL parameters") { + val extension = create { extensions => + extensions.injectParser(MyParser) + } + withSession(extension) { session => + assert(session.sql("SELECT ? AS col", Array[Any](42)).collect() === Array(Row(42))) + assert(session.sql("SELECT :p AS col", Map("p" -> 42)).collect() === Array(Row(42))) + // Markers with no value supplied are still rejected. + val e = intercept[AnalysisException](session.sql("SELECT ? AS col").collect()) + assert(e.getCondition === "UNBOUND_SQL_PARAMETER") + + // In legacy mode the markers are bound during analysis instead. + val legacyKey = SQLConf.LEGACY_PARAMETER_SUBSTITUTION_CONSTANTS_ONLY.key + session.conf.set(legacyKey, "true") + try { + assert(session.sql("SELECT ? AS col", Array[Any](42)).collect() === Array(Row(42))) + assert(session.sql("SELECT :p AS col", Map("p" -> 42)).collect() === Array(Row(42))) + } finally { + session.conf.unset(legacyKey) + } + } + } + test("inject function") { val extensions = create { extensions => extensions.injectFunction(MyExtensions.myFunction)