Make lazy vals strict.

sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala

@@ -48,13 +48,14 @@ private[sql] class SessionState(
     this(sparkSession, None)
   }
 
-  // Note: These are all lazy vals because they depend on each other (e.g. conf) and we
-  // want subclasses to override some of the fields. Otherwise, we would get a lot of NPEs.
+  // Note: Many of these vals depend on each other (e.g. conf) and should be initialized
+  // with an early initializer if we want subclasses to override some of the fields.
+  // Otherwise, we would get a lot of NPEs.
 
   /**
    * SQL-specific key-value configurations.
    */
-  lazy val conf: SQLConf = {
+  val conf: SQLConf = {
     parentSessionState.map(_.conf.copy).getOrElse(new SQLConf)
   }
 
@@ -74,7 +75,7 @@ private[sql] class SessionState(
     hadoopConf
   }
 
-  lazy val experimentalMethods: ExperimentalMethods = {
+  val experimentalMethods: ExperimentalMethods = {
     parentSessionState
       .map(_.experimentalMethods.copy)
       .getOrElse(new ExperimentalMethods)
@@ -83,14 +84,14 @@ private[sql] class SessionState(
   /**
    * Internal catalog for managing functions registered by the user.
    */
-  lazy val functionRegistry: FunctionRegistry = {
+  val functionRegistry: FunctionRegistry = {
     parentSessionState.map(_.functionRegistry.copy).getOrElse(FunctionRegistry.builtin.copy)
   }
 
   /**
    * A class for loading resources specified by a function.
    */
-  lazy val functionResourceLoader: FunctionResourceLoader = {
+  val functionResourceLoader: FunctionResourceLoader = {
     new FunctionResourceLoader {
       override def loadResource(resource: FunctionResource): Unit = {
         resource.resourceType match {
@@ -108,7 +109,7 @@ private[sql] class SessionState(
   /**
    * Internal catalog for managing table and database states.
    */
-  lazy val catalog: SessionCatalog = {
+  val catalog: SessionCatalog = {
     parentSessionState
       .map(_.catalog.copy)
       .getOrElse(new SessionCatalog(
@@ -125,12 +126,12 @@ private[sql] class SessionState(
    * Interface exposed to the user for registering user-defined functions.
    * Note that the user-defined functions must be deterministic.
    */
-  lazy val udf: UDFRegistration = new UDFRegistration(functionRegistry)
+  val udf: UDFRegistration = new UDFRegistration(functionRegistry)
 
   /**
    * Logical query plan analyzer for resolving unresolved attributes and relations.
    */
-  lazy val analyzer: Analyzer = {
+  val analyzer: Analyzer = {
     new Analyzer(catalog, conf) {
       override val extendedResolutionRules =
         new FindDataSourceTable(sparkSession) ::
@@ -149,12 +150,12 @@ private[sql] class SessionState(
   /**
    * Logical query plan optimizer.
    */
-  lazy val optimizer: Optimizer = new SparkOptimizer(catalog, conf, experimentalMethods)
+  val optimizer: Optimizer = new SparkOptimizer(catalog, conf, experimentalMethods)
 
   /**
    * Parser that extracts expressions, plans, table identifiers etc. from SQL texts.
    */
-  lazy val sqlParser: ParserInterface = new SparkSqlParser(conf)
+  val sqlParser: ParserInterface = new SparkSqlParser(conf)
 
   /**
    * Planner that converts optimized logical plans to physical plans.
@@ -166,12 +167,12 @@ private[sql] class SessionState(
    * An interface to register custom [[org.apache.spark.sql.util.QueryExecutionListener]]s
    * that listen for execution metrics.
    */
-  lazy val listenerManager: ExecutionListenerManager = new ExecutionListenerManager
+  val listenerManager: ExecutionListenerManager = new ExecutionListenerManager
 
   /**
    * Interface to start and stop [[StreamingQuery]]s.
    */
-  lazy val streamingQueryManager: StreamingQueryManager = {
+  val streamingQueryManager: StreamingQueryManager = {
     new StreamingQueryManager(sparkSession)
   }
 
@@ -187,8 +188,8 @@ private[sql] class SessionState(
   /**
    * Get an identical copy of the `SessionState` and associate it with the given `SparkSession`
    */
-  def copy(sc: SparkSession): SessionState = {
-    new SessionState(sc, Some(this))
+  def copy(associatedSparkSession: SparkSession): SessionState = {
+    new SessionState(associatedSparkSession, Some(this))
   }
 
   // ------------------------------------------------------

sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala

@@ -34,9 +34,8 @@ private[sql] class TestSparkSession(sc: SparkContext) extends SparkSession(sc) {
     this(new SparkConf)
   }
 
-  @transient
-  protected[sql] override lazy val sessionState: SessionState = new SessionState(self) {
-    override lazy val conf: SQLConf = {
+  class TestSessionState(sparkSession: SparkSession) extends {
+    override val conf: SQLConf = {
       new SQLConf {
         clear()
         override def clear(): Unit = {
@@ -46,7 +45,10 @@ private[sql] class TestSparkSession(sc: SparkContext) extends SparkSession(sc) {
         }
       }
     }
-  }
+  } with SessionState(sparkSession)
+
+  @transient
+  protected[sql] override lazy val sessionState: SessionState = new TestSessionState(this)
 
   // Needed for Java tests
   def loadTestData(): Unit = {

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

@@ -28,21 +28,29 @@ import org.apache.spark.sql.internal.SessionState
 /**
  * A class that holds all session-specific state in a given [[SparkSession]] backed by Hive.
  */
-private[hive] class HiveSessionState(sparkSession: SparkSession)
-  extends SessionState(sparkSession) {
+private[hive] class HiveSessionState(
+    sparkSession: SparkSession,
+    parentHiveSessionState: Option[HiveSessionState])
+  extends SessionState(sparkSession, parentHiveSessionState) {
 
   self =>
 
+  private[hive] def this(associatedSparkSession: SparkSession) = {
+    this(associatedSparkSession, None)
+  }
+
   /**
    * A Hive client used for interacting with the metastore.
    */
-  lazy val metadataHive: HiveClient =
-    sparkSession.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client.newSession()
+  val metadataHive: HiveClient =
+    parentHiveSessionState.map(_.metadataHive).getOrElse(
+      sparkSession.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
+        .newSession())
 
   /**
    * Internal catalog for managing table and database states.
    */
-  override lazy val catalog = {
+  override val catalog = {
     new HiveSessionCatalog(
       sparkSession.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog],
       sparkSession.sharedState.globalTempViewManager,
@@ -57,7 +65,7 @@ private[hive] class HiveSessionState(sparkSession: SparkSession)
   /**
    * An analyzer that uses the Hive metastore.
    */
-  override lazy val analyzer: Analyzer = {
+  override val analyzer: Analyzer = {
     new Analyzer(catalog, conf) {
       override val extendedResolutionRules =
         catalog.ParquetConversions ::
@@ -147,4 +155,8 @@ private[hive] class HiveSessionState(sparkSession: SparkSession)
     conf.getConf(HiveUtils.HIVE_THRIFT_SERVER_ASYNC)
   }
 
+  override def copy(associatedSparkSession: SparkSession): HiveSessionState = {
+    new HiveSessionState(associatedSparkSession, Some(this.asInstanceOf[HiveSessionState]))
+  }
+
 }