[VL][DELTA] Support Delta CDF scan offload

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

@@ -17,8 +17,8 @@
 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.config.{GlutenConfig, VeloxDeltaConfig}
+import org.apache.gluten.extension.{DeltaCDFScanStrategy, 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
@@ -35,6 +35,12 @@ class VeloxDeltaComponent extends Component {
   }
 
   override def injectRules(injector: Injector): Unit = {
+    injector.spark.injectPlannerStrategy(
+      spark =>
+        DeltaCDFScanStrategy(
+          spark,
+          () => new VeloxDeltaConfig(spark.sessionState.conf).enableChangeDataFeedScan))
+
     val legacy = injector.gluten.legacy
     // Deletion-vector scans need no Gluten-side logical preprocessing: Delta's own
     // PreprocessTableWithDVsStrategy injects the skip-row column and filter during physical

backends-velox/src-delta/main/scala/org/apache/gluten/config/VeloxDeltaConfig.scala

@@ -22,6 +22,8 @@ class VeloxDeltaConfig(conf: SQLConf) extends GlutenCoreConfig(conf) {
   import VeloxDeltaConfig._
 
   def enableNativeWrite: Boolean = getConf(ENABLE_NATIVE_WRITE)
+
+  def enableChangeDataFeedScan: Boolean = getConf(ENABLE_CHANGE_DATA_FEED_SCAN)
 }
 
 object VeloxDeltaConfig extends ConfigRegistry {
@@ -40,4 +42,11 @@ object VeloxDeltaConfig extends ConfigRegistry {
       .doc("Enable native Delta Lake write for Velox backend.")
       .booleanConf
       .createWithDefault(false)
+
+  val ENABLE_CHANGE_DATA_FEED_SCAN: ConfigEntry[Boolean] =
+    buildConf("spark.gluten.sql.columnar.backend.velox.delta.enableChangeDataFeedScan")
+      .experimental()
+      .doc("Enable Delta Lake change data feed scan offload for Velox backend.")
+      .booleanConf
+      .createWithDefault(true)
 }

backends-velox/src-delta/test/scala/org/apache/gluten/execution/VeloxDeltaSuite.scala

@@ -16,4 +16,37 @@
  */
 package org.apache.gluten.execution
 
-class VeloxDeltaSuite extends DeltaSuite
+import org.apache.gluten.config.VeloxDeltaConfig
+
+import org.apache.spark.sql.Row
+
+class VeloxDeltaSuite extends DeltaSuite {
+  testWithMinSparkVersion("delta: change data feed scan offload can be disabled", "3.2") {
+    withTable("delta_cdf_disabled") {
+      spark.sql("""
+                  |create table delta_cdf_disabled (id int, name string) using delta
+                  |tblproperties ("delta.enableChangeDataFeed" = "true")
+                  |""".stripMargin)
+      spark.sql("""
+                  |insert into delta_cdf_disabled values (1, "v1"), (2, "v2")
+                  |""".stripMargin)
+
+      withSQLConf(VeloxDeltaConfig.ENABLE_CHANGE_DATA_FEED_SCAN.key -> "false") {
+        val df = spark.sql("""
+                             |select id, name, _change_type, _commit_version
+                             |from table_changes('delta_cdf_disabled', 1)
+                             |""".stripMargin)
+        checkAnswer(
+          df,
+          Seq(
+            Row(1, "v1", "insert", 1L),
+            Row(2, "v2", "insert", 1L)))
+        assert(
+          df.queryExecution.executedPlan.collect {
+            case scan: DeltaScanTransformer => scan
+          }.isEmpty,
+          df.queryExecution.executedPlan)
+      }
+    }
+  }
+}

docs/get-started/VeloxDelta.md

@@ -28,6 +28,12 @@ Native Delta write is controlled by:
   - Default: `false`
   - Type: experimental
 
+Native change data feed scan offload is controlled by:
+
+- `spark.gluten.sql.columnar.backend.velox.delta.enableChangeDataFeedScan`
+  - Default: `true`
+  - Type: experimental
+
 | Feature | Delta minWriterVersion | Delta minReaderVersion | Iceberg format-version | Feature type | Supported by Gluten (Velox) |
 |---|---:|---:|---:|---|---|
 | Basic functionality | 2 | 1 | 1 | Writer | Yes |

gluten-delta/src-delta23/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala

@@ -0,0 +1,44 @@
+/*
+ * 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
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.delta.BatchCDFSchemaEndVersion
+import org.apache.spark.sql.delta.commands.cdc.CDCReader
+
+object DeltaCDFRelationHelper {
+  def changesToBatchDF(
+      relation: CDCReader.DeltaCDFRelation,
+      spark: SparkSession): DataFrame = {
+    val deltaLog = relation.snapshotWithSchemaMode.snapshot.deltaLog
+    val latestVersion = deltaLog.update().version
+    val endingVersionForBatchSchema =
+      relation.endingVersion.map(v => latestVersion.min(v)).getOrElse(latestVersion)
+    val snapshotForBatchSchema = relation.snapshotWithSchemaMode.schemaMode match {
+      case BatchCDFSchemaEndVersion => deltaLog.getSnapshotAt(endingVersionForBatchSchema)
+      case _ => relation.snapshotWithSchemaMode.snapshot
+    }
+    val endVersion = relation.endingVersion.getOrElse(latestVersion)
+
+    CDCReader.changesToBatchDF(
+      deltaLog,
+      relation.startingVersion.get,
+      endVersion,
+      spark,
+      readSchemaSnapshot = Some(snapshotForBatchSchema))
+  }
+}

gluten-delta/src-delta24/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala

@@ -0,0 +1,44 @@
+/*
+ * 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
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.delta.BatchCDFSchemaEndVersion
+import org.apache.spark.sql.delta.commands.cdc.CDCReader
+
+object DeltaCDFRelationHelper {
+  def changesToBatchDF(
+      relation: CDCReader.DeltaCDFRelation,
+      spark: SparkSession): DataFrame = {
+    val deltaLog = relation.snapshotWithSchemaMode.snapshot.deltaLog
+    val latestVersion = deltaLog.update().version
+    val endingVersionForBatchSchema =
+      relation.endingVersion.map(v => latestVersion.min(v)).getOrElse(latestVersion)
+    val snapshotForBatchSchema = relation.snapshotWithSchemaMode.schemaMode match {
+      case BatchCDFSchemaEndVersion => deltaLog.getSnapshotAt(endingVersionForBatchSchema)
+      case _ => relation.snapshotWithSchemaMode.snapshot
+    }
+    val endVersion = relation.endingVersion.getOrElse(latestVersion)
+
+    CDCReader.changesToBatchDF(
+      deltaLog,
+      relation.startingVersion.get,
+      endVersion,
+      spark,
+      readSchemaSnapshot = Some(snapshotForBatchSchema))
+  }
+}

gluten-delta/src-delta33/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala

@@ -0,0 +1,44 @@
+/*
+ * 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
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.delta.BatchCDFSchemaEndVersion
+import org.apache.spark.sql.delta.commands.cdc.CDCReader
+
+object DeltaCDFRelationHelper {
+  def changesToBatchDF(
+      relation: CDCReader.DeltaCDFRelation,
+      spark: SparkSession): DataFrame = {
+    val deltaLog = relation.snapshotWithSchemaMode.snapshot.deltaLog
+    val latestVersion = deltaLog.update().version
+    val endingVersionForBatchSchema =
+      relation.endingVersion.map(v => latestVersion.min(v)).getOrElse(latestVersion)
+    val snapshotForBatchSchema = relation.snapshotWithSchemaMode.schemaMode match {
+      case BatchCDFSchemaEndVersion => deltaLog.getSnapshotAt(endingVersionForBatchSchema)
+      case _ => relation.snapshotWithSchemaMode.snapshot
+    }
+    val endVersion = relation.endingVersion.getOrElse(latestVersion)
+
+    CDCReader.changesToBatchDF(
+      deltaLog,
+      relation.startingVersion.get,
+      endVersion,
+      spark,
+      readSchemaSnapshot = Some(snapshotForBatchSchema))
+  }
+}

gluten-delta/src-delta40/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala

@@ -0,0 +1,47 @@
+/*
+ * 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
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.delta.BatchCDFSchemaEndVersion
+import org.apache.spark.sql.delta.commands.cdc.CDCReader
+
+object DeltaCDFRelationHelper {
+  def changesToBatchDF(
+      relation: CDCReader.DeltaCDFRelation,
+      spark: SparkSession): DataFrame = {
+    val deltaLog = relation.snapshotWithSchemaMode.snapshot.deltaLog
+    val latestVersion = deltaLog.update(catalogTableOpt = relation.catalogTableOpt).version
+    val endingVersionForBatchSchema =
+      relation.endingVersion.map(v => latestVersion.min(v)).getOrElse(latestVersion)
+    val snapshotForBatchSchema = relation.snapshotWithSchemaMode.schemaMode match {
+      case BatchCDFSchemaEndVersion =>
+        deltaLog.getSnapshotAt(
+          endingVersionForBatchSchema,
+          catalogTableOpt = relation.catalogTableOpt)
+      case _ => relation.snapshotWithSchemaMode.snapshot
+    }
+    val endVersion = relation.endingVersion.getOrElse(latestVersion)
+
+    CDCReader.changesToBatchDF(
+      deltaLog,
+      relation.startingVersion.get,
+      endVersion,
+      spark,
+      readSchemaSnapshot = Some(snapshotForBatchSchema))
+  }
+}

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

@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference,
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.connector.read.streaming.SparkDataStream
 import org.apache.spark.sql.delta.{DeltaParquetFileFormat, NoMapping}
+import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeRemoveFileIndex}
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.datasources.{FilePartition, HadoopFsRelation}
 import org.apache.spark.sql.types.StructType
@@ -61,6 +62,27 @@ case class DeltaScanTransformer(
 
   override lazy val fileFormat: ReadFileFormat = ReadFileFormat.ParquetReadFormat
 
+  // Delta CDF over a deletion-vector-enabled table needs DV-aware, row-level reconciliation that
+  // the native scan path does not do yet: it would surface rows that are still live (not covered
+  // by the DV) as CDF `delete` change rows. Fall back to Spark for both CDF scan sides -- the add
+  // side (`CdcAddFileIndex`) and the remove side (`TahoeRemoveFileIndex`) -- whenever the touched
+  // files carry DVs. Normal (non-CDF) DV scans are unaffected: those apply the DV natively through
+  // the per-file split-info handoff and never reach this guard.
+  override protected def doValidateInternal(): ValidationResult = {
+    if (cdfFilesHaveDeletionVectors) {
+      return ValidationResult.failed(DeltaScanTransformer.DELETION_VECTOR_UNSUPPORTED)
+    }
+    super.doValidateInternal()
+  }
+
+  private def cdfFilesHaveDeletionVectors: Boolean = relation.location match {
+    case index: TahoeRemoveFileIndex =>
+      index.filesByVersion.exists(_.actions.exists(_.deletionVector != null))
+    case index: CdcAddFileIndex =>
+      index.addFiles.exists(_.deletionVector != null)
+    case _ => false
+  }
+
   // For Delta column-mapping tables, `dataFilters` on the scan node are LOGICAL-named so Delta's
   // file index (`PreparedDeltaFileIndex.matchingFiles`, `Snapshot.filesForScan`) can do partition
   // pruning and stats-based file skipping -- both resolve filter attrs against logical schemas.
@@ -138,6 +160,9 @@ 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,