[SPARK-29628][SQL] Forcibly create a temporary view in CREATE VIEW if referencing a temporary view

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

@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable
 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, StructType}
+import org.apache.spark.sql.types.MetadataBuilder
 import org.apache.spark.sql.util.SchemaUtils
 
 
@@ -110,19 +110,25 @@ case class CreateViewCommand(
 
   private def isTemporary = viewType == LocalTempView || viewType == GlobalTempView
 
-  // Disallows 'CREATE TEMPORARY VIEW IF NOT EXISTS' to be consistent with 'CREATE TEMPORARY TABLE'
-  if (allowExisting && isTemporary) {
-    throw new AnalysisException(
-      "It is not allowed to define a TEMPORARY view with IF NOT EXISTS.")
-  }
+  if (isTemporary) verifyTempView()
+
+  private def verifyTempView(): Unit = {
+    // Disallows 'CREATE TEMPORARY VIEW IF NOT EXISTS'
+    // to be consistent with 'CREATE TEMPORARY TABLE'
+    if (allowExisting) {
+      throw new AnalysisException(
+        "It is not allowed to define a TEMPORARY view with IF NOT EXISTS.")
+    }
 
-  // Temporary view names should NOT contain database prefix like "database.table"
-  if (isTemporary && name.database.isDefined) {
-    val database = name.database.get
-    throw new AnalysisException(
-      s"It is not allowed to add database prefix `$database` for the TEMPORARY view name.")
+    // Temporary view names should NOT contain database prefix like "database.table"
+    name.database.foreach {
+      database => throw new AnalysisException(
+        s"It is not allowed to add database prefix `$database` for the TEMPORARY view name.")
+    }
   }
 
+  private var isTempReferred = false
+
   override def run(sparkSession: SparkSession): Seq[Row] = {
     // If the plan cannot be analyzed, throw an exception and don't proceed.
     val qe = sparkSession.sessionState.executePlan(child)
@@ -136,12 +142,11 @@ case class CreateViewCommand(
         s"specified by CREATE VIEW (num: `${userSpecifiedColumns.length}`).")
     }
 
-    // When creating a permanent view, not allowed to reference temporary objects.
     // This should be called after `qe.assertAnalyzed()` (i.e., `child` can be resolved)
     verifyTemporaryObjectsNotExists(sparkSession)
 
     val catalog = sparkSession.sessionState.catalog
-    if (viewType == LocalTempView) {
+    if (viewType == LocalTempView || isTempReferred) {
       val aliasedPlan = aliasPlan(sparkSession, analyzedPlan)
       catalog.createTempView(name.table, aliasedPlan, overrideIfExists = replace)
     } else if (viewType == GlobalTempView) {
@@ -178,7 +183,8 @@ case class CreateViewCommand(
   }
 
   /**
-   * Permanent views are not allowed to reference temp objects, including temp function and views
+   * Permanent views are not allowed to reference temp function. When permanent view
+   * has a reference of temp view, it will be created as temp view [SPARK-29628].
    */
   private def verifyTemporaryObjectsNotExists(sparkSession: SparkSession): Unit = {
     import sparkSession.sessionState.analyzer.AsTableIdentifier
@@ -192,12 +198,20 @@ case class CreateViewCommand(
       // package (e.g., HiveGenericUDF).
       def verify(child: LogicalPlan) {
         child.collect {
-          // Disallow creating permanent views based on temporary views.
+          // Permanent views will be created as temporary view if based on temporary view.
           case UnresolvedRelation(AsTableIdentifier(ident))
             if sparkSession.sessionState.catalog.isTemporaryTable(ident) =>
-            // temporary views are only stored in the session catalog
-            throw new AnalysisException(s"Not allowed to create a permanent view $name by " +
-              s"referencing a temporary view $ident")
+            // Temporary views are only stored in the session catalog
+            if (sparkSession.sqlContext.conf.usePostgreSQLDialect) {
+              logInfo(s"View $name is based on temporary view $ident."
+                + s" $name will be created as temporary view")
+              verifyTempView()
+              isTempReferred = true
+            } else {
+              throw new AnalysisException(s"Not allowed to create a permanent view $name by " +
+                s"referencing a temporary view $ident")
+            }
+
           case other if !other.resolved => other.expressions.flatMap(_.collect {
             // Traverse subquery plan for any unresolved relations.
             case e: SubqueryExpression => verify(e.plan)

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

@@ -72,23 +72,15 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
   }
 
   test("create a permanent view on a temp view") {
-    withView("jtv1") {
+    withTempView("jtv1") {
       withTempView("temp_jtv1") {
         withGlobalTempView("global_temp_jtv1") {
           sql("CREATE TEMPORARY VIEW temp_jtv1 AS SELECT * FROM jt WHERE id > 3")
-          var e = intercept[AnalysisException] {
-            sql("CREATE VIEW jtv1 AS SELECT * FROM temp_jtv1 WHERE id < 6")
-          }.getMessage
-          assert(e.contains("Not allowed to create a permanent view `jtv1` by " +
-            "referencing a temporary view `temp_jtv1`"))
+          sql("CREATE VIEW jtv1 AS SELECT * FROM temp_jtv1 WHERE id < 6")
 
           val globalTempDB = spark.sharedState.globalTempViewManager.database
           sql("CREATE GLOBAL TEMP VIEW global_temp_jtv1 AS SELECT * FROM jt WHERE id > 0")
-          e = intercept[AnalysisException] {
-            sql(s"CREATE VIEW jtv1 AS SELECT * FROM $globalTempDB.global_temp_jtv1 WHERE id < 6")
-          }.getMessage
-          assert(e.contains(s"Not allowed to create a permanent view `jtv1` by referencing " +
-            s"a temporary view `global_temp`.`global_temp_jtv1`"))
+          sql(s"CREATE VIEW jtv2 AS SELECT * FROM $globalTempDB.global_temp_jtv1 WHERE id < 6")
         }
       }
     }