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 83ad97fdc4faf..eddcee169e377 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 906bb260b0ed0..785ca2f6f5978 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))
 
   /*
     =======================================================================================