This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 57a95dc847 [GLUTEN-12743][VL] Fall back Delta scans requiring
generated DV metadata (#13042)
57a95dc847 is described below
commit 57a95dc847c11e0e6103ae167a5246f358ac3417
Author: Felipe Pessoto <[email protected]>
AuthorDate: Thu Oct 15 08:31:03 2026 -0700
[GLUTEN-12743][VL] Fall back Delta scans requiring generated DV metadata
(#13042)
* [GLUTEN-12743][VL] Fall back generated Delta DV metadata scans
Co-authored-by: Copilot App <[email protected]>
* [GLUTEN-12743][VL] Fix Delta fallback CI
Co-authored-by: Copilot App <[email protected]>
* [GLUTEN-12743][VL] Strengthen Delta row-index assertions
Co-authored-by: Copilot App <[email protected]>
* [GLUTEN-12743][CI] Remove fixed Delta scan failures
Co-authored-by: Copilot App <[email protected]>
* [GLUTEN-12743][VL] Reorder Delta scan helpers
Co-authored-by: Copilot App <[email protected]>
* [GLUTEN-12743][VL] Preserve native Spark metadata row-index scans
Limit the generated DV metadata fallback to Delta internal columns,
exercise native metadata row-index reads through SQL, explain the
disabled-config path, and refresh test snapshots once.
Co-authored-by: Copilot App <[email protected]>
---------
Co-authored-by: Copilot App <[email protected]>
---
.../util/delta-spark-ut/known-failures.txt | 9 --
.../delta/DeltaDeletionVectorHandoffSuite.scala | 113 ++++++++++++++++++++-
.../delta/DeltaDeletionVectorHandoffSuite.scala | 111 +++++++++++++++++++-
.../apache/gluten/extension/OffloadDeltaScan.scala | 35 +++++--
4 files changed, 248 insertions(+), 20 deletions(-)
diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt
b/.github/workflows/util/delta-spark-ut/known-failures.txt
index 644df88b13..51bbf2b9ed 100644
--- a/.github/workflows/util/delta-spark-ut/known-failures.txt
+++ b/.github/workflows/util/delta-spark-ut/known-failures.txt
@@ -104,15 +104,6 @@
org.apache.spark.sql.delta.DeltaInsertIntoSQLByPathSuite#insertInto: timestamp p
org.apache.spark.sql.delta.DeltaInsertIntoSQLSuite#insertInto: timestamp
partition values with different precisions
org.apache.spark.sql.delta.DeltaLiteVacuumSuite#vacuum for cdc - delete
tombstones
org.apache.spark.sql.delta.DeltaLiteVacuumSuite#vacuum for cdc - update/merge
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=false, with rowIndexCol=true,
with vectorized Parquet reader=false, with readColumnarBatchAsRows=true
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=false, with rowIndexCol=true,
with vectorized Parquet reader=true, with readColumnarBatchAsRows=false
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=false, with rowIndexCol=true,
with vectorized Parquet reader=true, with readColumnarBatchAsRows=true
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=true, with rowIndexCol=false,
with vectorized Parquet reader=false, with readColumnarBatchAsRows=true
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=true, with rowIndexCol=false,
with vectorized Parquet reader=true, with readColumnarBatchAsRows=false
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=true, with rowIndexCol=false,
with vectorized Parquet reader=true, with readColumnarBatchAsRows=true
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=true, with rowIndexCol=true,
with vectorized Parquet reader=false, with readColumnarBatchAsRows=true
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=true, with rowIndexCol=true,
with vectorized Parquet reader=true, with readColumnarBatchAsRows=false
-org.apache.spark.sql.delta.DeltaParquetFileFormatSuite#isDeletionVectorsEnabled=false,
read DV metadata columns: with isRowDeletedCol=true, with rowIndexCol=true,
with vectorized Parquet reader=true, with readColumnarBatchAsRows=true
org.apache.spark.sql.delta.DeltaParquetFileFormatWithPredicatePushdownSuite#read
DV metadata columns: with rowIndexFilterType=IF_CONTAINED, with vectorized
Parquet reader=false, with readColumnarBatchAsRows=true
org.apache.spark.sql.delta.DeltaParquetFileFormatWithPredicatePushdownSuite#read
DV metadata columns: with rowIndexFilterType=IF_CONTAINED, with vectorized
Parquet reader=true, with readColumnarBatchAsRows=false
org.apache.spark.sql.delta.DeltaParquetFileFormatWithPredicatePushdownSuite#read
DV metadata columns: with rowIndexFilterType=IF_CONTAINED, with vectorized
Parquet reader=true, with readColumnarBatchAsRows=true
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 12f27c1e12..e75768578c 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
@@ -19,13 +19,16 @@ 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.{DataFrame, Dataset, QueryTest, Row}
+import org.apache.spark.sql.delta.files.TahoeLogFileIndex
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.execution.datasources.{HadoopFsRelation,
LogicalRelation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructField
import org.apache.spark.tags.ExtendedSQLTest
import org.apache.spark.util.SparkVersionUtil
@@ -70,6 +73,29 @@ class DeltaDeletionVectorHandoffSuite
s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
}
+ private def dataframeWithSyntheticColumns(
+ path: String,
+ syntheticFields: StructField*): DataFrame = {
+ val deltaLog = DeltaLog.forTable(spark, new Path(path))
+ val snapshot = deltaLog.update()
+ val metadata = snapshot.metadata
+ val fileIndex = TahoeLogFileIndex(spark, deltaLog)
+ val readingSchema = syntheticFields.foldLeft(metadata.schema)(_.add(_))
+ val relation = HadoopFsRelation(
+ fileIndex,
+ fileIndex.partitionSchema,
+ readingSchema,
+ bucketSpec = None,
+ DeltaParquetFileFormat(
+ snapshot.protocol,
+ metadata,
+ nullableRowTrackingFields = false,
+ optimizationsEnabled = false),
+ options = Map.empty
+ )(spark)
+ Dataset.ofRows(spark, LogicalRelation(relation))
+ }
+
test("Spark 3.5 Delta DV scan handoff should filter deleted rows") {
withTempDir {
tempDir =>
@@ -121,6 +147,91 @@ class DeltaDeletionVectorHandoffSuite
}
}
+ test("Delta generated row-index scan should fall back when metadata row
index is disabled") {
+ assume(SparkVersionUtil.gteSpark35, "generated row-index coverage targets
Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2)
+ .toDF("value")
+ .coalesce(1)
+ .sortWithinPartitions("value")
+ .write
+ .format("delta")
+ .save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val rowIndexDf =
+ dataframeWithSyntheticColumns(path,
DeltaParquetFileFormat.ROW_INDEX_STRUCT_FIELD)
+
+
assert(!containsNativeDeltaScan(rowIndexDf.queryExecution.executedPlan))
+ checkAnswer(
+ rowIndexDf.select("value",
DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME),
+ Seq(Row(0, 0L), Row(1, 1L), Row(2, 2L)))
+ }
+ }
+ }
+
+ test("Delta generated deleted-row scan should fall back for a DV-free file")
{
+ assume(SparkVersionUtil.gteSpark35, "generated DV metadata coverage
targets Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2).toDF("value").coalesce(1).write.format("delta").save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val deletedRowDf =
+ dataframeWithSyntheticColumns(path,
DeltaParquetFileFormat.IS_ROW_DELETED_STRUCT_FIELD)
+
+
assert(!containsNativeDeltaScan(deletedRowDf.queryExecution.executedPlan))
+ assert(
+ deletedRowDf
+ .select(DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME)
+ .collect()
+ .map(_.getByte(0))
+ .toSet === Set(0.toByte))
+ }
+ }
+ }
+
+ test("Delta metadata row-index scan should offload when useMetadataRowIndex
is false") {
+ assume(SparkVersionUtil.gteSpark35, "metadata row index is available in
Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2)
+ .toDF("id")
+ .coalesce(1)
+ .sortWithinPartitions("id")
+ .write
+ .format("delta")
+ .option("delta.enableDeletionVectors", "true")
+ .save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val rowIndexDf = spark.sql(s"SELECT id, _metadata.row_index FROM
delta.`$path`")
+
+ checkAnswer(rowIndexDf, Seq(Row(0, 0L), Row(1, 1L), Row(2, 2L)))
+ val plan = rowIndexDf.queryExecution.executedPlan
+ assert(containsNativeDeltaScan(plan), plan.treeString)
+ }
+ }
+ }
+
+ test("Delta ordinary scan should offload when metadata row index is
disabled") {
+ assume(SparkVersionUtil.gteSpark35, "generated DV metadata coverage
targets Spark 3.5+")
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2).toDF("value").coalesce(1).write.format("delta").save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val plan =
spark.read.format("delta").load(path).queryExecution.executedPlan
+ assert(containsNativeDeltaScan(plan))
+ }
+ }
+ }
+
Seq(true, false).foreach {
useMetadataRowIndex =>
test(
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 c9355e7c40..7e8247efe6 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
@@ -19,13 +19,17 @@ 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.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.delta.DataFrameUtils
+import org.apache.spark.sql.delta.files.TahoeLogFileIndex
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.execution.datasources.{HadoopFsRelation,
LogicalRelation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructField
import org.apache.spark.tags.ExtendedSQLTest
import org.apache.hadoop.fs.Path
@@ -69,6 +73,30 @@ class DeltaDeletionVectorHandoffSuite
s"ALTER TABLE delta.`$path` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
}
+ private def dataframeWithSyntheticColumns(
+ path: String,
+ syntheticFields: StructField*): DataFrame = {
+ val deltaLog = DeltaLog.forTable(spark, new Path(path))
+ val snapshot = deltaLog.update()
+ val metadata = snapshot.metadata
+ val fileIndex = TahoeLogFileIndex(spark, deltaLog, None)
+ val readingSchema = syntheticFields.foldLeft(metadata.schema)(_.add(_))
+ val relation = HadoopFsRelation(
+ fileIndex,
+ fileIndex.partitionSchema,
+ readingSchema,
+ bucketSpec = None,
+ DeltaParquetFileFormat(
+ snapshot.protocol,
+ metadata,
+ nullableRowTrackingConstantFields = false,
+ nullableRowTrackingGeneratedFields = false,
+ optimizationsEnabled = false),
+ options = Map.empty
+ )(spark)
+ DataFrameUtils.ofRows(spark, LogicalRelation(relation))
+ }
+
test("Spark 4 Delta DV scan should fall back when metadata row index is
disabled") {
withTempDir {
tempDir =>
@@ -140,6 +168,87 @@ class DeltaDeletionVectorHandoffSuite
}
}
+ test("Delta generated row-index scan should fall back when metadata row
index is disabled") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2)
+ .toDF("value")
+ .coalesce(1)
+ .sortWithinPartitions("value")
+ .write
+ .format("delta")
+ .save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val rowIndexDf =
+ dataframeWithSyntheticColumns(path,
DeltaParquetFileFormat.ROW_INDEX_STRUCT_FIELD)
+
+
assert(!containsNativeDeltaScan(rowIndexDf.queryExecution.executedPlan))
+ checkAnswer(
+ rowIndexDf.select("value",
DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME),
+ Seq(Row(0, 0L), Row(1, 1L), Row(2, 2L)))
+ }
+ }
+ }
+
+ test("Delta generated deleted-row scan should fall back for a DV-free file")
{
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2).toDF("value").coalesce(1).write.format("delta").save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val deletedRowDf =
+ dataframeWithSyntheticColumns(path,
DeltaParquetFileFormat.IS_ROW_DELETED_STRUCT_FIELD)
+
+
assert(!containsNativeDeltaScan(deletedRowDf.queryExecution.executedPlan))
+ assert(
+ deletedRowDf
+ .select(DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME)
+ .collect()
+ .map(_.getByte(0))
+ .toSet === Set(0.toByte))
+ }
+ }
+ }
+
+ test("Delta metadata row-index scan should offload when useMetadataRowIndex
is false") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2)
+ .toDF("id")
+ .coalesce(1)
+ .sortWithinPartitions("id")
+ .write
+ .format("delta")
+ .option("delta.enableDeletionVectors", "true")
+ .save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val rowIndexDf = spark.sql(s"SELECT id, _metadata.row_index FROM
delta.`$path`")
+
+ checkAnswer(rowIndexDf, Seq(Row(0, 0L), Row(1, 1L), Row(2, 2L)))
+ val plan = rowIndexDf.queryExecution.executedPlan
+ assert(containsNativeDeltaScan(plan), plan.treeString)
+ }
+ }
+ }
+
+ test("Delta ordinary scan should offload when metadata row index is
disabled") {
+ withTempDir {
+ tempDir =>
+ val path = tempDir.getCanonicalPath
+ Seq(0, 1, 2).toDF("value").coalesce(1).write.format("delta").save(path)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
+ val plan =
spark.read.format("delta").load(path).queryExecution.executedPlan
+ assert(containsNativeDeltaScan(plan))
+ }
+ }
+ }
+
Seq(true, false).foreach {
useMetadataRowIndex =>
test(
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 fd8ccf7842..73eae28414 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
@@ -35,9 +35,13 @@ case class OffloadDeltaScan(enableNativeDmlRowIndexScan:
Boolean) extends Offloa
// 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.
+ // Row-index columns used by Delta DML with either row-index generation path.
private val generatedRowIndexColumnNames =
Set(DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME,
parquetTemporaryRowIndexColumnName)
+ private val generatedDeletionVectorMetadataColumnNames =
+ Set(
+ DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME,
+ DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME)
// 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.
@@ -56,8 +60,9 @@ case class OffloadDeltaScan(enableNativeDmlRowIndexScan:
Boolean) extends Offloa
FallbackTags.add(scan, "fallback Spark 3.4 Delta DV scan")
scan
case scan: FileSourceScanExec
- if shouldFallbackDeletionVectorScanWithoutMetadataRowIndex(scan) =>
- FallbackTags.add(scan, "fallback Delta DV scan without metadata row
index")
+ if DeltaScanUtils.isDeltaScan(scan) &&
+ shouldFallbackGeneratedDeletionVectorMetadataScan(scan) =>
+ FallbackTags.add(scan, "fallback Delta scan requiring generated DV
metadata")
scan
case scan: FileSourceScanExec if DeltaScanUtils.isDeltaScan(scan) =>
DeltaScanTransformer(scan)
@@ -91,12 +96,25 @@ case class OffloadDeltaScan(enableNativeDmlRowIndexScan:
Boolean) extends Offloa
}))
}
+ private def scanReadsGeneratedDeletionVectorMetadataColumn(
+ scan: FileSourceScanExec): Boolean = {
+ scanReadsColumn(
+ scan,
+ (name, _) => generatedDeletionVectorMetadataColumnNames.contains(name))
+ }
+
private def scanReadsRowIndexColumn(scan: FileSourceScanExec): Boolean = {
+ scanReadsColumn(scan, isRowIndexColumn)
+ }
+
+ private def scanReadsColumn(
+ scan: FileSourceScanExec,
+ predicate: (String, DataType) => Boolean): 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)
+ case (name, dataType) => predicate(name, dataType)
}
}
@@ -116,20 +134,19 @@ case class OffloadDeltaScan(enableNativeDmlRowIndexScan:
Boolean) extends Offloa
containsDeletionVector(scan)
}
- private def shouldFallbackDeletionVectorScanWithoutMetadataRowIndex(
+ private def shouldFallbackGeneratedDeletionVectorMetadataScan(
scan: FileSourceScanExec): Boolean = {
if (!SparkVersionUtil.gteSpark35) {
return false
}
- // Delta DML tests force this path and rely on Spark's injected
- // row-index filter column for correctness. Keep it on Spark until the
native path can
- // prove the same contract for DML-generated DVs.
+ // Without metadata row indexes, Delta's JVM reader generates the internal
DV columns.
+ // Native scans bypass that synthesis; Spark's own metadata row index
remains supported.
val useMetadataRowIndex =
scan.relation.sparkSession.sessionState.conf
.getConfString(DeletionVectorsUseMetadataRowIndexKey, "true")
.toBoolean
- !useMetadataRowIndex && containsDeletionVector(scan)
+ !useMetadataRowIndex &&
scanReadsGeneratedDeletionVectorMetadataColumn(scan)
}
private def containsDeletionVector(scan: FileSourceScanExec): Boolean = {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
