Guard Delta CDF DV remove scans

Author: malinjawi

gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala

@@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.delta.files.TahoeRemoveFileIndex
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.datasources.HadoopFsRelation
 import org.apache.spark.sql.types.StructType
@@ -57,16 +58,27 @@ case class DeltaScanTransformer(
 
   override protected def doValidateInternal(): ValidationResult = {
     if (
-      requiredSchema.fields.exists(
-        _.name == "__delta_internal_is_row_deleted") || requiredSchema.fields.exists(
-        _.name == "__delta_internal_row_index")
+      requiredSchemaIncludesDeletionVectorColumns ||
+      cdfRemoveFilesHaveDeletionVectors
     ) {
-      return ValidationResult.failed(s"Deletion vector is not supported in native.")
+      return ValidationResult.failed(DeltaScanTransformer.DELETION_VECTOR_UNSUPPORTED)
     }
 
     super.doValidateInternal()
   }
 
+  private def requiredSchemaIncludesDeletionVectorColumns: Boolean = {
+    requiredSchema.fields.exists(
+      _.name == "__delta_internal_is_row_deleted") || requiredSchema.fields.exists(
+      _.name == "__delta_internal_row_index")
+  }
+
+  private def cdfRemoveFilesHaveDeletionVectors: Boolean = relation.location match {
+    case index: TahoeRemoveFileIndex =>
+      index.filesByVersion.exists(_.actions.exists(_.deletionVector != null))
+    case _ => false
+  }
+
   override def doCanonicalize(): DeltaScanTransformer = {
     DeltaScanTransformer(
       relation,
@@ -91,6 +103,8 @@ case class DeltaScanTransformer(
 
 object DeltaScanTransformer {
 
+  val DELETION_VECTOR_UNSUPPORTED = "Deletion vector is not supported in native."
+
   def apply(scanExec: FileSourceScanExec): DeltaScanTransformer = {
     new DeltaScanTransformer(
       scanExec.relation,

gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala

@@ -262,12 +262,17 @@ abstract class DeltaSuite extends WholeStageTransformerSuite {
                    |""".stripMargin)
 
       // Native DV scan support is separate; this regression locks down CDF correctness.
+      import org.apache.spark.sql.execution.GlutenImplicits._
       val df = runAndCompare(
         s"""
            |select id, name, _change_type
            |from table_changes('delta_cdf_dv', 0)
            |order by id, name, _change_type
            |""".stripMargin)
+      assert(
+        df.fallbackSummary.fallbackNodeToReason
+          .flatMap(_.values)
+          .exists(_.contains(DeltaScanTransformer.DELETION_VECTOR_UNSUPPORTED)))
       checkAnswer(
         df,
         Seq(