[SPARK-17338][SQL][FOLLOW-UP] add global temp view

## What changes were proposed in this pull request?

address post hoc review comments for https://github.com/apache/spark/pull/14897

## How was this patch tested?

N/A

Author: Wenchen Fan <wenchen@databricks.com>

Closes #15424 from cloud-fan/global-temp-view.

6 files changed, 24 insertions, 13 deletions

diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index e3d9a17469..ae72d37a0b 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -57,7 +57,9 @@ object MimaExcludes {
       ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.tableExists"),
       ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.functionExists"),
       // [SPARK-17338][SQL] add global temp view
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.dropGlobalTempView")
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.dropGlobalTempView"),
+      ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.sql.catalog.Catalog.dropTempView"),
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.dropTempView")
     )
   }
 
diff --git a/python/pyspark/sql/catalog.py b/python/pyspark/sql/catalog.py
index df3bf4254d..a36d02e0db 100644
--- a/python/pyspark/sql/catalog.py
+++ b/python/pyspark/sql/catalog.py
@@ -169,6 +169,10 @@ class Catalog(object):
     def dropTempView(self, viewName):
         """Drops the local temporary view with the given view name in the catalog.
         If the view has been cached before, then it will also be uncached.
+        Returns true if this view is dropped successfully, false otherwise.
+
+        Note that, the return type of this method was None in Spark 2.0, but changed to Boolean
+        in Spark 2.1.
 
         >>> spark.createDataFrame([(1, 1)]).createTempView("my_table")
         >>> spark.table("my_table").collect()
@@ -185,6 +189,7 @@ class Catalog(object):
     def dropGlobalTempView(self, viewName):
         """Drops the global temporary view with the given view name in the catalog.
         If the view has been cached before, then it will also be uncached.
+        Returns true if this view is dropped successfully, false otherwise.
 
         >>> spark.createDataFrame([(1, 1)]).createGlobalTempView("my_table")
         >>> spark.table("global_temp.my_table").collect()
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index e44e30ec64..5863c6a71c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -405,13 +405,17 @@ class SessionCatalog(
 
   /**
    * Drop a local temporary view.
+   *
+   * Returns true if this view is dropped successfully, false otherwise.
    */
-  def dropTempView(name: String): Unit = synchronized {
-    tempTables.remove(formatTableName(name))
+  def dropTempView(name: String): Boolean = synchronized {
+    tempTables.remove(formatTableName(name)).isDefined
   }
 
   /**
    * Drop a global temporary view.
+   *
+   * Returns true if this view is dropped successfully, false otherwise.
    */
   def dropGlobalTempView(name: String): Boolean = {
     globalTempViewManager.remove(formatTableName(name))
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 30349ba3cb..a7a84730a6 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
@@ -2494,7 +2494,7 @@ class Dataset[T] private[sql](
    * preserved database `_global_temp`, and we must use the qualified name to refer a global temp
    * view, e.g. `SELECT * FROM _global_temp.view1`.
    *
-   * @throws TempTableAlreadyExistsException if the view name already exists
+   * @throws AnalysisException if the view name already exists
    *
    * @group basic
    * @since 2.1.0
@@ -2508,12 +2508,7 @@ class Dataset[T] private[sql](
       viewName: String,
       replace: Boolean,
       global: Boolean): CreateViewCommand = {
-    val viewType = if (global) {
-      GlobalTempView
-    } else {
-      LocalTempView
-    }
-
+    val viewType = if (global) GlobalTempView else LocalTempView
     CreateViewCommand(
       name = sparkSession.sessionState.sqlParser.parseTableIdentifier(viewName),
       userSpecifiedColumns = Nil,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
index 717fb29190..18cba8ce28 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
@@ -269,10 +269,14 @@ abstract class Catalog {
    * created it, i.e. it will be automatically dropped when the session terminates. It's not
    * tied to any databases, i.e. we can't use `db1.view1` to reference a local temporary view.
    *
+   * Note that, the return type of this method was Unit in Spark 2.0, but changed to Boolean
+   * in Spark 2.1.
+   *
    * @param viewName the name of the view to be dropped.
+   * @return true if the view is dropped successfully, false otherwise.
    * @since 2.0.0
    */
-  def dropTempView(viewName: String): Unit
+  def dropTempView(viewName: String): Boolean
 
   /**
    * Drops the global temporary view with the given view name in the catalog.
@@ -284,6 +288,7 @@ abstract class Catalog {
    * view, e.g. `SELECT * FROM _global_temp.view1`.
    *
    * @param viewName the name of the view to be dropped.
+   * @return true if the view is dropped successfully, false otherwise.
    * @since 2.1.0
    */
   def dropGlobalTempView(viewName: String): Boolean
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
index c05bda3f1b..f6c297e91b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
@@ -371,8 +371,8 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
    * @group ddl_ops
    * @since 2.0.0
    */
-  override def dropTempView(viewName: String): Unit = {
-    sparkSession.sessionState.catalog.getTempView(viewName).foreach { tempView =>
+  override def dropTempView(viewName: String): Boolean = {
+    sparkSession.sessionState.catalog.getTempView(viewName).exists { tempView =>
       sparkSession.sharedState.cacheManager.uncacheQuery(Dataset.ofRows(sparkSession, tempView))
       sessionCatalog.dropTempView(viewName)
     }