[SC-5452] Do not assume that an ExternalCatalog is a HiveExternalCatalog

## What changes were proposed in this pull request?
The Spark extensions suite added a wrapper for the `ExternalCatalog`. This wrapper is injected into the `SharedState` when the spark extensions contain catalog hooks, and it sits between the `SharedState` and the actual `ExternalCatalog`. This - unfortunately - breaks `InsertIntoHiveTable` and a number tests, that assumes that the `ExternalCatalog` in the `SharedState` is always a `HiveExternalCatalog` in order to extract the `HiveClient`.

This PR fixes this problem by adding a utility method that properly extracts the `HiveClient` from the `SharedState`, and it updates all uses of the `spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client` with the new utility method.

## How was this patch tested?
Existing tests.

Author: Herman van Hovell <hvanhovell@databricks.com>

Closes apache#153 from hvanhovell/SC-5452.

Author: hvanhovell

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

@@ -36,13 +36,8 @@ private[hive] class HiveSessionState(sparkSession: SparkSession)
   /**
    * A Hive client used for interacting with the metastore.
    */
-  lazy val metadataHive: HiveClient = {
-    val catalog = sparkSession.sharedState.externalCatalog match {
-      case catalog: HiveExternalCatalog => catalog
-      case HookCallingExternalCatalog(catalog: HiveExternalCatalog) => catalog
-    }
-    catalog.client.newSession()
-  }
+  lazy val metadataHive: HiveClient =
+    HiveUtils.getHiveClient(sparkSession.sharedState).newSession()
 
   /**
    * Internal catalog for managing table and database states.

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

@@ -37,7 +37,7 @@ import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql._
 import org.apache.spark.sql.hive.client._
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{HookCallingExternalCatalog, SharedState, SQLConf}
 import org.apache.spark.sql.internal.SQLConf._
 import org.apache.spark.sql.internal.StaticSQLConf.{CATALOG_IMPLEMENTATION, WAREHOUSE_PATH}
 import org.apache.spark.sql.types._
@@ -455,4 +455,11 @@ private[spark] object HiveUtils extends Logging {
     case (decimal, DecimalType()) => decimal.toString
     case (other, tpe) if primitiveTypes contains tpe => other.toString
   }
+
+  /** Get the Hive client from a SharedState. */
+  def getHiveClient(state: SharedState): HiveClient = state.externalCatalog match {
+    case c: HiveExternalCatalog => c.client
+    case HookCallingExternalCatalog(c: HiveExternalCatalog) => c.client
+    case _ => sys.error(s"Cannot extract Hive client from $state")
+  }
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

@@ -131,7 +131,7 @@ case class InsertIntoHiveTable(
   def getExternalTmpPath(path: Path): Path = {
     import org.apache.spark.sql.hive.client.hive._
 
-    val hiveVersion = externalCatalog.asInstanceOf[HiveExternalCatalog].client.version
+    val hiveVersion = HiveUtils.getHiveClient(sqlContext.sharedState).version
     // Before Hive 1.1, when inserting into a table, Hive will create the staging directory under
     // a common scratch directory. After the writing is finished, Hive will simply empty the table
     // directory and move the staging directory to it.

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala

@@ -589,7 +589,7 @@ class HiveDDLCommandSuite extends PlanTest with SQLTestUtils with TestHiveSingle
   }
 
   test("SPARK-15887: hive-site.xml should be loaded") {
-    val hiveClient = spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+    val hiveClient = HiveUtils.getHiveClient(spark.sharedState)
     assert(hiveClient.getConf("hive.in.test", "") == "true")
   }
 }

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogBackwardCompatibilitySuite.scala

@@ -35,8 +35,7 @@ class HiveExternalCatalogBackwardCompatibilitySuite extends QueryTest
   with SQLTestUtils with TestHiveSingleton with BeforeAndAfterEach {
 
   // To test `HiveExternalCatalog`, we need to read/write the raw table meta from/to hive client.
-  val hiveClient: HiveClient =
-    spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+  val hiveClient: HiveClient = HiveUtils.getHiveClient(spark.sharedState)
 
   val tempDir = Utils.createTempDir().getCanonicalFile
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveSparkSubmitSuite.scala

@@ -396,8 +396,7 @@ object SetMetastoreURLTest extends Logging {
     }
 
     // HiveExternalCatalog is used when Hive support is enabled.
-    val actualMetastoreURL =
-      spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+    val actualMetastoreURL = HiveUtils.getHiveClient(spark.sharedState)
         .getConf("javax.jdo.option.ConnectionURL", "this_is_a_wrong_URL")
     logInfo(s"javax.jdo.option.ConnectionURL is $actualMetastoreURL")
 
@@ -823,7 +822,7 @@ object SPARK_18360 {
     val defaultDbLocation = spark.catalog.getDatabase("default").locationUri
     assert(new Path(defaultDbLocation) == new Path(spark.sharedState.warehousePath))
 
-    val hiveClient = spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+    val hiveClient = HiveUtils.getHiveClient(spark.sharedState)
 
     try {
       val tableMeta = CatalogTable(

sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala

@@ -53,8 +53,7 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv
 
   // To test `HiveExternalCatalog`, we need to read the raw table metadata(schema, partition
   // columns and bucket specification are still in table properties) from hive client.
-  private def hiveClient: HiveClient =
-    sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+  private def hiveClient: HiveClient = HiveUtils.getHiveClient(sharedState)
 
   test("persistent JSON table") {
     withTable("jsonTable") {

sql/hive/src/test/scala/org/apache/spark/sql/hive/ShowCreateTableSuite.scala

@@ -294,7 +294,7 @@ class ShowCreateTableSuite extends QueryTest with SQLTestUtils with TestHiveSing
   }
 
   private def createRawHiveTable(ddl: String): Unit = {
-    hiveContext.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client.runSqlHive(ddl)
+    HiveUtils.getHiveClient(hiveContext.sharedState).runSqlHive(ddl)
   }
 
   private def checkCreateTable(table: String): Unit = {

sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala

@@ -285,7 +285,7 @@ class StatisticsSuite extends StatisticsCollectionTestBase with TestHiveSingleto
       sql(s"analyze table $tableName compute STATISTICS FOR COLUMNS " + stats.keys.mkString(", "))
 
       // Validate statistics
-      val hiveClient = spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+      val hiveClient = HiveUtils.getHiveClient(spark.sharedState)
       val table = hiveClient.getTable("default", tableName)
 
       val props = table.properties.filterKeys(_.startsWith("spark.sql.statistics.colStats"))

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, TableAl
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.execution.command.DDLUtils
-import org.apache.spark.sql.hive.HiveExternalCatalog
+import org.apache.spark.sql.hive.HiveUtils
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.StaticSQLConf.CATALOG_IMPLEMENTATION
@@ -990,7 +990,7 @@ class HiveDDLSuite
     val indexName = tabName + "_index"
     withTable(tabName) {
       // Spark SQL does not support creating index. Thus, we have to use Hive client.
-      val client = spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+      val client = HiveUtils.getHiveClient(spark.sharedState)
       sql(s"CREATE TABLE $tabName(a int)")
 
       try {
@@ -1020,7 +1020,7 @@ class HiveDDLSuite
     val tabName = "tab1"
     withTable(tabName) {
       // Spark SQL does not support creating skewed table. Thus, we have to use Hive client.
-      val client = spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+      val client = HiveUtils.getHiveClient(spark.sharedState)
       client.runSqlHive(
         s"""
            |CREATE Table $tabName(col1 int, col2 int)

sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcSourceSuite.scala

@@ -22,7 +22,7 @@ import java.io.File
 import org.scalatest.BeforeAndAfterAll
 
 import org.apache.spark.sql.{QueryTest, Row}
-import org.apache.spark.sql.hive.HiveExternalCatalog
+import org.apache.spark.sql.hive.HiveUtils
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.sources._
 import org.apache.spark.sql.types._
@@ -153,7 +153,7 @@ abstract class OrcSuite extends QueryTest with TestHiveSingleton with BeforeAndA
   }
 
   test("SPARK-18220: read Hive orc table with varchar column") {
-    val hiveClient = spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+    val hiveClient = HiveUtils.getHiveClient(spark.sharedState)
     try {
       hiveClient.runSqlHive("CREATE TABLE orc_varchar(a VARCHAR(10)) STORED AS orc")
       hiveClient.runSqlHive("INSERT INTO TABLE orc_varchar SELECT 'a' FROM (SELECT 1) t")