[SPARK-52283][SQL] Declarative Pipelines DataflowGraph creation and resolution

[aakash-db]

What changes were proposed in this pull request?

This PR introduces the DataflowGraph, a container for Declarative Pipelines datasets and flows, as described in the Declarative Pipelines SPIP. It also adds functionality for

• Constructing a graph by registering a set of graph elements in succession (GraphRegistrationContext)
• "Resolving" a graph, which means resolving each of the flows within a graph. Resolving a flow means:
  • Validating that its plan can be successfully analyzed
  • Determining the schema of the data it will produce
  • Determining what upstream datasets within the graph it depends on

It also introduces various secondary changes:

• Changes to SparkBuild to support declarative pipelines.
• Updates to the pom.xml for the module.
• New error conditions

Why are the changes needed?

In order to implement Declarative Pipelines.

Does this PR introduce any user-facing change?

No changes to existing behavior.

How was this patch tested?

New test suites:

• ConnectValidPipelineSuite – test cases where the graph can be successfully resolved
• ConnectInvalidPipelineSuite – test cases where the graph fails to be resolved

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

No

[jonmio]

flushing some comments

[jonmio]

What is the purpose of this conf and do we really need it?

[jonmio]

Nit: Document return. I'm especially curious why this is a Seq and when processNode would return more than one element

[aakash-db]

Right now, it's mostly just for flexibility - in case one node maps to several in the future.

[cloud-fan]

TableIdentifier only supports 3-level namespace. Shall we use Seq[String] to better support DS v2, which can have an arbitrary level of namespace?

[cloud-fan]

Seq[String] is a bit hard to use here. We can switch to the DS v2 API after we create an encapsulation class to match TableIdentifier.

[sryza]

Hey @gengliangwang @LuciferYang – I just rebased the PR on master and pushed an update that fixes those log messages.

[LuciferYang]

[LuciferYang]

Why do we need this plugin? The pipelines module doesn't seem to have any .proto files that need compilation.

[LuciferYang]

The parent pom.xml has already configured this plugin.

[LuciferYang]

If I understand correctly, the assembly jar built this way will not include the code related to the spark-pipelines module. Is this what we expect? And does the assembly jar for pipelines really need to include grpc-*.jar? These should all be configured on demand.

In addition, if the pipelines module uses the assembly plugin, then corresponding configurations for the maven-shade-plugin should be made in the pom.xml file, and it should be ensured that their packaging results are consistent.

[LuciferYang]

What's the difference from before? Can we not modify this line?

[gengliangwang]

LGTM. Please address LuciferYang's comment in this PR or in a follow-up

[sryza]

@LuciferYang thanks for catching these issues in the pom.xml and SparkBuild. @gengliangwang @LuciferYang I created a followup PR which implements the pom.xml and SparkBuild changes that @LuciferYang raised.

If addressing the build issues in a followup is OK, I'd love to get this PR back in ASAP because there are several downstream PRs that depend on it and are currently in review.

[LuciferYang]

LGTM. Please address LuciferYang's comment in this PR or in a follow-up

Please give me a moment to manually verify the Maven compilation for this pr to avoid any failures in the Maven daily test after merging it.

[LuciferYang]

locally run build/mvn clean install -DskipTests -pl sql/pipelines -am

Could you make maven build happy @sryza ?

[LuciferYang]

I changed

spark/sql/pipelines/pom.xml

Lines 39 to 44 in 5ef6d4b

	<dependency>
	<groupId>org.apache.spark</groupId>
	<artifactId>spark-core_${scala.binary.version}</artifactId>
	<version>${project.version}</version>
	<scope>test</scope>
	</dependency>

to

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.binary.version}</artifactId>
            <version>${project.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.binary.version}</artifactId>
            <version>${project.version}</version>
            <type>test-jar</type>
            <scope>test</scope>
        </dependency>

and the Maven compilation passed. However, when I executed command build/mvn test -pl sql/pipelines to run the Maven test, there were test failures:

ConnectValidPipelineSuite:
org.apache.spark.sql.pipelines.graph.ConnectValidPipelineSuite *** ABORTED ***
  java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
This stopped SparkContext was created at:

org.apache.spark.sql.pipelines.graph.ConnectInvalidPipelineSuite.<init>(ConnectInvalidPipelineSuite.scala:30)
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:500)
java.base/java.lang.reflect.ReflectAccess.newInstance(ReflectAccess.java:128)
java.base/jdk.internal.reflect.ReflectionFactory.newInstance(ReflectionFactory.java:347)
java.base/java.lang.Class.newInstance(Class.java:645)
org.scalatest.tools.DiscoverySuite$.getSuiteInstance(DiscoverySuite.scala:66)
org.scalatest.tools.DiscoverySuite.$anonfun$nestedSuites$1(DiscoverySuite.scala:38)
scala.collection.immutable.Vector1.map(Vector.scala:2155)
scala.collection.immutable.Vector1.map(Vector.scala:386)
org.scalatest.tools.DiscoverySuite.<init>(DiscoverySuite.scala:37)
org.scalatest.tools.Runner$.genDiscoSuites$1(Runner.scala:1131)
org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:1225)
org.scalatest.tools.Runner$.$anonfun$runOptionallyWithPassFailReporter$24(Runner.scala:992)
org.scalatest.tools.Runner$.$anonfun$runOptionallyWithPassFailReporter$24$adapted(Runner.scala:970)
org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:1481)
org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:970)
org.scalatest.tools.Runner$.main(Runner.scala:775)

And it was stopped at:

org.apache.spark.sql.pipelines.utils.PipelineTest.afterAll(PipelineTest.scala:136)

[sryza]

Investigating now

[sryza]

I looked into this more deeply. What appears to be happening is:

• The test framework creates an instance of ConnectInvalidPipelineSuite
  • The initializer invokes SparkSession.builder.getOrCreate() to create a SparkSession and accompanying SparkContext
• The test framework creates an instance of ConnectValidPipelineSuite
  • The initializer invokes SparkSession.builder.getOrCreate() to create a SparkSession. It is returned the same SparkSession and backing SparkContext that ConnectInvalidPipelineSuite references.
• The test framework runs all tests inside ConnectInvalidPipelineSuite
• The test framework invokes the teardown code in ConnectInvalidPipelineSuite, which stops its SparkSession and underlying SparkContext
• The test framework tries to run beforeEach code in ConnectValidPipelineSuite, but the SparkSession it references is the same one that was referenced in ConnectInvalidPipelineSuite, and that has already been stopped

How does this work normally? There are several suites inside the repo that getOrCreate a SparkSession at instantiation time and stop it afterAll. Do they not run into this problem?

[LuciferYang]

diff --git a/sql/pipelines/pom.xml b/sql/pipelines/pom.xml
index 4ab7db079e4..5cbcebfbc8f 100644
--- a/sql/pipelines/pom.xml
+++ b/sql/pipelines/pom.xml
@@ -41,6 +41,12 @@
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
             <scope>test</scope>
         </dependency>
         <dependency>
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
index 01b2a91bb93..0d74146851c 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
@@ -29,7 +29,6 @@ import org.apache.spark.sql.types.{IntegerType, StructType}
  */
 class ConnectInvalidPipelineSuite extends PipelineTest {
 
-  import originalSpark.implicits._
   test("Missing source") {
     class P extends TestGraphRegistrationContext(spark) {
       registerView("b", query = readFlowFunc("a"))
@@ -144,6 +143,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Missing attribute in the schema") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("z")))
       registerView("b", query = sqlFlowFunc(spark, "SELECT x FROM a"))
@@ -157,6 +158,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Joining on a column with different names") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("b", query = dfFlowFunc(Seq("a", "b", "c").toDF("y")))
@@ -175,6 +178,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Writing to one table by unioning flows with different schemas") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("b", query = dfFlowFunc(Seq(true, false).toDF("x")))
@@ -225,6 +230,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Cyclic graph") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("b", query = sqlFlowFunc(spark, "SELECT * FROM a UNION SELECT * FROM d"))
@@ -246,6 +253,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Cyclic graph with materialized nodes") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerTable("a", query = Option(dfFlowFunc(Seq(1, 2, 3).toDF("x"))))
       registerTable(
@@ -270,6 +279,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Cyclic graph - second query makes it cyclic") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerTable("a", query = Option(dfFlowFunc(Seq(1, 2, 3).toDF("x"))))
       registerTable("b")
@@ -293,6 +304,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Cyclic graph - all named queries") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerTable("a", query = Option(dfFlowFunc(Seq(1, 2, 3).toDF("x"))))
       registerTable("b")
@@ -317,6 +330,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("view-table conf conflict") {
+    val session = spark
+    import session.implicits._
     val p = new TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1).toDF()), sqlConf = Map("x" -> "a-val"))
       registerTable("b", query = Option(readFlowFunc("a")), sqlConf = Map("x" -> "b-val"))
@@ -338,6 +353,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("view-view conf conflict") {
+    val session = spark
+    import session.implicits._
     val p = new TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1).toDF()), sqlConf = Map("x" -> "a-val"))
       registerView("b", query = dfFlowFunc(Seq(1).toDF()), sqlConf = Map("x" -> "b-val"))
@@ -364,6 +381,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("reading a complete view incrementally") {
+    val session = spark
+    import session.implicits._
     val p = new TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1).toDF()))
       registerTable("b", query = Option(readStreamFlowFunc("a")))
@@ -380,6 +399,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("reading an incremental view completely") {
+    val session = spark
+    import session.implicits._
     val p = new TestGraphRegistrationContext(spark) {
       val mem = MemoryStream[Int]
       mem.addData(1)
@@ -398,6 +419,8 @@ class ConnectInvalidPipelineSuite extends PipelineTest {
   }
 
   test("Inferred schema that isn't a subset of user-specified schema") {
+    val session = spark
+    import session.implicits._
     val graph1 = new TestGraphRegistrationContext(spark) {
       registerTable(
         "a",
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
index f8b5133ff16..d64c5f06cdf 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
@@ -32,9 +32,9 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
  */
 class ConnectValidPipelineSuite extends PipelineTest {
 
-  import originalSpark.implicits._
-
   test("Extra simple") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("b", query = dfFlowFunc(Seq(1, 2, 3).toDF("y")))
     }
@@ -44,6 +44,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Simple") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("b", query = sqlFlowFunc(spark, "SELECT x as y FROM a"))
@@ -65,6 +67,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Dependencies") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("c", query = sqlFlowFunc(spark, "SELECT y as z FROM b"))
@@ -78,6 +82,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Multi-hop schema merging") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView(
         "b",
@@ -93,6 +99,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Cross product join merges schema") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("b", query = dfFlowFunc(Seq(4, 5, 6).toDF("y")))
@@ -111,6 +119,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Real join merges schema") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq((1, "a"), (2, "b"), (3, "c")).toDF("x", "y")))
       registerView("b", query = dfFlowFunc(Seq((2, "m"), (3, "n"), (4, "o")).toDF("x", "z")))
@@ -132,6 +142,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Union of streaming and batch Dataframes") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       val ints = MemoryStream[Int]
       ints.addData(1, 2, 3, 4)
@@ -170,6 +182,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Union of two streaming Dataframes") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       val ints1 = MemoryStream[Int]
       ints1.addData(1, 2, 3, 4)
@@ -214,6 +228,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("MultipleInputs") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerView("b", query = dfFlowFunc(Seq(4, 5, 6).toDF("y")))
@@ -228,6 +244,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Connect retains and fuses confs") {
+    val session = spark
+    import session.implicits._
     // a -> b \
     //          d
     //      c /
@@ -250,6 +268,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Confs aren't fused past materialization points") {
+    val session = spark
+    import session.implicits._
     val p = new TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1).toDF("x")), Map("a" -> "a-val"))
       registerTable("b", query = Option(readFlowFunc("a")), Map("b" -> "b-val"))
@@ -276,6 +296,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Setting the same conf with the same value is totally cool") {
+    val session = spark
+    import session.implicits._
     val p = new TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")), Map("key" -> "val"))
       registerView("b", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")), Map("key" -> "val"))
@@ -290,6 +312,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Named query only") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       registerView("a", query = dfFlowFunc(Seq(1, 2, 3).toDF("x")))
       registerTable("b")
@@ -308,6 +332,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Default query and named query") {
+    val session = spark
+    import session.implicits._
     class P extends TestGraphRegistrationContext(spark) {
       val mem = MemoryStream[Int]
       registerView("a", query = dfFlowFunc(mem.toDF()))
@@ -348,6 +374,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Correct types of flows after connection") {
+    val session = spark
+    import session.implicits._
     val graph = new TestGraphRegistrationContext(spark) {
       val mem = MemoryStream[Int]
       mem.addData(1, 2)
@@ -402,6 +430,8 @@ class ConnectValidPipelineSuite extends PipelineTest {
   }
 
   test("Pipeline level default spark confs are applied with correct precedence") {
+    val session = spark
+    import session.implicits._
     val P = new TestGraphRegistrationContext(
       spark,
       Map("default.conf" -> "value")
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
index 981fa3cdcae..9039cc0e338 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
@@ -31,7 +31,7 @@ import org.scalatest.matchers.should.Matchers
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{Column, QueryTest, Row, TypedColumn}
-import org.apache.spark.sql.SparkSession.{clearActiveSession, setActiveSession}
+import org.apache.spark.sql.SparkSession.{clearActiveSession, clearDefaultSession, setActiveSession, setDefaultSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.classic.{DataFrame, Dataset, SparkSession, SQLContext}
 import org.apache.spark.sql.execution._
@@ -48,8 +48,7 @@ abstract class PipelineTest
 
   final protected val storageRoot = createTempDir()
 
-  var spark: SparkSession = createAndInitializeSpark()
-  val originalSpark: SparkSession = spark.cloneSession()
+  var spark: SparkSession = _
 
   implicit def sqlContext: SQLContext = spark.sqlContext
   def sql(text: String): DataFrame = spark.sql(text)
@@ -111,8 +110,23 @@ abstract class PipelineTest
   /** Set up the spark session before each test. */
   protected def initializeSparkBeforeEachTest(): Unit = {
     clearActiveSession()
-    spark = originalSpark.newSession()
+    clearDefaultSession()
+    spark = spark.newSession()
     setActiveSession(spark)
+    setDefaultSession(spark)
+  }
+
+  protected def initializeSession(): Unit = {
+    if (spark == null) {
+      spark = createAndInitializeSpark()
+    }
+  }
+
+  protected override def beforeAll(): Unit = {
+    initializeSession()
+
+    // Ensure we have initialized the context before calling parent code
+    super.beforeAll()
   }
 
   override def beforeEach(): Unit = {
@@ -133,7 +147,23 @@ abstract class PipelineTest
   }
 
   override def afterAll(): Unit = {
-    spark.stop()
+    try {
+      super.afterAll()
+    } finally {
+      try {
+        if (spark != null) {
+          try {
+            spark.sessionState.catalog.reset()
+          } finally {
+            spark.stop()
+            spark = null
+          }
+        }
+      } finally {
+        clearActiveSession()
+        clearDefaultSession()
+      }
+    }
   }
 
   protected def gridTest[A](testNamePrefix: String, testTags: Tag*)(params: Seq[A])(

I made some modifications to PipelineTest by referring to SharedSparkSession as above. Then I executed build/mvn clean install -pl sql/pipelines, and the compilation & testing were successful. The aforementioned modifications are for reference only.

[sryza]

Thanks for all the attention you're giving on this. I independently converged on a similar solution actually. I'm still a little bit confused about how this works for other tests that create a SparkSession more globally, like in https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/SqlBasedBenchmark.scala.

I pushed this change though. Hopefully adding the spark-core_ back in doesn't bring back the duplicate dependencies issue we were running into before 🤞 .

[sryza]

(I also pulled in the pom.xml and SparkBuild changes from the downstream PR)

[sryza]

I verified that these both work on the latest version:

build/mvn clean install -DskipTests -pl sql/pipelines -am

build/mvn test -pl sql/pipelines

[cloud-fan]

The docker test failure is unrelated, but I'll wait for @LuciferYang to do a final sign-off on these build changes.

[LuciferYang]

I verified that these both work on the latest version:

build/mvn clean install -DskipTests -pl sql/pipelines -am

build/mvn test -pl sql/pipelines

double checked, thanks @sryza

[gengliangwang]

@LuciferYang @sryza thanks for fixing the test failures!
Merging to master

[LuciferYang]

@aakash-db Here are some descriptions related to DBR, but since the current project is Apache Spark, could you revise these descriptions accordingly?

[sryza]

@LuciferYang thanks for catching this – we'll fix ASAP!

[sryza]

Fixed here: https://github.com/apache/spark/pull/51180/files

[pan3793]

Is it the same thing with the well-known "materialized view"?

[pan3793]

nvm, I figured it out after reading the code, it represents a View registered in the catalog.