[CORE][VL] Add OffloadProject to offload project having input_file_na…

…me's support considered (#6200)
apache · Jun 28, 2024 · 86449d0 · 86449d0
1 parent e684bf3
commit 86449d0
Show file tree

Hide file tree

Showing 19 changed files with 343 additions and 204 deletions.
diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
@@ -303,6 +303,8 @@ object VeloxBackendSettings extends BackendSettingsApi {
 
   override def supportNativeRowIndexColumn(): Boolean = true
 
+  override def supportNativeInputFileRelatedExpr(): Boolean = true
+
   override def supportExpandExec(): Boolean = true
 
   override def supportSortExec(): Boolean = true

diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala
@@ -19,7 +19,6 @@ package org.apache.gluten.backendsapi.velox
 import org.apache.gluten.GlutenNumaBindingInfo
 import org.apache.gluten.backendsapi.IteratorApi
 import org.apache.gluten.execution._
-import org.apache.gluten.extension.InputFileNameReplaceRule
 import org.apache.gluten.metrics.IMetrics
 import org.apache.gluten.sql.shims.SparkShimLoader
 import org.apache.gluten.substrait.plan.PlanNode
@@ -134,13 +133,6 @@ class VeloxIteratorApi extends IteratorApi with Logging {
         }
         val metadataColumn =
           SparkShimLoader.getSparkShims.generateMetadataColumns(file, metadataColumnNames)
-        metadataColumn.put(InputFileNameReplaceRule.replacedInputFileName, file.filePath.toString)
-        metadataColumn.put(
-          InputFileNameReplaceRule.replacedInputFileBlockStart,
-          file.start.toString)
-        metadataColumn.put(
-          InputFileNameReplaceRule.replacedInputFileBlockLength,
-          file.length.toString)
         metadataColumns.add(metadataColumn)
         val partitionColumn = new JHashMap[String, String]()
         for (i <- 0 until file.partitionValues.numFields) {

diff --git a/...ends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala b/...ends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
@@ -806,12 +806,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
    * @return
    */
   override def genExtendedColumnarValidationRules(): List[SparkSession => Rule[SparkPlan]] = {
-    val buf: ListBuffer[SparkSession => Rule[SparkPlan]] =
-      ListBuffer(BloomFilterMightContainJointRewriteRule.apply, ArrowScanReplaceRule.apply)
-    if (GlutenConfig.getConf.enableInputFileNameReplaceRule) {
-      buf += InputFileNameReplaceRule.apply
-    }
-    buf.result
+    List(BloomFilterMightContainJointRewriteRule.apply, ArrowScanReplaceRule.apply)
   }
 
   /**

diff --git a/backends-velox/src/main/scala/org/apache/gluten/extension/InputFileNameReplaceRule.scala b/backends-velox/src/main/scala/org/apache/gluten/extension/InputFileNameReplaceRule.scala
diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/ScalarFunctionsValidateSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/ScalarFunctionsValidateSuite.scala
@@ -645,13 +645,9 @@ class ScalarFunctionsValidateSuite extends FunctionsValidateTest {
   }
 
   test("Test input_file_name function") {
-    withSQLConf(
-      "spark.gluten.sql.enableInputFileNameReplaceRule" -> "true"
-    ) {
-      runQueryAndCompare("""SELECT input_file_name(), l_orderkey
-                           | from lineitem limit 100""".stripMargin) {
-        checkGlutenOperatorMatch[ProjectExecTransformer]
-      }
+    runQueryAndCompare("""SELECT input_file_name(), l_orderkey
+                         | from lineitem limit 100""".stripMargin) {
+      checkGlutenOperatorMatch[ProjectExecTransformer]
     }
   }
 

diff --git a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
@@ -44,6 +44,7 @@ trait BackendSettingsApi {
   def supportNativeWrite(fields: Array[StructField]): Boolean = true
   def supportNativeMetadataColumns(): Boolean = false
   def supportNativeRowIndexColumn(): Boolean = false
+  def supportNativeInputFileRelatedExpr(): Boolean = false
 
   def supportExpandExec(): Boolean = false
   def supportSortExec(): Boolean = false

diff --git a/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/MiscColumnarRules.scala b/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/MiscColumnarRules.scala
@@ -30,7 +30,7 @@ object MiscColumnarRules {
   object TransformPreOverrides {
     def apply(): TransformPreOverrides = {
       TransformPreOverrides(
-        List(OffloadFilter()),
+        List(OffloadProject(), OffloadFilter()),
         List(
           OffloadOthers(),
           OffloadAggregate(),