[GLUTEN-7690][CORE][CH][VL] Remove GlutenConfig from ColumnarRuleApplier

apache · Oct 26, 2024 · 395b0bf · 395b0bf
1 parent 619624a
commit 395b0bf
Show file tree

Hide file tree

Showing 9 changed files with 21 additions and 27 deletions.
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
@@ -16,6 +16,7 @@
  */
 package org.apache.gluten.backendsapi.clickhouse
 
+import org.apache.gluten.GlutenConfig.{EXTENDED_COLUMNAR_POST_RULES, EXTENDED_COLUMNAR_TRANSFORM_RULES}
 import org.apache.gluten.backendsapi.RuleApi
 import org.apache.gluten.extension._
 import org.apache.gluten.extension.columnar._
@@ -89,7 +90,7 @@ private object CHRuleApi {
     injector.injectTransform(
       c =>
         intercept(
-          SparkPlanRules.extendedColumnarRule(c.conf.extendedColumnarTransformRules)(c.session)))
+          SparkPlanRules.extendedColumnarRule(c.session.sessionState.conf.getConf(EXTENDED_COLUMNAR_TRANSFORM_RULES))(c.session)))
     injector.injectTransform(c => InsertTransitions(c.outputsColumnar))
 
     // Gluten columnar: Fallback policies.
@@ -101,14 +102,14 @@ private object CHRuleApi {
     SparkShimLoader.getSparkShims
       .getExtendedColumnarPostRules()
       .foreach(each => injector.injectPost(c => intercept(each(c.session))))
-    injector.injectPost(c => ColumnarCollapseTransformStages(c.conf))
+    injector.injectPost(c => ColumnarCollapseTransformStages())
     injector.injectTransform(
       c =>
-        intercept(SparkPlanRules.extendedColumnarRule(c.conf.extendedColumnarPostRules)(c.session)))
+        intercept(SparkPlanRules.extendedColumnarRule(c.session.sessionState.conf.getConf(EXTENDED_COLUMNAR_POST_RULES))(c.session)))
 
     // Gluten columnar: Final rules.
     injector.injectFinal(c => RemoveGlutenTableCacheColumnarToRow(c.session))
-    injector.injectFinal(c => GlutenFallbackReporter(c.conf, c.session))
+    injector.injectFinal(c => GlutenFallbackReporter(c.session))
     injector.injectFinal(_ => RemoveFallbackTagRule())
   }
 

diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
@@ -80,11 +80,11 @@ private object VeloxRuleApi {
     SparkShimLoader.getSparkShims
       .getExtendedColumnarPostRules()
       .foreach(each => injector.injectPost(c => each(c.session)))
-    injector.injectPost(c => ColumnarCollapseTransformStages(c.conf))
+    injector.injectPost(c => ColumnarCollapseTransformStages())
 
     // Gluten columnar: Final rules.
     injector.injectFinal(c => RemoveGlutenTableCacheColumnarToRow(c.session))
-    injector.injectFinal(c => GlutenFallbackReporter(c.conf, c.session))
+    injector.injectFinal(c => GlutenFallbackReporter(c.session))
     injector.injectFinal(_ => RemoveFallbackTagRule())
   }
 
@@ -116,9 +116,9 @@ private object VeloxRuleApi {
     SparkShimLoader.getSparkShims
       .getExtendedColumnarPostRules()
       .foreach(each => injector.inject(c => each(c.session)))
-    injector.inject(c => ColumnarCollapseTransformStages(c.conf))
+    injector.inject(c => ColumnarCollapseTransformStages())
     injector.inject(c => RemoveGlutenTableCacheColumnarToRow(c.session))
-    injector.inject(c => GlutenFallbackReporter(c.conf, c.session))
+    injector.inject(c => GlutenFallbackReporter(c.session))
     injector.inject(_ => RemoveFallbackTagRule())
   }
 }
diff --git a/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/ColumnarRuleApplier.scala b/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/ColumnarRuleApplier.scala
@@ -16,7 +16,6 @@
  */
 package org.apache.gluten.extension.columnar
 
-import org.apache.gluten.GlutenConfig
 import org.apache.gluten.extension.util.AdaptiveContext
 
 import org.apache.spark.sql.SparkSession
@@ -30,9 +29,5 @@ object ColumnarRuleApplier {
   class ColumnarRuleCall(
       val session: SparkSession,
       val ac: AdaptiveContext,
-      val outputsColumnar: Boolean) {
-    val conf: GlutenConfig = {
-      new GlutenConfig(session.sessionState.conf)
-    }
-  }
+      val outputsColumnar: Boolean) {}
 }
diff --git a/...trait/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseTransformStages.scala b/...trait/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseTransformStages.scala
@@ -16,7 +16,6 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.gluten.GlutenConfig
 import org.apache.gluten.backend.Backend
 import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.execution._
@@ -114,7 +113,6 @@ case class InputIteratorTransformer(child: SparkPlan) extends UnaryTransformSupp
  * generate/compile code.
  */
 case class ColumnarCollapseTransformStages(
-    glutenConfig: GlutenConfig,
     transformStageCounter: AtomicInteger = ColumnarCollapseTransformStages.transformStageCounter)
   extends Rule[SparkPlan] {
 

diff --git a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenFallbackReporter.scala b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenFallbackReporter.scala
@@ -16,7 +16,7 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.gluten.GlutenConfig
+import org.apache.gluten.GlutenConfig.{FALLBACK_REPORTER_ENABLED, VALIDATION_LOG_LEVEL}
 import org.apache.gluten.events.GlutenPlanFallbackEvent
 import org.apache.gluten.extension.GlutenPlan
 import org.apache.gluten.extension.columnar.FallbackTags
@@ -31,12 +31,10 @@ import org.apache.spark.sql.execution.ui.GlutenEventUtils
  * This rule is used to collect all fallback reason.
  *   1. print fallback reason for each plan node 2. post all fallback reason using one event
  */
-case class GlutenFallbackReporter(glutenConfig: GlutenConfig, spark: SparkSession)
-  extends Rule[SparkPlan]
-  with LogLevelUtil {
+case class GlutenFallbackReporter(spark: SparkSession) extends Rule[SparkPlan] with LogLevelUtil {
 
   override def apply(plan: SparkPlan): SparkPlan = {
-    if (!glutenConfig.enableFallbackReport) {
+    if (!spark.sessionState.conf.getConf(FALLBACK_REPORTER_ENABLED)) {
       return plan
     }
     printFallbackReason(plan)
@@ -52,12 +50,14 @@ case class GlutenFallbackReporter(glutenConfig: GlutenConfig, spark: SparkSessio
   }
 
   private def printFallbackReason(plan: SparkPlan): Unit = {
-    val validationLogLevel = glutenConfig.validationLogLevel
     plan.foreachUp {
       case _: GlutenPlan => // ignore
       case p: SparkPlan if FallbackTags.nonEmpty(p) =>
         val tag = FallbackTags.get(p)
-        logFallbackReason(validationLogLevel, p.nodeName, tag.reason())
+        logFallbackReason(
+          spark.sessionState.conf.getConf(VALIDATION_LOG_LEVEL),
+          p.nodeName,
+          tag.reason())
         // With in next round stage in AQE, the physical plan would be a new instance that
         // can not preserve the tag, so we need to set the fallback reason to logical plan.
         // Then we can be aware of the fallback reason for the whole plan.

diff --git a/...en-ut/spark32/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala b/...en-ut/spark32/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala
@@ -171,7 +171,7 @@ private object FallbackStrategiesSuite {
       List(c => ExpandFallbackPolicy(c.ac.isAdaptiveContext(), c.ac.originalPlan())),
       List(
         c => RemoveTopmostColumnarToRow(c.session, c.ac.isAdaptiveContext()),
-        _ => ColumnarCollapseTransformStages(GlutenConfig.getConf)
+        _ => ColumnarCollapseTransformStages()
       ),
       List(_ => RemoveFallbackTagRule())
     )

diff --git a/...en-ut/spark33/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala b/...en-ut/spark33/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala
@@ -181,7 +181,7 @@ private object FallbackStrategiesSuite {
       List(c => ExpandFallbackPolicy(c.ac.isAdaptiveContext(), c.ac.originalPlan())),
       List(
         c => RemoveTopmostColumnarToRow(c.session, c.ac.isAdaptiveContext()),
-        _ => ColumnarCollapseTransformStages(GlutenConfig.getConf)
+        _ => ColumnarCollapseTransformStages()
       ),
       List(_ => RemoveFallbackTagRule())
     )

diff --git a/...en-ut/spark34/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala b/...en-ut/spark34/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala
@@ -181,7 +181,7 @@ private object FallbackStrategiesSuite {
       List(c => ExpandFallbackPolicy(c.ac.isAdaptiveContext(), c.ac.originalPlan())),
       List(
         c => RemoveTopmostColumnarToRow(c.session, c.ac.isAdaptiveContext()),
-        _ => ColumnarCollapseTransformStages(GlutenConfig.getConf)
+        _ => ColumnarCollapseTransformStages()
       ),
       List(_ => RemoveFallbackTagRule())
     )

diff --git a/...en-ut/spark35/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala b/...en-ut/spark35/src/test/scala/org/apache/spark/sql/execution/FallbackStrategiesSuite.scala
@@ -182,7 +182,7 @@ private object FallbackStrategiesSuite {
       List(c => ExpandFallbackPolicy(c.ac.isAdaptiveContext(), c.ac.originalPlan())),
       List(
         c => RemoveTopmostColumnarToRow(c.session, c.ac.isAdaptiveContext()),
-        _ => ColumnarCollapseTransformStages(GlutenConfig.getConf)
+        _ => ColumnarCollapseTransformStages()
       ),
       List(_ => RemoveFallbackTagRule())
     )