You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "rdtr (via GitHub)" <gi...@apache.org> on 2026/10/13 06:48:37 UTC
[PR] [SPARK-57189][CONNECT] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
rdtr opened a new pull request, #59425:
URL: https://github.com/apache/spark/pull/59425
## What changes were proposed in this pull request?
This replaces https://github.com/apache/spark/pull/56241, which was closed as stale without a review. It is rebased on the current master with no other changes.
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 recursively inspects each reference's `SubqueryAlias.input` so nested `WITH_RELATIONS` cannot re-trigger the executing chain. For other relation types, the behavior is unchanged.
## Why are the changes needed?
Two bugs defeat the SDP guard:
1. **Bypassed guard.** The blocklist checks for `Command` subclasses (`CreateTableAsSelect`, `InsertIntoStatement`, etc.), but after execution the plan is wrapped as `CommandResult` which is not on the list. Commands have already mutated state by the time the guard runs.
Commands inside reference `SubqueryAlias` inputs also execute because `eagerlyExecuteCommands` walks the resolved plan tree.
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 three unit tests in `SparkDeclarativePipelinesServerSuite`:
- 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.
- Valid `SELECT` `WITH_RELATIONS` in pipeline analysis context is allowed (positive case).
Without the fix, the first two tests fail 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)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4319134491
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,127 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .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 = {
+ val withRelationsBuilder = proto.WithRelations.newBuilder().setRoot(sqlRelation(rootSql))
+ refs.foreach { case (alias, refSql) =>
+ withRelationsBuilder.addReferences(
+ proto.Relation
+ .newBuilder()
+ .setSubqueryAlias(
+ proto.SubqueryAlias
+ .newBuilder()
+ .setAlias(alias)
+ .setInput(sqlRelation(refSql))
+ .build())
+ .build())
+ }
+ proto.Relation.newBuilder().setWithRelations(withRelationsBuilder.build()).build()
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command at root of WITH_RELATIONS " +
+ "without executing it") {
+ 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,
+ 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")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command inside reference of " +
Review Comment:
Thanks, added in d4ed446e83. It fails if the recursion is replaced with a plain `transformRelation` check.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6341821855
Merged. Thank you @rdtr and @HyukjinKwon !
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6141509994
cc: @szehon-ho, and @dongjoon-hyun from #53156
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6139985653
@SCHJonathan @sryza @sarutak Could you take a look when you have time? This is about the SDP guard in `handleSqlCommand` from SPARK-54020: for `WITH_RELATIONS`, the guard currently runs the SQL before checking it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4319009978
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,127 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
Review Comment:
Nit, not blocking. This helper only sets `dataflowGraphId`, never `flowName`, so all three new tests exercise the path outside a pipeline flow function. The guard runs at the `hasPipelineAnalysisContext` check, which is before the `isInsidePipelineFlowFunction` early return, so the bug existed inside a flow function too: on the old code the guard ran `executeSQLWithRefs` (e.g. a `CREATE TABLE` root) and the resulting `CommandResult` passed the blocklist. There was no second execution there (the handler then returns the relation), but the command still ran. That is also the most common real shape: inside a query function, `spark.sql("... {df}", df=...)` sends exactly this WITH_RELATIONS with `SubqueryAlias` references, and `register_flow` wraps `flow.func()` in `add_pipeline_analysis_context(..., flow_name=flow.name)`.
Could the helper take an optional `flowName` and the two command tests run both with and without it (e.g. via `gridTest`)? On the old code the flow-function variant would also fail on both `intercept[RuntimeException]` and `assert(!tableExists)`; on the current code it pins that the guard keeps running before the flow-function return. This dovetails with @HyukjinKwon's suggestion below to give the helper a reference-input variant, so both are about widening what the new tests actually cover.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4319138159
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,127 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
Review Comment:
Thanks, done in 64236ec6b0. The helper takes an optional `flowName`, and the command tests run both inside and outside a flow function. All of them fail without the fix.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak closed pull request #59425: [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS
URL: https://github.com/apache/spark/pull/59425
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6277982413
@sarutak Thanks! Both points from @HyukjinKwon and your flow-function nit are addressed, and the PR description is updated.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4256050203
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,125 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .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 = {
+ val withRelationsBuilder = proto.WithRelations.newBuilder().setRoot(sqlRelation(rootSql))
+ refs.foreach { case (alias, refSql) =>
+ withRelationsBuilder.addReferences(
+ proto.Relation
+ .newBuilder()
+ .setSubqueryAlias(
+ proto.SubqueryAlias
+ .newBuilder()
+ .setAlias(alias)
+ .setInput(sqlRelation(refSql))
+ .build())
+ .build())
+ }
+ proto.Relation.newBuilder().setWithRelations(withRelationsBuilder.build()).build()
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command at root of WITH_RELATIONS " +
+ "without executing it") {
+ 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,
+ 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")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command inside reference of " +
+ "WITH_RELATIONS without executing it") {
+ 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,
+ 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")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand allows valid SELECT WITH_RELATIONS in pipeline " +
+ "analysis context") {
+ 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)
Review Comment:
This positive test passes both before and after the fix: reverting the production change does not turn it red (the old guard did notblock a plain SELECT either, so there was no over-block to begin with). It is therefore a forward-looking guard that protects the new recursive `blockUnsupportedSqlCommandIn` from a future regression where it would wrongly block a valid SELECT WITH_RELATIONS. That is worth protecting, but it is worth a one line comment (or a name that reflects "does not over-block a valid SELECT") so a future reader does not mistake it for the test that proves the double-execution/bypass fix. The two negative tests above are the fix-pinning ones: without the fix, `transformRelation` on the WITH_RELATIONS executes the CREATE TABLE and returns a `CommandResult` that is not on the blocklist, so both `intercept[RuntimeException]` and `assert(!tableExists)` fail. Those are non-vacuous. Only this positive case is a forward-looking non-regression check.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4262297685
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,125 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .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 = {
+ val withRelationsBuilder = proto.WithRelations.newBuilder().setRoot(sqlRelation(rootSql))
+ refs.foreach { case (alias, refSql) =>
+ withRelationsBuilder.addReferences(
+ proto.Relation
+ .newBuilder()
+ .setSubqueryAlias(
+ proto.SubqueryAlias
+ .newBuilder()
+ .setAlias(alias)
+ .setInput(sqlRelation(refSql))
+ .build())
+ .build())
+ }
+ proto.Relation.newBuilder().setWithRelations(withRelationsBuilder.build()).build()
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command at root of WITH_RELATIONS " +
+ "without executing it") {
+ 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,
+ 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")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command inside reference of " +
+ "WITH_RELATIONS without executing it") {
+ 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,
+ 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")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand allows valid SELECT WITH_RELATIONS in pipeline " +
+ "analysis context") {
+ 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)
Review Comment:
Good point, thanks. I renamed it to `"blockUnsupportedSqlCommand does not over-block a valid SELECT WITH_RELATIONS in pipeline analysis context"` and added a comment that it passes with and without the fix, and that the two tests above are the ones that verify the fix.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4319009978
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,127 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
Review Comment:
Nit, not blocking. This helper only sets `dataflowGraphId`, never `flowName`, so all three new
tests exercise the path outside a pipeline flow function. The guard runs at the
`hasPipelineAnalysisContext` check, which is before the `isInsidePipelineFlowFunction` early
return, so the bug existed inside a flow function too: on the old code the guard ran
`executeSQLWithRefs` (e.g. a `CREATE TABLE` root) and the resulting `CommandResult` passed the
blocklist. There was no second execution there (the handler then returns the relation), but the
command still ran. That is also the most common real shape: inside a query function,
`spark.sql("... {df}", df=...)` sends exactly this WITH_RELATIONS with `SubqueryAlias` references,
and `register_flow` wraps `flow.func()` in `add_pipeline_analysis_context(..., flow_name=flow.name)`.
Could the helper take an optional `flowName` and the two command tests run both with and without
it (e.g. via `gridTest`)? On the old code the flow-function variant would also fail on both
`intercept[RuntimeException]` and `assert(!tableExists)`; on the current code it pins that the guard
keeps running before the flow-function return. This dovetails with @HyukjinKwon's suggestion below
to give the helper a reference-input variant, so both are about widening what the new tests
actually cover.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4319132201
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -3198,6 +3203,29 @@ class SparkConnectPlanner(
}
}
+ /**
+ * Walks the relation tree and applies [[PipelinesHandler.blockUnsupportedSqlCommand]] to every
+ * component that would be executed by [[executeSQLWithRefs]] or [[executeSQL]] at the normal
+ * path. For a valid SQL-with-references [[proto.WithRelations]], the root SQL is parsed
+ * (without execution) and each reference's input is inspected recursively so nested
+ * [[proto.WithRelations]] cannot re-trigger the executing transformation chain. For other
+ * relation types, [[transformRelation]] produces an unresolved plan without side effects and is
Review Comment:
Thanks, fixed in d4ed446e83. The comments now cover only the top-level and direct-reference cases, and the scaladoc notes the wrapped case as a known limitation. I updated the PR description too.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6277612500
@rdtr Could you address @HyukjinKwon's two P3 comments when you get a chance? Both look correct
to me:
1. `SparkConnectPlanner.scala` (the `case _` fallback): `transformRelation` recurses into child
relations, so a valid SQL-with-references WITH_RELATIONS wrapped under another node (e.g.
`Limit(WithRelations{root: CREATE TABLE ...})`) still runs inside the guard, going through
`transformLimit`, then `transformWithRelations`, then `executeSQLWithRefs`. The "side-effect
free" / "nested WITH_RELATIONS cannot re-trigger" wording promises more than the two match arms
deliver; narrowing it to what the arms cover (top-level or direct reference input) and noting the
wrapped case as a known limitation would be accurate.
2. The three new tests never reach the recursive reference arm: `withRelationsRelation` always
builds reference inputs with `sqlRelation(...)`, so replacing the recursion with a non-recursive
`transformRelation` check would keep them green. A reference whose input is itself a
WITH_RELATIONS with a `CREATE TABLE` root would pin that arm.
I also left one related inline nit on the test helper (flow-function coverage) that fits naturally
with HyukjinKwon's helper-variant suggestion.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6141234106
Also, a few small things on the description before merge, since it lands in git log as
the commit message:
1. Accuracy: the first "Why are the changes needed?" bullet says the blocklist "checks for `Command` subclasses (`CreateTableAsSelect`, `InsertIntoStatement`, etc.)". Those two examples are actually the plans that are NOT `Command` subclasses: `CreateTableAsSelect extends V2CreateTableAsSelectPlan` and `InsertIntoStatement extends UnaryParsedStatement`, and `blockUnsupportedSqlCommand` checks them separately for exactly that reason (its own comment reads "Following commands are not subclasses of `Command` but have side effects"). Suggest rewording to distinguish the two, for example: "The blocklist rejects any `Command`, plus a few side-effecting plans that are not `Command` subclasses (`CreateTableAsSelect`, `InsertIntoStatement`, etc.)".
2. Minor mechanism accuracy (optional): the same bullet says commands inside reference `SubqueryAlias` inputs execute "because `eagerlyExecuteCommands` walks the resolved plan tree". For a reference input the command actually runs because `executeSQLWithRefs` eagerly executes each reference as its own query via `Dataset.ofRows(session, transformRelation(ref.input))` to register the temp view. `eagerlyExecuteCommands` is the tree-walk mechanism for a command nested within a single query plan, which is a different path. The conclusion (reference commands run as a side effect) is correct; only the stated mechanism is slightly off.
3. Template headings: the section headings use `##` but the Spark PR template uses `###` (`### What changes were proposed in this pull request?`, etc.). Minor, but worth matching the template.
4. Could you drop the opening line "This replaces #56241, which was closed as stale without a review. It is rebased on the current master with no other changes."? That context is useful in the PR conversation, but it becomes noise in the permanent commit message. The PR link history already records the lineage.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6141425769
The linter failure seems relevant to this change.
https://github.com/rdtr/spark/actions/runs/38358754437/job/115168705636
Could you fix it?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "rdtr (via GitHub)" <gi...@apache.org>.
rdtr commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6152067810
@sarutak Thank you for the detailed review!
- Lint: fixed in 3921af28e9. scalafmt only rewrapped one doc comment in `SparkConnectPlanner`. `dev/lint-scala` passes locally.
- Description: I updated all four points. The blocklist sentence now says it rejects any `Command` plus a few side-effecting plans that are not `Command` subclasses, the reference case now points to `executeSQLWithRefs` running each reference with `Dataset.ofRows`, the headings use `###`, and the opening line about #56241 is removed.
- The top-node limitation: agreed that it is out of scope here. I'd like to fix it as a follow-up after this PR, by checking the whole parsed plan. I'll open a JIRA for it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59425:
URL: https://github.com/apache/spark/pull/59425#discussion_r4270893370
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala:
##########
@@ -1164,4 +1164,127 @@ class SparkDeclarativePipelinesServerSuite
relation.getCommand.getSqlCommand.getInput)
}
}
+
+ // Helper to build a SqlCommand ExecutePlanRequest with PipelineAnalysisContext.
+ private def buildSqlCommandRequestWithPipelineContext(
+ graphId: String,
+ input: proto.Relation): proto.ExecutePlanRequest = {
+ val context = proto.PipelineAnalysisContext
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .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 = {
+ val withRelationsBuilder = proto.WithRelations.newBuilder().setRoot(sqlRelation(rootSql))
+ refs.foreach { case (alias, refSql) =>
+ withRelationsBuilder.addReferences(
+ proto.Relation
+ .newBuilder()
+ .setSubqueryAlias(
+ proto.SubqueryAlias
+ .newBuilder()
+ .setAlias(alias)
+ .setInput(sqlRelation(refSql))
+ .build())
+ .build())
+ }
+ proto.Relation.newBuilder().setWithRelations(withRelationsBuilder.build()).build()
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command at root of WITH_RELATIONS " +
+ "without executing it") {
+ 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,
+ 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")
+ }
+ }
+ }
+
+ test(
+ "SPARK-57189: blockUnsupportedSqlCommand catches Command inside reference of " +
Review Comment:
**Nit (P3):** None of the new tests reaches the recursive arm for references. `withRelationsRelation` always builds the reference input with `sqlRelation(...)`, so this test goes through `case _` and `transformSql`. If the recursion were replaced with `PipelinesHandler.blockUnsupportedSqlCommand(transformRelation(ref.getSubqueryAlias.getInput))`, all three tests would still pass. A reference whose input is itself a valid SQL-with-references `WITH_RELATIONS` with a `CREATE TABLE` root would then run inside the guard again.
That shape is reachable from PySpark: a DataFrame returned by `spark.sql("...", df=...)` carries its `WITH_RELATIONS` relation, and it becomes a reference input when you pass it to another `spark.sql` call. Could you add a case where `ref0`'s input is a `WITH_RELATIONS` with root `CREATE TABLE $tableName ...` and one plain reference, and assert the same error and `!tableExists`? The helper would need a variant that takes a `proto.Relation` for the reference input.
**Verification:**
- **Regression:** With the reference recursion replaced by a non-recursive transformRelation check, the nested CREATE TABLE executes in the guard and the new test fails on its error or table-existence assertion.
- **Behavior:** With the PR's recursive walker, the new test passes, and the existing SPARK-57189 tests still pass.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -3198,6 +3203,29 @@ class SparkConnectPlanner(
}
}
+ /**
+ * Walks the relation tree and applies [[PipelinesHandler.blockUnsupportedSqlCommand]] to every
+ * component that would be executed by [[executeSQLWithRefs]] or [[executeSQL]] at the normal
+ * path. For a valid SQL-with-references [[proto.WithRelations]], the root SQL is parsed
+ * (without execution) and each reference's input is inspected recursively so nested
+ * [[proto.WithRelations]] cannot re-trigger the executing transformation chain. For other
+ * relation types, [[transformRelation]] produces an unresolved plan without side effects and is
Review Comment:
**Nit (P3):** This says the `case _` branch is side-effect free, but `transformRelation(rel)` still recurses into child relations. So a valid SQL-with-references `WithRelations` nested under any other node still gets executed here. For example, a reference whose input is `Limit(WithRelations{root: SQL "CREATE TABLE t ...", refs: [...]})` goes `transformLimit` -> `transformWithRelations` -> `transformSqlWithRefs` -> `executeSQLWithRefs`. The CREATE TABLE runs inside the guard, the blocklist only sees the `Limit`, and the normal path runs it again. Dataset-building wrappers such as `SHOW_STRING` behave the same way.
The guard behaved this way before this PR too, so it is not a regression. But this sentence, the "nested [[proto.WithRelations]] cannot re-trigger the executing transformation chain" sentence above it, and the new call-site comment in `handleSqlCommand` now promise more than the walker delivers. It also matters for the planned follow-up: checking the whole parsed plan would not catch this, because the execution happens before there is a plan to inspect.
Could you narrow the wording to what the two arms actually cover (a valid SQL-with-references `WithRelations` at the top level or as a direct reference input) and note the wrapped case as a known limitation?
**Verification:**
- **Inspection:** The revised comments make no unconditional side-effect-free or no-re-trigger claim for the fallback branch, and they describe coverage consistently with the code's two match arms.
- **Compile:** The connect server module still compiles, and the Scala style checks accept the edited comments.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57189][CONNECT][SDP] Fix double execution and guard bypass in handleSqlCommand for WITH_RELATIONS [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #59425:
URL: https://github.com/apache/spark/pull/59425#issuecomment-6341772152
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/aa0b584e1bf552d73bb0673834338810fe769884
- merged into branch-4.x https://github.com/apache/spark/commit/4af3371cd59a0794e371384460729b1cc2e6ea69
- merged into branch-4.3 https://github.com/apache/spark/commit/a9a1bc7a5e8df60fe4faf993d45825debb20b032
- merged into branch-4.2 https://github.com/apache/spark/commit/5258c7e1f9467fdf35737810dd791b7d6745cd93
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org