use delta write rule

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala

@@ -70,7 +70,6 @@ private object VeloxRuleApi {
     injector.injectTransform(c => RewriteTransformer.apply(c.session))
     injector.injectTransform(_ => PushDownFilterToScan)
     injector.injectTransform(_ => PushDownInputFileExpression.PostOffload)
-    injector.injectTransform(_ => ReplaceDeltaTransformer())
     injector.injectTransform(_ => EnsureLocalSortRequirements)
     injector.injectTransform(_ => EliminateLocalSort)
     injector.injectTransform(_ => CollapseProjectExecTransformer)

gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaRewriteTransformerRules.scala

@@ -16,22 +16,22 @@
  */
 package org.apache.gluten.extension
 
-import org.apache.gluten.execution.{DeltaScanTransformer, ProjectExecTransformer}
-import org.apache.gluten.extension.DeltaRewriteTransformerRules.columnMappingRule
+import org.apache.gluten.execution.{DeltaFilterExecTransformer, DeltaProjectExecTransformer, DeltaScanTransformer, ProjectExecTransformer}
+import org.apache.gluten.extension.DeltaRewriteTransformerRules.{columnMappingRule, filterRule, projectRule}
 import org.apache.gluten.extension.columnar.RewriteTransformerRules
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Expression}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.delta.{DeltaColumnMapping, DeltaParquetFileFormat, NoMapping}
-import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.{FilterExec, ProjectExec, SparkPlan}
 import org.apache.spark.sql.execution.datasources.FileFormat
 
 import scala.collection.mutable.ListBuffer
 
 class DeltaRewriteTransformerRules extends RewriteTransformerRules {
-  override def rules: Seq[Rule[SparkPlan]] = columnMappingRule :: Nil
+  override def rules: Seq[Rule[SparkPlan]] = columnMappingRule :: filterRule :: projectRule :: Nil
 }
 
 object DeltaRewriteTransformerRules {
@@ -58,13 +58,32 @@ object DeltaRewriteTransformerRules {
         transformColumnMappingPlan(p)
     }
 
+  val filterRule: Rule[SparkPlan] = (plan: SparkPlan) =>
+    plan.transformUp {
+      case FilterExec(condition, child) if condition.exists(containsIncrementMetricExpr) =>
+        DeltaFilterExecTransformer(condition, child)
+    }
+
+  val projectRule: Rule[SparkPlan] = (plan: SparkPlan) =>
+    plan.transformUp {
+      case ProjectExec(projectList, child) if projectList.exists(containsIncrementMetricExpr) =>
+        DeltaProjectExecTransformer(projectList, child)
+    }
+
   private def isDeltaColumnMappingFileFormat(fileFormat: FileFormat): Boolean = fileFormat match {
     case d: DeltaParquetFileFormat if d.columnMappingMode != NoMapping =>
       true
     case _ =>
       false
   }
 
+  private def containsIncrementMetricExpr(expr: Expression): Boolean = {
+    expr match {
+      case e if e.prettyName == "increment_metric" => true
+      case _ => expr.children.exists(containsIncrementMetricExpr)
+    }
+  }
+
   /**
    * This method is only used for Delta ColumnMapping FileFormat(e.g. nameMapping and idMapping)
    * transform the metadata of Delta into Parquet's, each plan should only be transformed once.