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

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -98,6 +98,13 @@ package object config {
     .checkValues(Set("hive", "in-memory"))
     .createWithDefault("in-memory")
 
+  // Note: This is a SQL config but needs to be in core because it's cross-session and can not put
+  // in SQLConf.
+  private[spark] val GLOBAL_TEMP_DATABASE = ConfigBuilder("spark.sql.globalTempDatabase")
+    .internal()
+    .stringConf
+    .createWithDefault("global_temp")
+
   private[spark] val LISTENER_BUS_EVENT_QUEUE_SIZE =
     ConfigBuilder("spark.scheduler.listenerbus.eventqueue.size")
       .intConf

project/MimaExcludes.scala

@@ -46,7 +46,9 @@ object MimaExcludes {
       // [SPARK-16967] Move Mesos to Module
       ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.SparkMasterRegex.MESOS_REGEX"),
       // [SPARK-16240] ML persistence backward compatibility for LDA
-      ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.ml.clustering.LDA$")
+      ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.ml.clustering.LDA$"),
+      // [SPARK-17338][SQL] add global temp view
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.dropGlobalTempView")
     )
   }
 

python/pyspark/sql/context.py

@@ -386,7 +386,7 @@ def tables(self, dbName=None):
         >>> sqlContext.registerDataFrameAsTable(df, "table1")
         >>> df2 = sqlContext.tables()
         >>> df2.filter("tableName = 'table1'").first()
-        Row(tableName=u'table1', isTemporary=True)
+        Row(database=u'', tableName=u'table1', isTemporary=True)
         """
         if dbName is None:
             return DataFrame(self._ssql_ctx.tables(), self)

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -111,11 +111,12 @@ statement
     | ALTER TABLE tableIdentifier RECOVER PARTITIONS                   #recoverPartitions
     | DROP TABLE (IF EXISTS)? tableIdentifier PURGE?                   #dropTable
     | DROP VIEW (IF EXISTS)? tableIdentifier                           #dropTable
-    | CREATE (OR REPLACE)? TEMPORARY? VIEW (IF NOT EXISTS)? tableIdentifier
+    | CREATE (OR REPLACE)? (GLOBAL? TEMPORARY)?
+        VIEW (IF NOT EXISTS)? tableIdentifier
         identifierCommentList? (COMMENT STRING)?
         (PARTITIONED ON identifierList)?
         (TBLPROPERTIES tablePropertyList)? AS query                    #createView
-    | CREATE (OR REPLACE)? TEMPORARY VIEW
+    | CREATE (OR REPLACE)? GLOBAL? TEMPORARY VIEW
         tableIdentifier ('(' colTypeList ')')? tableProvider
         (OPTIONS tablePropertyList)?                                   #createTempViewUsing
     | ALTER VIEW tableIdentifier AS? query                             #alterViewQuery
@@ -669,7 +670,7 @@ nonReserved
     | MAP | ARRAY | STRUCT
     | LATERAL | WINDOW | REDUCE | TRANSFORM | USING | SERDE | SERDEPROPERTIES | RECORDREADER
     | DELIMITED | FIELDS | TERMINATED | COLLECTION | ITEMS | KEYS | ESCAPED | LINES | SEPARATED
-    | EXTENDED | REFRESH | CLEAR | CACHE | UNCACHE | LAZY | TEMPORARY | OPTIONS
+    | EXTENDED | REFRESH | CLEAR | CACHE | UNCACHE | LAZY | GLOBAL | TEMPORARY | OPTIONS
     | GROUPING | CUBE | ROLLUP
     | EXPLAIN | FORMAT | LOGICAL | FORMATTED | CODEGEN
     | TABLESAMPLE | USE | TO | BUCKET | PERCENTLIT | OUT | OF
@@ -857,6 +858,7 @@ CACHE: 'CACHE';
 UNCACHE: 'UNCACHE';
 LAZY: 'LAZY';
 FORMATTED: 'FORMATTED';
+GLOBAL: 'GLOBAL';
 TEMPORARY: 'TEMPORARY' | 'TEMP';
 OPTIONS: 'OPTIONS';
 UNSET: 'UNSET';

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

@@ -458,12 +458,12 @@ class Analyzer(
         i.copy(table = EliminateSubqueryAliases(lookupTableFromCatalog(u)))
       case u: UnresolvedRelation =>
         val table = u.tableIdentifier
-        if (table.database.isDefined && conf.runSQLonFile &&
+        if (table.database.isDefined && conf.runSQLonFile && !catalog.isTemporaryTable(table) &&
             (!catalog.databaseExists(table.database.get) || !catalog.tableExists(table))) {
-          // If the table does not exist, and the database part is specified, and we support
-          // running SQL directly on files, then let's just return the original UnresolvedRelation.
-          // It is possible we are matching a query like "select * from parquet.`/path/to/query`".
-          // The plan will get resolved later.
+          // If the database part is specified, and we support running SQL directly on files, and
+          // it's not a temporary view, and the table does not exist, then let's just return the
+          // original UnresolvedRelation. It is possible we are matching a query like "select *
+          // from parquet.`/path/to/query`". The plan will get resolved later.
           // Note that we are testing (!db_exists || !table_exists) because the catalog throws
           // an exception from tableExists if the database does not exist.
           u

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/GlobalTempViewManager.scala

@@ -0,0 +1,121 @@
+/*
+ * 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.catalyst.catalog
+
+import javax.annotation.concurrent.GuardedBy
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.TempTableAlreadyExistsException
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.StringUtils
+
+
+/**
+ * A thread-safe manager for global temporary views, providing atomic operations to manage them,
+ * e.g. create, update, remove, etc.
+ *
+ * Note that, the view name is always case-sensitive here, callers are responsible to format the
+ * view name w.r.t. case-sensitive config.
+ *
+ * @param database The system preserved virtual database that keeps all the global temporary views.
+ */
+class GlobalTempViewManager(val database: String) {
+
+  /** List of view definitions, mapping from view name to logical plan. */
+  @GuardedBy("this")
+  private val viewDefinitions = new mutable.HashMap[String, LogicalPlan]
+
+  /**
+   * Returns the global view definition which matches the given name, or None if not found.
+   */
+  def get(name: String): Option[LogicalPlan] = synchronized {
+    viewDefinitions.get(name)
+  }
+
+  /**
+   * Creates a global temp view, or issue an exception if the view already exists and
+   * `overrideIfExists` is false.
+   */
+  def create(
+      name: String,
+      viewDefinition: LogicalPlan,
+      overrideIfExists: Boolean): Unit = synchronized {
+    if (!overrideIfExists && viewDefinitions.contains(name)) {
+      throw new TempTableAlreadyExistsException(name)
+    }
+    viewDefinitions.put(name, viewDefinition)
+  }
+
+  /**
+   * Updates the global temp view if it exists, returns true if updated, false otherwise.
+   */
+  def update(
+      name: String,
+      viewDefinition: LogicalPlan): Boolean = synchronized {
+    if (viewDefinitions.contains(name)) {
+      viewDefinitions.put(name, viewDefinition)
+      true
+    } else {
+      false
+    }
+  }
+
+  /**
+   * Removes the global temp view if it exists, returns true if removed, false otherwise.
+   */
+  def remove(name: String): Boolean = synchronized {
+    viewDefinitions.remove(name).isDefined
+  }
+
+  /**
+   * Renames the global temp view if the source view exists and the destination view not exists, or
+   * issue an exception if the source view exists but the destination view already exists. Returns
+   * true if renamed, false otherwise.
+   */
+  def rename(oldName: String, newName: String): Boolean = synchronized {
+    if (viewDefinitions.contains(oldName)) {
+      if (viewDefinitions.contains(newName)) {
+        throw new AnalysisException(
+          s"rename temporary view from '$oldName' to '$newName': destination view already exists")
+      }
+
+      val viewDefinition = viewDefinitions(oldName)
+      viewDefinitions.remove(oldName)
+      viewDefinitions.put(newName, viewDefinition)
+      true
+    } else {
+      false
+    }
+  }
+
+  /**
+   * Lists the names of all global temporary views.
+   */
+  def listViewNames(pattern: String): Seq[String] = synchronized {
+    StringUtils.filterPattern(viewDefinitions.keys.toSeq, pattern)
+  }
+
+  /**
+   * Clears all the global temporary views.
+   */
+  def clear(): Unit = synchronized {
+    viewDefinitions.clear()
+  }
+}