felipepessoto commented on code in PR #13042:
URL: https://github.com/apache/gluten/pull/13042#discussion_r4277855610
##########
backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala:
##########
@@ -70,6 +74,28 @@ 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 metadata = deltaLog.snapshot.metadata
+ val fileIndex = TahoeLogFileIndex(spark, deltaLog)
+ val readingSchema = syntheticFields.foldLeft(metadata.schema)(_.add(_))
+ val relation = HadoopFsRelation(
+ fileIndex,
+ fileIndex.partitionSchema,
+ readingSchema,
+ bucketSpec = None,
+ DeltaParquetFileFormat(
+ deltaLog.snapshot.protocol,
Review Comment:
Updated both helpers to capture `val snapshot = deltaLog.update()` once and
use that snapshot for both metadata and protocol. This removes the deprecated
accessor and keeps the two values from the same snapshot.
##########
gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala:
##########
@@ -116,20 +132,17 @@ 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.
val useMetadataRowIndex =
scan.relation.sparkSession.sessionState.conf
.getConfString(DeletionVectorsUseMetadataRowIndexKey, "true")
.toBoolean
- !useMetadataRowIndex && containsDeletionVector(scan)
+ !useMetadataRowIndex &&
scanReadsGeneratedDeletionVectorMetadataColumn(scan)
Review Comment:
Added the rationale in `shouldFallbackGeneratedDeletionVectorMetadataScan`:
the disabled setting requires Delta's JVM-generated internal columns, whose
synthesis is bypassed by the native scan. Spark's own metadata row index is
explicitly excluded from this fallback.
##########
backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala:
##########
@@ -69,6 +74,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 metadata = deltaLog.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(
+ deltaLog.snapshot.protocol,
Review Comment:
Updated the src-delta40 helper as well: it now calls `deltaLog.update()`
once and uses the resulting snapshot's metadata and protocol, with no
deprecated `deltaLog.snapshot` calls.
##########
gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala:
##########
@@ -38,6 +38,8 @@ case class OffloadDeltaScan(enableNativeDmlRowIndexScan:
Boolean) extends Offloa
// Row-index columns Delta generates as top-level scan outputs.
private val generatedRowIndexColumnNames =
Set(DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME,
parquetTemporaryRowIndexColumnName)
+ private val generatedDeletionVectorMetadataColumnNames =
+ generatedRowIndexColumnNames +
DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME
Review Comment:
Agreed; fixed in 76ac03f44. The new fallback set now contains only
`ROW_INDEX_COLUMN_NAME` and `IS_ROW_DELETED_COLUMN_NAME`.
`_tmp_metadata_row_index` remains in the existing DML row-index detector, but
no longer triggers this JVM-synthesis fallback. I also corrected the shared
set's comment to describe both row-index generation paths.
##########
gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala:
##########
Review Comment:
Added the explanation immediately above the config read: with metadata row
indexes disabled, Delta generates its internal DV columns in the JVM reader,
and native scans bypass that synthesis. The comment also distinguishes Spark's
own metadata row index, which remains supported natively.
##########
backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala:
##########
@@ -121,6 +147,93 @@ 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 temporary row-index scan should fall back when metadata row
index is disabled") {
+ assume(SparkVersionUtil.gteSpark35, "temporary 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)
+ val temporaryRowIndexField =
+ StructField(ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME,
LongType)
+
+ withSQLConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX.key
-> "false") {
Review Comment:
Replaced the test in both src-delta33 and src-delta40 with a real `SELECT
id, _metadata.row_index` under `useMetadataRowIndex=false`. The fixture enables
the table's DV feature so the metadata field is exposed, without creating a
deletion vector. Each test checks the complete `(id, row_index)` results and
then asserts a native Delta scan in the executed plan. The new Spark 3.5/Delta
3.3 case passed locally alongside the two synthetic-column fallback cases and
the ordinary native-scan case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
