detect cyclic references for CREATE OR REPLACE VIEW and subqueries.

Author: jiangxb1987

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala

@@ -23,7 +23,7 @@ import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedFunction, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
-import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.{Alias, SubqueryExpression}
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, View}
 import org.apache.spark.sql.types.MetadataBuilder
@@ -154,6 +154,10 @@ case class CreateViewCommand(
       } else if (tableMetadata.tableType != CatalogTableType.VIEW) {
         throw new AnalysisException(s"$name is not a view")
       } else if (replace) {
+        // Detect cyclic view reference on CREATE OR REPLACE VIEW.
+        val viewIdent = tableMetadata.identifier
+        checkCyclicViewReference(analyzedPlan, Seq(viewIdent), viewIdent)
+
         // Handles `CREATE OR REPLACE VIEW v0 AS SELECT ...`
         catalog.alterTable(prepareTable(sparkSession, analyzedPlan))
       } else {
@@ -283,17 +287,9 @@ case class AlterViewAsCommand(
       throw new AnalysisException(s"${viewMeta.identifier} is not a view.")
     }
 
-    // Detect cyclic view references, a cyclic view reference may be created by the following
-    // queries:
-    // CREATE VIEW testView AS SELECT id FROM tbl
-    // CREATE VIEW testView2 AS SELECT id FROM testView
-    // ALTER VIEW testView AS SELECT * FROM testView2
-    // In the above example, a reference cycle (testView -> testView2 -> testView) exsits.
-    //
-    // We disallow cyclic view references by checking that in ALTER VIEW command, when the
-    // `analyzedPlan` contains the same `View` node with the altered view, we should prevent the
-    // behavior and throw an AnalysisException.
-    checkCyclicViewReference(analyzedPlan, Seq(viewMeta.identifier), viewMeta.identifier)
+    // Detect cyclic view reference on ALTER VIEW.
+    val viewIdent = viewMeta.identifier
+    checkCyclicViewReference(analyzedPlan, Seq(viewIdent), viewIdent)
 
     val newProperties = generateViewProperties(viewMeta.properties, session, analyzedPlan)
 
@@ -304,38 +300,6 @@ case class AlterViewAsCommand(
 
     session.sessionState.catalog.alterTable(updatedViewMeta)
   }
-
-  /**
-   * Recursively search the logical plan to detect cyclic view references, throw an
-   * AnalysisException if cycle detected.
-   *
-   * @param plan the logical plan we detect cyclic view references from.
-   * @param path the path between the altered view and current node.
-   * @param viewIdent the table identifier of the altered view, we compare two views by the
-   *                  `desc.identifier`.
-   */
-  private def checkCyclicViewReference(
-      plan: LogicalPlan,
-      path: Seq[TableIdentifier],
-      viewIdent: TableIdentifier): Unit = {
-    plan match {
-      case v: View =>
-        val ident = v.desc.identifier
-        val newPath = path :+ ident
-        // If the table identifier equals to the `viewIdent`, current view node is the same with
-        // the altered view. We detect a view reference cycle, should throw an AnalysisException.
-        if (ident == viewIdent) {
-          throw new AnalysisException(s"Recursive view $viewIdent detected " +
-            s"(cycle: ${newPath.mkString(" -> ")})")
-        } else {
-          v.children.foreach { child =>
-            checkCyclicViewReference(child, newPath, viewIdent)
-          }
-        }
-      case _ =>
-        plan.children.foreach(child => checkCyclicViewReference(child, path, viewIdent))
-    }
-  }
 }
 
 object ViewHelper {
@@ -402,4 +366,51 @@ object ViewHelper {
       generateViewDefaultDatabase(viewDefaultDatabase) ++
       generateQueryColumnNames(queryOutput)
   }
+
+  /**
+   * Recursively search the logical plan to detect cyclic view references, throw an
+   * AnalysisException if cycle detected.
+   *
+   * A cyclic view reference is a cycle of reference dependencies, for example, if the following
+   * statements are executed:
+   * CREATE VIEW testView AS SELECT id FROM tbl
+   * CREATE VIEW testView2 AS SELECT id FROM testView
+   * ALTER VIEW testView AS SELECT * FROM testView2
+   * The view `testView` references `testView2`, and `testView2` also references `testView`,
+   * therefore a reference cycle (testView -> testView2 -> testView) exists.
+   *
+   * @param plan the logical plan we detect cyclic view references from.
+   * @param path the path between the altered view and current node.
+   * @param viewIdent the table identifier of the altered view, we compare two views by the
+   *                  `desc.identifier`.
+   */
+  def checkCyclicViewReference(
+      plan: LogicalPlan,
+      path: Seq[TableIdentifier],
+      viewIdent: TableIdentifier): Unit = {
+    plan match {
+      case v: View =>
+        val ident = v.desc.identifier
+        val newPath = path :+ ident
+        // If the table identifier equals to the `viewIdent`, current view node is the same with
+        // the altered view. We detect a view reference cycle, should throw an AnalysisException.
+        if (ident == viewIdent) {
+          throw new AnalysisException(s"Recursive view $viewIdent detected " +
+            s"(cycle: ${newPath.mkString(" -> ")})")
+        } else {
+          v.children.foreach { child =>
+            checkCyclicViewReference(child, newPath, viewIdent)
+          }
+        }
+      case _ =>
+        plan.children.foreach(child => checkCyclicViewReference(child, path, viewIdent))
+    }
+
+    // Detect cyclic references from subqueries.
+    plan.expressions.foreach { expr =>
+      if (expr.isInstanceOf[SubqueryExpression]) {
+        checkCyclicViewReference(expr.asInstanceOf[SubqueryExpression].plan, path, viewIdent)
+      }
+    }
+  }
 }

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

@@ -628,6 +628,20 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
       }.getMessage
       assert(e2.contains("Recursive view `default`.`view1` detected (cycle: `default`.`view1` " +
         "-> `default`.`view3` -> `default`.`view2` -> `default`.`view1`)"))
+
+      // Detect cyclic view reference on CREATE OR REPLACE VIEW.
+      val e3 = intercept[AnalysisException] {
+        sql("CREATE OR REPLACE VIEW view1 AS SELECT * FROM view2")
+      }.getMessage
+      assert(e3.contains("Recursive view `default`.`view1` detected (cycle: `default`.`view1` " +
+        "-> `default`.`view2` -> `default`.`view1`)"))
+
+      // Detect cyclic view reference from subqueries.
+      val e4 = intercept[AnalysisException] {
+        sql("ALTER VIEW view1 AS SELECT * FROM jt WHERE EXISTS (SELECT 1 FROM view2)")
+      }.getMessage
+      assert(e4.contains("Recursive view `default`.`view1` detected (cycle: `default`.`view1` " +
+        "-> `default`.`view2` -> `default`.`view1`)"))
     }
   }
 }