[SPARK-33141][SQL][FOLLOW-UP] Store the max nested view depth in AnalysisContext

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

This is a followup of #30289. It removes the hack in `View.effectiveSQLConf`, by putting the max nested view depth in `AnalysisContext`. Then we don't get the max nested view depth from the active SQLConf, which keeps changing during nested view resolution.

### Why are the changes needed?

remove hacks.

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

No

### How was this patch tested?

If I just remove the hack, `SimpleSQLViewSuite.restrict the nested level of a view` fails. With this fix, it passes again.

Closes #30575 from cloud-fan/view.

Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit acc211d)
Signed-off-by: Wenchen Fan <[email protected]>

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -87,8 +87,8 @@ object FakeV2SessionCatalog extends TableCatalog {
 }
 
 /**
- * Provides a way to keep state during the analysis, this enables us to decouple the concerns
- * of analysis environment from the catalog.
+ * Provides a way to keep state during the analysis, mostly for resolving views. This enables us to
+ * decouple the concerns of analysis environment from the catalog.
  * The state that is kept here is per-query.
  *
  * Note this is thread local.
@@ -98,13 +98,21 @@ object FakeV2SessionCatalog extends TableCatalog {
  *                            views.
  * @param nestedViewDepth The nested depth in the view resolution, this enables us to limit the
  *                        depth of nested views.
+ * @param maxNestedViewDepth The maximum allowed depth of nested view resolution.
  * @param relationCache A mapping from qualified table names to resolved relations. This can ensure
  *                      that the table is resolved only once if a table is used multiple times
  *                      in a query.
+ * @param referredTempViewNames All the temp view names referred by the current view we are
+ *                              resolving. It's used to make sure the relation resolution is
+ *                              consistent between view creation and view resolution. For example,
+ *                              if `t` was a permanent table when the current view was created, it
+ *                              should still be a permanent table when resolving the current view,
+ *                              even if a temp view `t` has been created.
  */
 case class AnalysisContext(
     catalogAndNamespace: Seq[String] = Nil,
     nestedViewDepth: Int = 0,
+    maxNestedViewDepth: Int = -1,
     relationCache: mutable.Map[Seq[String], LogicalPlan] = mutable.Map.empty,
     referredTempViewNames: Seq[Seq[String]] = Seq.empty)
 
@@ -118,14 +126,20 @@ object AnalysisContext {
 
   private def set(context: AnalysisContext): Unit = value.set(context)
 
-  def withAnalysisContext[A](
-      catalogAndNamespace: Seq[String], referredTempViewNames: Seq[Seq[String]])(f: => A): A = {
+  def withAnalysisContext[A](viewDesc: CatalogTable)(f: => A): A = {
     val originContext = value.get()
+    val maxNestedViewDepth = if (originContext.maxNestedViewDepth == -1) {
+      // Here we start to resolve views, get `maxNestedViewDepth` from configs.
+      SQLConf.get.maxNestedViewDepth
+    } else {
+      originContext.maxNestedViewDepth
+    }
     val context = AnalysisContext(
-      catalogAndNamespace,
+      viewDesc.viewCatalogAndNamespace,
       originContext.nestedViewDepth + 1,
+      maxNestedViewDepth,
       originContext.relationCache,
-      referredTempViewNames)
+      viewDesc.viewReferredTempViewNames)
     set(context)
     try f finally { set(originContext) }
   }
@@ -1034,18 +1048,19 @@ class Analyzer(override val catalogManager: CatalogManager)
       // operator.
       case view @ View(desc, isTempView, _, child) if !child.resolved =>
         // Resolve all the UnresolvedRelations and Views in the child.
-        val newChild = AnalysisContext.withAnalysisContext(
-          desc.viewCatalogAndNamespace, desc.viewReferredTempViewNames) {
-            if (AnalysisContext.get.nestedViewDepth > conf.maxNestedViewDepth) {
-              view.failAnalysis(s"The depth of view ${desc.identifier} exceeds the maximum " +
-                s"view resolution depth (${conf.maxNestedViewDepth}). Analysis is aborted to " +
-                s"avoid errors. Increase the value of ${SQLConf.MAX_NESTED_VIEW_DEPTH.key} to " +
-                "work around this.")
-            }
-            SQLConf.withExistingConf(View.effectiveSQLConf(desc.viewSQLConfigs, isTempView)) {
-              executeSameContext(child)
-            }
+        val newChild = AnalysisContext.withAnalysisContext(desc) {
+          val nestedViewDepth = AnalysisContext.get.nestedViewDepth
+          val maxNestedViewDepth = AnalysisContext.get.maxNestedViewDepth
+          if (nestedViewDepth > maxNestedViewDepth) {
+            view.failAnalysis(s"The depth of view ${desc.identifier} exceeds the maximum " +
+              s"view resolution depth ($maxNestedViewDepth). Analysis is aborted to " +
+              s"avoid errors. Increase the value of ${SQLConf.MAX_NESTED_VIEW_DEPTH.key} to " +
+              "work around this.")
+          }
+          SQLConf.withExistingConf(View.effectiveSQLConf(desc.viewSQLConfigs, isTempView)) {
+            executeSameContext(child)
           }
+        }
         view.copy(child = newChild)
       case p @ SubqueryAlias(_, view: View) =>
         p.copy(child = resolveViews(view))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -483,9 +483,6 @@ object View {
     for ((k, v) <- configs) {
       sqlConf.settings.put(k, v)
     }
-    // We should respect the current maxNestedViewDepth cause the view resolving are executed
-    // from top to down.
-    sqlConf.setConf(SQLConf.MAX_NESTED_VIEW_DEPTH, activeConf.maxNestedViewDepth)
     sqlConf
   }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala

@@ -704,31 +704,6 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
     }
   }
 
-  test("restrict the nested level of a view") {
-    val viewNames = Array.range(0, 11).map(idx => s"view$idx")
-    withView(viewNames: _*) {
-      sql("CREATE VIEW view0 AS SELECT * FROM jt")
-      Array.range(0, 10).foreach { idx =>
-        sql(s"CREATE VIEW view${idx + 1} AS SELECT * FROM view$idx")
-      }
-
-      withSQLConf(MAX_NESTED_VIEW_DEPTH.key -> "10") {
-        val e = intercept[AnalysisException] {
-          sql("SELECT * FROM view10")
-        }.getMessage
-        assert(e.contains("The depth of view `default`.`view0` exceeds the maximum view " +
-          "resolution depth (10). Analysis is aborted to avoid errors. Increase the value " +
-          s"of ${MAX_NESTED_VIEW_DEPTH.key} to work around this."))
-      }
-
-      val e = intercept[IllegalArgumentException] {
-        withSQLConf(MAX_NESTED_VIEW_DEPTH.key -> "0") {}
-      }.getMessage
-      assert(e.contains("The maximum depth of a view reference in a nested view must be " +
-        "positive."))
-    }
-  }
-
   test("permanent view should be case-preserving") {
     withView("v") {
       sql("CREATE VIEW v AS SELECT 1 as aBc")

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewTestSuite.scala

@@ -121,7 +121,7 @@ abstract class SQLViewTestSuite extends QueryTest with SQLTestUtils {
   test("change current database should not change view behavior") {
     withTable("t") {
       Seq(2, 3, 1).toDF("c1").write.format("parquet").saveAsTable("t")
-      val viewName = createView("v1", "SELECT * from t")
+      val viewName = createView("v1", "SELECT * FROM t")
       withView(viewName) {
         withTempDatabase { db =>
           sql(s"USE $db")
@@ -135,7 +135,7 @@ abstract class SQLViewTestSuite extends QueryTest with SQLTestUtils {
   test("view should read the new data if table is updated") {
     withTable("t") {
       Seq(2, 3, 1).toDF("c1").write.format("parquet").saveAsTable("t")
-      val viewName = createView("v1", "SELECT c1 from t", Seq("c1"))
+      val viewName = createView("v1", "SELECT c1 FROM t", Seq("c1"))
       withView(viewName) {
         Seq(9, 7, 8).toDF("c1").write.mode("overwrite").format("parquet").saveAsTable("t")
         checkViewOutput(viewName, Seq(Row(9), Row(7), Row(8)))
@@ -146,7 +146,7 @@ abstract class SQLViewTestSuite extends QueryTest with SQLTestUtils {
   test("add column for table should not affect view output") {
     withTable("t") {
       Seq(2, 3, 1).toDF("c1").write.format("parquet").saveAsTable("t")
-      val viewName = createView("v1", "SELECT * from t")
+      val viewName = createView("v1", "SELECT * FROM t")
       withView(viewName) {
         sql("ALTER TABLE t ADD COLUMN (c2 INT)")
         checkViewOutput(viewName, Seq(Row(2), Row(3), Row(1)))
@@ -157,8 +157,8 @@ abstract class SQLViewTestSuite extends QueryTest with SQLTestUtils {
   test("check cyclic view reference on CREATE OR REPLACE VIEW") {
     withTable("t") {
       Seq(2, 3, 1).toDF("c1").write.format("parquet").saveAsTable("t")
-      val viewName1 = createView("v1", "SELECT * from t")
-      val viewName2 = createView("v2", s"SELECT * from $viewName1")
+      val viewName1 = createView("v1", "SELECT * FROM t")
+      val viewName2 = createView("v2", s"SELECT * FROM $viewName1")
       withView(viewName2, viewName1) {
         val e = intercept[AnalysisException] {
           createView("v1", s"SELECT * FROM $viewName2", replace = true)
@@ -171,8 +171,8 @@ abstract class SQLViewTestSuite extends QueryTest with SQLTestUtils {
   test("check cyclic view reference on ALTER VIEW") {
     withTable("t") {
       Seq(2, 3, 1).toDF("c1").write.format("parquet").saveAsTable("t")
-      val viewName1 = createView("v1", "SELECT * from t")
-      val viewName2 = createView("v2", s"SELECT * from $viewName1")
+      val viewName1 = createView("v1", "SELECT * FROM t")
+      val viewName2 = createView("v2", s"SELECT * FROM $viewName1")
       withView(viewName2, viewName1) {
         val e = intercept[AnalysisException] {
           sql(s"ALTER VIEW $viewName1 AS SELECT * FROM $viewName2")
@@ -181,6 +181,24 @@ abstract class SQLViewTestSuite extends QueryTest with SQLTestUtils {
       }
     }
   }
+
+  test("restrict the nested level of a view") {
+    val viewNames = scala.collection.mutable.ArrayBuffer.empty[String]
+    val view0 = createView("view0", "SELECT 1")
+    viewNames += view0
+    for (i <- 1 to 10) {
+      viewNames += createView(s"view$i", s"SELECT * FROM ${viewNames.last}")
+    }
+    withView(viewNames.reverse: _*) {
+      withSQLConf(MAX_NESTED_VIEW_DEPTH.key -> "10") {
+        val e = intercept[AnalysisException] {
+          sql(s"SELECT * FROM ${viewNames.last}")
+        }.getMessage
+        assert(e.contains("exceeds the maximum view resolution depth (10)"))
+        assert(e.contains(s"Increase the value of ${MAX_NESTED_VIEW_DEPTH.key}"))
+      }
+    }
+  }
 }
 
 class LocalTempViewTestSuite extends SQLViewTestSuite with SharedSparkSession {