You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sa...@apache.org on 2026/10/22 13:19:08 UTC
(spark) branch branch-4.x updated: [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS
This is an automated email from the ASF dual-hosted git repository.
sarutak 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 4af3371cd59a [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS
4af3371cd59a is described below
commit 4af3371cd59a0794e371384460729b1cc2e6ea69
Author: rdtr <no...@apache.org>
AuthorDate: Thu Oct 22 22:18:43 2026 +0900
[SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS
### What changes were proposed in this pull request?
In `SparkConnectPlanner.handleSqlCommand`, the SDP guard previously received a `queryPlan` built by `transformRelation(relation)`. For `WITH_RELATIONS` matching `isValidSQLWithRefs`, that chain routes through `executeSQLWithRefs` → `session.sql(...)` and eagerly executes
the root SQL (and any Commands resolved from references), producing a `CommandResult` that bypasses the blocklist. The normal execution path then runs the same query again.
This PR introduces a recursive helper `blockUnsupportedSqlCommandIn(rel)` that walks the relation tree and inspects each component the normal handler would execute, without triggering execution. For a valid SQL-with-references `WITH_RELATIONS`, it parses the root SQL with
`transformSql` to get an unresolved plan and checks each reference's `SubqueryAlias.input` the same way, so a valid SQL-with-references `WITH_RELATIONS` used directly as a reference input is also parsed instead of executed. For other relation types, the behavior is unchanged.
Known limitation: a SQL-with-references `WITH_RELATIONS` nested under another relation (for example a `Limit`), or wrapped in a Dataset-building relation such as `SHOW_STRING`, still goes through `transformRelation` and is executed inside the guard, as before this PR.
### Why are the changes needed?
Two bugs defeat the SDP guard:
1. **Bypassed guard.** The blocklist rejects any `Command`, plus a few side-effecting plans that are not `Command` subclasses (`CreateTableAsSelect`, `InsertIntoStatement`, etc.). But after execution the plan is wrapped as `CommandResult`, which is neither, so it passes. Commands have already changed state by the time the guard runs.
Commands inside reference `SubqueryAlias` inputs also execute, because `executeSQLWithRefs` runs each reference as its own query (`Dataset.ofRows(session, transformRelation(ref.input))`) to register its temp view.
2. **Double execution.** After the guard, `handleSqlCommand` falls through to the normal execution path and runs `executeSQLWithRefs` again, causing duplicate side effects.
The guard should match the runtime's execution surface: inspect both the root SQL and each reference's input, without triggering execution itself.
### Does this PR introduce _any_ user-facing change?
Yes. SDP requests containing blocked Commands inside `WITH_RELATIONS` (at the root or in a reference's input) now fail with `UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND` before any side effects occur.
### How was this patch tested?
Added tests in `SparkDeclarativePipelinesServerSuite`. The three command tests run both outside and inside a pipeline flow function (with a flow name in the pipeline analysis context), because the guard runs before the flow-function early return:
- Command at the root of `WITH_RELATIONS` is blocked, and the target table is asserted not to exist (verifying no side effect).
- Command embedded in a `SubqueryAlias` reference input is blocked, and the target table is asserted not to exist.
- Command at the root of a `WITH_RELATIONS` that is itself a reference input is blocked, and the target table is asserted not to exist. This covers the recursive check of reference inputs.
- A valid `SELECT` `WITH_RELATIONS` in pipeline analysis context is not blocked. This test passes with and without the fix. It keeps the new recursive check from blocking a valid query in the future.
Without the fix, the three command tests fail in both variants because the target table is created as a side effect of the guard.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59425 from rdtr/fix-handle-sql-command-with-relations.
Authored-by: rdtr <no...@apache.org>
Signed-off-by: Kousuke Saruta <sa...@apache.org>
(cherry picked from commit aa0b584e1bf552d73bb0673834338810fe769884)
Signed-off-by: Kousuke Saruta <sa...@apache.org>
---
.../sql/connect/planner/SparkConnectPlanner.scala | 31 +++-
.../SparkDeclarativePipelinesServerSuite.scala | 175 +++++++++++++++++++++
2 files changed, 205 insertions(+), 1 deletion(-)
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 f59eee0dbab3..f18e953a6f7d 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
@@ -3152,8 +3152,13 @@ class SparkConnectPlanner(
}
// Block unsupported SQL commands if the request comes from Spark Declarative Pipelines.
+ // For a valid SQL-with-references WITH_RELATIONS, transformRelation would route
+ // through executeSQLWithRefs and eagerly run the root SQL as a side effect, producing
+ // a CommandResult that bypasses the blocklist. blockUnsupportedSqlCommandIn checks such
+ // a relation at the top level or as a direct reference input without executing it. See
+ // its scaladoc for a known limitation.
if (PipelineAnalysisContextUtils.hasPipelineAnalysisContext(userContext)) {
- PipelinesHandler.blockUnsupportedSqlCommand(queryPlan = transformRelation(relation))
+ blockUnsupportedSqlCommandIn(relation)
}
// If the spark.sql() is called inside a pipeline flow function, we don't need to execute
@@ -3256,6 +3261,30 @@ class SparkConnectPlanner(
}
}
+ /**
+ * Applies [[PipelinesHandler.blockUnsupportedSqlCommand]] to a relation without executing a
+ * valid SQL-with-references [[proto.WithRelations]] at the top level or as a direct reference
+ * input. For such a relation, the root SQL is parsed (not executed) and each reference input is
+ * checked with this method again. Any other relation is passed to [[transformRelation]].
+ *
+ * Known limitation: [[transformRelation]] still executes a SQL-with-references
+ * [[proto.WithRelations]] that is nested under another relation (for example a `Limit`) or
+ * wrapped in a Dataset-building relation such as `SHOW_STRING`, as it did before this check.
+ */
+ private def blockUnsupportedSqlCommandIn(rel: proto.Relation): Unit = {
+ rel.getRelTypeCase match {
+ case proto.Relation.RelTypeCase.WITH_RELATIONS
+ if isValidSQLWithRefs(rel.getWithRelations) =>
+ val withRel = rel.getWithRelations
+ PipelinesHandler.blockUnsupportedSqlCommand(transformSql(withRel.getRoot.getSql))
+ withRel.getReferencesList.asScala.foreach { ref =>
+ blockUnsupportedSqlCommandIn(ref.getSubqueryAlias.getInput)
+ }
+ case _ =>
+ PipelinesHandler.blockUnsupportedSqlCommand(transformRelation(rel))
+ }
+ }
+
private def isValidSQLWithRefs(query: proto.WithRelations): Boolean = {
query.getRoot.getRelTypeCase match {
case proto.Relation.RelTypeCase.SQL =>
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
index 4ada35c7629e..1842985024b2 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
@@ -1164,4 +1164,179 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext. A flow name
+ // marks the request as coming from inside a pipeline flow function.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation,
+ flowName: Option[String] = None): proto.ExecutePlanRequest = {
+ val contextBuilder = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ flowName.foreach(contextBuilder.setFlowName)
+ val context = contextBuilder.build()
+ val userContext = proto.UserContext
+ .newBuilder()
+ .addExtensions(com.google.protobuf.Any.pack(context))
+ .setUserId("test_user")
+ .build()
+ val plan = proto.Plan
+ .newBuilder()
+ .setCommand(
+ proto.Command
+ .newBuilder()
+ .setSqlCommand(proto.SqlCommand.newBuilder().setInput(input).build())
+ .build())
+ .build()
+ proto.ExecutePlanRequest
+ .newBuilder()
+ .setUserContext(userContext)
+ .setPlan(plan)
+ .setSessionId(UUID.randomUUID().toString)
+ .build()
+ }
+
+ // Helper to build a SQL Relation.
+ private def sqlRelation(query: String): proto.Relation =
+ proto.Relation
+ .newBuilder()
+ .setSql(proto.SQL.newBuilder().setQuery(query))
+ .build()
+
+ // Helper to build a WITH_RELATIONS Relation with the given root SQL and reference SQLs.
+ private def withRelationsRelation(
+ rootSql: String,
+ refs: Seq[(String, String)]): proto.Relation =
+ withRelationsRelationOf(
+ rootSql,
+ refs.map { case (alias, refSql) =>
+ alias -> sqlRelation(refSql)
+ })
+
+ // Helper to build a WITH_RELATIONS Relation with the given root SQL and reference inputs.
+ private def withRelationsRelationOf(
+ rootSql: String,
+ refs: Seq[(String, proto.Relation)]): proto.Relation = {
+ val withRelationsBuilder = proto.WithRelations.newBuilder().setRoot(sqlRelation(rootSql))
+ refs.foreach { case (alias, input) =>
+ withRelationsBuilder.addReferences(
+ proto.Relation
+ .newBuilder()
+ .setSubqueryAlias(
+ proto.SubqueryAlias
+ .newBuilder()
+ .setAlias(alias)
+ .setInput(input)
+ .build())
+ .build())
+ }
+ proto.Relation.newBuilder().setWithRelations(withRelationsBuilder.build()).build()
+ }
+
+ namedGridTest(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command at root of WITH_RELATIONS " +
+ "without executing it")(
+ Map("outside a flow function" -> None, "inside a flow function" -> Some("flow1"))) {
+ flowName =>
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val tableName = "spark_catalog.default.test_with_relations_root_table"
+ try {
+ sql(s"DROP TABLE IF EXISTS $tableName")
+ val request = buildSqlCommandRequestWithPipelineContext(
+ graphId,
+ flowName = flowName,
+ input = withRelationsRelation(
+ rootSql = s"CREATE TABLE $tableName (id INT) USING parquet",
+ refs = Seq("ref0" -> "SELECT 1 AS id")))
+ val ex = intercept[RuntimeException] {
+ stub.executePlan(request).next()
+ }
+ assert(ex.getMessage.contains("UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND"))
+ // Verify the guard caught the Command before any side effect could materialize.
+ assert(!spark.catalog.tableExists(tableName))
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+
+ namedGridTest(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command inside reference of " +
+ "WITH_RELATIONS without executing it")(
+ Map("outside a flow function" -> None, "inside a flow function" -> Some("flow1"))) {
+ flowName =>
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val tableName = "spark_catalog.default.test_with_relations_ref_table"
+ try {
+ sql(s"DROP TABLE IF EXISTS $tableName")
+ val request = buildSqlCommandRequestWithPipelineContext(
+ graphId,
+ flowName = flowName,
+ input = withRelationsRelation(
+ rootSql = "SELECT * FROM ref0",
+ refs = Seq("ref0" -> s"CREATE TABLE $tableName (id INT) USING parquet")))
+ val ex = intercept[RuntimeException] {
+ stub.executePlan(request).next()
+ }
+ assert(ex.getMessage.contains("UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND"))
+ // Verify the Command in the reference did not execute as a side effect.
+ assert(!spark.catalog.tableExists(tableName))
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+
+ namedGridTest(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command in a WITH_RELATIONS used as a " +
+ "reference input without executing it")(
+ Map("outside a flow function" -> None, "inside a flow function" -> Some("flow1"))) {
+ flowName =>
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val tableName = "spark_catalog.default.test_with_relations_nested_ref_table"
+ try {
+ sql(s"DROP TABLE IF EXISTS $tableName")
+ // A DataFrame from spark.sql("...", df=...) carries a WITH_RELATIONS relation, and it
+ // becomes a reference input when passed to another spark.sql call.
+ val nested = withRelationsRelation(
+ rootSql = s"CREATE TABLE $tableName (id INT) USING parquet",
+ refs = Seq("inner0" -> "SELECT 1 AS id"))
+ val request = buildSqlCommandRequestWithPipelineContext(
+ graphId,
+ flowName = flowName,
+ input = withRelationsRelationOf(
+ rootSql = "SELECT * FROM ref0",
+ refs = Seq("ref0" -> nested)))
+ val ex = intercept[RuntimeException] {
+ stub.executePlan(request).next()
+ }
+ assert(ex.getMessage.contains("UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND"))
+ // Verify the Command in the nested WITH_RELATIONS did not execute as a side effect.
+ assert(!spark.catalog.tableExists(tableName))
+ } finally {
+ sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand does not over-block a valid SELECT " +
+ "WITH_RELATIONS in pipeline analysis context") {
+ // This passes with and without the fix. It keeps the new recursive check from blocking a
+ // valid SELECT in the future. The two tests above are the ones that verify the fix.
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val request = buildSqlCommandRequestWithPipelineContext(
+ graphId,
+ withRelationsRelation(
+ rootSql = "SELECT * FROM ref0",
+ refs = Seq("ref0" -> "SELECT 1 AS id")))
+ val response = stub.executePlan(request).next()
+ assert(response.hasSqlCommandResult)
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org