SPARK-25003: Address Reviewer Comments

Author: RussellSpitzer

python/pyspark/sql/session.py

@@ -212,16 +212,13 @@ def __init__(self, sparkContext, jsparkSession=None):
         self._sc = sparkContext
         self._jsc = self._sc._jsc
         self._jvm = self._sc._jvm
-
         if jsparkSession is None:
             if self._jvm.SparkSession.getDefaultSession().isDefined() \
                     and not self._jvm.SparkSession.getDefaultSession().get() \
                         .sparkContext().isStopped():
                 jsparkSession = self._jvm.SparkSession.getDefaultSession().get()
             else:
-                extensions = self._sc._jvm.org.apache.spark.sql\
-                    .SparkSessionExtensions(self._jsc.getConf())
-                jsparkSession = self._jvm.SparkSession(self._jsc.sc(), extensions)
+                jsparkSession = self._jvm.SparkSession(self._jsc.sc())
 
         self._jsparkSession = jsparkSession
         self._jwrapped = self._jsparkSession.sqlContext()

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

@@ -86,10 +86,7 @@ class SparkSession private(
 
   private[sql] def this(sc: SparkContext) {
     this(sc, None, None, new SparkSessionExtensions)
-  }
-
-  private[sql] def this(sc: SparkContext, extensions: SparkSessionExtensions) {
-    this(sc, None, None, extensions)
+    SparkSession.applyExtensionsFromConf(sc.getConf, this.extensions)
   }
 
   sparkContext.assertNotStopped()
@@ -939,7 +936,7 @@ object SparkSession extends Logging {
           // Do not update `SparkConf` for existing `SparkContext`, as it's shared by all sessions.
         }
 
-        SparkSessionExtensions.applyExtensionsFromConf(sparkContext.conf, extensions)
+        applyExtensionsFromConf(sparkContext.conf, extensions)
 
         session = new SparkSession(sparkContext, None, None, extensions)
         options.foreach { case (k, v) => session.initialSessionOptions.put(k, v) }
@@ -1124,4 +1121,27 @@ object SparkSession extends Logging {
       SparkSession.clearDefaultSession()
     }
   }
+
+  /**
+   * Initialize extensions if the user has defined a configurator class in their SparkConf.
+   * This class will be applied to the extensions passed into this function.
+   */
+  private[sql] def applyExtensionsFromConf(conf: SparkConf, extensions: SparkSessionExtensions) {
+    val extensionConfOption = conf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS)
+    if (extensionConfOption.isDefined) {
+      val extensionConfClassName = extensionConfOption.get
+      try {
+        val extensionConfClass = Utils.classForName(extensionConfClassName)
+        val extensionConf = extensionConfClass.newInstance()
+          .asInstanceOf[SparkSessionExtensions => Unit]
+        extensionConf(extensions)
+      } catch {
+        // Ignore the error if we cannot find the class or when the class has the wrong type.
+        case e@(_: ClassCastException |
+                _: ClassNotFoundException |
+                _: NoClassDefFoundError) =>
+          logWarning(s"Cannot use $extensionConfClassName to configure session extensions.", e)
+      }
+    }
+  }
 }

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

@@ -19,14 +19,10 @@ package org.apache.spark.sql
 
 import scala.collection.mutable
 
-import org.apache.spark.SparkConf
 import org.apache.spark.annotation.{DeveloperApi, Experimental, InterfaceStability}
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.parser.ParserInterface
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.internal.StaticSQLConf
-import org.apache.spark.util.Utils
 
 /**
  * :: Experimental ::
@@ -70,11 +66,6 @@ class SparkSessionExtensions {
   type StrategyBuilder = SparkSession => Strategy
   type ParserBuilder = (SparkSession, ParserInterface) => ParserInterface
 
-  private[sql] def this(conf: SparkConf) {
-    this()
-    SparkSessionExtensions.applyExtensionsFromConf(conf, this)
-  }
-
   private[this] val resolutionRuleBuilders = mutable.Buffer.empty[RuleBuilder]
 
   /**
@@ -178,29 +169,3 @@ class SparkSessionExtensions {
     parserBuilders += builder
   }
 }
-
-object SparkSessionExtensions extends Logging {
-
-  /**
-   * Initialize extensions if the user has defined a configurator class in their SparkConf.
-   * This class will be applied to the extensions passed into this function.
-   */
-  private[sql] def applyExtensionsFromConf(conf: SparkConf, extensions: SparkSessionExtensions) {
-    val extensionConfOption = conf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS)
-    if (extensionConfOption.isDefined) {
-      val extensionConfClassName = extensionConfOption.get
-      try {
-        val extensionConfClass = Utils.classForName(extensionConfClassName)
-        val extensionConf = extensionConfClass.newInstance()
-          .asInstanceOf[SparkSessionExtensions => Unit]
-        extensionConf(extensions)
-      } catch {
-        // Ignore the error if we cannot find the class or when the class has the wrong type.
-        case e@(_: ClassCastException |
-                _: ClassNotFoundException |
-                _: NoClassDefFoundError) =>
-          logWarning(s"Cannot use $extensionConfClassName to configure session extensions.", e)
-      }
-    }
-  }
-}