This is an automated email from the ASF dual-hosted git repository.
zhouyuan 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 2fe84a1a00 [VL][Delta] Add an opt-out for the native Delta DV DML
row-index scan (#12215)
2fe84a1a00 is described below
commit 2fe84a1a00842c519e1c3f12571114b10850c291
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Thu Sep 3 18:11:15 2026 +0300
[VL][Delta] Add an opt-out for the native Delta DV DML row-index scan
(#12215)
---
.../gluten/component/VeloxDeltaComponent.scala | 8 +-
.../apache/gluten/config/VeloxDeltaConfig.scala | 12 +
.../delta/DeltaDeletionVectorHandoffSuite.scala | 237 +++++++++++++++++++-
.../delta/DeltaDeletionVectorHandoffSuite.scala | 241 +++++++++++++++++++--
docs/get-started/VeloxDelta.md | 8 +
.../gluten/extension/DeltaPostTransformRules.scala | 9 +-
.../apache/gluten/extension/DeltaScanUtils.scala | 40 ++++
.../apache/gluten/extension/OffloadDeltaScan.scala | 65 ++++--
8 files changed, 572 insertions(+), 48 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 d9bdc4e392..d10ef08e41 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
@@ -52,7 +52,13 @@ class VeloxDeltaComponent extends Component {
// offloads, and DeltaScanTransformer materializes the per-file DV
payloads for Velox.
legacy.injectTransform {
c =>
- val offload = Seq(OffloadDeltaScan(), OffloadDeltaProject(),
OffloadDeltaFilter())
+ val offload = Seq(
+ OffloadDeltaScan(
+ enableNativeDmlRowIndexScan =
+ new VeloxDeltaConfig(c.sqlConf).enableNativeDmlRowIndexScan),
+ OffloadDeltaProject(),
+ OffloadDeltaFilter()
+ )
.map(_.toStrcitRule())
HeuristicTransform.Simple(
Validators.newValidator(new GlutenConfig(c.sqlConf), offload),
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 566a17aab7..948eec1f7f 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
@@ -23,6 +23,8 @@ class VeloxDeltaConfig(conf: SQLConf) extends
GlutenCoreConfig(conf) {
def enableNativeWrite: Boolean = getConf(ENABLE_NATIVE_WRITE)
+ def enableNativeDmlRowIndexScan: Boolean =
getConf(ENABLE_NATIVE_DML_ROW_INDEX_SCAN)
+
def enableChangeDataFeedScan: Boolean = getConf(ENABLE_CHANGE_DATA_FEED_SCAN)
}
@@ -43,6 +45,16 @@ object VeloxDeltaConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(false)
+ val ENABLE_NATIVE_DML_ROW_INDEX_SCAN: ConfigEntry[Boolean] =
+
buildConf("spark.gluten.sql.columnar.backend.velox.delta.enableNativeDmlRowIndexScan")
+ .experimental()
+ .doc(
+ "Enable the native Delta DELETE/UPDATE/MERGE target row-index scan for
Velox. When " +
+ "disabled, the DML target scan that produces file paths and row
indexes for " +
+ "deletion-vector writes stays on Spark; other scans are unaffected.")
+ .booleanConf
+ .createWithDefault(true)
+
val ENABLE_CHANGE_DATA_FEED_SCAN: ConfigEntry[Boolean] =
buildConf("spark.gluten.sql.columnar.backend.velox.delta.enableChangeDataFeedScan")
.experimental()
diff --git
a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
index 6b44941b45..12f27c1e12 100644
---
a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
+++
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
@@ -16,37 +16,65 @@
*/
package org.apache.spark.sql.delta
+import org.apache.gluten.config.VeloxDeltaConfig
import org.apache.gluten.execution.DeltaScanTransformer
import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
import org.apache.spark.sql.delta.test.{DeltaSQLCommandTest, DeltaSQLTestUtils}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.tags.ExtendedSQLTest
+import org.apache.spark.util.SparkVersionUtil
import org.apache.hadoop.fs.Path
+import java.io.File
+
@ExtendedSQLTest
class DeltaDeletionVectorHandoffSuite
extends QueryTest
with SharedSparkSession
with DeltaSQLTestUtils
- with DeltaSQLCommandTest {
+ with DeltaSQLCommandTest
+ with AdaptiveSparkPlanHelper {
import testImplicits._
+ private def containsNativeDeltaScan(plan: SparkPlan): Boolean = {
+ collectWithSubqueries(plan) { case scan: DeltaScanTransformer => scan
}.nonEmpty
+ }
+
+ private def captureDeletePlans(path: String, predicate: String):
Seq[SparkPlan] = {
+ DeltaTestUtils.withAllPlansCaptured(spark) {
+ spark.sql(s"DELETE FROM delta.`$path` WHERE $predicate").collect()
+ }.map(_.executedPlan)
+ }
+
+ private def activeDvCardinality(path: String): Long = {
+ val log = DeltaLog.forTable(spark, new Path(path))
+ log.update().allFiles.collect().flatMap(
+ file => Option(file.deletionVector).map(_.cardinality)).sum
+ }
+
+ private def writeDvTable(path: String, rows: Seq[(Int, String)]): Unit = {
+ rows
+ .toDF("id", "value")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(path)
+ spark.sql(
+ s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ }
+
test("Spark 3.5 Delta DV scan handoff should filter deleted rows") {
withTempDir {
tempDir =>
val path = tempDir.getCanonicalPath
- Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
- .toDF("id", "value")
- .coalesce(1)
- .write
- .format("delta")
- .save(path)
-
- spark.sql(
- s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)")
val log = DeltaLog.forTable(spark, new Path(path))
@@ -58,7 +86,8 @@ class DeltaDeletionVectorHandoffSuite
val df = spark.read.format("delta").load(path)
val executedPlan = df.queryExecution.executedPlan
- val nativeScans = executedPlan.collect { case scan:
DeltaScanTransformer => scan }
+ val nativeScans =
+ collectWithSubqueries(executedPlan) { case scan:
DeltaScanTransformer => scan }
assert(nativeScans.nonEmpty)
val planText = executedPlan.toString()
assert(!planText.contains("__delta_internal_is_row_deleted"))
@@ -72,4 +101,190 @@ class DeltaDeletionVectorHandoffSuite
assert(metrics("dvPayloadReadTime").value > 0L)
}
}
+
+ test("Delta metadata row-index predicate should not be stripped from a
native scan") {
+ assume(SparkVersionUtil.gteSpark35, "metadata row index is available in
Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ val df = spark.sql(
+ s"SELECT id, _metadata.row_index AS row_index FROM delta.`$path` " +
+ "WHERE _metadata.row_index = 2")
+ val rows = df.collect()
+ val executedPlan = df.queryExecution.executedPlan
+ val planText = executedPlan.treeString
+ assert(containsNativeDeltaScan(executedPlan), planText)
+ assert(rows.length === 1, planText)
+ assert(rows.head.getLong(1) === 2L, planText)
+ }
+ }
+
+ Seq(true, false).foreach {
+ useMetadataRowIndex =>
+ test(
+ "Delta DV DELETE should write correct deletion vectors, " +
+ s"metadata row index=$useMetadataRowIndex") {
+ assume(SparkVersionUtil.gteSpark35, "DV DML coverage targets Spark
3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ withSQLConf(
+ DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key ->
+ useMetadataRowIndex.toString) {
+ val executedPlans = captureDeletePlans(path, "id IN (3, 4)")
+ val planText = executedPlans.map(_.treeString).mkString("\n\n")
+ // With the metadata row index, the DML target scan offloads
like any other DV
+ // scan; without it, Delta relies on Spark's injected row-index
filter column and
+ // the scan stays on Spark.
+ assert(
+ executedPlans.exists(containsNativeDeltaScan) ===
useMetadataRowIndex,
+ planText)
+
+ assert(activeDvCardinality(path) === 2L)
+ checkAnswer(spark.read.format("delta").load(path), Seq((1, "a"),
(2, "b")).toDF())
+ }
+ }
+ }
+ }
+
+ test("Delta DV repeated DELETE over an existing DV should accumulate deleted
rows") {
+ assume(SparkVersionUtil.gteSpark35, "DV DML coverage targets Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = new File(tempDir, "delta table with
spaces").getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"), (5,
"e"), (6, "f")))
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "true") {
+ // Delete the LEADING rows first: after DV {0, 1} masks them, every
surviving row's
+ // absolute row index differs from its post-mask position. A scan
that renumbered row
+ // indexes after applying the existing DV would emit {0, 1} for the
second DELETE
+ // instead of {2, 3}, failing both the cardinality and the result
checks below.
+ val firstDeletePlans = captureDeletePlans(path, "id IN (1, 2)")
+ assert(
+ firstDeletePlans.exists(containsNativeDeltaScan),
+ firstDeletePlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 2L)
+
+ // The second DELETE scans files that already carry a DV and must
merge into it.
+ val secondDeletePlans = captureDeletePlans(path, "id IN (3, 4)")
+ assert(
+ secondDeletePlans.exists(containsNativeDeltaScan),
+ secondDeletePlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 4L)
+
+ checkAnswer(spark.read.format("delta").load(path), Seq((5, "e"), (6,
"f")).toDF())
+ }
+ }
+ }
+
+ // MERGE puts joins between the target scan and Delta's BitmapAggregator.
With a shuffle join
+ // the scan lands in its own AQE query stage, so this covers target-scan
offload both with and
+ // without the rest of the DML plan visible in the same stage.
+ Seq(true, false).foreach {
+ broadcastJoin =>
+ test(s"Delta DV MERGE should write correct deletion vectors, broadcast
join=$broadcastJoin") {
+ assume(SparkVersionUtil.gteSpark35, "DV DML coverage targets Spark
3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ withTempView("merge_source") {
+ Seq((3, "c2"), (4, "d2")).toDF("id",
"value").createOrReplaceTempView("merge_source")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key ->
+ (if (broadcastJoin) "10485760" else "-1"),
+ DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key ->
"true"
+ ) {
+ val executedPlans = DeltaTestUtils.withAllPlansCaptured(spark)
{
+ spark
+ .sql(s"""MERGE INTO delta.`$path` AS t
+ |USING merge_source AS s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN DELETE""".stripMargin)
+ .collect()
+ }.map(_.executedPlan)
+ assert(
+ executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ }
+
+ val log = DeltaLog.forTable(spark, new Path(path))
+ assert(log.update().allFiles.collect().exists(_.deletionVector
!= null))
+ checkAnswer(
+ spark.read.format("delta").load(path),
+ Seq((1, "a"), (2, "b")).toDF())
+ }
+ }
+ }
+ }
+
+ test("Delta DV DML row-index scan should stay on Spark when disabled") {
+ assume(SparkVersionUtil.gteSpark35, "DV DML coverage targets Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ withSQLConf(
+ DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key -> "true",
+ VeloxDeltaConfig.ENABLE_NATIVE_DML_ROW_INDEX_SCAN.key -> "false") {
+ val executedPlans = captureDeletePlans(path, "id IN (3, 4)")
+ assert(
+ !executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 2L)
+
+ // The fallback is scoped to the DML target scan: a plain read of
the same table keeps
+ // offloading even while the config is off.
+ val df = spark.read.format("delta").load(path)
+ assert(containsNativeDeltaScan(df.queryExecution.executedPlan))
+ checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
+ }
+
+ // The config is read per query, so re-enabling in the same session
restores DML offload.
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "true") {
+ val executedPlans = captureDeletePlans(path, "id = 2")
+ assert(
+ executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 3L)
+ checkAnswer(spark.read.format("delta").load(path), Seq((1,
"a")).toDF())
+ }
+ }
+ }
+
+ test("Delta non-DV DML should offload with a user column named row_index
when disabled") {
+ assume(SparkVersionUtil.gteSpark35, "DV DML coverage targets Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ // Deletion vectors are deliberately left off: this DELETE rewrites
whole files and never
+ // reads a generated row index. The scoped fallback must not claim it
merely because the
+ // table has a user column called row_index -- that name is only
Delta's row index when it
+ // appears inside the file metadata struct.
+ Seq((1, "a", 10L), (2, "b", 20L), (3, "c", 30L))
+ .toDF("id", "value", "row_index")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(path)
+
+ withSQLConf(VeloxDeltaConfig.ENABLE_NATIVE_DML_ROW_INDEX_SCAN.key ->
"false") {
+ val executedPlans = captureDeletePlans(path, "id = 3")
+ assert(
+ executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ }
+
+ checkAnswer(
+ spark.read.format("delta").load(path),
+ Seq((1, "a", 10L), (2, "b", 20L)).toDF())
+ }
+ }
}
diff --git
a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
index 159e848df0..c9355e7c40 100644
---
a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
+++
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
@@ -16,38 +16,64 @@
*/
package org.apache.spark.sql.delta
+import org.apache.gluten.config.VeloxDeltaConfig
import org.apache.gluten.execution.DeltaScanTransformer
import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.delta.sources.DeltaSQLConf
import org.apache.spark.sql.delta.test.{DeltaSQLCommandTest, DeltaSQLTestUtils}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.tags.ExtendedSQLTest
import org.apache.hadoop.fs.Path
+import java.io.File
+
@ExtendedSQLTest
class DeltaDeletionVectorHandoffSuite
extends QueryTest
with SharedSparkSession
with DeltaSQLTestUtils
- with DeltaSQLCommandTest {
+ with DeltaSQLCommandTest
+ with AdaptiveSparkPlanHelper {
import testImplicits._
+ private def containsNativeDeltaScan(plan: SparkPlan): Boolean = {
+ collectWithSubqueries(plan) { case scan: DeltaScanTransformer => scan
}.nonEmpty
+ }
+
+ private def captureDeletePlans(path: String, predicate: String):
Seq[SparkPlan] = {
+ DeltaTestUtils.withAllPlansCaptured(spark) {
+ spark.sql(s"DELETE FROM delta.`$path` WHERE $predicate").collect()
+ }.map(_.executedPlan)
+ }
+
+ private def activeDvCardinality(path: String): Long = {
+ val log = DeltaLog.forTable(spark, new Path(path))
+ log.update().allFiles.collect().flatMap(
+ file => Option(file.deletionVector).map(_.cardinality)).sum
+ }
+
+ private def writeDvTable(path: String, rows: Seq[(Int, String)]): Unit = {
+ rows
+ .toDF("id", "value")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(path)
+ spark.sql(
+ s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ }
+
test("Spark 4 Delta DV scan should fall back when metadata row index is
disabled") {
withTempDir {
tempDir =>
val path = tempDir.getCanonicalPath
- Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
- .toDF("id", "value")
- .coalesce(1)
- .write
- .format("delta")
- .save(path)
-
- spark.sql(
- s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)")
val log = DeltaLog.forTable(spark, new Path(path))
@@ -58,7 +84,7 @@ class DeltaDeletionVectorHandoffSuite
withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
val df = spark.read.format("delta").load(path)
val executedPlan = df.queryExecution.executedPlan
- assert(executedPlan.collect { case _: DeltaScanTransformer => true
}.isEmpty)
+ assert(!containsNativeDeltaScan(executedPlan))
checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
}
}
@@ -68,15 +94,7 @@ class DeltaDeletionVectorHandoffSuite
withTempDir {
tempDir =>
val path = tempDir.getCanonicalPath
- Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
- .toDF("id", "value")
- .coalesce(1)
- .write
- .format("delta")
- .save(path)
-
- spark.sql(
- s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)")
val log = DeltaLog.forTable(spark, new Path(path))
@@ -88,7 +106,8 @@ class DeltaDeletionVectorHandoffSuite
val df = spark.read.format("delta").load(path)
val executedPlan = df.queryExecution.executedPlan
- val nativeScans = executedPlan.collect { case scan:
DeltaScanTransformer => scan }
+ val nativeScans =
+ collectWithSubqueries(executedPlan) { case scan:
DeltaScanTransformer => scan }
assert(nativeScans.nonEmpty)
val planText = executedPlan.toString()
assert(!planText.contains("__delta_internal_is_row_deleted"))
@@ -102,4 +121,184 @@ class DeltaDeletionVectorHandoffSuite
assert(metrics("dvPayloadReadTime").value > 0L)
}
}
+
+ test("Delta metadata row-index predicate should not be stripped from a
native scan") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ val df = spark.sql(
+ s"SELECT id, _metadata.row_index AS row_index FROM delta.`$path` " +
+ "WHERE _metadata.row_index = 2")
+ val rows = df.collect()
+ val executedPlan = df.queryExecution.executedPlan
+ val planText = executedPlan.treeString
+ assert(containsNativeDeltaScan(executedPlan), planText)
+ assert(rows.length === 1, planText)
+ assert(rows.head.getLong(1) === 2L, planText)
+ }
+ }
+
+ Seq(true, false).foreach {
+ useMetadataRowIndex =>
+ test(
+ "Delta DV DELETE should write correct deletion vectors, " +
+ s"metadata row index=$useMetadataRowIndex") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ withSQLConf(
+ DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key ->
+ useMetadataRowIndex.toString) {
+ val executedPlans = captureDeletePlans(path, "id IN (3, 4)")
+ val planText = executedPlans.map(_.treeString).mkString("\n\n")
+ // With the metadata row index, the DML target scan offloads
like any other DV
+ // scan; without it, Delta relies on Spark's injected row-index
filter column and
+ // the scan stays on Spark.
+ assert(
+ executedPlans.exists(containsNativeDeltaScan) ===
useMetadataRowIndex,
+ planText)
+
+ assert(activeDvCardinality(path) === 2L)
+ checkAnswer(spark.read.format("delta").load(path), Seq((1, "a"),
(2, "b")).toDF())
+ }
+ }
+ }
+ }
+
+ test("Delta DV repeated DELETE over an existing DV should accumulate deleted
rows") {
+ withTempDir {
+ tempDir =>
+ val path = new File(tempDir, "delta table with
spaces").getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"), (5,
"e"), (6, "f")))
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "true") {
+ // Delete the LEADING rows first: after DV {0, 1} masks them, every
surviving row's
+ // absolute row index differs from its post-mask position. A scan
that renumbered row
+ // indexes after applying the existing DV would emit {0, 1} for the
second DELETE
+ // instead of {2, 3}, failing both the cardinality and the result
checks below.
+ val firstDeletePlans = captureDeletePlans(path, "id IN (1, 2)")
+ assert(
+ firstDeletePlans.exists(containsNativeDeltaScan),
+ firstDeletePlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 2L)
+
+ // The second DELETE scans files that already carry a DV and must
merge into it.
+ val secondDeletePlans = captureDeletePlans(path, "id IN (3, 4)")
+ assert(
+ secondDeletePlans.exists(containsNativeDeltaScan),
+ secondDeletePlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 4L)
+
+ checkAnswer(spark.read.format("delta").load(path), Seq((5, "e"), (6,
"f")).toDF())
+ }
+ }
+ }
+
+ // MERGE puts joins between the target scan and Delta's BitmapAggregator.
With a shuffle join
+ // the scan lands in its own AQE query stage, so this covers target-scan
offload both with and
+ // without the rest of the DML plan visible in the same stage.
+ Seq(true, false).foreach {
+ broadcastJoin =>
+ test(s"Delta DV MERGE should write correct deletion vectors, broadcast
join=$broadcastJoin") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ withTempView("merge_source") {
+ Seq((3, "c2"), (4, "d2")).toDF("id",
"value").createOrReplaceTempView("merge_source")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key ->
+ (if (broadcastJoin) "10485760" else "-1"),
+ DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key ->
"true"
+ ) {
+ val executedPlans = DeltaTestUtils.withAllPlansCaptured(spark)
{
+ spark
+ .sql(s"""MERGE INTO delta.`$path` AS t
+ |USING merge_source AS s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN DELETE""".stripMargin)
+ .collect()
+ }.map(_.executedPlan)
+ assert(
+ executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ }
+
+ val log = DeltaLog.forTable(spark, new Path(path))
+ assert(log.update().allFiles.collect().exists(_.deletionVector
!= null))
+ checkAnswer(
+ spark.read.format("delta").load(path),
+ Seq((1, "a"), (2, "b")).toDF())
+ }
+ }
+ }
+ }
+
+ test("Delta DV DML row-index scan should stay on Spark when disabled") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ writeDvTable(path, Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")))
+
+ withSQLConf(
+ DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key -> "true",
+ VeloxDeltaConfig.ENABLE_NATIVE_DML_ROW_INDEX_SCAN.key -> "false") {
+ val executedPlans = captureDeletePlans(path, "id IN (3, 4)")
+ assert(
+ !executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 2L)
+
+ // The fallback is scoped to the DML target scan: a plain read of
the same table keeps
+ // offloading even while the config is off.
+ val df = spark.read.format("delta").load(path)
+ assert(containsNativeDeltaScan(df.queryExecution.executedPlan))
+ checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
+ }
+
+ // The config is read per query, so re-enabling in the same session
restores DML offload.
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "true") {
+ val executedPlans = captureDeletePlans(path, "id = 2")
+ assert(
+ executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ assert(activeDvCardinality(path) === 3L)
+ checkAnswer(spark.read.format("delta").load(path), Seq((1,
"a")).toDF())
+ }
+ }
+ }
+
+ test("Delta non-DV DML should offload with a user column named row_index
when disabled") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ // Deletion vectors are deliberately left off: this DELETE rewrites
whole files and never
+ // reads a generated row index. The scoped fallback must not claim it
merely because the
+ // table has a user column called row_index -- that name is only
Delta's row index when it
+ // appears inside the file metadata struct.
+ Seq((1, "a", 10L), (2, "b", 20L), (3, "c", 30L))
+ .toDF("id", "value", "row_index")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(path)
+
+ withSQLConf(VeloxDeltaConfig.ENABLE_NATIVE_DML_ROW_INDEX_SCAN.key ->
"false") {
+ val executedPlans = captureDeletePlans(path, "id = 3")
+ assert(
+ executedPlans.exists(containsNativeDeltaScan),
+ executedPlans.map(_.treeString).mkString("\n\n"))
+ }
+
+ checkAnswer(
+ spark.read.format("delta").load(path),
+ Seq((1, "a", 10L), (2, "b", 20L)).toDF())
+ }
+ }
}
diff --git a/docs/get-started/VeloxDelta.md b/docs/get-started/VeloxDelta.md
index 3c594b9b97..6a73a27403 100644
--- a/docs/get-started/VeloxDelta.md
+++ b/docs/get-started/VeloxDelta.md
@@ -34,6 +34,14 @@ Native change data feed scan offload is controlled by:
- Default: `true`
- Type: experimental
+The native DELETE/UPDATE/MERGE deletion-vector target scan is controlled by:
+
+- `spark.gluten.sql.columnar.backend.velox.delta.enableNativeDmlRowIndexScan`
+ - Default: `true`
+ - Type: experimental
+ - When disabled, the DML target scan that produces file paths and row
indexes for
+ deletion-vector writes stays on Spark; other scans are unaffected.
+
| 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/DeltaPostTransformRules.scala
b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala
index fb694c70d9..a6cfc42916 100644
---
a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala
+++
b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala
@@ -56,6 +56,13 @@ object DeltaPostTransformRules {
private val deletionVectorDeletedRowColumnName =
"__delta_internal_is_row_deleted"
private val deletionVectorRowIndexColumnName = "__delta_internal_row_index"
+ // Spark 3.5+ exposes this as
ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME.
+ private val parquetTemporaryRowIndexColumnName = "_tmp_metadata_row_index"
+ private val deletionVectorRowIndexColumnNames =
+ Set(
+ deletionVectorRowIndexColumnName,
+ DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME,
+ parquetTemporaryRowIndexColumnName)
private val deletionVectorInternalColumnNames =
Set(deletionVectorDeletedRowColumnName, deletionVectorRowIndexColumnName)
@@ -213,7 +220,7 @@ object DeltaPostTransformRules {
}
private def referencesDeletionVectorRowIndex(expr: Expression): Boolean = {
- expr.references.exists(_.name == deletionVectorRowIndexColumnName)
+ expr.references.exists(attr =>
deletionVectorRowIndexColumnNames.contains(attr.name))
}
private def tagRowIndexRequiredSubtrees(plan: SparkPlan): Unit = {
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaScanUtils.scala
b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaScanUtils.scala
new file mode 100644
index 0000000000..d5e02ba3de
--- /dev/null
+++
b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaScanUtils.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.delta.DeltaParquetFileFormat
+import org.apache.spark.sql.delta.files.TahoeFileIndex
+import org.apache.spark.sql.delta.stats.PreparedDeltaFileIndex
+import org.apache.spark.sql.execution.FileSourceScanExec
+
+/** Structural checks shared by the Delta offload rules for recognizing Delta
table scans. */
+object DeltaScanUtils {
+ def isDeltaScan(scan: FileSourceScanExec): Boolean = {
+ isDeltaFileIndex(scan) || isDeltaParquetScan(scan)
+ }
+
+ private def isDeltaParquetScan(scan: FileSourceScanExec): Boolean = {
+ val fileFormatClass = scan.relation.fileFormat.getClass
+ fileFormatClass == classOf[DeltaParquetFileFormat] ||
+ fileFormatClass.getSimpleName == "GlutenDeltaParquetFileFormat"
+ }
+
+ private def isDeltaFileIndex(scan: FileSourceScanExec): Boolean = {
+ scan.relation.location.isInstanceOf[TahoeFileIndex] ||
+ scan.relation.location.isInstanceOf[PreparedDeltaFileIndex]
+ }
+}
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 4664da8769..fd8ccf7842 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
@@ -20,22 +20,38 @@ import org.apache.gluten.execution.DeltaScanTransformer
import org.apache.gluten.extension.columnar.FallbackTags
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.{DeltaParquetFileFormat, SnapshotDescriptor}
import
org.apache.spark.sql.delta.commands.DeletionVectorUtils.deletionVectorsReadable
-import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeFileIndex,
TahoeRemoveFileIndex}
+import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeBatchFileIndex,
TahoeFileIndex, TahoeRemoveFileIndex}
import org.apache.spark.sql.delta.stats.PreparedDeltaFileIndex
import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.types.{DataType, StructType}
import org.apache.spark.util.SparkVersionUtil
-case class OffloadDeltaScan() extends OffloadSingleNode {
+case class OffloadDeltaScan(enableNativeDmlRowIndexScan: Boolean) extends
OffloadSingleNode {
private val DeletionVectorsUseMetadataRowIndexKey =
"spark.databricks.delta.deletionVectors.useMetadataRowIndex"
+ // Spark 3.5+ exposes this as
ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME.
+ private val parquetTemporaryRowIndexColumnName = "_tmp_metadata_row_index"
+ // Row-index columns Delta generates as top-level scan outputs.
+ private val generatedRowIndexColumnNames =
+ Set(DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME,
parquetTemporaryRowIndexColumnName)
+ // ParquetFileFormat.ROW_INDEX, the generated field Delta adds to the file
metadata struct in
+ // PreprocessTableWithDVs when deletionVectors.useMetadataRowIndex is on.
Only meaningful nested
+ // under _metadata -- a user column may legitimately be called row_index.
+ private val metadataRowIndexFieldName = "row_index"
+ // TahoeBatchFileIndex.actionType as set by Delta's DELETE, UPDATE and MERGE
commands.
+ private val dmlActionTypes = Set("delete", "update", "merge")
+
override def offload(plan: SparkPlan): SparkPlan = plan match {
case scan: FileSourceScanExec if isDeltaLogScan(scan) =>
FallbackTags.add(scan, "fallback Delta _delta_log scan")
scan
+ case scan: FileSourceScanExec if shouldFallbackDeletionVectorDmlScan(scan)
=>
+ FallbackTags.add(scan, "fallback Delta DV DML row-index scan by
configuration")
+ scan
case scan: FileSourceScanExec if
shouldFallbackSpark34DeletionVectorScan(scan) =>
FallbackTags.add(scan, "fallback Spark 3.4 Delta DV scan")
scan
@@ -43,24 +59,45 @@ case class OffloadDeltaScan() extends OffloadSingleNode {
if shouldFallbackDeletionVectorScanWithoutMetadataRowIndex(scan) =>
FallbackTags.add(scan, "fallback Delta DV scan without metadata row
index")
scan
- case scan: FileSourceScanExec if isDeltaScan(scan) =>
+ case scan: FileSourceScanExec if DeltaScanUtils.isDeltaScan(scan) =>
DeltaScanTransformer(scan)
case other => other
}
- private def isDeltaScan(scan: FileSourceScanExec): Boolean = {
- isDeltaFileIndex(scan) || isDeltaParquetScan(scan)
+ /**
+ * The scoped escape hatch: with the config off, the DELETE/UPDATE/MERGE
target scan that produces
+ * file paths and row indexes for deletion-vector writes stays on Spark,
while every other scan
+ * keeps offloading. The whole check lives here, on the scan alone: Delta
builds every DML target
+ * relation over a [[TahoeBatchFileIndex]] carrying the command name, which
survives AQE stage
+ * splits and arbitrary join placement, and only DV-writing DML reads a
row-index column from that
+ * relation; DML that rewrites whole files does not, and remains eligible
for native execution.
+ */
+ private def shouldFallbackDeletionVectorDmlScan(scan: FileSourceScanExec):
Boolean = {
+ !enableNativeDmlRowIndexScan && isDmlTargetScan(scan) &&
scanReadsRowIndexColumn(scan)
}
- private def isDeltaParquetScan(scan: FileSourceScanExec): Boolean = {
- val fileFormatClass = scan.relation.fileFormat.getClass
- fileFormatClass == classOf[DeltaParquetFileFormat] ||
- fileFormatClass.getSimpleName == "GlutenDeltaParquetFileFormat"
+ private def isDmlTargetScan(scan: FileSourceScanExec): Boolean = {
+ scan.relation.location match {
+ case index: TahoeBatchFileIndex =>
dmlActionTypes.contains(index.actionType)
+ case _ => false
+ }
}
- private def isDeltaFileIndex(scan: FileSourceScanExec): Boolean = {
- scan.relation.location.isInstanceOf[TahoeFileIndex] ||
- scan.relation.location.isInstanceOf[PreparedDeltaFileIndex]
+ private def isRowIndexColumn(name: String, dataType: DataType): Boolean = {
+ generatedRowIndexColumnNames.contains(name) ||
+ (name == FileFormat.METADATA_NAME && (dataType match {
+ case struct: StructType =>
struct.fieldNames.contains(metadataRowIndexFieldName)
+ case _ => false
+ }))
+ }
+
+ private def scanReadsRowIndexColumn(scan: FileSourceScanExec): Boolean = {
+ val outputFields = scan.output.iterator.map(attribute => (attribute.name,
attribute.dataType))
+ val requiredFields =
+ scan.requiredSchema.fields.iterator.map(field => (field.name,
field.dataType))
+ (outputFields ++ requiredFields).exists {
+ case (name, dataType) => isRowIndexColumn(name, dataType)
+ }
}
private def isDeltaLogScan(scan: FileSourceScanExec): Boolean = {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
