zhztheplayer commented on code in PR #7124:
URL: https://github.com/apache/incubator-gluten/pull/7124#discussion_r1746538276
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala:
##########
@@ -52,6 +52,7 @@ private object VeloxRuleApi {
def injectLegacy(injector: LegacyInjector): Unit = {
// Gluten columnar: Transform rules.
injector.injectTransform(_ => RemoveTransitions)
+ injector.injectTransform(_ => PushDownInputFileExpressionBeforeLeaf)
Review Comment:
Let's do a rename for the two rules. Since they seem to always be used
together as a rule pair.
```suggestion
injector.injectTransform(_ => PushDownInputFileExpression.PreOffload)
```
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.extension.columnar
+
+import org.apache.gluten.execution.{BatchScanExecTransformer,
FileSourceScanExecTransformer}
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference,
Expression, InputFileBlockLength, InputFileBlockStart, InputFileName,
NamedExpression}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.{DeserializeToObjectExec, LeafExecNode,
ProjectExec, SerializeFromObjectExec, SparkPlan, UnionExec}
+
+import scala.collection.mutable
+import scala.collection.mutable.Map
+
+/**
+ * The Spark implementations of
input_file_name/input_file_block_start/input_file_block_length uses
+ * a thread local to stash the file name and retrieve it from the function. If
there is a
+ * transformer node between project input_file_function and scan, the result
of input_file_name is
+ * an empty string. So we should push down input_file_function to transformer
scan or add fallback
+ * project of input_file_function before fallback scan.
+ *
+ * Two rules are involved:
+ * - Before offload, add new project before leaf node and push down input
file expression to the
+ * new project
+ * - After offload, if scan be offloaded, push down input file expression
into scan and remove
+ * project
+ */
+object PushDownInputFileExpression {
+ def containsInputFileRelatedExpr(expr: Expression): Boolean = {
+ expr match {
+ case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength
=> true
+ case _ => expr.children.exists(containsInputFileRelatedExpr)
+ }
+ }
+}
+
+object PushDownInputFileExpressionBeforeLeaf extends Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
+ case ProjectExec(projectList, child)
+ if
projectList.exists(PushDownInputFileExpression.containsInputFileRelatedExpr) =>
+ val replacedExprs = mutable.Map[String, Alias]()
+ val newProjectList = projectList.map {
+ expr => rewriteExpr(expr, replacedExprs).asInstanceOf[NamedExpression]
+ }
+ val newChild = addMetadataCol(child, replacedExprs)
+ ProjectExec(newProjectList, newChild)
+ }
+
+ private def rewriteExpr(expr: Expression, replacedExprs: Map[String,
Alias]): Expression = {
Review Comment:
Could use `replacedExprs: mutable.Map[String, Alias]`
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala:
##########
@@ -64,6 +65,7 @@ private object VeloxRuleApi {
injector.injectTransform(_ => TransformPreOverrides())
injector.injectTransform(_ => RemoveNativeWriteFilesSortAndProject())
injector.injectTransform(c => RewriteTransformer.apply(c.session))
+ injector.injectTransform(_ => PushDownInputFileExpressionToScan)
Review Comment:
Similarly,
```suggestion
injector.injectTransform(_ => PushDownInputFileExpression.PostOffload)
```
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.extension.columnar
+
+import org.apache.gluten.execution.{BatchScanExecTransformer,
FileSourceScanExecTransformer}
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference,
Expression, InputFileBlockLength, InputFileBlockStart, InputFileName,
NamedExpression}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.{DeserializeToObjectExec, LeafExecNode,
ProjectExec, SerializeFromObjectExec, SparkPlan, UnionExec}
+
+import scala.collection.mutable
+import scala.collection.mutable.Map
+
+/**
+ * The Spark implementations of
input_file_name/input_file_block_start/input_file_block_length uses
+ * a thread local to stash the file name and retrieve it from the function. If
there is a
+ * transformer node between project input_file_function and scan, the result
of input_file_name is
+ * an empty string. So we should push down input_file_function to transformer
scan or add fallback
+ * project of input_file_function before fallback scan.
+ *
+ * Two rules are involved:
+ * - Before offload, add new project before leaf node and push down input
file expression to the
+ * new project
+ * - After offload, if scan be offloaded, push down input file expression
into scan and remove
+ * project
+ */
+object PushDownInputFileExpression {
+ def containsInputFileRelatedExpr(expr: Expression): Boolean = {
+ expr match {
+ case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength
=> true
+ case _ => expr.children.exists(containsInputFileRelatedExpr)
+ }
+ }
+}
+
+object PushDownInputFileExpressionBeforeLeaf extends Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
+ case ProjectExec(projectList, child)
+ if
projectList.exists(PushDownInputFileExpression.containsInputFileRelatedExpr) =>
+ val replacedExprs = mutable.Map[String, Alias]()
+ val newProjectList = projectList.map {
+ expr => rewriteExpr(expr, replacedExprs).asInstanceOf[NamedExpression]
+ }
+ val newChild = addMetadataCol(child, replacedExprs)
+ ProjectExec(newProjectList, newChild)
+ }
+
+ private def rewriteExpr(expr: Expression, replacedExprs: Map[String,
Alias]): Expression = {
+ expr match {
+ case _: InputFileName =>
+ replacedExprs
+ .getOrElseUpdate(expr.prettyName, Alias(InputFileName(),
expr.prettyName)())
+ .toAttribute
+ case _: InputFileBlockStart =>
+ replacedExprs
+ .getOrElseUpdate(expr.prettyName, Alias(InputFileBlockStart(),
expr.prettyName)())
+ .toAttribute
+ case _: InputFileBlockLength =>
+ replacedExprs
+ .getOrElseUpdate(expr.prettyName, Alias(InputFileBlockLength(),
expr.prettyName)())
+ .toAttribute
+ case other =>
+ other.withNewChildren(other.children.map(child => rewriteExpr(child,
replacedExprs)))
+ }
+ }
+
+ def addMetadataCol(plan: SparkPlan, replacedExprs: Map[String, Alias]):
SparkPlan = plan match {
Review Comment:
Could use replacedExprs: mutable.Map[String, Alias]
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
