From db45b4cc3614965980571c6203074af845e4d249 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Sun, 21 Aug 2016 23:45:37 -0700
Subject: [PATCH 1/7] fix.

---
 .../spark/sql/execution/command/views.scala   |  8 ++-
 .../sql/hive/execution/SQLViewSuite.scala     | 61 ++++++++++++++++++-
 2 files changed, 67 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index e397cfa058e24..19a1d8565ff61 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -105,7 +105,13 @@ case class CreateViewCommand(
     }
     val sessionState = sparkSession.sessionState
 
-    if (isTemporary) {
+    // 1) CREATE VIEW: create a temp view when users explicitly specify the keyword TEMPORARY;
+    //                 otherwise, create a permanent view no matter whether the temporary view
+    //                 with the same name exists or not.
+    // 2) ALTER VIEW: alter the temporary view if the temp view exists; otherwise, try to alter
+    //                the permanent view. Here, it follows the same resolution like DROP VIEW,
+    //                since users are unable to specify the keyword TEMPORARY.
+    if (isTemporary || (replace && sessionState.catalog.isTemporaryTable(name))) {
       createTemporaryView(sparkSession, analyzedPlan)
     } else {
       // Adds default database for permanent table if it doesn't exist, so that tableExists()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index 6a80664417911..14166f2532f04 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -18,6 +18,8 @@
 package org.apache.spark.sql.hive.execution
 
 import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.CatalogTableType
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.test.SQLTestUtils
 
@@ -204,10 +206,67 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     }
   }
 
-  test("should allow CREATE permanent VIEW when a TEMPORARY VIEW with same name exists") {
+  test("ALTER VIEW: alter a temporary view when a permanent VIEW with same name exists") {
+    verifyAlterViewWithIdenticalName(isTempAlteredView = true)
+  }
+
+  test("ALTER VIEW: alter a persistent view when a temp VIEW with same name exists") {
+    verifyAlterViewWithIdenticalName(isTempAlteredView = false)
+  }
+
+  private def verifyAlterViewWithIdenticalName (isTempAlteredView: Boolean) = {
+    withView("testView", "default.testView") {
+      val catalog = spark.sessionState.catalog
+      val oldViewQuery = "SELECT id FROM jt"
+      val newViewQuery = "SELECT  id, id1 FROM jt"
+      sql(s"CREATE VIEW default.testView AS $oldViewQuery")
+      sql(s"CREATE TEMPORARY VIEW testView AS $oldViewQuery")
+      if (isTempAlteredView) {
+        // When the database is not specified, we will first try to alter the temporary view
+        sql(s"ALTER VIEW testView AS $newViewQuery")
+      } else {
+        // When the database is specified, we will try to alter the permanent view, no matter
+        // whether the temporary view with the same name exists or not.
+        sql(s"ALTER VIEW default.testView AS $newViewQuery")
+      }
+
+      val persistentView = catalog.getTableMetadata(
+        TableIdentifier(table = "testView", database = Some("default")))
+      assert(persistentView.tableType == CatalogTableType.VIEW)
+      val tempView = catalog.getTableMetadata(TableIdentifier("testView"))
+      assert(tempView.tableType == CatalogTableType.VIEW)
+      assert(tempView.viewOriginalText.isEmpty)
+
+      if (isTempAlteredView) {
+        // View Text of the persistent view default.testView is changed
+        assert(persistentView.viewOriginalText == Option(oldViewQuery))
+        // temp view testView is changed
+        checkAnswer(
+          sql(newViewQuery),
+          sql("select * from testView"))
+      } else {
+        // View Text of the persistent view default.testView is changed
+        assert(persistentView.viewOriginalText == Option(newViewQuery))
+        // temp view testView is not changed
+        checkAnswer(
+          sql(oldViewQuery),
+          sql("select * from testView"))
+      }
+    }
+  }
+
+  test("CREATE VIEW: should allow CREATE permanent VIEW when a temp VIEW with same name exists") {
     withView("testView", "default.testView") {
       sql("CREATE TEMPORARY VIEW testView AS SELECT id FROM jt")
       sql("CREATE VIEW testView AS SELECT id FROM jt")
+
+      // Both temporary and permanent view have been successfully created.
+      val catalog = spark.sessionState.catalog
+      val persistentView = catalog.getTableMetadata(
+        TableIdentifier(table = "testView", database = Some("default")))
+      assert(persistentView.tableType == CatalogTableType.VIEW)
+      val tempView = catalog.getTableMetadata(TableIdentifier("testView"))
+      assert(tempView.tableType == CatalogTableType.VIEW)
     }
   }
 

From c5add2cbbcc3cbbce1ff09155da27b145c204ee1 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Wed, 24 Aug 2016 21:26:10 -0700
Subject: [PATCH 2/7] address comments.

---
 .../apache/spark/sql/hive/execution/SQLViewSuite.scala | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index 14166f2532f04..2e2a413bb8f77 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -207,18 +207,18 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
   }
 
   test("ALTER VIEW: alter a temporary view when a permanent VIEW with same name exists") {
-    verifyAlterViewWithIdenticalName(isTempAlteredView = true)
+    alterTempView(isTempAlteredView = true)
   }
 
   test("ALTER VIEW: alter a persistent view when a temp VIEW with same name exists") {
-    verifyAlterViewWithIdenticalName(isTempAlteredView = false)
+    alterTempView(isTempAlteredView = false)
   }
 
-  private def verifyAlterViewWithIdenticalName (isTempAlteredView: Boolean) = {
+  private def alterTempView (isTempAlteredView: Boolean) = {
     withView("testView", "default.testView") {
       val catalog = spark.sessionState.catalog
       val oldViewQuery = "SELECT id FROM jt"
-      val newViewQuery = "SELECT  id, id1 FROM jt"
+      val newViewQuery = "SELECT id, id1 FROM jt"
       sql(s"CREATE VIEW default.testView AS $oldViewQuery")
       sql(s"CREATE TEMPORARY VIEW testView AS $oldViewQuery")
       if (isTempAlteredView) {
@@ -238,7 +238,7 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       assert(tempView.viewOriginalText.isEmpty)
 
       if (isTempAlteredView) {
-        // View Text of the persistent view default.testView is changed
+        // View Text of the persistent view default.testView is not changed
         assert(persistentView.viewOriginalText == Option(oldViewQuery))
         // temp view testView is changed
         checkAnswer(

From 51dbf727f1bb0b71e813f32a064e0d13e33843fc Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Thu, 25 Aug 2016 23:40:37 -0700
Subject: [PATCH 3/7] address comments

---
 .../scala/org/apache/spark/sql/Dataset.scala  |  6 +++---
 .../spark/sql/execution/SparkSqlParser.scala  | 12 ++++++++----
 .../spark/sql/execution/command/views.scala   |  6 ++++--
 .../sql/hive/execution/SQLViewSuite.scala     | 19 +++++++++++++++++++
 4 files changed, 34 insertions(+), 9 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
index 6da99ce0dd683..2ddd98689037e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
@@ -2429,7 +2429,7 @@ class Dataset[T] private[sql](
    */
   @throws[AnalysisException]
   def createTempView(viewName: String): Unit = withPlan {
-    createViewCommand(viewName, replace = false)
+    createTempViewCommand(viewName, replace = false)
   }
 
   /**
@@ -2440,10 +2440,10 @@ class Dataset[T] private[sql](
    * @since 2.0.0
    */
   def createOrReplaceTempView(viewName: String): Unit = withPlan {
-    createViewCommand(viewName, replace = true)
+    createTempViewCommand(viewName, replace = true)
   }
 
-  private def createViewCommand(viewName: String, replace: Boolean): CreateViewCommand = {
+  private def createTempViewCommand(viewName: String, replace: Boolean): CreateViewCommand = {
     CreateViewCommand(
       name = sparkSession.sessionState.sqlParser.parseTableIdentifier(viewName),
       userSpecifiedColumns = Nil,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 71c3bd31e02e4..568e6e4b4ea91 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -1263,7 +1263,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
         Option(ctx.tablePropertyList).map(visitPropertyKeyValues).getOrElse(Map.empty),
         allowExisting = ctx.EXISTS != null,
         replace = ctx.REPLACE != null,
-        isTemporary = ctx.TEMPORARY != null
+        isTemporary = ctx.TEMPORARY != null,
+        isAlterViewAsSelect = false
       )
     }
   }
@@ -1281,7 +1282,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       properties = Map.empty,
       allowExisting = false,
       replace = true,
-      isTemporary = false)
+      isTemporary = false,
+      isAlterViewAsSelect = true)
   }
 
   /**
@@ -1296,7 +1298,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       properties: Map[String, String],
       allowExisting: Boolean,
       replace: Boolean,
-      isTemporary: Boolean): LogicalPlan = {
+      isTemporary: Boolean,
+      isAlterViewAsSelect: Boolean): LogicalPlan = {
     val originalText = source(query)
     CreateViewCommand(
       visitTableIdentifier(name),
@@ -1307,7 +1310,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       plan(query),
       allowExisting = allowExisting,
       replace = replace,
-      isTemporary = isTemporary)
+      isTemporary = isTemporary,
+      isAlterViewAsSelect = isAlterViewAsSelect)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index 19a1d8565ff61..c46c7a3b5ccef 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -49,6 +49,7 @@ import org.apache.spark.sql.types.StructType
  *                 at the end of current Spark session. Existing permanent relations with the same
  *                 name are not visible to the current session while the temporary view exists,
  *                 unless they are specified with full qualified table name with database prefix.
+  * @param isAlterViewAsSelect if true, this original DDL command is ALTER VIEW AS SELECT
  */
 case class CreateViewCommand(
     name: TableIdentifier,
@@ -59,7 +60,8 @@ case class CreateViewCommand(
     child: LogicalPlan,
     allowExisting: Boolean,
     replace: Boolean,
-    isTemporary: Boolean)
+    isTemporary: Boolean,
+    isAlterViewAsSelect: Boolean = false)
   extends RunnableCommand {
 
   override protected def innerChildren: Seq[QueryPlan[_]] = Seq(child)
@@ -111,7 +113,7 @@ case class CreateViewCommand(
     // 2) ALTER VIEW: alter the temporary view if the temp view exists; otherwise, try to alter
     //                the permanent view. Here, it follows the same resolution like DROP VIEW,
     //                since users are unable to specify the keyword TEMPORARY.
-    if (isTemporary || (replace && sessionState.catalog.isTemporaryTable(name))) {
+    if (isTemporary || (isAlterViewAsSelect && sessionState.catalog.isTemporaryTable(name))) {
       createTemporaryView(sparkSession, analyzedPlan)
     } else {
       // Adds default database for permanent table if it doesn't exist, so that tableExists()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index 2e2a413bb8f77..d2fac04f64fdc 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -203,6 +203,25 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     withView("testView", "default.testView") {
       sql("CREATE VIEW testView AS SELECT id FROM jt")
       sql("CREATE TEMPORARY VIEW testView AS SELECT id FROM jt")
+      checkAnswer(
+        sql("SHOW TABLES 'testView*'"),
+        Row("testview", true) :: Row("testview", false) :: Nil)
+    }
+
+    withView("testView", "default.testView") {
+      sql("CREATE VIEW testView AS SELECT id FROM jt")
+      sql("CREATE OR REPLACE TEMPORARY VIEW testView AS SELECT id FROM jt")
+      checkAnswer(
+        sql("SHOW TABLES 'testView*'"),
+        Row("testview", true) :: Row("testview", false) :: Nil)
+    }
+
+    withView("testView", "default.testView") {
+      sql("CREATE VIEW testView AS SELECT id FROM jt")
+      sql("CREATE OR REPLACE VIEW testView AS SELECT id FROM jt")
+      checkAnswer(
+        sql("SHOW TABLES 'testView*'"),
+        Row("testview", false) :: Nil)
     }
   }
 

From d84db62dfd343654fdb8e64badf862eea3f08832 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Fri, 26 Aug 2016 23:41:08 -0700
Subject: [PATCH 4/7] address comments.

---
 .../scala/org/apache/spark/sql/Dataset.scala  |  3 +-
 .../spark/sql/execution/SparkSqlParser.scala  | 17 +++++++----
 .../spark/sql/execution/SparkStrategies.scala |  2 --
 .../spark/sql/execution/command/views.scala   | 29 +++++++++----------
 .../spark/sql/hive/HiveDDLCommandSuite.scala  | 15 +++++++++-
 5 files changed, 41 insertions(+), 25 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
index 2ddd98689037e..8b665e8d31cb2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
@@ -2451,8 +2451,7 @@ class Dataset[T] private[sql](
       properties = Map.empty,
       originalText = None,
       child = logicalPlan,
-      allowExisting = false,
-      replace = replace,
+      if (replace) SaveMode.Overwrite else SaveMode.ErrorIfExists,
       isTemporary = true)
   }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 568e6e4b4ea91..f4a26a0522d3c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -1261,7 +1261,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
         userSpecifiedColumns,
         ctx.query,
         Option(ctx.tablePropertyList).map(visitPropertyKeyValues).getOrElse(Map.empty),
-        allowExisting = ctx.EXISTS != null,
+        ignoreIfExists = ctx.EXISTS != null,
         replace = ctx.REPLACE != null,
         isTemporary = ctx.TEMPORARY != null,
         isAlterViewAsSelect = false
@@ -1280,7 +1280,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       userSpecifiedColumns = Seq.empty,
       query = ctx.query,
       properties = Map.empty,
-      allowExisting = false,
+      ignoreIfExists = false,
       replace = true,
       isTemporary = false,
       isAlterViewAsSelect = true)
@@ -1296,11 +1296,19 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       userSpecifiedColumns: Seq[(String, Option[String])],
       query: QueryContext,
       properties: Map[String, String],
-      allowExisting: Boolean,
+      ignoreIfExists: Boolean,
       replace: Boolean,
       isTemporary: Boolean,
       isAlterViewAsSelect: Boolean): LogicalPlan = {
+    val mode = (replace, ignoreIfExists) match {
+      case (true, false) => SaveMode.Overwrite
+      case (false, true) => SaveMode.Ignore
+      case (false, false) => SaveMode.ErrorIfExists
+      case _ => throw new ParseException(
+        "CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed.", ctx)
+    }
     val originalText = source(query)
+
     CreateViewCommand(
       visitTableIdentifier(name),
       userSpecifiedColumns,
@@ -1308,8 +1316,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       properties,
       Some(originalText),
       plan(query),
-      allowExisting = allowExisting,
-      replace = replace,
+      mode,
       isTemporary = isTemporary,
       isAlterViewAsSelect = isAlterViewAsSelect)
   }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 4aaf454285f4f..efff77c3c235c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -450,8 +450,6 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
             query)
         ExecutedCommandExec(cmd) :: Nil
 
-      case c: CreateTempViewUsing => ExecutedCommandExec(c) :: Nil
-
       case _ => Nil
     }
   }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index c46c7a3b5ccef..cdca8a9fdbed2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.command
 
 import scala.util.control.NonFatal
 
-import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.{AnalysisException, Row, SaveMode, SparkSession}
 import org.apache.spark.sql.catalyst.{SQLBuilder, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute}
@@ -41,10 +41,11 @@ import org.apache.spark.sql.types.StructType
  *                     Dataset API.
  * @param child the logical plan that represents the view; this is used to generate a canonicalized
  *              version of the SQL that can be saved in the catalog.
- * @param allowExisting if true, and if the view already exists, noop; if false, and if the view
- *                already exists, throws analysis exception.
- * @param replace if true, and if the view already exists, updates it; if false, and if the view
- *                already exists, throws analysis exception.
+ * @param mode only three modes are applicable here: Ignore, Overwrite and ErrorIfExists. If the
+ *             view already exists, CreateViewCommand behaves based on the mode:
+ *             1) Overwrite: update the view;
+ *             2) Ignore: noop;
+ *             3) ErrorIfExists throws analysis exception.
  * @param isTemporary if true, the view is created as a temporary view. Temporary views are dropped
  *                 at the end of current Spark session. Existing permanent relations with the same
  *                 name are not visible to the current session while the temporary view exists,
@@ -58,8 +59,7 @@ case class CreateViewCommand(
     properties: Map[String, String],
     originalText: Option[String],
     child: LogicalPlan,
-    allowExisting: Boolean,
-    replace: Boolean,
+    mode: SaveMode,
     isTemporary: Boolean,
     isAlterViewAsSelect: Boolean = false)
   extends RunnableCommand {
@@ -71,17 +71,16 @@ case class CreateViewCommand(
 
   override def output: Seq[Attribute] = Seq.empty[Attribute]
 
+  assert(mode != SaveMode.Append,
+    "CREATE or ALTER VIEW can only use ErrorIfExists, Ignore or Overwrite.")
+
   if (!isTemporary) {
     require(originalText.isDefined,
       "The table to created with CREATE VIEW must have 'originalText'.")
   }
 
-  if (allowExisting && replace) {
-    throw new AnalysisException("CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed.")
-  }
-
   // Disallows 'CREATE TEMPORARY VIEW IF NOT EXISTS' to be consistent with 'CREATE TEMPORARY TABLE'
-  if (allowExisting && isTemporary) {
+  if (mode == SaveMode.Ignore && isTemporary) {
     throw new AnalysisException(
       "It is not allowed to define a TEMPORARY view with IF NOT EXISTS.")
   }
@@ -123,14 +122,14 @@ case class CreateViewCommand(
 
       if (sessionState.catalog.tableExists(qualifiedName)) {
         val tableMetadata = sessionState.catalog.getTableMetadata(qualifiedName)
-        if (allowExisting) {
+        if (mode == SaveMode.Ignore) {
           // Handles `CREATE VIEW IF NOT EXISTS v0 AS SELECT ...`. Does nothing when the target view
           // already exists.
         } else if (tableMetadata.tableType != CatalogTableType.VIEW) {
           throw new AnalysisException(
             "Existing table is not a view. The following is an existing table, " +
               s"not a view: $qualifiedName")
-        } else if (replace) {
+        } else if (mode == SaveMode.Overwrite) {
           // Handles `CREATE OR REPLACE VIEW v0 AS SELECT ...`
           sessionState.catalog.alterTable(prepareTable(sparkSession, analyzedPlan))
         } else {
@@ -162,7 +161,7 @@ case class CreateViewCommand(
       sparkSession.sessionState.executePlan(Project(projectList, analyzedPlan)).analyzed
     }
 
-    catalog.createTempView(name.table, logicalPlan, replace)
+    catalog.createTempView(name.table, logicalPlan, mode == SaveMode.Overwrite)
   }
 
   /**
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index 54e27b6f73502..643cd74d53b5d 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -468,13 +468,25 @@ class HiveDDLCommandSuite extends PlanTest {
   test("create view -- basic") {
     val v1 = "CREATE VIEW view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
-    assert(!command.allowExisting)
+    assert(command.mode == SaveMode.ErrorIfExists)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.originalText == Some("SELECT * FROM tab1"))
     assert(command.userSpecifiedColumns.isEmpty)
   }
 
+  test("create view -- IF NOT EXISTS") {
+    val v1 = "CREATE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
+    val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
+    assert(command.mode == SaveMode.Ignore)
+
+    val v2 = "CREATE OR REPLACE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
+    val e = intercept[ParseException] {
+      parser.parsePlan(v2).asInstanceOf[CreateViewCommand]
+    }.getMessage
+    assert(e.contains("CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed"))
+  }
+
   test("create view - full") {
     val v1 =
       """
@@ -485,6 +497,7 @@ class HiveDDLCommandSuite extends PlanTest {
         |AS SELECT * FROM tab1
       """.stripMargin
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
+    assert(command.mode == SaveMode.Overwrite)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.userSpecifiedColumns == Seq("col1" -> None, "col3" -> Some("hello")))

From 47fdf1f217d4ea53e01827877620616be4d0efee Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Sat, 27 Aug 2016 23:32:28 -0700
Subject: [PATCH 5/7] add ViewType

---
 .../java/org/apache/spark/sql/ViewType.java   | 39 +++++++++++++++++++
 .../scala/org/apache/spark/sql/Dataset.scala  |  2 +-
 .../spark/sql/execution/SparkSqlParser.scala  |  5 ++-
 .../spark/sql/execution/command/views.scala   | 26 +++++--------
 .../spark/sql/hive/HiveDDLCommandSuite.scala  | 11 +++++-
 5 files changed, 63 insertions(+), 20 deletions(-)
 create mode 100644 sql/core/src/main/java/org/apache/spark/sql/ViewType.java

diff --git a/sql/core/src/main/java/org/apache/spark/sql/ViewType.java b/sql/core/src/main/java/org/apache/spark/sql/ViewType.java
new file mode 100644
index 0000000000000..6bb00f1175359
--- /dev/null
+++ b/sql/core/src/main/java/org/apache/spark/sql/ViewType.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql;
+
+/**
+ * ViewType is used to specify the type of views.
+ */
+public enum ViewType {
+  /**
+   * Temporary means local temporary views. The views are session-scoped and automatically dropped
+   * when the session terminates. Do not qualify a temporary table with a schema name. Existing
+   * permanent tables or views with the same name are not visible while the temporary view exists,
+   * unless they are referenced with schema-qualified names.
+   */
+  Temporary,
+  /**
+   * Permanent means global permanent views. The views are global-scoped and accessible by all
+   * sessions. The permanent views stays until they are explicitly dropped.
+   */
+  Permanent,
+  /**
+   * Any means the view type is unknown.
+   */
+  Any
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
index 8b665e8d31cb2..88633da09448c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
@@ -2452,7 +2452,7 @@ class Dataset[T] private[sql](
       originalText = None,
       child = logicalPlan,
       if (replace) SaveMode.Overwrite else SaveMode.ErrorIfExists,
-      isTemporary = true)
+      ViewType.Temporary)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index f4a26a0522d3c..6649a87f9fb2c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
 import org.antlr.v4.runtime.{ParserRuleContext, Token}
 import org.antlr.v4.runtime.tree.TerminalNode
 
-import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.{SaveMode, ViewType}
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.parser._
@@ -1307,6 +1307,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       case _ => throw new ParseException(
         "CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed.", ctx)
     }
+    val viewType = if (isTemporary) ViewType.Temporary else ViewType.Any
     val originalText = source(query)
 
     CreateViewCommand(
@@ -1317,7 +1318,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       Some(originalText),
       plan(query),
       mode,
-      isTemporary = isTemporary,
+      viewType,
       isAlterViewAsSelect = isAlterViewAsSelect)
   }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index cdca8a9fdbed2..e79d9ebe54b8f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.command
 
 import scala.util.control.NonFatal
 
-import org.apache.spark.sql.{AnalysisException, Row, SaveMode, SparkSession}
+import org.apache.spark.sql.{AnalysisException, Row, SaveMode, SparkSession, ViewType}
 import org.apache.spark.sql.catalyst.{SQLBuilder, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute}
@@ -41,16 +41,13 @@ import org.apache.spark.sql.types.StructType
  *                     Dataset API.
  * @param child the logical plan that represents the view; this is used to generate a canonicalized
  *              version of the SQL that can be saved in the catalog.
- * @param mode only three modes are applicable here: Ignore, Overwrite and ErrorIfExists. If the
+ * @param mode only three modes are supported here: Ignore, Overwrite and ErrorIfExists. If the
  *             view already exists, CreateViewCommand behaves based on the mode:
  *             1) Overwrite: update the view;
  *             2) Ignore: noop;
  *             3) ErrorIfExists throws analysis exception.
- * @param isTemporary if true, the view is created as a temporary view. Temporary views are dropped
- *                 at the end of current Spark session. Existing permanent relations with the same
- *                 name are not visible to the current session while the temporary view exists,
- *                 unless they are specified with full qualified table name with database prefix.
-  * @param isAlterViewAsSelect if true, this original DDL command is ALTER VIEW AS SELECT
+ * @param viewType the type of this view.
+ * @param isAlterViewAsSelect if true, this original DDL command is ALTER VIEW AS SELECT
  */
 case class CreateViewCommand(
     name: TableIdentifier,
@@ -60,7 +57,7 @@ case class CreateViewCommand(
     originalText: Option[String],
     child: LogicalPlan,
     mode: SaveMode,
-    isTemporary: Boolean,
+    viewType: ViewType,
     isAlterViewAsSelect: Boolean = false)
   extends RunnableCommand {
 
@@ -74,19 +71,14 @@ case class CreateViewCommand(
   assert(mode != SaveMode.Append,
     "CREATE or ALTER VIEW can only use ErrorIfExists, Ignore or Overwrite.")
 
-  if (!isTemporary) {
-    require(originalText.isDefined,
-      "The table to created with CREATE VIEW must have 'originalText'.")
-  }
-
   // Disallows 'CREATE TEMPORARY VIEW IF NOT EXISTS' to be consistent with 'CREATE TEMPORARY TABLE'
-  if (mode == SaveMode.Ignore && isTemporary) {
+  if (mode == SaveMode.Ignore && viewType == ViewType.Temporary) {
     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) {
+  if (viewType == ViewType.Temporary && 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.")
@@ -112,7 +104,9 @@ case class CreateViewCommand(
     // 2) ALTER VIEW: alter the temporary view if the temp view exists; otherwise, try to alter
     //                the permanent view. Here, it follows the same resolution like DROP VIEW,
     //                since users are unable to specify the keyword TEMPORARY.
-    if (isTemporary || (isAlterViewAsSelect && sessionState.catalog.isTemporaryTable(name))) {
+    if (viewType == ViewType.Temporary ||
+        (viewType != ViewType.Permanent && isAlterViewAsSelect &&
+          sessionState.catalog.isTemporaryTable(name))) {
       createTemporaryView(sparkSession, analyzedPlan)
     } else {
       // Adds default database for permanent table if it doesn't exist, so that tableExists()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index 643cd74d53b5d..f691d7e9aa429 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.hive
 
-import org.apache.spark.sql.{AnalysisException, SaveMode}
+import org.apache.spark.sql.{AnalysisException, SaveMode, ViewType}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType}
@@ -469,6 +469,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE VIEW view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.ErrorIfExists)
+    assert(command.viewType == ViewType.Any)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.originalText == Some("SELECT * FROM tab1"))
@@ -479,6 +480,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Ignore)
+    assert(command.viewType == ViewType.Any)
 
     val v2 = "CREATE OR REPLACE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
     val e = intercept[ParseException] {
@@ -498,6 +500,7 @@ class HiveDDLCommandSuite extends PlanTest {
       """.stripMargin
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Overwrite)
+    assert(command.viewType == ViewType.Any)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.userSpecifiedColumns == Seq("col1" -> None, "col3" -> Some("hello")))
@@ -506,6 +509,12 @@ class HiveDDLCommandSuite extends PlanTest {
     assert(command.comment == Some("BLABLA"))
   }
 
+  test("create temporary view") {
+    val v1 = "CREATE TEMPORARY VIEW testView AS SELECT id FROM jt"
+    val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
+    assert(command.viewType == ViewType.Temporary)
+  }
+
   test("create view -- partitioned view") {
     val v1 = "CREATE VIEW view1 partitioned on (ds, hr) as select * from srcpart"
     intercept[ParseException] {

From 5606344584c1c45cf49d2a95158c55193e777a9e Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Sat, 27 Aug 2016 23:53:28 -0700
Subject: [PATCH 6/7] remove isAlterViewAsSelect

---
 .../spark/sql/execution/SparkSqlParser.scala  | 15 +++++------
 .../spark/sql/execution/command/views.scala   |  7 ++---
 .../spark/sql/hive/HiveDDLCommandSuite.scala  | 26 ++++++++++++++++---
 3 files changed, 31 insertions(+), 17 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 6649a87f9fb2c..ced88afd5e035 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -1254,6 +1254,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
           ic.identifier.getText -> Option(ic.STRING).map(string)
         }
       }
+      val viewType = if (ctx.TEMPORARY != null) ViewType.Temporary else ViewType.Permanent
       createView(
         ctx,
         ctx.tableIdentifier,
@@ -1263,8 +1264,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
         Option(ctx.tablePropertyList).map(visitPropertyKeyValues).getOrElse(Map.empty),
         ignoreIfExists = ctx.EXISTS != null,
         replace = ctx.REPLACE != null,
-        isTemporary = ctx.TEMPORARY != null,
-        isAlterViewAsSelect = false
+        viewType
       )
     }
   }
@@ -1282,8 +1282,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       properties = Map.empty,
       ignoreIfExists = false,
       replace = true,
-      isTemporary = false,
-      isAlterViewAsSelect = true)
+      ViewType.Any
+    )
   }
 
   /**
@@ -1298,8 +1298,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       properties: Map[String, String],
       ignoreIfExists: Boolean,
       replace: Boolean,
-      isTemporary: Boolean,
-      isAlterViewAsSelect: Boolean): LogicalPlan = {
+      viewType: ViewType): LogicalPlan = {
     val mode = (replace, ignoreIfExists) match {
       case (true, false) => SaveMode.Overwrite
       case (false, true) => SaveMode.Ignore
@@ -1307,7 +1306,6 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       case _ => throw new ParseException(
         "CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed.", ctx)
     }
-    val viewType = if (isTemporary) ViewType.Temporary else ViewType.Any
     val originalText = source(query)
 
     CreateViewCommand(
@@ -1318,8 +1316,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       Some(originalText),
       plan(query),
       mode,
-      viewType,
-      isAlterViewAsSelect = isAlterViewAsSelect)
+      viewType)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index e79d9ebe54b8f..eff976775af3c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -47,7 +47,6 @@ import org.apache.spark.sql.types.StructType
  *             2) Ignore: noop;
  *             3) ErrorIfExists throws analysis exception.
  * @param viewType the type of this view.
- * @param isAlterViewAsSelect if true, this original DDL command is ALTER VIEW AS SELECT
  */
 case class CreateViewCommand(
     name: TableIdentifier,
@@ -57,8 +56,7 @@ case class CreateViewCommand(
     originalText: Option[String],
     child: LogicalPlan,
     mode: SaveMode,
-    viewType: ViewType,
-    isAlterViewAsSelect: Boolean = false)
+    viewType: ViewType)
   extends RunnableCommand {
 
   override protected def innerChildren: Seq[QueryPlan[_]] = Seq(child)
@@ -105,8 +103,7 @@ case class CreateViewCommand(
     //                the permanent view. Here, it follows the same resolution like DROP VIEW,
     //                since users are unable to specify the keyword TEMPORARY.
     if (viewType == ViewType.Temporary ||
-        (viewType != ViewType.Permanent && isAlterViewAsSelect &&
-          sessionState.catalog.isTemporaryTable(name))) {
+        (viewType != ViewType.Permanent && sessionState.catalog.isTemporaryTable(name))) {
       createTemporaryView(sparkSession, analyzedPlan)
     } else {
       // Adds default database for permanent table if it doesn't exist, so that tableExists()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index f691d7e9aa429..771801964bf42 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -469,7 +469,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE VIEW view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.ErrorIfExists)
-    assert(command.viewType == ViewType.Any)
+    assert(command.viewType == ViewType.Permanent)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.originalText == Some("SELECT * FROM tab1"))
@@ -480,7 +480,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Ignore)
-    assert(command.viewType == ViewType.Any)
+    assert(command.viewType == ViewType.Permanent)
 
     val v2 = "CREATE OR REPLACE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
     val e = intercept[ParseException] {
@@ -500,7 +500,7 @@ class HiveDDLCommandSuite extends PlanTest {
       """.stripMargin
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Overwrite)
-    assert(command.viewType == ViewType.Any)
+    assert(command.viewType == ViewType.Permanent)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.userSpecifiedColumns == Seq("col1" -> None, "col3" -> Some("hello")))
@@ -513,6 +513,26 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE TEMPORARY VIEW testView AS SELECT id FROM jt"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.viewType == ViewType.Temporary)
+    assert(command.mode == SaveMode.ErrorIfExists)
+    assert(command.name.database.isEmpty)
+    assert(command.name.table == "testView")
+    assert(command.originalText == Option("SELECT id FROM jt"))
+    assert(command.userSpecifiedColumns.isEmpty)
+    assert(command.comment.isEmpty)
+    assert(command.properties.isEmpty)
+  }
+
+  test("alter view as select") {
+    val v1 = "ALTER VIEW testView AS SELECT id FROM jt"
+    val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
+    assert(command.mode == SaveMode.Overwrite)
+    assert(command.viewType == ViewType.Any)
+    assert(command.name.database.isEmpty)
+    assert(command.name.table == "testView")
+    assert(command.originalText == Option("SELECT id FROM jt"))
+    assert(command.userSpecifiedColumns.isEmpty)
+    assert(command.comment.isEmpty)
+    assert(command.properties.isEmpty)
   }
 
   test("create view -- partitioned view") {

From baa8a1feee78a354e719cfa35d5123d0dd38ad1f Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Mon, 29 Aug 2016 11:14:04 -0700
Subject: [PATCH 7/7] address comments

---
 .../java/org/apache/spark/sql/ViewType.java   | 39 --------
 .../scala/org/apache/spark/sql/Dataset.scala  |  6 +-
 .../spark/sql/execution/SparkSqlParser.scala  |  9 +-
 .../spark/sql/execution/command/views.scala   | 99 +++++++++++--------
 .../spark/sql/hive/HiveDDLCommandSuite.scala  | 12 +--
 5 files changed, 73 insertions(+), 92 deletions(-)
 delete mode 100644 sql/core/src/main/java/org/apache/spark/sql/ViewType.java

diff --git a/sql/core/src/main/java/org/apache/spark/sql/ViewType.java b/sql/core/src/main/java/org/apache/spark/sql/ViewType.java
deleted file mode 100644
index 6bb00f1175359..0000000000000
--- a/sql/core/src/main/java/org/apache/spark/sql/ViewType.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql;
-
-/**
- * ViewType is used to specify the type of views.
- */
-public enum ViewType {
-  /**
-   * Temporary means local temporary views. The views are session-scoped and automatically dropped
-   * when the session terminates. Do not qualify a temporary table with a schema name. Existing
-   * permanent tables or views with the same name are not visible while the temporary view exists,
-   * unless they are referenced with schema-qualified names.
-   */
-  Temporary,
-  /**
-   * Permanent means global permanent views. The views are global-scoped and accessible by all
-   * sessions. The permanent views stays until they are explicitly dropped.
-   */
-  Permanent,
-  /**
-   * Any means the view type is unknown.
-   */
-  Any
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
index 88633da09448c..c9cf5bbcdfe7b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
@@ -42,8 +42,8 @@ import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.util.usePrettyExpression
 import org.apache.spark.sql.execution.{FileRelation, LogicalRDD, QueryExecution, SQLExecution}
-import org.apache.spark.sql.execution.command.{CreateViewCommand, ExplainCommand}
-import org.apache.spark.sql.execution.datasources.{CreateTable, LogicalRelation}
+import org.apache.spark.sql.execution.command.{CreateViewCommand, ExplainCommand, TemporaryView}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.execution.datasources.json.JacksonGenerator
 import org.apache.spark.sql.execution.python.EvaluatePython
 import org.apache.spark.sql.streaming.{DataStreamWriter, StreamingQuery}
@@ -2452,7 +2452,7 @@ class Dataset[T] private[sql](
       originalText = None,
       child = logicalPlan,
       if (replace) SaveMode.Overwrite else SaveMode.ErrorIfExists,
-      ViewType.Temporary)
+      TemporaryView)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 6a0d1f463936c..c3c0628a13f50 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
 import org.antlr.v4.runtime.{ParserRuleContext, Token}
 import org.antlr.v4.runtime.tree.TerminalNode
 
-import org.apache.spark.sql.{SaveMode, ViewType}
+import org.apache.spark.sql.SaveMode
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.parser._
@@ -1254,7 +1254,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
           ic.identifier.getText -> Option(ic.STRING).map(string)
         }
       }
-      val viewType = if (ctx.TEMPORARY != null) ViewType.Temporary else ViewType.Permanent
+      val viewType = if (ctx.TEMPORARY != null) TemporaryView else PermanentView
       createView(
         ctx,
         ctx.tableIdentifier,
@@ -1270,7 +1270,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
   }
 
   /**
-   * Alter the query of a view. This creates a [[CreateViewCommand]] command.
+   * Alter the query of a view. This creates a [[CreateViewCommand]] command. The view type could be
+   * either temporary or permanent.
    */
   override def visitAlterViewQuery(ctx: AlterViewQueryContext): LogicalPlan = withOrigin(ctx) {
     createView(
@@ -1282,7 +1283,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
       properties = Map.empty,
       ignoreIfExists = false,
       replace = true,
-      ViewType.Any
+      AnyTypeView
     )
   }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index 1f90b17eac7a7..da0868dbf9901 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.command
 
 import scala.util.control.NonFatal
 
-import org.apache.spark.sql.{AnalysisException, Row, SaveMode, SparkSession, ViewType}
+import org.apache.spark.sql.{AnalysisException, Row, SaveMode, SparkSession}
 import org.apache.spark.sql.catalyst.{SQLBuilder, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute}
@@ -70,13 +70,13 @@ case class CreateViewCommand(
     "CREATE or ALTER VIEW can only use ErrorIfExists, Ignore or Overwrite.")
 
   // Disallows 'CREATE TEMPORARY VIEW IF NOT EXISTS' to be consistent with 'CREATE TEMPORARY TABLE'
-  if (mode == SaveMode.Ignore && viewType == ViewType.Temporary) {
+  if (mode == SaveMode.Ignore && viewType == TemporaryView) {
     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 (viewType == ViewType.Temporary && name.database.isDefined) {
+  if (viewType == TemporaryView && 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.")
@@ -96,45 +96,40 @@ case class CreateViewCommand(
     }
     val sessionState = sparkSession.sessionState
 
-    // 1) CREATE VIEW: create a temp view when users explicitly specify the keyword TEMPORARY;
-    //                 otherwise, create a permanent view no matter whether the temporary view
-    //                 with the same name exists or not.
-    // 2) ALTER VIEW: alter the temporary view if the temp view exists; otherwise, try to alter
-    //                the permanent view. Here, it follows the same resolution like DROP VIEW,
-    //                since users are unable to specify the keyword TEMPORARY.
-    if (viewType == ViewType.Temporary ||
-        (viewType != ViewType.Permanent && sessionState.catalog.isTemporaryTable(name))) {
-      createTemporaryView(sparkSession, analyzedPlan)
-    } else {
-      // Adds default database for permanent table if it doesn't exist, so that tableExists()
-      // only check permanent tables.
-      val database = name.database.getOrElse(sessionState.catalog.getCurrentDatabase)
-      val qualifiedName = name.copy(database = Option(database))
-
-      if (sessionState.catalog.tableExists(qualifiedName)) {
-        val tableMetadata = sessionState.catalog.getTableMetadata(qualifiedName)
-        if (mode == SaveMode.Ignore) {
-          // Handles `CREATE VIEW IF NOT EXISTS v0 AS SELECT ...`. Does nothing when the target view
-          // already exists.
-        } else if (tableMetadata.tableType != CatalogTableType.VIEW) {
-          throw new AnalysisException(
-            "Existing table is not a view. The following is an existing table, " +
-              s"not a view: $qualifiedName")
-        } else if (mode == SaveMode.Overwrite) {
-          // Handles `CREATE OR REPLACE VIEW v0 AS SELECT ...`
-          sessionState.catalog.alterTable(prepareTable(sparkSession, analyzedPlan))
+    viewType match {
+      case TemporaryView =>
+        createTemporaryView(sparkSession, analyzedPlan)
+      case AnyTypeView if sessionState.catalog.isTemporaryTable(name) =>
+        createTemporaryView(sparkSession, analyzedPlan)
+      case _ =>
+        // Adds default database for permanent table if it doesn't exist, so that tableExists()
+        // only check permanent tables.
+        val database = name.database.getOrElse(sessionState.catalog.getCurrentDatabase)
+        val qualifiedName = name.copy(database = Option(database))
+        if (sessionState.catalog.tableExists(qualifiedName)) {
+          val tableMetadata = sessionState.catalog.getTableMetadata(qualifiedName)
+          if (mode == SaveMode.Ignore) {
+            // Handles `CREATE VIEW IF NOT EXISTS v0 AS SELECT ...`. Does nothing when the target
+            // view already exists.
+          } else if (tableMetadata.tableType != CatalogTableType.VIEW) {
+            throw new AnalysisException(
+              "Existing table is not a view. The following is an existing table, " +
+                s"not a view: $qualifiedName")
+          } else if (mode == SaveMode.Overwrite) {
+            // Handles `CREATE OR REPLACE VIEW v0 AS SELECT ...`
+            sessionState.catalog.alterTable(prepareTable(sparkSession, analyzedPlan))
+          } else {
+            // Handles `CREATE VIEW v0 AS SELECT ...`. Throws exception when the target view already
+            // exists.
+            throw new AnalysisException(
+              s"View $qualifiedName already exists. If you want to update the view definition, " +
+                "please use ALTER VIEW AS or CREATE OR REPLACE VIEW AS")
+          }
         } else {
-          // Handles `CREATE VIEW v0 AS SELECT ...`. Throws exception when the target view already
-          // exists.
-          throw new AnalysisException(
-            s"View $qualifiedName already exists. If you want to update the view definition, " +
-              "please use ALTER VIEW AS or CREATE OR REPLACE VIEW AS")
+          // Create the view if it doesn't exist.
+          sessionState.catalog.createTable(
+            prepareTable(sparkSession, analyzedPlan), ignoreIfExists = false)
         }
-      } else {
-        // Create the view if it doesn't exist.
-        sessionState.catalog.createTable(
-          prepareTable(sparkSession, analyzedPlan), ignoreIfExists = false)
-      }
     }
     Seq.empty[Row]
   }
@@ -200,3 +195,27 @@ case class CreateViewCommand(
     )
   }
 }
+
+/**
+ * The trait used to represent the type of a view.
+ */
+sealed trait ViewType
+
+/**
+ * Temporary means local temporary views. The views are session-scoped and automatically dropped
+ * when the session terminates. Do not qualify a temporary table with a schema name. Existing
+ * permanent tables or views with the same name are not visible while the temporary view exists,
+ * unless they are referenced with schema-qualified names.
+ */
+case object TemporaryView extends ViewType
+
+/**
+ * Permanent means global permanent views. The views are global-scoped and accessible by all
+ * sessions. The permanent views stays until they are explicitly dropped.
+ */
+case object PermanentView extends ViewType
+
+/**
+ * Any means the view type is unknown.
+ */
+case object AnyTypeView extends ViewType
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index 771801964bf42..1e42759f4be07 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.hive
 
-import org.apache.spark.sql.{AnalysisException, SaveMode, ViewType}
+import org.apache.spark.sql.{AnalysisException, SaveMode}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType}
@@ -469,7 +469,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE VIEW view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.ErrorIfExists)
-    assert(command.viewType == ViewType.Permanent)
+    assert(command.viewType == PermanentView)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.originalText == Some("SELECT * FROM tab1"))
@@ -480,7 +480,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "CREATE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Ignore)
-    assert(command.viewType == ViewType.Permanent)
+    assert(command.viewType == PermanentView)
 
     val v2 = "CREATE OR REPLACE VIEW IF NOT EXISTS view1 AS SELECT * FROM tab1"
     val e = intercept[ParseException] {
@@ -500,7 +500,7 @@ class HiveDDLCommandSuite extends PlanTest {
       """.stripMargin
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Overwrite)
-    assert(command.viewType == ViewType.Permanent)
+    assert(command.viewType == PermanentView)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "view1")
     assert(command.userSpecifiedColumns == Seq("col1" -> None, "col3" -> Some("hello")))
@@ -512,7 +512,7 @@ class HiveDDLCommandSuite extends PlanTest {
   test("create temporary view") {
     val v1 = "CREATE TEMPORARY VIEW testView AS SELECT id FROM jt"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
-    assert(command.viewType == ViewType.Temporary)
+    assert(command.viewType == TemporaryView)
     assert(command.mode == SaveMode.ErrorIfExists)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "testView")
@@ -526,7 +526,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val v1 = "ALTER VIEW testView AS SELECT id FROM jt"
     val command = parser.parsePlan(v1).asInstanceOf[CreateViewCommand]
     assert(command.mode == SaveMode.Overwrite)
-    assert(command.viewType == ViewType.Any)
+    assert(command.viewType == AnyTypeView)
     assert(command.name.database.isEmpty)
     assert(command.name.table == "testView")
     assert(command.originalText == Option("SELECT id FROM jt"))