This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 2608b6bc1517 [SPARK-57402][SQL][FOLLOWUP] Allow pipeline definitions
in batch operation checks
2608b6bc1517 is described below
commit 2608b6bc151764168119213a5fc56594f6604f27
Author: Wenchen Fan <[email protected]>
AuthorDate: Wed Jul 22 20:19:14 2026 +0800
[SPARK-57402][SQL][FOLLOWUP] Allow pipeline definitions in batch operation
checks
### What changes were proposed in this pull request?
Followup to https://github.com/apache/spark/pull/57176.
Exclude streaming-table and flow definition plans from the generic batch
streaming-source check.
Add Catalyst tests covering all three definition forms and document why the
check applies only to
root pipeline commands while excluding materialized views.
### Why are the changes needed?
`UnsupportedOperationChecker` says streaming sources are allowed within
Spark Declarative Pipeline definitions, but currently rejects those definition
plans before the pipelines subsystem can interpret them.
### Does this PR introduce _any_ user-facing change?
No. This fixes validation consistency for pipeline-definition logical plans
introduced on the unreleased master branch.
### How was this patch tested?
Ran before the latest review follow-up:
`./build/sbt 'catalyst/testOnly
org.apache.spark.sql.catalyst.analysis.UnsupportedOperationsSuite'`
All 218 tests passed. The latest follow-up adds the third AUTO CDC
regression case; the suite has
not been rerun locally since that addition.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
Closes #57345 from cloud-fan/SPARK-57402-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../analysis/UnsupportedOperationChecker.scala | 12 +++++++
.../analysis/UnsupportedOperationsSuite.scala | 39 ++++++++++++++++++++++
2 files changed, 51 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
index 83ad97fdc4fa..eddcee169e37 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
@@ -37,7 +37,19 @@ import org.apache.spark.sql.streaming.{GroupStateTimeout,
OutputMode}
*/
object UnsupportedOperationChecker extends Logging {
+ // Pipeline definition commands are always root plans and must reach
SparkStrategies.Pipelines
+ // for their dedicated errors. Materialized views are intentionally excluded
because their
+ // queries are batch queries and must continue to reject streaming sources
here.
+ private def isPipelineDefinition(plan: LogicalPlan): Boolean = plan match {
+ case _: CreateStreamingTableAsSelect | _: CreateStreamingTableAutoCdc |
+ _: CreateFlowCommand => true
+ case _ => false
+ }
+
def checkForBatch(plan: LogicalPlan): Unit = {
+ if (isPipelineDefinition(plan)) {
+ return
+ }
plan.foreachUp {
case p if p.isStreaming =>
throwError("Queries with streaming sources must be executed with
writeStream.start(), or " +
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala
index 906bb260b0ed..785ca2f6f597 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala
@@ -65,6 +65,45 @@ class UnsupportedOperationsSuite extends SparkFunSuite with
SQLHelper {
streamRelation.select($"`count(*)`"),
Seq("with streaming source", "start"))
+ private val emptyTableSpec = TableSpec(
+ properties = Map.empty,
+ provider = None,
+ options = Map.empty,
+ location = None,
+ comment = None,
+ collation = None,
+ serde = None,
+ external = false)
+
+ assertSupportedInBatchPlan(
+ "streaming table definition with streaming source",
+ CreateStreamingTableAsSelect(
+ name = batchRelation,
+ columns = Nil,
+ partitioning = Nil,
+ tableSpec = emptyTableSpec,
+ query = streamRelation,
+ originalText = "",
+ ifNotExists = false))
+
+ assertSupportedInBatchPlan(
+ "flow definition with streaming source",
+ CreateFlowCommand(batchRelation, streamRelation, comment = None))
+
+ assertSupportedInBatchPlan(
+ "AUTO CDC streaming table definition with streaming source",
+ CreateStreamingTableAutoCdc(
+ name = batchRelation,
+ columns = Nil,
+ partitioning = Nil,
+ tableSpec = emptyTableSpec,
+ ifNotExists = false,
+ source = streamRelation,
+ keys = Nil,
+ deleteCondition = None,
+ sequenceByExpr = attribute,
+ includeColumns = None,
+ excludeColumns = None))
/*
=======================================================================================
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
