[VL][Delta] Support Delta CDF scan offload

Delta change-data-feed reads enter Spark as a CDCReader.DeltaCDFRelation,
which does not have the FileSourceScanExec + DeltaParquetFileFormat shape
that Gluten's Delta scan offload recognizes. Add a Gluten Delta planner
strategy (wired from VeloxDeltaComponent) that recognizes DeltaCDFRelation,
expands it through Delta's own CDF batch planning, and rewrites the
projection/filter attributes onto the expanded plan, so the existing Delta
scan offload path can plan the underlying CDF file scans as
DeltaScanTransformer.

CDF over a deletion-vector-enabled table is kept on Spark: native CDF lacks
the DV-aware row-level reconciliation, so an offloaded CDF scan would emit
still-live rows as `delete` change rows. DeltaScanTransformer falls back for
both CDF scan sides -- the add side (CdcAddFileIndex) and the remove side
(TahoeRemoveFileIndex) -- when the touched files carry DVs. Normal (non-CDF)
DV scans are unaffected and continue to apply the DV natively.

Addresses #12195.

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.{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,8 @@ class VeloxDeltaComponent extends Component {
   }
 
   override def injectRules(injector: Injector): Unit = {
+    injector.spark.injectPlannerStrategy(DeltaCDFScanStrategy(_))
+
     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

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,

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

@@ -0,0 +1,98 @@
+/*
+ * 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.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation, LogicalPlan, Project}
+import org.apache.spark.sql.delta.commands.cdc.CDCReader
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+
+case class DeltaCDFScanStrategy(spark: SparkSession) extends SparkStrategy {
+  override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+    case PhysicalOperation(projects, filters, relation: LogicalRelation) =>
+      relation.relation match {
+        case cdfRelation: CDCReader.DeltaCDFRelation =>
+          planCDFRelation(relation, cdfRelation, projects, filters).map(planLater).toSeq
+        case _ => Nil
+      }
+    case _ => Nil
+  }
+
+  private def planCDFRelation(
+      relation: LogicalRelation,
+      cdfRelation: CDCReader.DeltaCDFRelation,
+      projects: Seq[NamedExpression],
+      filters: Seq[Expression]): Option[LogicalPlan] = {
+    if (cdfRelation.startingVersion.isEmpty) {
+      return Some(projectAndFilter(LocalRelation(relation.output), projects, filters))
+    }
+
+    val cdfPlan =
+      DeltaCDFRelationHelper.changesToBatchDF(cdfRelation, spark).queryExecution.analyzed
+    val cdfOutput = cdfPlan.output
+    val rewrittenFilters = filters.map(rewriteExpression(_, cdfOutput))
+    val rewrittenProjects = projects.map(rewriteProject(_, cdfOutput))
+    Some(projectAndFilter(cdfPlan, rewrittenProjects, rewrittenFilters))
+  }
+
+  private def projectAndFilter(
+      child: LogicalPlan,
+      projects: Seq[NamedExpression],
+      filters: Seq[Expression]): LogicalPlan = {
+    val filtered = filters.reduceOption(org.apache.spark.sql.catalyst.expressions.And) match {
+      case Some(condition) => Filter(condition, child)
+      case None => child
+    }
+    Project(projects, filtered)
+  }
+
+  private def rewriteProject(
+      project: NamedExpression,
+      cdfOutput: Seq[Attribute]): NamedExpression = {
+    project match {
+      case attr: AttributeReference =>
+        Alias(
+          resolveCDFAttribute(attr, cdfOutput),
+          attr.name)(
+          exprId = attr.exprId,
+          qualifier = attr.qualifier,
+          explicitMetadata = Some(attr.metadata))
+      case other =>
+        rewriteExpression(other, cdfOutput).asInstanceOf[NamedExpression]
+    }
+  }
+
+  private def rewriteExpression(expr: Expression, cdfOutput: Seq[Attribute]): Expression = {
+    expr.transform {
+      case attr: AttributeReference => resolveCDFAttribute(attr, cdfOutput)
+    }
+  }
+
+  private def resolveCDFAttribute(
+      attr: AttributeReference,
+      cdfOutput: Seq[Attribute]): Attribute = {
+    cdfOutput
+      .find(output => spark.sessionState.conf.resolver(output.name, attr.name))
+      .getOrElse(
+        throw new IllegalArgumentException(
+          s"Unable to resolve CDF attribute ${attr.name} from " +
+            s"${cdfOutput.map(_.name).mkString("[", ", ", "]")}"))
+  }
+}