[SPARK-51058][PYTHON] Avoid using jvm.SparkSession

### What changes were proposed in this pull request?

Avoids using `jvm.SparkSession` style, to improve Py4J performance similar to #49312, #49313, and #49412.

### Why are the changes needed?

To reduce the overhead of Py4J calls.

```py
import time

def benchmark(f, _n=10, *args, **kwargs):
    start = time.time()

    for i in range(_n):
        f(*args, **kwargs)

    print(time.time() - start)
```

```py
from pyspark.context import SparkContext
jvm = SparkContext._jvm

def f():
  return jvm.SparkSession

benchmark(f, 10000)  # -> 3.578310251235962
```

```py
from pyspark.context import SparkContext
jvm = SparkContext._jvm

def g():
  return getattr(jvm, "org.apache.spark.sql.classic.SparkSession")

benchmark(g, 10000)  # -> 0.254807710647583
```

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

The existing tests should pass.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #49760 from ueshin/issues/SPARK-51058/spark_session.

Authored-by: Takuya Ueshin <ueshin@databricks.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

Author: ueshin

python/pyspark/ml/connect/tuning.py

@@ -180,7 +180,7 @@ def single_task() -> Tuple[int, float]:
             if not is_remote():
                 # Active session is thread-local variable, in background thread the active session
                 # is not set, the following line sets it as the main thread active session.
-                active_session._jvm.SparkSession.setActiveSession(  # type: ignore[union-attr]
+                SparkSession._get_j_spark_session_class(active_session._jvm).setActiveSession(
                     active_session._jsparkSession
                 )
 

python/pyspark/sql/session.py

@@ -71,7 +71,7 @@
 from pyspark.errors import PySparkValueError, PySparkTypeError, PySparkRuntimeError
 
 if TYPE_CHECKING:
-    from py4j.java_gateway import JavaObject
+    from py4j.java_gateway import JavaClass, JavaObject, JVMView
     import pyarrow as pa
     from pyspark.core.context import SparkContext
     from pyspark.core.rdd import RDD
@@ -542,9 +542,8 @@ def getOrCreate(self) -> "SparkSession":
                     # by all sessions.
                     session = SparkSession(sc, options=self._options)
                 else:
-                    getattr(
-                        getattr(session._jvm, "SparkSession$"), "MODULE$"
-                    ).applyModifiableSettings(session._jsparkSession, self._options)
+                    module = SparkSession._get_j_spark_session_module(session._jvm)
+                    module.applyModifiableSettings(session._jsparkSession, self._options)
                 return session
 
         # Spark Connect-specific API
@@ -612,21 +611,20 @@ def __init__(
 
         assert self._jvm is not None
 
+        jSparkSessionClass = SparkSession._get_j_spark_session_class(self._jvm)
+        jSparkSessionModule = SparkSession._get_j_spark_session_module(self._jvm)
+
         if jsparkSession is None:
             if (
-                self._jvm.SparkSession.getDefaultSession().isDefined()
-                and not self._jvm.SparkSession.getDefaultSession().get().sparkContext().isStopped()
+                jSparkSessionClass.getDefaultSession().isDefined()
+                and not jSparkSessionClass.getDefaultSession().get().sparkContext().isStopped()
             ):
-                jsparkSession = self._jvm.SparkSession.getDefaultSession().get()
-                getattr(getattr(self._jvm, "SparkSession$"), "MODULE$").applyModifiableSettings(
-                    jsparkSession, options
-                )
+                jsparkSession = jSparkSessionClass.getDefaultSession().get()
+                jSparkSessionModule.applyModifiableSettings(jsparkSession, options)
             else:
-                jsparkSession = self._jvm.SparkSession(self._jsc.sc(), options)
+                jsparkSession = jSparkSessionClass(self._jsc.sc(), options)
         else:
-            getattr(getattr(self._jvm, "SparkSession$"), "MODULE$").applyModifiableSettings(
-                jsparkSession, options
-            )
+            jSparkSessionModule.applyModifiableSettings(jsparkSession, options)
         self._jsparkSession = jsparkSession
         _monkey_patch_RDD(self)
         install_exception_handler()
@@ -637,8 +635,8 @@ def __init__(
             SparkSession._instantiatedSession = self
             SparkSession._activeSession = self
             assert self._jvm is not None
-            self._jvm.SparkSession.setDefaultSession(self._jsparkSession)
-            self._jvm.SparkSession.setActiveSession(self._jsparkSession)
+            jSparkSessionClass.setDefaultSession(self._jsparkSession)
+            jSparkSessionClass.setActiveSession(self._jsparkSession)
 
         self._profiler_collector = AccumulatorProfilerCollector()
 
@@ -649,6 +647,14 @@ def _should_update_active_session() -> bool:
             or SparkSession._instantiatedSession._sc._jsc is None
         )
 
+    @staticmethod
+    def _get_j_spark_session_class(jvm: "JVMView") -> "JavaClass":
+        return getattr(jvm, "org.apache.spark.sql.classic.SparkSession")
+
+    @staticmethod
+    def _get_j_spark_session_module(jvm: "JVMView") -> "JavaObject":
+        return getattr(getattr(jvm, "org.apache.spark.sql.classic.SparkSession$"), "MODULE$")
+
     def _repr_html_(self) -> str:
         return """
             <div>
@@ -721,9 +727,10 @@ def getActiveSession(cls) -> Optional["SparkSession"]:
             return None
         else:
             assert sc._jvm is not None
-            if sc._jvm.SparkSession.getActiveSession().isDefined():
+            jSparkSessionClass = SparkSession._get_j_spark_session_class(sc._jvm)
+            if jSparkSessionClass.getActiveSession().isDefined():
                 if SparkSession._should_update_active_session():
-                    SparkSession(sc, sc._jvm.SparkSession.getActiveSession().get())
+                    SparkSession(sc, jSparkSessionClass.getActiveSession().get())
                 return SparkSession._activeSession
             else:
                 return None
@@ -1501,7 +1508,7 @@ def createDataFrame(  # type: ignore[misc]
         """
         SparkSession._activeSession = self
         assert self._jvm is not None
-        self._jvm.SparkSession.setActiveSession(self._jsparkSession)
+        SparkSession._get_j_spark_session_class(self._jvm).setActiveSession(self._jsparkSession)
         if isinstance(data, DataFrame):
             raise PySparkTypeError(
                 errorClass="INVALID_TYPE",
@@ -2000,8 +2007,9 @@ def stop(self) -> None:
         self._sc.stop()
         # We should clean the default session up. See SPARK-23228.
         assert self._jvm is not None
-        self._jvm.SparkSession.clearDefaultSession()
-        self._jvm.SparkSession.clearActiveSession()
+        jSparkSessionClass = SparkSession._get_j_spark_session_class(self._jvm)
+        jSparkSessionClass.clearDefaultSession()
+        jSparkSessionClass.clearActiveSession()
         SparkSession._instantiatedSession = None
         SparkSession._activeSession = None
         SQLContext._instantiatedContext = None