malinjawi commented on code in PR #13042:
URL: https://github.com/apache/gluten/pull/13042#discussion_r4256428510
##########
gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala:
##########
Review Comment:
Nit: the old comment explaining the flag check got dropped.
Could you add a line saying why this only applies when
`useMetadataRowIndex=false` would help the next reader?
##########
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:
If the temp name comes out, this test would flip (same in src-delta40).
A real `SELECT id, _metadata.row_index` with the flag off, asserting a
native scan would cover it better.
##########
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:
Nit: the old comment explaining the flag check got dropped.
Could you add a line saying why this only applies when
useMetadataRowIndex=false would help the next reader?
##########
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:
This set inherits `parquetTemporaryRowIndexColumnName` (Spark's
`_tmp_metadata_row_index`) from `generatedRowIndexColumnNames`, and I don't
think it belongs here. Delta's reader only uses that name when
`useMetadataRowIndex=true`; with the flag off it's just Spark's parquet row
index, which the native scan already fills correctly. Keeping it means
`_metadata.row_index` reads lose native execution when the flag is off.
Partly my fault: `generatedRowIndexColumnNames` is from my #12215, and its
comment ("Row-index columns Delta generates") is misleading. Suggest building
this set from just `ROW_INDEX_COLUMN_NAME` and `IS_ROW_DELETED_COLUMN_NAME`.
Not a blocker tho but could be better in long run.
##########
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:
nit:`deltaLog.snapshot` is deprecated.
Same as the src-delta33 copy: `deltaLog.snapshot`
##########
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:
Nit: `deltaLog.snapshot` is deprecated in all the Delta versions we build
(3.3.2, 4.0.1, 4.1.0), and CI logs the warning.
`deltaLog.unsafeVolatileSnapshot` is the direct replacement, or
`deltaLog.update()` once to match the rest of the suite.
--
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]
