You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/23 12:22:23 UTC
(spark) branch branch-4.2 updated: [SPARK-60495][CONNECT][SDP] Block `EXECUTE IMMEDIATE`, `CALL`, and `IDENTIFIER` clause commands in the SDP SQL guard
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 22fde39f8ec5 [SPARK-60495][CONNECT][SDP] Block `EXECUTE IMMEDIATE`, `CALL`, and `IDENTIFIER` clause commands in the SDP SQL guard
22fde39f8ec5 is described below
commit 22fde39f8ec5fcfd929c1b57291a0573b494652b
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Fri Oct 23 05:21:31 2026 -0700
[SPARK-60495][CONNECT][SDP] Block `EXECUTE IMMEDIATE`, `CALL`, and `IDENTIFIER` clause commands in the SDP SQL guard
### What changes were proposed in this pull request?
This PR makes `PipelinesHandler.blockUnsupportedSqlCommand`, the guard for `spark.sql(...)` in Spark Declarative Pipelines (SDP), reject three kinds of statements. The parser does not produce a `Command` for them, but they run a command or a procedure once they are analyzed.
| Statement | Top node of the parsed plan | What happens during analysis |
|---|---|---|
| `EXECUTE IMMEDIATE '<sql>'` | `UnresolvedExecuteImmediate` | `ResolveExecuteImmediate` parses and analyzes `<sql>`, and a command payload then runs as `ExecuteImmediateCommand`. |
| `CALL cat.ns.proc(...)` | `Call` (an `ExecutableDuringAnalysis`) | `InvokeProcedures` runs the procedure. |
| A statement named by an `IDENTIFIER` clause that the parser does not build yet, e.g. `USE SCHEMA IDENTIFIER('a' \|\| 'b')`, `CREATE TABLE IDENTIFIER('a' \|\| 'b') (id INT)`, or `CREATE TEMPORARY VIEW IDENTIFIER('v') AS SELECT 1` | `PlanWithUnresolvedIdentifier` | `ResolveIdentifierClause` evaluates the clause and builds the actual command. |
The guard now does the following.
- It blocks `UnresolvedExecuteImmediate` and `ExecutableDuringAnalysis`, the trait that `Call` implements.
- For a top-level `PlanWithUnresolvedIdentifier`, it builds the statement through the node's `planBuilder` with a placeholder name and checks that statement with the same allowlist and blocklist. As a result, `USE SCHEMA IDENTIFIER('a' || 'b')` is rejected as `SetNamespaceCommand`, like `USE SCHEMA IDENTIFIER('ab')`, while read-only forms such as `SHOW COLUMNS FROM IDENTIFIER(...)` and `TABLE IDENTIFIER(...)` are still allowed.
Because the change is in `blockUnsupportedSqlCommand`, it covers both a plain SQL relation and the root SQL of a SQL-with-references `WITH_RELATIONS` (SPARK-57189).
Like the existing checks, the new checks are applied to the top node of the parsed plan. Nested cases such as `WITH ... INSERT ...` are out of scope here. They are the top-node limitation discussed in https://github.com/apache/spark/pull/59425.
### Why are the changes needed?
The guard checks only the class of the top node of the parsed plan. The statements above pass that check, and `session.sql` then analyzes them, which runs the command or the procedure. For example, the following line in a pipeline definition file creates a table while the pipeline is being registered, although `spark.sql("CREATE TABLE ...")` is rejected.
```python
spark.sql("EXECUTE IMMEDIATE 'CREATE TABLE spark_catalog.default.t (id INT)'")
```
Similarly, `spark.sql("USE SCHEMA IDENTIFIER('some_' || 'schema')")` runs `SetNamespaceCommand` although `USE SCHEMA some_schema` is rejected, and `CALL` runs a procedure of a configured `ProcedureCatalog`. Inside a flow function, the guard lets these statements pass as well.
### Does this PR introduce _any_ user-facing change?
Yes. In an SDP pipeline definition, `spark.sql(...)` with `EXECUTE IMMEDIATE`, `CALL`, or a command named by such an `IDENTIFIER` clause now fails with errors like the following instead of running.
```
[UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND] 'UnresolvedExecuteImmediate' is not supported in spark.sql("...") API in Spark Declarative Pipeline. SQLSTATE: 0A000
[UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND] 'Call' is not supported in spark.sql("...") API in Spark Declarative Pipeline. SQLSTATE: 0A000
[UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND] 'SetNamespaceCommand' is not supported in spark.sql("...") API in Spark Declarative Pipeline. SQLSTATE: 0A000
```
This includes `EXECUTE IMMEDIATE` of a query such as `EXECUTE IMMEDIATE 'SELECT 1'`, because its SQL string is parsed only during analysis. This is a bug fix for the SDP SQL guard, which has existed since Apache Spark [4.1.0](https://github.com/apache/spark/releases/tag/v4.1.0) (2025-12-11).
### How was this patch tested?
Pass the CIs with the newly added test cases in `PythonPipelineSuite`.
- `unsupportedSqlCommandList`: `EXECUTE IMMEDIATE`, `CALL`, `USE SCHEMA` with a non-literal `IDENTIFIER` clause, and `CREATE TEMPORARY VIEW` with a literal `IDENTIFIER` clause. Each is checked outside and inside a query function. Without the fix, these 8 test cases fail because the guard does not reject the statements. For example, outside a query function, `EXECUTE IMMEDIATE` creates the table, and `USE SCHEMA IDENTIFIER(...)` runs `SetNamespaceCommand` and fails with `SCHEMA_NOT_FOUND`.
- `supportedSqlCommandList`: `SHOW COLUMNS FROM IDENTIFIER(...)` and `TABLE IDENTIFIER(...)` with a non-literal `IDENTIFIER` clause, to make sure that the new check does not block them.
```
$ build/sbt 'connect/testOnly org.apache.spark.sql.connect.pipelines.PythonPipelineSuite -- -z "SQL command"'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59736 from dongjoon-hyun/SPARK-60495.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit c09a4ad5bde1ab1b34b9f059866e1bdef36e605d)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/sql/connect/pipelines/PipelinesHandler.scala | 16 ++++++++++++++--
.../sql/connect/pipelines/PythonPipelineSuite.scala | 8 +++++++-
2 files changed, 21 insertions(+), 3 deletions(-)
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
index 671f9552628f..776204cd76f0 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
@@ -29,9 +29,9 @@ import org.apache.spark.connect.proto.PipelineCommand.DefineFlow.AutoCdcFlowDeta
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{AnalysisException, Column}
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{PlanWithUnresolvedIdentifier, UnresolvedAttribute, UnresolvedExecuteImmediate, UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.{Command, CreateNamespace, CreateTable, CreateTableAsSelect, CreateView, DescribeRelation, DescribeTablePartition, DropView, InsertIntoDir, InsertIntoStatement, LogicalPlan, RenameTable, ShowColumns, ShowCreateTable, ShowFunctions, ShowTableProperties, ShowTables, ShowViews, UnresolvedInsert}
+import org.apache.spark.sql.catalyst.plans.logical.{Command, CreateNamespace, CreateTable, CreateTableAsSelect, CreateView, DescribeRelation, DescribeTablePartition, DropView, ExecutableDuringAnalysis, InsertIntoDir, InsertIntoStatement, LogicalPlan, RenameTable, ShowColumns, ShowCreateTable, ShowFunctions, ShowTableProperties, ShowTables, ShowViews, UnresolvedInsert}
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.connect.common.DataTypeProtoConverter
import org.apache.spark.sql.connect.service.SessionHolder
@@ -150,6 +150,14 @@ private[connect] object PipelinesHandler extends Logging {
* set of read-only operations (e.g., SHOW, DESCRIBE).
*/
def blockUnsupportedSqlCommand(queryPlan: LogicalPlan): Unit = {
+ queryPlan match {
+ // The parser can leave a statement named by an IDENTIFIER clause, such as
+ // IDENTIFIER('a' || 'b'), unbuilt until analysis evaluates the clause. Build the statement
+ // with a placeholder name and check it instead.
+ case p: PlanWithUnresolvedIdentifier =>
+ blockUnsupportedSqlCommand(p.planBuilder(Seq("placeholder"), p.children))
+ case _ =>
+ }
val allowlistedCommands = Set(
classOf[DescribeRelation],
classOf[DescribeTablePartition],
@@ -168,6 +176,10 @@ private[connect] object PipelinesHandler extends Logging {
// Disable all [[Command]] except the ones that are explicitly allowlisted
// in "allowlistedCommands".
queryPlan.isInstanceOf[Command] ||
+ // EXECUTE IMMEDIATE runs a SQL string that is parsed only during analysis, and CALL runs a
+ // procedure during analysis.
+ queryPlan.isInstanceOf[UnresolvedExecuteImmediate] ||
+ queryPlan.isInstanceOf[ExecutableDuringAnalysis] ||
// Following commands are not subclasses of [[Command]] but have side effects.
queryPlan.isInstanceOf[CreateTableAsSelect] ||
queryPlan.isInstanceOf[CreateTable] ||
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
index 834e2d8144e1..4c9d2f495766 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
@@ -1411,13 +1411,17 @@ class PythonPipelineSuite
private val unsupportedSqlCommandList: Seq[String] = Seq(
"SET CATALOG some_catalog",
"USE SCHEMA some_schema",
+ "USE SCHEMA IDENTIFIER('some_' || 'schema')",
"SET `test_conf` = `true`",
"CREATE TABLE some_table (id INT)",
+ "EXECUTE IMMEDIATE 'CREATE TABLE some_table (id INT)'",
"CREATE VIEW some_view AS SELECT * FROM some_table",
+ "CREATE TEMPORARY VIEW IDENTIFIER('some_view') AS SELECT 1",
"INSERT INTO some_table VALUES (1)",
"ALTER TABLE some_table RENAME TO some_new_table",
"CREATE NAMESPACE some_namespace",
"DROP VIEW some_view",
+ "CALL some_catalog.some_schema.some_procedure()",
"CREATE MATERIALIZED VIEW some_view AS SELECT * FROM some_table",
"CREATE STREAMING TABLE some_table AS SELECT * FROM some_table")
@@ -1455,12 +1459,14 @@ class PythonPipelineSuite
"SHOW TBLPROPERTIES spark_catalog.default.src",
"SHOW NAMESPACES",
"SHOW COLUMNS FROM spark_catalog.default.src",
+ "SHOW COLUMNS FROM IDENTIFIER('spark_catalog.default.' || 'src')",
"SHOW FUNCTIONS",
"SHOW VIEWS",
"SHOW CATALOGS",
"SHOW CREATE TABLE spark_catalog.default.src",
"SELECT * FROM RANGE(5)",
- "SELECT * FROM spark_catalog.default.src")
+ "SELECT * FROM spark_catalog.default.src",
+ "TABLE IDENTIFIER('spark_catalog.default.' || 'src')")
gridTest("Supported SQL command outside query function should work")(supportedSqlCommandList) {
supportedSqlCommand =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org