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..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,8 +17,8 @@ package org.apache.gluten.component import org.apache.gluten.backendsapi.velox.VeloxBackend -import org.apache.gluten.config.GlutenConfig -import org.apache.gluten.extension.{DeltaPostTransformRules, OffloadDeltaFilter, OffloadDeltaProject, OffloadDeltaScan} +import org.apache.gluten.config.{GlutenConfig, VeloxDeltaConfig} +import org.apache.gluten.extension.{DeltaCDFScanStrategy, DeltaPostTransformRules, OffloadDeltaFilter, OffloadDeltaProject, OffloadDeltaScan} import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform import org.apache.gluten.extension.columnar.validator.Validators import org.apache.gluten.extension.injector.Injector @@ -35,6 +35,12 @@ class VeloxDeltaComponent extends Component { } override def injectRules(injector: Injector): Unit = { + injector.spark.injectPlannerStrategy( + spark => + DeltaCDFScanStrategy( + spark, + () => new VeloxDeltaConfig(spark.sessionState.conf).enableChangeDataFeedScan)) + val legacy = injector.gluten.legacy // Deletion-vector scans need no Gluten-side logical preprocessing: Delta's own // PreprocessTableWithDVsStrategy injects the skip-row column and filter during physical 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-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 86667d988ff..c2cb6604db8 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..045a758b155 --- /dev/null +++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaCDFScanStrategy.scala @@ -0,0 +1,153 @@ +/* + * 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.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 + +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 + // 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, + 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)) + + // This strategy expands the CDF relation during physical planning, after Spark's normal + // 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 + } + 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/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 => 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 9f8e994f8d0..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 @@ -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 @@ -363,6 +363,207 @@ 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))) + 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) + } + } + + 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) + + // 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)) + testWithMinSparkVersion("column mapping with complex type", "3.2") { withTable("t1") { val simpleNestedSchema = new StructType()