[SPARK-19877][SQL] Restrict the nested level of a view

## What changes were proposed in this pull request?

We should restrict the nested level of a view, to avoid stack overflow exception during the view resolution.

## How was this patch tested?

Add new test case in `SQLViewSuite`.

Author: jiangxingbo <jiangxb1987@gmail.com>

Closes #17241 from jiangxb1987/view-depth.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SimpleCatalystConf.scala

@@ -41,7 +41,8 @@ case class SimpleCatalystConf(
     override val joinReorderEnabled: Boolean = false,
     override val joinReorderDPThreshold: Int = 12,
     override val warehousePath: String = "/user/hive/warehouse",
-    override val sessionLocalTimeZone: String = TimeZone.getDefault().getID)
+    override val sessionLocalTimeZone: String = TimeZone.getDefault().getID,
+    override val maxNestedViewDepth: Int = 100)
   extends SQLConf {
 
   override def clone(): SimpleCatalystConf = this.copy()

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

@@ -58,13 +58,12 @@ object SimpleAnalyzer extends Analyzer(
  *
  * @param defaultDatabase The default database used in the view resolution, this overrules the
  *                        current catalog database.
- * @param nestedViewLevel The nested level in the view resolution, this enables us to limit the
+ * @param nestedViewDepth The nested depth in the view resolution, this enables us to limit the
  *                        depth of nested views.
- *                        TODO Limit the depth of nested views.
  */
 case class AnalysisContext(
     defaultDatabase: Option[String] = None,
-    nestedViewLevel: Int = 0)
+    nestedViewDepth: Int = 0)
 
 object AnalysisContext {
   private val value = new ThreadLocal[AnalysisContext]() {
@@ -77,7 +76,7 @@ object AnalysisContext {
   def withAnalysisContext[A](database: Option[String])(f: => A): A = {
     val originContext = value.get()
     val context = AnalysisContext(defaultDatabase = database,
-      nestedViewLevel = originContext.nestedViewLevel + 1)
+      nestedViewDepth = originContext.nestedViewDepth + 1)
     set(context)
     try f finally { set(originContext) }
   }
@@ -598,6 +597,12 @@ class Analyzer(
       case view @ View(desc, _, child) if !child.resolved =>
         // Resolve all the UnresolvedRelations and Views in the child.
         val newChild = AnalysisContext.withAnalysisContext(desc.viewDefaultDatabase) {
+          if (AnalysisContext.get.nestedViewDepth > conf.maxNestedViewDepth) {
+            view.failAnalysis(s"The depth of view ${view.desc.identifier} exceeds the maximum " +
+              s"view resolution depth (${conf.maxNestedViewDepth}). Analysis is aborted to " +
+              "avoid errors. Increase the value of spark.sql.view.maxNestedViewDepth to work " +
+              "aroud this.")
+          }
           execute(child)
         }
         view.copy(child = newChild)

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -571,6 +571,19 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val MAX_NESTED_VIEW_DEPTH =
+    buildConf("spark.sql.view.maxNestedViewDepth")
+      .internal()
+      .doc("The maximum depth of a view reference in a nested view. A nested view may reference " +
+        "other nested views, the dependencies are organized in a directed acyclic graph (DAG). " +
+        "However the DAG depth may become too large and cause unexpected behavior. This " +
+        "configuration puts a limit on this: when the depth of a view exceeds this value during " +
+        "analysis, we terminate the resolution to avoid potential errors.")
+      .intConf
+      .checkValue(depth => depth > 0, "The maximum depth of a view reference in a nested view " +
+        "must be positive.")
+      .createWithDefault(100)
+
   val STREAMING_FILE_COMMIT_PROTOCOL_CLASS =
     buildConf("spark.sql.streaming.commitProtocolClass")
       .internal()
@@ -932,6 +945,8 @@ class SQLConf extends Serializable with Logging {
 
   def joinReorderDPThreshold: Int = getConf(SQLConf.JOIN_REORDER_DP_THRESHOLD)
 
+  def maxNestedViewDepth: Int = getConf(SQLConf.MAX_NESTED_VIEW_DEPTH)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */

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

@@ -644,4 +644,29 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
         "-> `default`.`view2` -> `default`.`view1`)"))
     }
   }
+
+  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("spark.sql.view.maxNestedViewDepth" -> "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 " +
+          "of spark.sql.view.maxNestedViewDepth to work aroud this."))
+      }
+
+      val e = intercept[IllegalArgumentException] {
+        withSQLConf("spark.sql.view.maxNestedViewDepth" -> "0") {}
+      }.getMessage
+      assert(e.contains("The maximum depth of a view reference in a nested view must be " +
+        "positive."))
+    }
+  }
 }