[SPARK-23809][SQL] Active SparkSession should be set by getOrCreate

Currently, the active spark session is set inconsistently (e.g., in createDataFrame, prior to query execution). Many places in spark also incorrectly query active session when they should be calling activeSession.getOrElse(defaultSession) and so might get None even if a Spark session exists.

The semantics here can be cleaned up if we also set the active session when the default session is set.

Related: https://github.com/apache/spark/pull/20926/files

Unit test, existing test. Note that if apache#20926 merges first we should also update the tests there.

Author: Eric Liang <[email protected]>

Closes apache#20927 from ericl/active-session-cleanup.

(cherry picked from commit 359375e)

NOTE: This cherry-pick includes only some of the original changes, because LIHADOOP-54684 already made some identical
changes as part of test resolution. This now ensures the entirety of the original SPARK-23809 code is backported.
Also note that when SPARK-23809 was originally backported to branch-2.3 in PR apache#20971, the new `active` API was _not_
included since new APIs shouldn't generally be added in patch releases. That new API is exactly what is needed,
so we backport directly from the original commit.

RB=2575134
BUG=BDP-6088
G=spark-reviewers
R=mmuralid,wyzhang,smahadik
A=mmuralid,wyzhang

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -1042,6 +1042,17 @@ object SparkSession extends Logging {
    */
   def getDefaultSession: Option[SparkSession] = Option(defaultSession.get)
 
+  /**
+   * Returns the currently active SparkSession, otherwise the default one. If there is no default
+   * SparkSession, throws an exception.
+   *
+   * @since 2.4.0
+   */
+  def active: SparkSession = {
+    getActiveSession.getOrElse(getDefaultSession.getOrElse(
+      throw new IllegalStateException("No active or default Spark session found")))
+  }
+
   ////////////////////////////////////////////////////////////////////////////////////////
   // Private methods from now on
   ////////////////////////////////////////////////////////////////////////////////////////

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

@@ -58,6 +58,16 @@ class SparkSessionBuilderSuite extends SparkFunSuite with BeforeAndAfterEach {
     assert(SparkSession.getActiveSession == Some(session))
   }
 
+  test("get active or default session") {
+    val session = SparkSession.builder().master("local").getOrCreate()
+    assert(SparkSession.active == session)
+    SparkSession.clearActiveSession()
+    assert(SparkSession.active == session)
+    SparkSession.clearDefaultSession()
+    intercept[IllegalStateException](SparkSession.active)
+    session.stop()
+  }
+
   test("config options are propagated to existing SparkSession") {
     val session1 = SparkSession.builder().master("local").config("spark-config1", "a").getOrCreate()
     assert(session1.conf.get("spark-config1") == "a")