still detach SessionState

Author: AngersZhuuuu

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

@@ -294,6 +294,7 @@ private[hive] class HiveClientImpl(
   def withHiveState[A](f: => A): A = retryLocked {
     val original = Thread.currentThread().getContextClassLoader
     val originalConfLoader = state.getConf.getClassLoader
+    val originState = SessionState.get()
     // The classloader in clientLoader could be changed after addJar, always use the latest
     // classloader. We explicitly set the context class loader since "conf.setClassLoader" does
     // not do that, and the Hive client libraries may need to load classes defined by the client's
@@ -311,6 +312,9 @@ private[hive] class HiveClientImpl(
     val ret = try f finally {
       state.getConf.setClassLoader(originalConfLoader)
       Thread.currentThread().setContextClassLoader(original)
+      if (originState != null) {
+        SessionState.setCurrentSessionState(originState)
+      }
       HiveCatalogMetrics.incrementHiveClientCalls(1)
     }
     ret

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

@@ -87,8 +87,6 @@ case class HiveTableScanExec(
     BindReferences.bindReference(pred, relation.partitionCols)
   }
 
-  @transient private lazy val  hiveClient: HiveClient = sparkSession.sharedState.externalCatalog
-    .unwrapped.asInstanceOf[HiveExternalCatalog].client
   @transient private lazy val hiveQlTable = HiveClientImpl.toHiveTable(relation.tableMeta)
   @transient private lazy val tableDesc = new TableDesc(
     hiveQlTable.getInputFormatClass,
@@ -97,7 +95,7 @@ case class HiveTableScanExec(
 
   // Create a local copy of hadoopConf,so that scan specific modifications should not impact
   // other queries
-  @transient private lazy val hadoopConf = hiveClient.withHiveState {
+  @transient private lazy val hadoopConf = {
     val c = sparkSession.sessionState.newHadoopConf()
     // append columns ids and names before broadcast
     addColumnMetadataToConf(c)