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 00:16:41 UTC
(spark) branch branch-4.3 updated: [SPARK-60481][CONNECT][SDP] Block `INSERT OVERWRITE DIRECTORY` in `spark.sql` of SDP
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 34120b1b8597 [SPARK-60481][CONNECT][SDP] Block `INSERT OVERWRITE DIRECTORY` in `spark.sql` of SDP
34120b1b8597 is described below
commit 34120b1b8597b9df6d3826d0776c217f6b47830f
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Oct 22 17:15:58 2026 -0700
[SPARK-60481][CONNECT][SDP] Block `INSERT OVERWRITE DIRECTORY` in `spark.sql` of SDP
### What changes were proposed in this pull request?
This PR adds `InsertIntoDir` to the side-effecting plans that `PipelinesHandler.blockUnsupportedSqlCommand` rejects. As a result, `INSERT OVERWRITE [LOCAL] DIRECTORY` in `spark.sql(...)` of a Spark Declarative Pipelines (SDP) definition fails with `UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND`, like `INSERT INTO` and `INSERT OVERWRITE TABLE`.
Like the existing entries, the new entry is matched against 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 SDP guard rejects any `Command`, plus a few side-effecting plans that are not `Command` subclasses, such as `InsertIntoStatement` and `UnresolvedInsert`.
`INSERT OVERWRITE [LOCAL] DIRECTORY ... SELECT ...` is parsed into `InsertIntoDir`. It is a `UnaryNode`, not a `Command`, and it is missing from the list, so the guard lets it pass. During analysis, it is converted into a command such as `InsertIntoDataSourceDirCommand` or `InsertIntoHiveDirCommand`, which runs eagerly and overwrites the directory.
For example, the following line in a pipeline definition file overwrites the directory while the pipeline is being registered.
```python
spark.sql("INSERT OVERWRITE DIRECTORY '/path/to/dir' USING parquet SELECT 1")
```
Inside a flow function, the guard lets the same statement pass as well.
### Does this PR introduce _any_ user-facing change?
Yes. `INSERT OVERWRITE [LOCAL] DIRECTORY` in `spark.sql(...)` of SDP now fails with the following error instead of overwriting the directory. 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).
```
[UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND] 'InsertIntoDir' is not supported in spark.sql("...") API in Spark Declarative Pipeline. SQLSTATE: 0A000
```
### How was this patch tested?
Pass the CIs with the newly added test case in `SparkDeclarativePipelinesServerSuite`. Like the SPARK-57189 tests, it runs both outside and inside a pipeline flow function. It checks that `INSERT OVERWRITE DIRECTORY` into a temporary directory fails with `UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND` and that nothing is written into the directory.
Without the fix, both variants fail because no exception is thrown. Outside a flow function, the statement is executed and the write job is committed into the temporary directory.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59719 from dongjoon-hyun/SPARK-60481.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 15659c5fea0a35ab865ece5a0579bcbaa8ebf5b4)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../sql/connect/pipelines/PipelinesHandler.scala | 3 ++-
.../SparkDeclarativePipelinesServerSuite.scala | 23 ++++++++++++++++++++++
2 files changed, 25 insertions(+), 1 deletion(-)
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 718d1670bf8c..56029ab33f64 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
@@ -31,7 +31,7 @@ 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.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.{Command, CreateNamespace, CreateTable, CreateTableAsSelect, CreateView, DescribeRelation, DescribeTablePartition, DropView, 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, 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
@@ -174,6 +174,7 @@ private[connect] object PipelinesHandler extends Logging {
queryPlan.isInstanceOf[CreateView] ||
queryPlan.isInstanceOf[InsertIntoStatement] ||
queryPlan.isInstanceOf[UnresolvedInsert] ||
+ queryPlan.isInstanceOf[InsertIntoDir] ||
queryPlan.isInstanceOf[RenameTable] ||
queryPlan.isInstanceOf[CreateNamespace] ||
queryPlan.isInstanceOf[DropView]
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 d95d683cc57c..1a4d469bf8f1 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
@@ -1218,4 +1218,27 @@ class SparkDeclarativePipelinesServerSuite
assert(response.hasSqlCommandResult)
}
}
+
+ namedGridTest(
+ "SPARK-60481: blockUnsupportedSqlCommand catches INSERT OVERWRITE DIRECTORY " +
+ "without executing it")(
+ Map("outside a flow function" -> None, "inside a flow function" -> Some("flow1"))) {
+ flowName =>
+ withRawBlockingStub { implicit stub =>
+ withTempDir { dir =>
+ val graphId = createDataflowGraph
+ val request = buildSqlCommandRequestWithPipelineContext(
+ graphId,
+ flowName = flowName,
+ input = sqlRelation(
+ s"INSERT OVERWRITE DIRECTORY '${dir.toURI}' USING parquet SELECT 1 AS id"))
+ val ex = intercept[RuntimeException] {
+ stub.executePlan(request).next()
+ }
+ assert(ex.getMessage.contains("UNSUPPORTED_PIPELINE_SPARK_SQL_COMMAND"))
+ // Verify the directory was not overwritten as a side effect.
+ assert(dir.listFiles().isEmpty)
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org