This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new ddca36e75a [VL] Delta: Support Delta CDF scan offload (#12218)
ddca36e75a is described below
commit ddca36e75aeeba347b9ef9e66e2688f277dc574c
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Thu Aug 13 15:32:19 2026 +0300
[VL] Delta: Support Delta CDF scan offload (#12218)
---
.../util/delta-spark-ut/known-failures.txt | 13 ++
.../gluten/component/VeloxDeltaComponent.scala | 14 +-
.../apache/gluten/config/VeloxDeltaConfig.scala | 9 +
.../apache/gluten/execution/VeloxDeltaSuite.scala | 35 +++-
docs/get-started/VeloxDelta.md | 6 +
.../gluten/extension/DeltaCDFRelationHelper.scala | 44 +++++
.../gluten/extension/DeltaCDFRelationHelper.scala | 44 +++++
.../gluten/extension/DeltaCDFRelationHelper.scala | 44 +++++
.../gluten/extension/DeltaCDFRelationHelper.scala | 47 +++++
.../gluten/execution/DeltaScanTransformer.scala | 25 +++
.../apache/gluten/extension/DeltaCDFScanRule.scala | 157 ++++++++++++++++
.../apache/gluten/extension/OffloadDeltaScan.scala | 8 +-
.../org/apache/gluten/execution/DeltaSuite.scala | 203 ++++++++++++++++++++-
13 files changed, 644 insertions(+), 5 deletions(-)
diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt
b/.github/workflows/util/delta-spark-ut/known-failures.txt
index bc55624d49..810c5bc58f 100644
--- a/.github/workflows/util/delta-spark-ut/known-failures.txt
+++ b/.github/workflows/util/delta-spark-ut/known-failures.txt
@@ -73,6 +73,19 @@
org.apache.spark.sql.delta.DeltaAlterTableByNameIdColumnMappingSuite#CHANGE COLU
org.apache.spark.sql.delta.DeltaAlterTableByNameNameColumnMappingSuite#CHANGE
COLUMN - case insensitive - column mapping name mode
org.apache.spark.sql.delta.DeltaAlterTableByNameNameColumnMappingSuite#CHANGE
COLUMN - move to first (nested) - column mapping name mode
org.apache.spark.sql.delta.DeltaArbitraryColumnNameSuite#create table
+org.apache.spark.sql.delta.DeltaCDCIdColumnMappingSuite#filters with special
characters in name should be pushed down - column mapping id mode
+org.apache.spark.sql.delta.DeltaCDCNameColumnMappingSuite#filters with special
characters in name should be pushed down - column mapping name mode
+org.apache.spark.sql.delta.DeltaCDCSQLIdColumnMappingSuite#filters with
special characters in name should be pushed down - column mapping id mode
+org.apache.spark.sql.delta.DeltaCDCSQLNameColumnMappingSuite#filters with
special characters in name should be pushed down - column mapping name mode
+org.apache.spark.sql.delta.DeltaCDCSQLSuite#select individual column should
push down filters
+org.apache.spark.sql.delta.DeltaCDCSQLWithCatalogOwnedBatch100Suite#select
individual column should push down filters
+org.apache.spark.sql.delta.DeltaCDCSQLWithCatalogOwnedBatch1Suite#select
individual column should push down filters
+org.apache.spark.sql.delta.DeltaCDCSQLWithCatalogOwnedBatch2Suite#select
individual column should push down filters
+org.apache.spark.sql.delta.DeltaCDCScalaSuite#filters should be pushed down
+org.apache.spark.sql.delta.DeltaCDCScalaWithCatalogOwnedBatch100Suite#filters
should be pushed down
+org.apache.spark.sql.delta.DeltaCDCScalaWithCatalogOwnedBatch1Suite#filters
should be pushed down
+org.apache.spark.sql.delta.DeltaCDCScalaWithCatalogOwnedBatch2Suite#filters
should be pushed down
+org.apache.spark.sql.delta.DeltaCDCScalaWithDeletionVectorsSuite#filters
should be pushed down
org.apache.spark.sql.delta.DeltaCDCStreamDeletionVectorSuite#cdc streams with
noop merge
org.apache.spark.sql.delta.DeltaCDCStreamSuite#cdc streams with noop merge
org.apache.spark.sql.delta.DeltaCDCStreamWithCatalogManagedBatch100Suite#cdc
streams with noop merge
diff --git
a/backends-velox/src-delta/main/scala/org/apache/gluten/component/VeloxDeltaComponent.scala
b/backends-velox/src-delta/main/scala/org/apache/gluten/component/VeloxDeltaComponent.scala
index 164cf52886..d9bdc4e392 100644
---
a/backends-velox/src-delta/main/scala/org/apache/gluten/component/VeloxDeltaComponent.scala
+++
b/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.{DeltaCDFScanRule, 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,16 @@ class VeloxDeltaComponent extends Component {
}
override def injectRules(injector: Injector): Unit = {
+ // Expands Delta CDF relations while the plan is still logical, so the
Delta file scans they
+ // read reach the offload rules below and Spark's optimizer handles their
predicate pushdown.
+ // Must not run earlier than the optimizer: analysis is eager at Dataset
creation, and an
+ // open-ended CDF range has to resolve to the latest version at execution
time.
+ injector.spark.injectOptimizerRule(
+ spark =>
+ DeltaCDFScanRule(
+ 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
diff --git
a/backends-velox/src-delta/main/scala/org/apache/gluten/config/VeloxDeltaConfig.scala
b/backends-velox/src-delta/main/scala/org/apache/gluten/config/VeloxDeltaConfig.scala
index 99c2d2c26a..566a17aab7 100644
---
a/backends-velox/src-delta/main/scala/org/apache/gluten/config/VeloxDeltaConfig.scala
+++
b/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)
}
diff --git
a/backends-velox/src-delta/test/scala/org/apache/gluten/execution/VeloxDeltaSuite.scala
b/backends-velox/src-delta/test/scala/org/apache/gluten/execution/VeloxDeltaSuite.scala
index d7a12d1fc5..851cf1dcbc 100644
---
a/backends-velox/src-delta/test/scala/org/apache/gluten/execution/VeloxDeltaSuite.scala
+++
b/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 {
+ test("delta: change data feed scan offload can be disabled") {
+ 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)
+ }
+ }
+ }
+}
diff --git a/docs/get-started/VeloxDelta.md b/docs/get-started/VeloxDelta.md
index 69964e7890..3c594b9b97 100644
--- a/docs/get-started/VeloxDelta.md
+++ b/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 |
diff --git
a/gluten-delta/src-delta23/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
b/gluten-delta/src-delta23/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
new file mode 100644
index 0000000000..30bdb75e99
--- /dev/null
+++
b/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))
+ }
+}
diff --git
a/gluten-delta/src-delta24/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
b/gluten-delta/src-delta24/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
new file mode 100644
index 0000000000..30bdb75e99
--- /dev/null
+++
b/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))
+ }
+}
diff --git
a/gluten-delta/src-delta33/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
b/gluten-delta/src-delta33/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
new file mode 100644
index 0000000000..30bdb75e99
--- /dev/null
+++
b/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))
+ }
+}
diff --git
a/gluten-delta/src-delta40/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
b/gluten-delta/src-delta40/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala
new file mode 100644
index 0000000000..f1f84cd153
--- /dev/null
+++
b/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))
+ }
+}
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
index 86667d988f..c2cb6604db 100644
---
a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
+++
b/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,
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanRule.scala
b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanRule.scala
new file mode 100644
index 0000000000..f4b993cece
--- /dev/null
+++
b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanRule.scala
@@ -0,0 +1,157 @@
+/*
+ * 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,
NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation,
LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.delta.actions.{AddFile, RemoveFile}
+import org.apache.spark.sql.delta.commands.cdc.CDCReader
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+
+import scala.util.control.NonFatal
+
+/**
+ * Expands Delta's `DeltaCDFRelation` into the underlying change-data plan so
the Delta file scans
+ * it reads become visible to Gluten's normal scan offload rules. Without
this, Spark plans the CDF
+ * relation as a `RowDataSourceScanExec` whose file scans are hidden inside an
RDD, and no columnar
+ * rule can reach them.
+ *
+ * This runs as a logical optimizer rule. Only an output-restoring projection
is needed: the
+ * replacement keeps the original relation's expression IDs, and because the
rule participates in
+ * the operator optimization batch, Spark's own optimizer then pushes the
surrounding filters into
+ * the expanded file scans. Expanding later -- during physical planning --
would require re-running
+ * the optimizer out of band and hand-preserving expression IDs against
operators that were already
+ * planned, which is fragile (a collapsed alias there surfaces as `Couldn't
find <attr>` at shuffle
+ * binding time).
+ *
+ * The phase also has to be no earlier than the optimizer. Analysis runs
eagerly when a Dataset is
+ * created, while optimization is deferred to the first action, and Delta
resolves an open-ended CDF
+ * range (no ending version) to the latest version at execution time.
Expanding during analysis
+ * would snapshot that range at Dataset creation and silently drop commits
made before the action --
+ * see `DeltaCDCSQLSuite."ending version not specified resolves to latest at
execution time"`.
+ */
+case class DeltaCDFScanRule(spark: SparkSession, offloadEnabled: () => Boolean)
+ extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!offloadEnabled()) {
+ return plan
+ }
+
+ // transformUp, not resolveOperatorsUp: by the optimizer the plan is
already marked analyzed,
+ // and resolveOperatorsUp skips analyzed subtrees, so it would never reach
the relation. The
+ // analyzer's assertion against transformUp does not apply outside the
analyzer. Expansion is
+ // idempotent -- the replacement holds no DeltaCDFRelation -- so
re-running the operator
+ // optimization batch to fixed point cannot expand twice.
+ plan.transformUp {
+ case relation: LogicalRelation =>
+ relation.relation match {
+ case cdfRelation: CDCReader.DeltaCDFRelation =>
+ expandOrDecline(relation, cdfRelation)
+ case _ => relation
+ }
+ }
+ }
+
+ // Leave the relation untouched on any failure so Delta's own CDF path runs.
This rule executes
+ // during analysis, so an escaping exception would fail the query outright
-- including for
+ // metadata-only calls such as `explain()` -- where vanilla Spark would have
succeeded. Offload is
+ // an optimization; it must never be the reason a CDF query stops working.
+ private def expandOrDecline(
+ relation: LogicalRelation,
+ cdfRelation: CDCReader.DeltaCDFRelation): LogicalPlan = {
+ try {
+ if (changesContainDeletionVectors(cdfRelation)) {
+ relation
+ } else {
+ expand(relation, cdfRelation)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning("Failed to expand Delta CDF relation, falling back to
Delta's CDF scan.", e)
+ relation
+ }
+ }
+
+ // Delta CDF over a commit that uses deletion vectors needs Delta's
DV-aware, row-level
+ // reconciliation: a deleted row stays physically in its data file and is
masked by a DV, so the
+ // CDF `delete` rows for a commit are only the rows the DV newly marks.
Expanding the analyzed
+ // CDF batch plan here loses that reconciliation on some Delta versions
(e.g. Delta 2.4 / Spark
+ // 3.4), where the remove side would then surface every row of a
logically-removed file as a
+ // `delete` change row instead of just the DV-marked ones.
+ //
+ // The table property is not a reliable guard: it controls whether future
writes may create DVs,
+ // while existing DVs can remain after the property is disabled. Inspect the
AddFile/RemoveFile
+ // actions in the requested CDF interval instead. Decline to expand only
ranges that actually
+ // contain DVs and let Delta's DeltaCDFRelation apply them correctly. The
matching physical-scan
+ // guard in DeltaScanTransformer remains as a backstop.
+ private def changesContainDeletionVectors(
+ cdfRelation: CDCReader.DeltaCDFRelation): Boolean = {
+ cdfRelation.startingVersion.exists {
+ startVersion =>
+ val changes =
+
cdfRelation.snapshotWithSchemaMode.snapshot.deltaLog.getChanges(startVersion)
+ val changesInRange = cdfRelation.endingVersion match {
+ case Some(endVersion) => changes.takeWhile { case (version, _) =>
version <= endVersion }
+ case None => changes
+ }
+ changesInRange.exists {
+ case (_, actions) =>
+ actions.exists {
+ case file: AddFile => file.deletionVector != null
+ case file: RemoveFile => file.deletionVector != null
+ case _ => false
+ }
+ }
+ }
+ }
+
+ private def expand(
+ relation: LogicalRelation,
+ cdfRelation: CDCReader.DeltaCDFRelation): LogicalPlan = {
+ // Delta represents an empty CDF range with `emptyCDFRelation`, whose
`startingVersion` is unset
+ // and whose scan is an empty RDD. There is nothing to expand, and asking
Delta for the changes
+ // would fail on `startingVersion.get`.
+ if (cdfRelation.startingVersion.isEmpty) {
+ return LocalRelation(relation.output)
+ }
+
+ val cdfPlan =
+ DeltaCDFRelationHelper.changesToBatchDF(cdfRelation,
spark).queryExecution.analyzed
+ val cdfOutput = cdfPlan.output
+ // Bind the expanded plan back to the expression IDs the rest of the query
already references.
+ val projects: Seq[NamedExpression] = relation.output.map {
+ attr =>
+ Alias(resolveCDFAttribute(attr, cdfOutput), attr.name)(
+ exprId = attr.exprId,
+ qualifier = attr.qualifier,
+ explicitMetadata = Some(attr.metadata))
+ }
+ Project(projects, cdfPlan)
+ }
+
+ private def resolveCDFAttribute(attr: Attribute, 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("[", ", ", "]")}"))
+ }
+}
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala
b/gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala
index ebafb0c08c..4664da8769 100644
---
a/gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala
+++
b/gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala
@@ -23,7 +23,7 @@ import
org.apache.gluten.extension.columnar.offload.OffloadSingleNode
import org.apache.spark.sql.delta.DeltaParquetFileFormat
import org.apache.spark.sql.delta.SnapshotDescriptor
import
org.apache.spark.sql.delta.commands.DeletionVectorUtils.deletionVectorsReadable
-import org.apache.spark.sql.delta.files.TahoeFileIndex
+import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeFileIndex,
TahoeRemoveFileIndex}
import org.apache.spark.sql.delta.stats.PreparedDeltaFileIndex
import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
import org.apache.spark.util.SparkVersionUtil
@@ -97,6 +97,12 @@ case class OffloadDeltaScan() extends OffloadSingleNode {
private def containsDeletionVector(scan: FileSourceScanExec): Boolean = {
scan.relation.location match {
+ // CDF indexes expose the exact actions in the requested range. Use
those instead of the
+ // table-level protocol capability so a DV-capable table can still
offload DV-free ranges.
+ case index: TahoeRemoveFileIndex =>
+ index.filesByVersion.exists(_.actions.exists(_.deletionVector != null))
+ case index: CdcAddFileIndex =>
+ index.addFiles.exists(_.deletionVector != null)
case preparedIndex: PreparedDeltaFileIndex =>
preparedIndex.preparedScan.files.exists(_.deletionVector != null)
case index: TahoeFileIndex =>
diff --git
a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
index b2f4122039..138d318cff 100644
--- a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
+++ b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
@@ -19,7 +19,7 @@ package org.apache.gluten.execution
import org.apache.gluten.extension.DeltaPostTransformRules
import org.apache.spark.SparkConf
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.types._
import org.apache.spark.util.SparkVersionUtil
@@ -350,6 +350,207 @@ abstract class DeltaSuite extends
WholeStageTransformerSuite {
}
}
+ test("delta: change data feed read") {
+ withTable("delta_cdf") {
+ spark.sql(s"""
+ |create table delta_cdf (id int, name string) using delta
+ |tblproperties ("delta.enableChangeDataFeed" = "true")
+ |""".stripMargin)
+ spark.sql(s"""
+ |insert into delta_cdf values (1, "v1"), (2, "v2")
+ |""".stripMargin)
+ spark.sql(s"""
+ |update delta_cdf set name = "v2_updated" where id = 2
+ |""".stripMargin)
+ spark.sql(s"""
+ |delete from delta_cdf where id = 1
+ |""".stripMargin)
+
+ val tableChangesFromZeroDF = runAndCompare(
+ s"""
+ |select id, name, _change_type, _commit_version
+ |from table_changes('delta_cdf', 0)
+ |order by _commit_version, id, name, _change_type
+ |""".stripMargin)
+ checkCDFRead(tableChangesFromZeroDF)
+
+ val tableChangesDF = runAndCompare(
+ s"""
+ |select id, name, _change_type, _commit_version
+ |from table_changes('delta_cdf', 1)
+ |order by _commit_version, id, name, _change_type
+ |""".stripMargin)
+ checkCDFRead(tableChangesDF)
+
+ val filteredCDF = runAndCompare(
+ s"""
+ |select id, name, _change_type, _commit_version
+ |from table_changes('delta_cdf', 1)
+ |where _commit_version = 2 and id = 2
+ |order by name, _change_type
+ |""".stripMargin)
+ checkCDFRead(
+ filteredCDF,
+ Seq(
+ Row(2, "v2", "update_preimage", 2L),
+ Row(2, "v2_updated", "update_postimage", 2L)))
+ assert(
+ collect(filteredCDF.queryExecution.executedPlan) {
+ case scan: DeltaScanTransformer =>
+ scan.dataFilters.exists(_.references.exists(_.name == "id"))
+ }.contains(true),
+ filteredCDF.queryExecution.executedPlan
+ )
+
+ val boundedCDF = runAndCompare(
+ s"""
+ |select id, name, _change_type, _commit_version
+ |from table_changes('delta_cdf', 1, 2)
+ |order by _commit_version, id, name, _change_type
+ |""".stripMargin)
+ checkCDFRead(
+ boundedCDF,
+ Seq(
+ Row(1, "v1", "insert", 1L),
+ Row(2, "v2", "insert", 1L),
+ Row(2, "v2", "update_preimage", 2L),
+ Row(2, "v2_updated", "update_postimage", 2L)))
+
+ val readChangeFeedDF = compareCDFDataFrame(
+ () =>
+ spark.read
+ .format("delta")
+ .option("readChangeFeed", "true")
+ .option("startingVersion", "1")
+ .table("delta_cdf")
+ .selectExpr("id", "name", "_change_type", "_commit_version")
+ .orderBy("_commit_version", "id", "name", "_change_type"))
+ checkCDFRead(readChangeFeedDF)
+ }
+ }
+
+ test("delta: change data feed read with column mapping") {
+ withTable("delta_cdf_cm") {
+ spark.sql(s"""
+ |create table delta_cdf_cm (id int, name string) using delta
+ |tblproperties (
+ | "delta.enableChangeDataFeed" = "true",
+ | "delta.columnMapping.mode" = "name")
+ |""".stripMargin)
+ spark.sql(s"""
+ |insert into delta_cdf_cm values (1, "v1"), (2, "v2")
+ |""".stripMargin)
+ spark.sql(s"""
+ |update delta_cdf_cm set name = "v2_updated" where id = 2
+ |""".stripMargin)
+ spark.sql(s"""
+ |delete from delta_cdf_cm where id = 1
+ |""".stripMargin)
+
+ val df = runAndCompare(
+ s"""
+ |select id, name, _change_type, _commit_version
+ |from table_changes('delta_cdf_cm', 1)
+ |order by _commit_version, id, name, _change_type
+ |""".stripMargin)
+ checkCDFRead(df)
+ }
+ }
+
+ testWithMinSparkVersion("delta: change data feed read with deletion
vectors", "3.4") {
+ withTable("delta_cdf_dv") {
+ spark.sql(s"""
+ |create table delta_cdf_dv (id int, name string) using delta
+ |tblproperties (
+ | "delta.enableChangeDataFeed" = "true",
+ | "delta.enableDeletionVectors" = "true")
+ |""".stripMargin)
+ spark.sql(s"""
+ |insert into delta_cdf_dv values (1, "v1"), (2, "v2"), (3,
"v3")
+ |""".stripMargin)
+
+ // Enabling DV writes does not mean this CDF range contains a DV. The
insert-only range must
+ // still be eligible for native scan offload.
+ val insertOnlyDF = runAndCompare(
+ s"""
+ |select id, name, _change_type
+ |from table_changes('delta_cdf_dv', 0, 1)
+ |order by id, name, _change_type
+ |""".stripMargin)
+ assert(
+ collect(insertOnlyDF.queryExecution.executedPlan) {
+ case _: DeltaScanTransformer => true
+ }.nonEmpty,
+ insertOnlyDF.queryExecution.executedPlan)
+ checkAnswer(
+ insertOnlyDF,
+ Seq(
+ Row(1, "v1", "insert"),
+ Row(2, "v2", "insert"),
+ Row(3, "v3", "insert")))
+
+ spark.sql(s"""
+ |delete from delta_cdf_dv where id = 2
+ |""".stripMargin)
+ spark.sql(s"""
+ |alter table delta_cdf_dv set tblproperties (
+ | "delta.enableDeletionVectors" = "false")
+ |""".stripMargin)
+
+ // Disabling future DV writes does not remove existing DVs. This range
contains the DV-backed
+ // delete, so Gluten keeps the whole CDF read on Spark and lets Delta
perform row-level
+ // reconciliation. Still-live rows must not be surfaced as `delete`
change rows.
+ val df = runAndCompare(
+ s"""
+ |select id, name, _change_type
+ |from table_changes('delta_cdf_dv', 0)
+ |order by id, name, _change_type
+ |""".stripMargin)
+ assert(
+ collect(df.queryExecution.executedPlan) { case d: DeltaScanTransformer
=> d }.isEmpty,
+ df.queryExecution.executedPlan)
+ checkAnswer(
+ df,
+ Seq(
+ Row(1, "v1", "insert"),
+ Row(2, "v2", "delete"),
+ Row(2, "v2", "insert"),
+ Row(3, "v3", "insert")))
+ }
+ }
+
+ private def compareCDFDataFrame(dataframe: () => DataFrame): DataFrame = {
+ var expected: Seq[Row] = null
+ withSQLConf(vanillaSparkConfs(): _*) {
+ expected = dataframe().collect()
+ }
+ val df = dataframe()
+ checkAnswer(df, expected)
+ df
+ }
+
+ private def checkCDFRead(
+ df: DataFrame,
+ expectedRows: Seq[Row] = allCDFRows): Unit = {
+ // Delta CDF expansion can keep a Spark-side branch for synthesized change
rows; this PR
+ // verifies the Delta file scans in the expanded plan are transformed.
+ checkLengthAndPlan(df, expectedRows.length)
+ checkAnswer(
+ df,
+ expectedRows)
+ assert(
+ collect(df.queryExecution.executedPlan) { case _: DeltaScanTransformer
=> true }.nonEmpty,
+ df.queryExecution.executedPlan)
+ }
+
+ private def allCDFRows: Seq[Row] =
+ Seq(
+ Row(1, "v1", "insert", 1L),
+ Row(2, "v2", "insert", 1L),
+ Row(2, "v2", "update_preimage", 2L),
+ Row(2, "v2_updated", "update_postimage", 2L),
+ Row(1, "v1", "delete", 3L))
+
test("column mapping with complex type") {
withTable("t1") {
val simpleNestedSchema = new StructType()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
