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

### What changes were proposed in this pull request?

When user runs a pipeline, throw a `RUN_EMPTY_PIPELINE` exception if the pipeline source directory does not contain any tables or persisted views.

* Add checks in `GraphRegistrationContext. toDataflowGraph` to throw the exception if the pipeline does not have any tables or persisted views
* Modify test cases that currently register only temporary view to register persisted views instead since pipelines that only include temporary views are invalid now.
* Add additional test cases to ensure the exception is thrown correctly.

### Why are the changes needed?

In Spark Declarative Pipelines, using the CLI tool users run a pipeline that is defined from the configured pipeline root directory. This directory contains information such as the pipeline spec, and the source code files (Python, SQL) that define the pipeline tables/flows/views.

It’s possible the user tries to run a pipeline defined from a pipeline directory whose source files don’t actually define any tables or views.

An exception should be thrown if the pipeline does not have any tables or views, to inform the user they should double check that they are running the pipeline in the correct directory. The previous behavior is that the pipeline just run to completion without emitting any info.

### Does this PR introduce _any_ user-facing change?

Yes, this is an additive non-breaking behavior change. However, SDP has not been released, so no user should be impacted by this change.

### How was this patch tested?

Created additional test case to verify that the exception is indeed thrown.

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #51445 from JiaqiWang18/SPARK-52759-empty-pipeline-fails.

Lead-authored-by: Jacky Wang <[email protected]>
Co-authored-by: Jacky Wang <[email protected]>
Signed-off-by: Sandy Ryza <[email protected]>

Author: 2 people

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