[VL][Delta] Guard DV DML row-index scans

Native DELETE/UPDATE/MERGE deletion-vector support deliberately keeps the
target row-index scan on Spark until native row-index execution is proven
for the required Delta table shapes. Detect those DML row-index scan
shapes and keep the small scan subtree (and its parent filter/project) off
the native path, so the offload path stays correctness-first while the
write side is built out.

- add DeltaDeletionVectorDmlUtils to detect and tag Delta DV DML
  row-index scans (row-index / file-path / bitmap-aggregator references)
- gate OffloadDeltaScan to fall back on DML row-index scans unless native
  Delta write and native DML row-index scan are both enabled
- add keepDmlRowIndexFallbackSubtreeOnSpark post-transform rule to keep the
  filter/project above a fallen-back DML scan on Spark, avoiding
  row<->columnar transitions right before Delta's JVM bitmap path
- cover the fallback shape with DeltaDeletionVectorHandoffSuite cases and a
  DeltaDmlRowIndexScanBenchmark for Spark 3.5 / Delta 3.3 and Spark 4.0

Author: malinjawi

backends-velox/src-delta/main/scala/org/apache/gluten/component/VeloxDeltaComponent.scala

@@ -18,7 +18,7 @@ package org.apache.gluten.component
 
 import org.apache.gluten.backendsapi.velox.VeloxBackend
 import org.apache.gluten.config.GlutenConfig
-import org.apache.gluten.extension.{DeltaPostTransformRules, OffloadDeltaFilter, OffloadDeltaProject, OffloadDeltaScan}
+import org.apache.gluten.extension.{DeltaDeletionVectorDmlUtils, DeltaPostTransformRules, OffloadDeltaFilter, OffloadDeltaProject, OffloadDeltaScan}
 import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
 import org.apache.gluten.extension.columnar.validator.Validators
 import org.apache.gluten.extension.injector.Injector
@@ -40,6 +40,11 @@ class VeloxDeltaComponent extends Component {
     // PreprocessTableWithDVsStrategy injects the skip-row column and filter during physical
     // planning, DeltaPostTransformRules.nativeDeletionVectorRule strips them when the scan
     // offloads, and DeltaScanTransformer materializes the per-file DV payloads for Velox.
+    //
+    // For native DELETE/UPDATE/MERGE, the DML target row-index scan is deliberately kept on Spark
+    // until native row-index execution is proven; tag those scans here so the post-transform rules
+    // can keep the small subtree off the native path.
+    legacy.injectPreTransform(_ => DeltaDeletionVectorDmlUtils.tagDmlRowIndexScans)
     legacy.injectTransform {
       c =>
         val offload = Seq(OffloadDeltaScan(), OffloadDeltaProject(), OffloadDeltaFilter())

backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala

@@ -16,15 +16,22 @@
  */
 package org.apache.spark.sql.delta
 
-import org.apache.gluten.execution.DeltaScanTransformer
+import org.apache.gluten.execution.{DeltaScanTransformer, FilterExecTransformerBase, ProjectExecTransformerBase}
+import org.apache.gluten.extension.DeltaDeletionVectorDmlUtils
+import org.apache.gluten.extension.columnar.FallbackTags
 
 import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.{DeltaSQLCommandTest, DeltaSQLTestUtils}
+import org.apache.spark.sql.execution.{FileSourceScanExec, FilterExec, ProjectExec, SparkPlan}
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.tags.ExtendedSQLTest
+import org.apache.spark.util.SparkVersionUtil
 
 import org.apache.hadoop.fs.Path
 
+import java.io.File
+
 @ExtendedSQLTest
 class DeltaDeletionVectorHandoffSuite
   extends QueryTest
@@ -34,6 +41,93 @@ class DeltaDeletionVectorHandoffSuite
 
   import testImplicits._
 
+  private val DmlFallbackReason = "fallback Delta DV DML row-index scan"
+  private val DmlRowIndexColumnNames =
+    Seq("__delta_internal_row_index", "_tmp_metadata_row_index", "rowIndexCol")
+
+  private def containsDmlFallbackScan(plan: SparkPlan): Boolean = {
+    plan.exists {
+      case scan: FileSourceScanExec =>
+        DeltaDeletionVectorDmlUtils.isDeletionVectorDmlRowIndexScan(scan) &&
+        FallbackTags.getOption(scan).exists(_.reason().contains(DmlFallbackReason))
+      case _ => false
+    }
+  }
+
+  private def hasSparkParentOverDmlFallbackScan(plan: SparkPlan): Boolean = {
+    plan.exists {
+      case ProjectExec(_, child) if containsDmlFallbackScan(child) => true
+      case FilterExec(_, child) if containsDmlFallbackScan(child) => true
+      case _ => false
+    }
+  }
+
+  private def hasNativeParentOverDmlFallbackScan(plan: SparkPlan): Boolean = {
+    plan.exists {
+      case project: ProjectExecTransformerBase if containsDmlFallbackScan(project.child) => true
+      case filter: FilterExecTransformerBase if containsDmlFallbackScan(filter.child) => true
+      case _ => false
+    }
+  }
+
+  private def containsDmlRowIndexTargetScanText(plan: SparkPlan): Boolean = {
+    val planText = plan.treeString
+    planText.contains("FileScan parquet") &&
+    planText.contains("file_path") &&
+    DmlRowIndexColumnNames.exists(planText.contains) &&
+    (planText.contains("TahoeBatchFileIndex") || planText.contains("PreparedDeltaFileIndex"))
+  }
+
+  private def captureDeletePlans(
+      path: String,
+      predicate: String,
+      useMetadataRowIndex: Boolean): Seq[SparkPlan] = {
+    var executedPlans: Seq[SparkPlan] = Seq.empty
+    withSQLConf(
+      DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key ->
+        useMetadataRowIndex.toString,
+      "spark.gluten.sql.columnar.backend.velox.delta.enableNativeWrite" -> "false",
+      "spark.gluten.sql.delta.enableNativeDmlRowIndexScan" -> "false"
+    ) {
+      executedPlans = DeltaTestUtils.withAllPlansCaptured(spark) {
+        spark.sql(s"DELETE FROM delta.`$path` WHERE $predicate").collect()
+      }.map(_.executedPlan)
+    }
+    executedPlans
+  }
+
+  private def assertSparkDmlFallback(executedPlans: Seq[SparkPlan]): Unit = {
+    val planText = executedPlans.map(_.treeString).mkString("\n\n")
+    if (executedPlans.exists(containsDmlFallbackScan)) {
+      assert(executedPlans.exists(hasSparkParentOverDmlFallbackScan), planText)
+      assert(!executedPlans.exists(hasNativeParentOverDmlFallbackScan), planText)
+    } else {
+      assert(executedPlans.exists(containsDmlRowIndexTargetScanText), planText)
+    }
+  }
+
+  private def assertReadPlanAfterDmlFallback(path: String, useMetadataRowIndex: Boolean): Unit = {
+    withSQLConf(
+      DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key -> useMetadataRowIndex.toString) {
+      val df = spark.read.format("delta").load(path)
+      val executedPlan = df.queryExecution.executedPlan
+      val planText = executedPlan.treeString
+      if (useMetadataRowIndex) {
+        assert(executedPlan.collect { case _: DeltaScanTransformer => true }.nonEmpty, planText)
+        assert(!planText.contains(DmlFallbackReason), planText)
+      } else {
+        assert(executedPlan.collect { case _: DeltaScanTransformer => true }.isEmpty, planText)
+      }
+      checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
+    }
+  }
+
+  private def activeDvCardinality(path: String): Long = {
+    val log = DeltaLog.forTable(spark, new Path(path))
+    log.update().allFiles.collect().flatMap(
+      file => Option(file.deletionVector).map(_.cardinality)).sum
+  }
+
   test("Spark 3.5 Delta DV scan handoff should filter deleted rows") {
     withTempDir {
       tempDir =>
@@ -65,4 +159,70 @@ class DeltaDeletionVectorHandoffSuite
         checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
     }
   }
+
+  Seq(true, false).foreach {
+    useMetadataRowIndex =>
+      test(
+        "Delta DV DML row-index scan should fall back with Spark project/filter, " +
+          s"metadata row index=$useMetadataRowIndex") {
+        assume(SparkVersionUtil.gteSpark35, "DML row-index scan fallback is Spark 3.5+ coverage")
+        withTempDir {
+          tempDir =>
+            val path = tempDir.getCanonicalPath
+            Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+              .toDF("id", "value")
+              .coalesce(1)
+              .write
+              .format("delta")
+              .save(path)
+
+            spark.sql(
+              s"ALTER TABLE delta.`$path` SET TBLPROPERTIES " +
+                "('delta.enableDeletionVectors' = true)")
+
+            var executedPlans: Seq[SparkPlan] = Seq.empty
+            withSQLConf(
+              DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key ->
+                useMetadataRowIndex.toString,
+              "spark.gluten.sql.columnar.backend.velox.delta.enableNativeWrite" -> "false",
+              "spark.gluten.sql.delta.enableNativeDmlRowIndexScan" -> "false"
+            ) {
+              executedPlans = DeltaTestUtils.withAllPlansCaptured(spark) {
+                spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)").collect()
+              }.map(_.executedPlan)
+            }
+            assertSparkDmlFallback(executedPlans)
+
+            val log = DeltaLog.forTable(spark, new Path(path))
+            assert(log.update().allFiles.collect().exists(_.deletionVector != null))
+            assertReadPlanAfterDmlFallback(path, useMetadataRowIndex)
+        }
+      }
+  }
+
+  test("Delta DV DML row-index scan should fall back when updating an existing DV") {
+    assume(SparkVersionUtil.gteSpark35, "DML row-index scan fallback is Spark 3.5+ coverage")
+    withTempDir {
+      tempDir =>
+        val path = new File(tempDir, "delta table with spaces").getCanonicalPath
+        Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"), (5, "e"), (6, "f"))
+          .toDF("id", "value")
+          .coalesce(1)
+          .write
+          .format("delta")
+          .save(path)
+
+        spark.sql(
+          s"ALTER TABLE delta.`$path` SET TBLPROPERTIES " +
+            "('delta.enableDeletionVectors' = true)")
+
+        assertSparkDmlFallback(captureDeletePlans(path, "id IN (5, 6)", useMetadataRowIndex = true))
+        assert(activeDvCardinality(path) === 2L)
+
+        assertSparkDmlFallback(captureDeletePlans(path, "id IN (3, 4)", useMetadataRowIndex = true))
+        assert(activeDvCardinality(path) === 4L)
+
+        assertReadPlanAfterDmlFallback(path, useMetadataRowIndex = true)
+    }
+  }
 }