Fix tests

The biggest change here is moving HiveMetastoreCatalog from
HiveCatalog (the external one) to HiveSessionCatalog (the session
specific one). This is needed because HMC depends on a lot of
session specific things for, e.g. creating data source tables.
This was failing tests that do things with multiple sessions,
i.e. HiveQuerySuite.

Author: Andrew Or

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

@@ -34,7 +34,7 @@ class UnresolvedException[TreeType <: TreeNode[_]](tree: TreeType, function: Str
   errors.TreeNodeException(tree, s"Invalid call to $function on unresolved object", null)
 
 /**
- * Holds the name of a relation that has yet to be looked up in a [[Catalog]].
+ * Holds the name of a relation that has yet to be looked up in a catalog.
  */
 case class UnresolvedRelation(
     tableIdentifier: TableIdentifier,

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

@@ -201,17 +201,17 @@ class InMemoryCatalog extends ExternalCatalog {
   }
 
   override def listTables(db: String): Seq[String] = synchronized {
-    requireDbExists(db)
-    catalog(db).tables.keySet.toSeq
+    if (databaseExists(db)) {
+      catalog(db).tables.keySet.toSeq
+    } else {
+      Seq()
+    }
   }
 
   override def listTables(db: String, pattern: String): Seq[String] = synchronized {
-    requireDbExists(db)
     filterPattern(listTables(db), pattern)
   }
 
-  override def refreshTable(db: String, table: String): Unit = { /* no-op */ }
-
   // --------------------------------------------------------------------------
   // Partitions
   // --------------------------------------------------------------------------

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

@@ -41,7 +41,13 @@ class SessionCatalog(externalCatalog: ExternalCatalog, caseSensitiveAnalysis: Bo
   // specify the database (e.g. DROP TABLE my_table). In these cases we must first
   // check whether the temporary table or function exists, then, if not, operate on
   // the corresponding item in the current database.
-  protected[this] var currentDb = "default"
+  protected[this] var currentDb = {
+    val defaultName = "default"
+    val defaultDbDefinition = CatalogDatabase(defaultName, "default database", "", Map())
+    // Initialize default database if it doesn't already exist
+    createDatabase(defaultDbDefinition, ignoreIfExists = true)
+    defaultName
+  }
 
   /**
    * Format table name, taking into account case sensitivity.
@@ -239,7 +245,7 @@ class SessionCatalog(externalCatalog: ExternalCatalog, caseSensitiveAnalysis: Bo
     if (name.database.isDefined || !tempTables.containsKey(table)) {
       externalCatalog.tableExists(db, table)
     } else {
-      true
+      true // it's a temporary table
     }
   }
 
@@ -264,11 +270,7 @@ class SessionCatalog(externalCatalog: ExternalCatalog, caseSensitiveAnalysis: Bo
   /**
    * Refresh the cache entry for a metastore table, if any.
    */
-  def refreshTable(name: TableIdentifier): Unit = {
-    val db = name.database.getOrElse(currentDb)
-    val table = formatTableName(name.table)
-    externalCatalog.refreshTable(db, table)
-  }
+  def refreshTable(name: TableIdentifier): Unit = { /* no-op */ }
 
   /**
    * Return a temporary table exactly as it was stored.

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

@@ -97,8 +97,6 @@ abstract class ExternalCatalog {
 
   def listTables(db: String, pattern: String): Seq[String]
 
-  def refreshTable(db: String, table: String)
-
   // --------------------------------------------------------------------------
   // Partitions
   // --------------------------------------------------------------------------

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala

@@ -161,14 +161,10 @@ class AnalysisSuite extends AnalysisTest {
   }
 
   test("resolve relations") {
-    assertAnalysisError(
-      UnresolvedRelation(TableIdentifier("tAbLe"), None), Seq("Table not found: tAbLe"))
-
+    assertAnalysisError(UnresolvedRelation(TableIdentifier("tAbLe"), None), Seq())
     checkAnalysis(UnresolvedRelation(TableIdentifier("TaBlE"), None), testRelation)
-
     checkAnalysis(
       UnresolvedRelation(TableIdentifier("tAbLe"), None), testRelation, caseSensitive = false)
-
     checkAnalysis(
       UnresolvedRelation(TableIdentifier("TaBlE"), None), testRelation, caseSensitive = false)
   }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveCatalog.scala

@@ -27,24 +27,16 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.NoSuchItemException
 import org.apache.spark.sql.catalyst.catalog._
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources.BucketSpec
 import org.apache.spark.sql.hive.client.HiveClient
-import org.apache.spark.sql.types.StructType
 
 
 /**
  * A persistent implementation of the system catalog using Hive.
  * All public methods must be synchronized for thread-safety.
  */
-private[spark] class HiveCatalog(client: HiveClient) extends ExternalCatalog with Logging {
+private[spark] class HiveCatalog(var client: HiveClient) extends ExternalCatalog with Logging {
   import ExternalCatalog._
 
-  // Legacy catalog for handling data source tables.
-  // TODO: integrate this in a better way; it's confusing to have a catalog in a catalog.
-  private var metastoreCatalog: HiveMetastoreCatalog = _
-
   // Exceptions thrown by the hive client that we would like to wrap
   private val clientExceptions = Set(
     classOf[HiveException].getCanonicalName,
@@ -93,19 +85,14 @@ private[spark] class HiveCatalog(client: HiveClient) extends ExternalCatalog wit
     withClient { getTable(db, table) }
   }
 
-  private def requireInitialized(): Unit = {
-    require(metastoreCatalog != null, "catalog not yet initialized!")
-  }
-
   /**
-   * Initialize [[HiveMetastoreCatalog]] when the [[HiveContext]] is ready.
-   * This is needed to avoid initialization order cycles with [[HiveContext]].
+   * Switch our client to one that belongs to the new session.
    */
-  def initialize(hiveContext: HiveContext): Unit = {
-    metastoreCatalog = new HiveMetastoreCatalog(client, hiveContext)
+  def newSession(newClient: HiveClient): this.type = {
+    client = newClient
+    this
   }
 
-
   // --------------------------------------------------------------------------
   // Databases
   // --------------------------------------------------------------------------
@@ -207,19 +194,13 @@ private[spark] class HiveCatalog(client: HiveClient) extends ExternalCatalog wit
   }
 
   override def listTables(db: String): Seq[String] = withClient {
-    requireDbExists(db)
     client.listTables(db)
   }
 
   override def listTables(db: String, pattern: String): Seq[String] = withClient {
-    requireDbExists(db)
     client.listTables(db, pattern)
   }
 
-  override def refreshTable(db: String, table: String): Unit = {
-    refreshTable(TableIdentifier(table, Some(db)))
-  }
-
   // --------------------------------------------------------------------------
   // Partitions
   // --------------------------------------------------------------------------
@@ -319,69 +300,4 @@ private[spark] class HiveCatalog(client: HiveClient) extends ExternalCatalog wit
     client.listFunctions(db, pattern)
   }
 
-
-  // ----------------------------------------------------------------
-  // | Methods and fields for interacting with HiveMetastoreCatalog |
-  // ----------------------------------------------------------------
-
-  lazy val ParquetConversions: Rule[LogicalPlan] = {
-    requireInitialized()
-    metastoreCatalog.ParquetConversions
-  }
-
-  lazy val CreateTables: Rule[LogicalPlan] = {
-    requireInitialized()
-    metastoreCatalog.CreateTables
-  }
-
-  lazy val PreInsertionCasts: Rule[LogicalPlan] = {
-    requireInitialized()
-    metastoreCatalog.PreInsertionCasts
-  }
-
-  def refreshTable(table: TableIdentifier): Unit = {
-    requireInitialized()
-    metastoreCatalog.refreshTable(table)
-  }
-
-  def invalidateTable(table: TableIdentifier): Unit = {
-    requireInitialized()
-    metastoreCatalog.invalidateTable(table)
-  }
-
-  def invalidateCache(): Unit = {
-    requireInitialized()
-    metastoreCatalog.cachedDataSourceTables.invalidateAll()
-  }
-
-  def createDataSourceTable(
-      table: TableIdentifier,
-      userSpecifiedSchema: Option[StructType],
-      partitionColumns: Array[String],
-      bucketSpec: Option[BucketSpec],
-      provider: String,
-      options: Map[String, String],
-      isExternal: Boolean): Unit = {
-    requireInitialized()
-    metastoreCatalog.createDataSourceTable(
-      table, userSpecifiedSchema, partitionColumns, bucketSpec, provider, options, isExternal)
-  }
-
-  def lookupRelation(table: TableIdentifier, alias: Option[String]): LogicalPlan = {
-    requireInitialized()
-    metastoreCatalog.lookupRelation(table, alias)
-  }
-
-  def hiveDefaultTableFilePath(table: TableIdentifier): String = {
-    requireInitialized()
-    metastoreCatalog.hiveDefaultTableFilePath(table)
-  }
-
-  // For testing only
-  private[hive] def getCachedDataSourceTable(table: TableIdentifier): LogicalPlan = {
-    requireInitialized()
-    val key = metastoreCatalog.getQualifiedTableName(table)
-    metastoreCatalog.cachedDataSourceTables.getIfPresent(key)
-  }
-
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala

@@ -114,23 +114,22 @@ class HiveContext private[hive](
 
   logDebug("create HiveContext")
 
-  // Initialize catalog after context creation to avoid initialization ordering issues
-  hiveCatalog.initialize(this)
-
   /**
    * Returns a new HiveContext as new session, which will have separated SQLConf, UDF/UDAF,
    * temporary tables and SessionState, but sharing the same CacheManager, IsolatedClientLoader
    * and Hive client (both of execution and metadata) with existing HiveContext.
    */
   override def newSession(): HiveContext = {
+    val newExecutionHive = executionHive.newSession()
+    val newMetadataHive = metadataHive.newSession()
     new HiveContext(
       sc = sc,
       cacheManager = cacheManager,
       listener = listener,
-      executionHive = executionHive.newSession(),
-      metadataHive = metadataHive.newSession(),
+      executionHive = newExecutionHive,
+      metadataHive = newMetadataHive,
       isRootContext = false,
-      hiveCatalog = hiveCatalog)
+      hiveCatalog = hiveCatalog.newSession(newMetadataHive))
   }
 
   @transient
@@ -211,12 +210,12 @@ class HiveContext private[hive](
    */
   def refreshTable(tableName: String): Unit = {
     val tableIdent = sessionState.sqlParser.parseTableIdentifier(tableName)
-    hiveCatalog.refreshTable(tableIdent)
+    sessionState.sessionCatalog.refreshTable(tableIdent)
   }
 
   protected[hive] def invalidateTable(tableName: String): Unit = {
     val tableIdent = sessionState.sqlParser.parseTableIdentifier(tableName)
-    hiveCatalog.invalidateTable(tableIdent)
+    sessionState.sessionCatalog.invalidateTable(tableIdent)
   }
 
   /**

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionCatalog.scala

@@ -20,15 +20,29 @@ package org.apache.spark.sql.hive
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.SessionCatalog
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.BucketSpec
+import org.apache.spark.sql.hive.client.HiveClient
+import org.apache.spark.sql.types.StructType
 
 
-class HiveSessionCatalog(hiveCatalog: HiveCatalog) extends SessionCatalog(hiveCatalog) {
+class HiveSessionCatalog(
+    externalCatalog: HiveCatalog,
+    client: HiveClient,
+    context: HiveContext,
+    caseSensitiveAnalysis: Boolean)
+  extends SessionCatalog(externalCatalog, caseSensitiveAnalysis) {
+
+  override def setCurrentDatabase(db: String): Unit = {
+    super.setCurrentDatabase(db)
+    client.setCurrentDatabase(db)
+  }
 
   override def lookupRelation(name: TableIdentifier, alias: Option[String]): LogicalPlan = {
     val table = formatTableName(name.table)
     if (name.database.isDefined || !tempTables.containsKey(table)) {
       val newName = name.copy(table = table)
-      hiveCatalog.lookupRelation(newName, alias)
+      metastoreCatalog.lookupRelation(newName, alias)
     } else {
       val relation = tempTables.get(table)
       val tableWithQualifiers = SubqueryAlias(table, relation)
@@ -38,4 +52,52 @@ class HiveSessionCatalog(hiveCatalog: HiveCatalog) extends SessionCatalog(hiveCa
     }
   }
 
+  // ----------------------------------------------------------------
+  // | Methods and fields for interacting with HiveMetastoreCatalog |
+  // ----------------------------------------------------------------
+
+  // Catalog for handling data source tables. TODO: This really doesn't belong here since it is
+  // essentially a cache for metastore tables. However, it relies on a lot of session-specific
+  // things so it would be a lot of work to split its functionality between HiveSessionCatalog
+  // and HiveCatalog. We should still do it at some point...
+  private val metastoreCatalog = new HiveMetastoreCatalog(client, context)
+
+  val ParquetConversions: Rule[LogicalPlan] = metastoreCatalog.ParquetConversions
+  val CreateTables: Rule[LogicalPlan] = metastoreCatalog.CreateTables
+  val PreInsertionCasts: Rule[LogicalPlan] = metastoreCatalog.PreInsertionCasts
+
+  override def refreshTable(name: TableIdentifier): Unit = {
+    metastoreCatalog.refreshTable(name)
+  }
+
+  def invalidateTable(name: TableIdentifier): Unit = {
+    metastoreCatalog.invalidateTable(name)
+  }
+
+  def invalidateCache(): Unit = {
+    metastoreCatalog.cachedDataSourceTables.invalidateAll()
+  }
+
+  def createDataSourceTable(
+      name: TableIdentifier,
+      userSpecifiedSchema: Option[StructType],
+      partitionColumns: Array[String],
+      bucketSpec: Option[BucketSpec],
+      provider: String,
+      options: Map[String, String],
+      isExternal: Boolean): Unit = {
+    metastoreCatalog.createDataSourceTable(
+      name, userSpecifiedSchema, partitionColumns, bucketSpec, provider, options, isExternal)
+  }
+
+  def hiveDefaultTableFilePath(name: TableIdentifier): String = {
+    metastoreCatalog.hiveDefaultTableFilePath(name)
+  }
+
+  // For testing only
+  private[hive] def getCachedDataSourceTable(table: TableIdentifier): LogicalPlan = {
+    val key = metastoreCatalog.getQualifiedTableName(table)
+    metastoreCatalog.cachedDataSourceTables.getIfPresent(key)
+  }
+
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala

@@ -37,7 +37,10 @@ private[hive] class HiveSessionState(ctx: HiveContext) extends SessionState(ctx)
   /**
    * Internal catalog for managing table and database states.
    */
-  override lazy val sessionCatalog = new HiveSessionCatalog(ctx.hiveCatalog)
+  override lazy val sessionCatalog = {
+    new HiveSessionCatalog(
+      ctx.hiveCatalog, ctx.metadataHive, ctx, caseSensitiveAnalysis = false)
+  }
 
   /**
    * Internal catalog for managing functions registered by the user.
@@ -53,9 +56,9 @@ private[hive] class HiveSessionState(ctx: HiveContext) extends SessionState(ctx)
   override lazy val analyzer: Analyzer = {
     new Analyzer(sessionCatalog, functionRegistry, conf) {
       override val extendedResolutionRules =
-        ctx.hiveCatalog.ParquetConversions ::
-        ctx.hiveCatalog.CreateTables ::
-        ctx.hiveCatalog.PreInsertionCasts ::
+        sessionCatalog.ParquetConversions ::
+        sessionCatalog.CreateTables ::
+        sessionCatalog.PreInsertionCasts ::
         python.ExtractPythonUDFs ::
         PreInsertCastAndRename ::
         DataSourceAnalysis ::