Port HIVE-10415

Author: wangyum

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.hive.thriftserver
 import java.util.Locale
 import java.util.concurrent.atomic.AtomicBoolean
 
+import org.apache.hadoop.hive.common.ServerUtils
 import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars
 import org.apache.hive.service.cli.thrift.{ThriftBinaryCLIService, ThriftHttpCLIService}
@@ -101,6 +102,8 @@ object HiveThriftServer2 extends Logging {
       SparkSQLEnv.sqlContext.sessionState.newHadoopConf())
 
     try {
+      // Cleanup the scratch dir before starting
+      ServerUtils.cleanUpScratchDir(executionHive.conf)
       val server = new HiveThriftServer2(SparkSQLEnv.sqlContext)
       server.init(executionHive.conf)
       server.start()

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala

@@ -924,6 +924,39 @@ class SingleSessionSuite extends HiveThriftJdbcTest {
   }
 }
 
+class HiveThriftCleanUpScratchDirSuite extends HiveThriftJdbcTest{
+  var tempScratchDir: File = _
+
+  override protected def beforeAll(): Unit = {
+    tempScratchDir = Utils.createTempDir()
+    tempScratchDir.setWritable(true, false)
+    assert(tempScratchDir.list().isEmpty)
+    new File(tempScratchDir.getAbsolutePath + File.separator + "SPARK-31626").createNewFile()
+    assert(tempScratchDir.list().nonEmpty)
+    super.beforeAll()
+  }
+
+  override def mode: ServerMode.Value = ServerMode.binary
+
+  override protected def extraConf: Seq[String] =
+    s" --hiveconf ${ConfVars.HIVE_START_CLEANUP_SCRATCHDIR}=true " ::
+       s"--hiveconf ${ConfVars.SCRATCHDIR}=${tempScratchDir.getAbsolutePath}" :: Nil
+
+  test("Cleanup the Hive scratchdir when starting the Hive Server") {
+    assert(!tempScratchDir.exists())
+    withJdbcStatement() { statement =>
+      val rs = statement.executeQuery("SELECT id FROM range(1)")
+      assert(rs.next())
+      assert(rs.getLong(1) === 0L)
+    }
+  }
+
+  override protected def afterAll(): Unit = {
+    Utils.deleteRecursively(tempScratchDir)
+    super.afterAll()
+  }
+}
+
 class HiveThriftHttpServerSuite extends HiveThriftJdbcTest {
   override def mode: ServerMode.Value = ServerMode.http