set hive.query.reexecution.enabled=false

Author: wangyum

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -712,55 +712,47 @@ private[hive] class HiveClientImpl(
    * in the sequence is one row.
    * Since upgrading the built-in Hive to 2.3, hive-llap-client is needed when
    * running MapReduce jobs with `runHive`.
+   * Since HIVE-17626(Hive 3.0.0), need to set hive.query.reexecution.enabled=false.
    */
   protected def runHive(cmd: String, maxRows: Int = 1000): Seq[String] = withHiveState {
-    logDebug(s"Running hiveql '$cmd'")
-    if (cmd.toLowerCase(Locale.ROOT).startsWith("set")) { logDebug(s"Changing config: $cmd") }
-
-    def runProcessor(proc: CommandProcessor, tokens: Array[String], cmd_1: String): Seq[String] = {
-      if (state.out != null) {
-        // scalastyle:off println
-        state.out.println(tokens(0) + " " + cmd_1)
-        // scalastyle:on println
+    def closeDriver(driver: Driver): Unit = {
+      // Since HIVE-18238(Hive 3.0.0), the Driver.close function's return type changed
+      // and the CommandProcessorFactory.clean function removed.
+      driver.getClass.getMethod("close").invoke(driver)
+      if (version != hive.v3_1) {
+        CommandProcessorFactory.clean(conf)
       }
-      Seq(proc.run(cmd_1).getResponseCode.toString)
     }
 
+    logDebug(s"Running hiveql '$cmd'")
+    if (cmd.toLowerCase(Locale.ROOT).startsWith("set")) { logDebug(s"Changing config: $cmd") }
     try {
       val cmd_trimmed: String = cmd.trim()
       val tokens: Array[String] = cmd_trimmed.split("\\s+")
       // The remainder of the command.
       val cmd_1: String = cmd_trimmed.substring(tokens(0).length()).trim()
       val proc = shim.getCommandProcessor(tokens(0), conf)
       proc match {
-        // HIVE-18238(Hive 3.0.0) changed the close() function return type.
-        // This change is not compatible with the built-in Hive.
-        case driver: Driver if version != hive.v3_1 =>
+        case driver: Driver =>
           val response: CommandProcessorResponse = driver.run(cmd)
           // Throw an exception if there is an error in query processing.
           if (response.getResponseCode != 0) {
-            driver.close()
-            CommandProcessorFactory.clean(conf)
+            closeDriver(driver)
             throw new QueryExecutionException(response.getErrorMessage)
           }
           driver.setMaxRows(maxRows)
 
           val results = shim.getDriverResults(driver)
-          driver.close()
-          CommandProcessorFactory.clean(conf)
+          closeDriver(driver)
           results
 
-        case _: SetProcessor | _: ResetProcessor =>
-          runProcessor(proc, tokens, cmd_1)
-        case _: AddResourceProcessor | _: DeleteResourceProcessor | _: ListResourceProcessor =>
-          runProcessor(proc, tokens, cmd_1)
-        case _: CompileProcessor | _: CryptoProcessor | _: DfsProcessor | _: ReloadProcessor =>
-          runProcessor(proc, tokens, cmd_1)
-
-        case unsupportedProcessor =>
-          val className = unsupportedProcessor.getClass.getCanonicalName
-          throw new AnalysisException(
-            s"Dose not support Hive ${version.fullVersion} processor: $className")
+        case _ =>
+          if (state.out != null) {
+            // scalastyle:off println
+            state.out.println(tokens(0) + " " + cmd_1)
+            // scalastyle:on println
+          }
+          Seq(proc.run(cmd_1).getResponseCode.toString)
       }
     } catch {
       case e: Exception =>
@@ -872,20 +864,31 @@ private[hive] class HiveClientImpl(
   }
 
   def reset(): Unit = withHiveState {
-    client.getAllTables("default").asScala.foreach { t =>
-      logDebug(s"Deleting table $t")
-      val table = client.getTable("default", t)
+    val allTables = client.getAllTables("default")
+    val (mvs, others) = allTables.asScala.map(t => client.getTable("default", t))
+      .partition(_.getTableType.toString.equals("MATERIALIZED_VIEW"))
+
+    // Remove materialized view first, otherwise caused a violation of foreign key constraint.
+    mvs.foreach { table =>
+      val tableName = table.getTableName
+      logDebug(s"Deleting materialized view $tableName")
+      client.dropTable("default", table.getTableName)
+    }
+
+    others.foreach { table =>
+      val tableName = table.getTableName
+      logDebug(s"Deleting table $tableName")
       try {
-        client.getIndexes("default", t, 255).asScala.foreach { index =>
-          shim.dropIndex(client, "default", t, index.getIndexName)
+        client.getIndexes("default", tableName, 255).asScala.foreach { index =>
+          shim.dropIndex(client, "default", tableName, index.getIndexName)
         }
         if (!table.isIndexTable) {
-          client.dropTable("default", t)
+          client.dropTable("default", tableName)
         }
       } catch {
         case _: NoSuchMethodError =>
           // HIVE-18448 Hive 3.0 remove index APIs
-          client.dropTable("default", t)
+          client.dropTable("default", tableName)
       }
     }
     client.getAllDatabases.asScala.filterNot(_ == "default").foreach { db =>

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/Hive_3_1_ClientSuite.scala

@@ -123,9 +123,11 @@ class VersionsSuite extends SparkFunSuite with Logging {
         hadoopConf.set("datanucleus.schema.autoCreateAll", "true")
         hadoopConf.set("hive.metastore.schema.verification", "false")
       }
-      // Since Hive 3.0, HIVE-19310 skipped `ensureDbInit` if `hive.in.test=false`.
       if (version == "3.1") {
+        // Since Hive 3.0, HIVE-19310 skipped `ensureDbInit` if `hive.in.test=false`.
         hadoopConf.set("hive.in.test", "true")
+        // Since HIVE-17626(Hive 3.0.0), need to set hive.query.reexecution.enabled=false.
+        hadoopConf.set("hive.query.reexecution.enabled", "false")
       }
       client = buildClient(version, hadoopConf, HiveUtils.formatTimeVarsForHiveClient(hadoopConf))
       if (versionSpark != null) versionSpark.reset()
@@ -584,10 +586,11 @@ class VersionsSuite extends SparkFunSuite with Logging {
 
     test(s"$version: sql read hive materialized view") {
       // HIVE-14249 Since Hive 2.3.0, materialized view is supported.
-      // But skip Hive 3.1 because of SPARK-27074.
-      if (version == "2.3") {
+      if (version == "2.3" || version == "3.1") {
+        val disableRewrite = if (version == "2.3") "" else "DISABLE REWRITE"
         client.runSqlHive("CREATE TABLE materialized_view_tbl (c1 INT)")
-        client.runSqlHive("CREATE MATERIALIZED VIEW mv1 AS SELECT * FROM materialized_view_tbl")
+        client.runSqlHive(
+          s"CREATE MATERIALIZED VIEW mv1 $disableRewrite AS SELECT * FROM materialized_view_tbl")
         val e = intercept[AnalysisException](versionSpark.table("mv1").collect()).getMessage
         assert(e.contains("Hive materialized view is not supported"))
       }