[SPARK-52759][SDP][SQL] Throw exception if pipeline has no tables or persisted views

common/utils/src/main/resources/error/error-conditions.json

@@ -4930,6 +4930,13 @@
     ],
     "sqlState" : "22023"
   },
+  "RUN_EMPTY_PIPELINE" : {
+    "message" : [
+      "Pipelines are expected to have at least one non-temporary dataset defined (tables, persisted views) but no non-temporary datasets were found in your pipeline.",
+      "Please verify that you have included the expected source files, and that your source code includes table definitions (e.g., CREATE MATERIALIZED VIEW in SQL code, @sdp.table in python code)."
+    ],
+    "sqlState" : "42617"
+  },
   "SCALAR_FUNCTION_NOT_COMPATIBLE" : {
     "message" : [
       "ScalarFunction <scalarFunc> not overrides method 'produceResult(InternalRow)' with custom implementation."

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala

@@ -339,8 +339,12 @@ class PythonPipelineSuite
         TableIdentifier("st", Some("some_schema"), Some("some_catalog"))))
   }
 
-  test("view works") {
+  test("temporary views works") {
+    // A table is defined since pipeline with only temporary views is invalid.
     val graph = buildGraph(s"""
+         |@sdp.table
+         |def mv_1():
+         |  return spark.range(5)
          |@sdp.temporary_view
          |def view_1():
          |  return spark.range(5)
@@ -354,9 +358,9 @@ class PythonPipelineSuite
          |  return spark.read.table("view_1")
          |""".stripMargin).resolve()
     // views are temporary views, so they're not fully qualified.
-    assert(graph.tables.isEmpty)
     assert(
-      graph.flows.map(_.identifier.unquotedString).toSet == Set("view_1", "view_2", "view_3"))
+      Set("view_1", "view_2", "view_3").subsetOf(
+        graph.flows.map(_.identifier.unquotedString).toSet))
     // dependencies are correctly resolved view_2 reading from view_1
     assert(
       graph.resolvedFlow(TableIdentifier("view_2")).inputs.contains(TableIdentifier("view_1")))
@@ -416,6 +420,43 @@ class PythonPipelineSuite
         .map(_.identifier) == Seq(graphIdentifier("a"), graphIdentifier("something")))
   }
 
+  test("create pipeline without table will throw RUN_EMPTY_PIPELINE exception") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        buildGraph(s"""
+            |spark.range(1)
+            |""".stripMargin)
+      },
+      condition = "RUN_EMPTY_PIPELINE",
+      parameters = Map.empty)
+  }
+
+  test("create pipeline with only temp view will throw RUN_EMPTY_PIPELINE exception") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        buildGraph(s"""
+            |@sdp.temporary_view
+            |def view_1():
+            |  return spark.range(5)
+            |""".stripMargin)
+      },
+      condition = "RUN_EMPTY_PIPELINE",
+      parameters = Map.empty)
+  }
+
+  test("create pipeline with only flow will throw RUN_EMPTY_PIPELINE exception") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        buildGraph(s"""
+            |@sdp.append_flow(target = "a")
+            |def flow():
+            |  return spark.range(5)
+            |""".stripMargin)
+      },
+      condition = "RUN_EMPTY_PIPELINE",
+      parameters = Map.empty)
+  }
+
   /**
    * Executes Python code in a separate process and returns the exit code.
    *

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphRegistrationContext.scala

@@ -50,6 +50,13 @@ class GraphRegistrationContext(
   }
 
   def toDataflowGraph: DataflowGraph = {
+    if (tables.isEmpty && views.collect { case v: PersistedView =>
+        v
+      }.isEmpty) {
+      throw new AnalysisException(
+        errorClass = "RUN_EMPTY_PIPELINE",
+        messageParameters = Map.empty)
+    }
     val qualifiedTables = tables.toSeq.map { t =>
       t.copy(
         identifier = GraphIdentifierManager