From 582641ac67b456b0314e2f4c1a39b95cd868597b Mon Sep 17 00:00:00 2001 From: malinjawi Date: Tue, 16 Jun 2026 12:46:44 +0300 Subject: [PATCH 1/6] [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. Re-planning Delta's analyzed CDF batch plan also drops that reconciliation on some Delta versions (Delta 2.4 / Spark 3.4), where the remove side would surface every row of a logically removed file as a `delete` change row. The planner strategy therefore declines to intercept CDF reads whose table has deletion vectors enabled and leaves them to Delta's own DeltaCDFRelation scan, which reconciles DVs correctly on every supported Delta version; DeltaScanTransformer additionally guards both CDF scan sides -- the add side (CdcAddFileIndex) and the remove side (TahoeRemoveFileIndex) -- as a backstop. Normal (non-CDF) DV scans are unaffected and continue to apply the DV natively. Addresses #12195. --- .../component/VeloxDeltaComponent.scala | 4 +- .../extension/DeltaCDFRelationHelper.scala | 44 +++++ .../extension/DeltaCDFRelationHelper.scala | 44 +++++ .../extension/DeltaCDFRelationHelper.scala | 44 +++++ .../extension/DeltaCDFRelationHelper.scala | 47 +++++ .../execution/DeltaScanTransformer.scala | 25 +++ .../extension/DeltaCDFScanStrategy.scala | 112 ++++++++++++ .../apache/gluten/execution/DeltaSuite.scala | 172 +++++++++++++++++- 8 files changed, 490 insertions(+), 2 deletions(-) create mode 100644 gluten-delta/src-delta23/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala create mode 100644 gluten-delta/src-delta24/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala create mode 100644 gluten-delta/src-delta33/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala create mode 100644 gluten-delta/src-delta40/main/scala/org/apache/gluten/extension/DeltaCDFRelationHelper.scala create mode 100644 gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala 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 164cf528860..b96f07cce05 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 @@ -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 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 00000000000..30bdb75e99a --- /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 00000000000..30bdb75e99a --- /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 00000000000..30bdb75e99a --- /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 00000000000..f1f84cd1537 --- /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 6ac644622dc..1ae44749c37 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/DeltaCDFScanStrategy.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala new file mode 100644 index 00000000000..4d23a335ef2 --- /dev/null +++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala @@ -0,0 +1,112 @@ +/* + * 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 if !touchesDeletionVectors(cdfRelation) => + planCDFRelation(relation, cdfRelation, projects, filters).map(planLater).toSeq + case _ => Nil + } + case _ => Nil + } + + // Delta CDF over a deletion-vector table 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. Re-planning 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. Decline to intercept such reads and let Spark serve them through + // Delta's own DeltaCDFRelation scan, which applies DVs correctly. Offloading loses nothing here: + // the native CDF scan path does not support DV reconciliation either (see the matching guard in + // DeltaScanTransformer). + private def touchesDeletionVectors(cdfRelation: CDCReader.DeltaCDFRelation): Boolean = + cdfRelation.snapshotWithSchemaMode.snapshot.metadata.configuration + .get("delta.enableDeletionVectors") + .exists(_.equalsIgnoreCase("true")) + + 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("[", ", ", "]")}")) + } +} 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 c138bed1cdf..7919e926606 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 @@ -17,7 +17,7 @@ package org.apache.gluten.execution 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 @@ -361,6 +361,176 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { } } + testWithMinSparkVersion("delta: change data feed read", "3.2") { + 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))) + + 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) + } + } + + testWithMinSparkVersion("delta: change data feed read with column mapping", "3.2") { + 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) + spark.sql(s""" + |delete from delta_cdf_dv where id = 2 + |""".stripMargin) + + // CDF over a deletion-vector table needs Delta's DV-aware row-level reconciliation, which the + // native scan path does not do. Gluten keeps the whole CDF read on Spark (no + // DeltaScanTransformer offload) so that still-live, DV-masked rows are not surfaced as + // `delete` change rows. The result must still match vanilla Spark. + 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)) + testWithMinSparkVersion("column mapping with complex type", "3.2") { withTable("t1") { val simpleNestedSchema = new StructType() From 3b462034a2eb3ccf7427ddbc839280f5744c2363 Mon Sep 17 00:00:00 2001 From: malinjawi Date: Sun, 19 Jul 2026 13:48:15 +0300 Subject: [PATCH 2/6] [VL][Delta] Check CDF actions for deletion vectors Inspect AddFile and RemoveFile actions in the requested CDF version range instead of relying on the table property, which only controls future DV writes.\n\nCover both DV-enabled ranges without DV actions and existing DV actions after the property is disabled. --- .../extension/DeltaCDFScanStrategy.scala | 49 +++++++++++++------ .../apache/gluten/execution/DeltaSuite.scala | 32 ++++++++++-- 2 files changed, 63 insertions(+), 18 deletions(-) diff --git a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala index 4d23a335ef2..17be38aef21 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala +++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala @@ -20,6 +20,7 @@ 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.actions.{AddFile, RemoveFile} 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 @@ -28,26 +29,46 @@ 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 !touchesDeletionVectors(cdfRelation) => + case cdfRelation: CDCReader.DeltaCDFRelation + if !changesContainDeletionVectors(cdfRelation) => planCDFRelation(relation, cdfRelation, projects, filters).map(planLater).toSeq case _ => Nil } case _ => Nil } - // Delta CDF over a deletion-vector table 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. Re-planning 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. Decline to intercept such reads and let Spark serve them through - // Delta's own DeltaCDFRelation scan, which applies DVs correctly. Offloading loses nothing here: - // the native CDF scan path does not support DV reconciliation either (see the matching guard in - // DeltaScanTransformer). - private def touchesDeletionVectors(cdfRelation: CDCReader.DeltaCDFRelation): Boolean = - cdfRelation.snapshotWithSchemaMode.snapshot.metadata.configuration - .get("delta.enableDeletionVectors") - .exists(_.equalsIgnoreCase("true")) + // 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. Re-planning 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 intercept 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 planCDFRelation( relation: LogicalRelation, 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 add22d78aef..9e024cb1fae 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 @@ -474,14 +474,38 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { 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) - // CDF over a deletion-vector table needs Delta's DV-aware row-level reconciliation, which the - // native scan path does not do. Gluten keeps the whole CDF read on Spark (no - // DeltaScanTransformer offload) so that still-live, DV-masked rows are not surfaced as - // `delete` change rows. The result must still match vanilla Spark. + // 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 049e608a34b981e658266db8ab62fa254b7f8b5a Mon Sep 17 00:00:00 2001 From: malinjawi Date: Sun, 19 Jul 2026 15:00:39 +0300 Subject: [PATCH 3/6] [VL][Delta] Check CDF indexes for deletion vectors --- .../org/apache/gluten/extension/OffloadDeltaScan.scala | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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 ebafb0c08c3..4664da8769b 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 => From d740b0da2e4f458d0c251cb042e03ebc5fbe2f3e Mon Sep 17 00:00:00 2001 From: malinjawi Date: Fri, 24 Jul 2026 15:21:09 +0300 Subject: [PATCH 4/6] [VL][Delta] Preserve CDF filter pushdown --- .../org/apache/gluten/extension/DeltaCDFScanStrategy.scala | 6 +++++- .../scala/org/apache/gluten/execution/DeltaSuite.scala | 7 +++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala index 17be38aef21..bfb511f1a7e 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala +++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala @@ -84,7 +84,11 @@ case class DeltaCDFScanStrategy(spark: SparkSession) extends SparkStrategy { val cdfOutput = cdfPlan.output val rewrittenFilters = filters.map(rewriteExpression(_, cdfOutput)) val rewrittenProjects = projects.map(rewriteProject(_, cdfOutput)) - Some(projectAndFilter(cdfPlan, rewrittenProjects, rewrittenFilters)) + val rewrittenPlan = projectAndFilter(cdfPlan, rewrittenProjects, rewrittenFilters) + + // This strategy expands the CDF relation during physical planning, after Spark's normal + // optimizer pass. Optimize the resolved replacement so predicates can reach its file scans. + Some(spark.sessionState.optimizer.execute(rewrittenPlan)) } private def projectAndFilter( 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 9e024cb1fae..5d7377658e0 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 @@ -407,6 +407,13 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { 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""" From 65178209a09fa7cc768f46cdb532add64aac936a Mon Sep 17 00:00:00 2001 From: malinjawi Date: Fri, 24 Jul 2026 15:31:48 +0300 Subject: [PATCH 5/6] [VL][Delta] Add CDF offload guard --- .../component/VeloxDeltaComponent.scala | 8 +++-- .../gluten/config/VeloxDeltaConfig.scala | 9 +++++ .../gluten/execution/VeloxDeltaSuite.scala | 35 ++++++++++++++++++- docs/get-started/VeloxDelta.md | 6 ++++ .../extension/DeltaCDFScanStrategy.scala | 27 ++++++++------ 5 files changed, 72 insertions(+), 13 deletions(-) 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 b96f07cce05..c7146071770 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,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 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 99c2d2c26a7..566a17aab7a 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 d7a12d1fc5d..83810b9708d 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 { + 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) + } + } + } +} diff --git a/docs/get-started/VeloxDelta.md b/docs/get-started/VeloxDelta.md index 69964e78908..3c594b9b976 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/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala index bfb511f1a7e..219f8549036 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala +++ b/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 From bf2ac70f5aec2fffc8a95c8dc8a6bcc2ffcd0ad4 Mon Sep 17 00:00:00 2001 From: malinjawi Date: Fri, 24 Jul 2026 16:16:19 +0300 Subject: [PATCH 6/6] [VL][Delta] Preserve CDF output attributes --- .../gluten/extension/DeltaCDFScanStrategy.scala | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala index 219f8549036..045a758b155 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala +++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala @@ -91,22 +91,31 @@ case class DeltaCDFScanStrategy(spark: SparkSession, offloadEnabled: () => Boole val cdfOutput = cdfPlan.output val rewrittenFilters = filters.map(rewriteExpression(_, cdfOutput)) val rewrittenProjects = projects.map(rewriteProject(_, cdfOutput)) - val rewrittenPlan = projectAndFilter(cdfPlan, rewrittenProjects, rewrittenFilters) // This strategy expands the CDF relation during physical planning, after Spark's normal - // optimizer pass. Optimize the resolved replacement so predicates can reach its file scans. - Some(spark.sessionState.optimizer.execute(rewrittenPlan)) + // optimizer pass. Optimize the resolved child so predicates can reach its file scans, but keep + // the output-restoring projection outside the optimizer: parent operators were planned against + // the original relation's expression IDs, and Spark 3.3 can remove these aliases as redundant. + val optimizedChild = + spark.sessionState.optimizer.execute(filter(cdfPlan, rewrittenFilters)) + Some(Project(rewrittenProjects, optimizedChild)) } private def projectAndFilter( child: LogicalPlan, projects: Seq[NamedExpression], filters: Seq[Expression]): LogicalPlan = { + Project(projects, filter(child, filters)) + } + + private def filter( + child: LogicalPlan, + 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) + filtered } private def rewriteProject(