[VL][Delta] Add CDF offload guard

Author: malinjawi

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

@@ -17,7 +17,7 @@
 package org.apache.gluten.component
 
 import org.apache.gluten.backendsapi.velox.VeloxBackend
-import org.apache.gluten.config.GlutenConfig
+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
@@ -35,7 +35,11 @@ class VeloxDeltaComponent extends Component {
   }
 
   override def injectRules(injector: Injector): Unit = {
-    injector.spark.injectPlannerStrategy(DeltaCDFScanStrategy(_))
+    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

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/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala

@@ -25,16 +25,23 @@ 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
-            if !changesContainDeletionVectors(cdfRelation) =>
-          planCDFRelation(relation, cdfRelation, projects, filters).map(planLater).toSeq
-        case _ => Nil
-      }
-    case _ => Nil
+case class DeltaCDFScanStrategy(spark: SparkSession, offloadEnabled: () => Boolean)
+  extends SparkStrategy {
+  override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
+    if (!offloadEnabled()) {
+      return Nil
+    }
+
+    plan match {
+      case PhysicalOperation(projects, filters, relation: LogicalRelation) =>
+        relation.relation match {
+          case cdfRelation: CDCReader.DeltaCDFRelation
+              if !changesContainDeletionVectors(cdfRelation) =>
+            planCDFRelation(relation, cdfRelation, projects, filters).map(planLater).toSeq
+          case _ => Nil
+        }
+      case _ => Nil
+    }
   }
 
   // Delta CDF over a commit that uses deletion vectors needs Delta's DV-aware, row-level