[GLUTEN][VL] Read Delta deletion vectors through Velox

.github/workflows/util/delta-spark-ut/known-failures.txt

@@ -100,7 +100,6 @@ org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#SC-86916: invalidate
 org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#SC-86916: read/write Delta paths using DataFrame should pick up Hadoop file system options
 org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#all operations should propagate Hadoop file system options
 org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#operations without Hadoop options should fail for fake:// filesystem
-org.apache.spark.sql.delta.DeltaFastDropFeatureSuite#Vacuum does not delete deletion vector files.generateDVTombstones: false
 org.apache.spark.sql.delta.DeltaGenerateSymlinkManifestSuite#incremental manifest: failure to generate manifest throws exception
 org.apache.spark.sql.delta.DeltaGenerateSymlinkManifestSuite#special partition column values
 org.apache.spark.sql.delta.DeltaHistoryManagerSuite#data skipping still works with time travel
@@ -735,10 +734,8 @@ org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.PartitionLikeDataSkippingColumnMappingSuite#partition-like data skipping for expression COALESCE: COALESCE(TO_DATE(S.b), c) = '1976-07-03' - column mapping id mode
 org.apache.spark.sql.delta.stats.StatsCollectionSuite#recompute stats multiple columns and files
-org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening BIGINT -> DECIMAL(20,0), partitioned=true
 org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening DATE -> TIMESTAMP_NTZ, partitioned=false
 org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening DATE -> TIMESTAMP_NTZ, partitioned=true
-org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening DECIMAL(9,2) -> DECIMAL(19,3), partitioned=true
 org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening FLOAT -> DOUBLE, partitioned=true
 org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening INT -> DOUBLE, partitioned=true
 org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#unsupported type changes DOUBLE -> FLOAT, partitioned=true

backends-velox/src-delta/test/scala/org/apache/gluten/delta/DeltaDeletionVectorDeferredReadTests.scala

@@ -0,0 +1,287 @@
+/*
+ * 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.delta
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat}
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.test.SharedSparkSession
+
+import org.apache.hadoop.fs.Path
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}
+import java.util.concurrent.{CountDownLatch, Executors}
+
+import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.concurrent.duration._
+
+final private[delta] case class TestDeletionVectorFile(
+    relativePath: String,
+    fileSize: Long,
+    encodedDescriptor: String,
+    storageType: String,
+    absolutePath: String,
+    offset: Long,
+    payloadSize: Long,
+    cardinality: Long)
+
+/** Shared executor-side DV payload tests for the Delta 3.3 and Delta 4.0 source profiles. */
+trait DeltaDeletionVectorDeferredReadTests {
+  self: QueryTest with SharedSparkSession =>
+
+  import testImplicits._
+
+  protected def loadDeletionVectorFile(tablePath: Path): TestDeletionVectorFile
+
+  protected def deletionVectorMetadata(encodedDescriptor: String): Map[String, Object]
+
+  protected def encodeDeletionVectorDescriptor(descriptor: DeletionVectorDescriptor): String
+
+  protected def partitionedFileWithMetadata(
+      tablePath: String,
+      relativeFilePath: String,
+      fileSize: Long,
+      metadata: Map[String, Object]): PartitionedFile
+
+  protected def normalizeDeletionVectorOptions(
+      partitionedFile: PartitionedFile,
+      tablePath: Path,
+      readTime: SQLMetric,
+      readBytes: SQLMetric,
+      readAttempts: SQLMetric): DeltaFileReadOptions
+
+  protected def normalizeDeletionVectorOptions(
+      partitionedFile: PartitionedFile,
+      tablePath: Path): DeltaFileReadOptions
+
+  test("defers on-disk DV reads through serialization and coalesces concurrent materialization") {
+    withTempDir {
+      tempDir =>
+        val tablePath = new Path(tempDir.getCanonicalPath, "table")
+        val unrelatedPath = new Path(tempDir.getCanonicalPath, "unrelated")
+        Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+          .toDF("id", "value")
+          .coalesce(1)
+          .write
+          .format("delta")
+          .save(tablePath.toString)
+
+        spark.sql(
+          s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)")
+        spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)")
+
+        val dataFile = loadDeletionVectorFile(tablePath)
+        assert(dataFile.storageType == "u")
+        val partitionedFile = partitionedFileWithMetadata(
+          unrelatedPath.toString,
+          dataFile.relativePath,
+          dataFile.fileSize,
+          deletionVectorMetadata(dataFile.encodedDescriptor)
+        )
+
+        val readTime = SQLMetrics.createNanoTimingMetric(spark.sparkContext, "DV read time")
+        val readBytes = SQLMetrics.createSizeMetric(spark.sparkContext, "DV read bytes")
+        val readAttempts = SQLMetrics.createMetric(spark.sparkContext, "DV read attempts")
+        val options = normalizeDeletionVectorOptions(
+          partitionedFile,
+          tablePath,
+          readTime,
+          readBytes,
+          readAttempts)
+        assert(!options.isDeletionVectorPayloadMaterialized)
+
+        val executorCopy = javaRoundTrip(options)
+        assert(!executorCopy.isDeletionVectorPayloadMaterialized)
+        assert(executorCopy.serializedDeletionVector.nonEmpty)
+        assert(executorCopy.isDeletionVectorPayloadMaterialized)
+        assert(!options.isDeletionVectorPayloadMaterialized)
+
+        val start = new CountDownLatch(1)
+        val pool = Executors.newFixedThreadPool(8)
+        implicit val executionContext: ExecutionContext =
+          ExecutionContext.fromExecutorService(pool)
+        val reads = (1 to 16).map {
+          _ =>
+            Future {
+              start.await()
+              options.serializedDeletionVector
+            }
+        }
+        start.countDown()
+        val payloads =
+          try {
+            Await.result(Future.sequence(reads), 30.seconds)
+          } finally {
+            pool.shutdownNow()
+          }
+
+        assert(payloads.head.nonEmpty)
+        assert(payloads.forall(_ eq payloads.head))
+        assert(options.isDeletionVectorPayloadMaterialized)
+        assert(readAttempts.value == 1L)
+        assert(readBytes.value == payloads.head.length.toLong)
+        assert(readTime.value > 0L)
+    }
+  }
+
+  test("keeps inline DV payloads eager without filesystem access") {
+    val bitmap = new RoaringBitmapArray()
+    bitmap.add(3L)
+    bitmap.add(7L)
+    val expectedPayload = bitmap.serializeAsByteArray(RoaringBitmapArrayFormat.Portable)
+    val descriptor = DeletionVectorDescriptor.inlineInLog(expectedPayload, cardinality = 2L)
+    val tablePath = new Path("unsupported-inline-dv-test://authority/table")
+    val partitionedFile = partitionedFileWithMetadata(
+      tablePath.toString,
+      "data.parquet",
+      fileSize = 0L,
+      metadata = deletionVectorMetadata(encodeDeletionVectorDescriptor(descriptor)))
+
+    val readTime = SQLMetrics.createNanoTimingMetric(spark.sparkContext, "DV read time")
+    val readBytes = SQLMetrics.createSizeMetric(spark.sparkContext, "DV read bytes")
+    val readAttempts = SQLMetrics.createMetric(spark.sparkContext, "DV read attempts")
+    val options = normalizeDeletionVectorOptions(
+      partitionedFile,
+      tablePath,
+      readTime,
+      readBytes,
+      readAttempts)
+
+    assert(options.isDeletionVectorPayloadMaterialized)
+    assert(options.serializedDeletionVector.sameElements(expectedPayload))
+    assert(readAttempts.value == 0L)
+    assert(readBytes.value == 0L)
+    assert(readTime.value == 0L)
+  }
+
+  test("does not cache failed deferred DV reads") {
+    withTempDir {
+      tempDir =>
+        val tablePath = new Path(tempDir.getCanonicalPath, "table")
+        Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+          .toDF("id", "value")
+          .coalesce(1)
+          .write
+          .format("delta")
+          .save(tablePath.toString)
+
+        spark.sql(
+          s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)")
+        spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)")
+
+        val dataFile = loadDeletionVectorFile(tablePath)
+        val partitionedFile = partitionedFileWithMetadata(
+          tablePath.toString,
+          dataFile.relativePath,
+          dataFile.fileSize,
+          deletionVectorMetadata(dataFile.encodedDescriptor)
+        )
+
+        val readTime = SQLMetrics.createNanoTimingMetric(spark.sparkContext, "DV read time")
+        val readBytes = SQLMetrics.createSizeMetric(spark.sparkContext, "DV read bytes")
+        val readAttempts = SQLMetrics.createMetric(spark.sparkContext, "DV read attempts")
+        val options = normalizeDeletionVectorOptions(
+          partitionedFile,
+          tablePath,
+          readTime,
+          readBytes,
+          readAttempts)
+
+        val dvPath = new Path(dataFile.absolutePath)
+        val backupPath = new Path(dvPath.toString + ".retry-test-backup")
+        val fs = dvPath.getFileSystem(spark.sessionState.newHadoopConf())
+        assert(fs.rename(dvPath, backupPath))
+        try {
+          intercept[Exception] {
+            options.serializedDeletionVector
+          }
+          assert(!options.isDeletionVectorPayloadMaterialized)
+          assert(readAttempts.value == 1L)
+          assert(readBytes.value == 0L)
+        } finally {
+          assert(fs.rename(backupPath, dvPath))
+        }
+
+        val payload = options.serializedDeletionVector
+        assert(payload.nonEmpty)
+        assert(options.isDeletionVectorPayloadMaterialized)
+        assert(readAttempts.value == 2L)
+        assert(readBytes.value == payload.length.toLong)
+        assert(readTime.value > 0L)
+    }
+  }
+
+  test("passes authoritative on-disk DV descriptors to native without JVM reads") {
+    withTempDir {
+      tempDir =>
+        val tablePath = new Path(tempDir.getCanonicalPath, "table")
+        Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+          .toDF("id", "value")
+          .coalesce(1)
+          .write
+          .format("delta")
+          .save(tablePath.toString)
+        spark.sql(
+          s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)")
+        spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)")
+
+        val dataFile = loadDeletionVectorFile(tablePath)
+        val partitionedFile = partitionedFileWithMetadata(
+          tablePath.toString,
+          dataFile.relativePath,
+          dataFile.fileSize,
+          deletionVectorMetadata(dataFile.encodedDescriptor)
+        )
+
+        withSQLConf(
+          GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") {
+          val options = normalizeDeletionVectorOptions(partitionedFile, tablePath)
+          assert(options.hasNativeDeletionVectorDescriptor)
+          assert(!options.isDeletionVectorPayloadMaterialized)
+          val descriptor = options.nativeDeletionVectorDescriptor
+          assert(descriptor.absolutePath == dataFile.absolutePath)
+          assert(descriptor.offset == dataFile.offset)
+          assert(descriptor.payloadSize == dataFile.payloadSize)
+          val error = intercept[IllegalStateException](options.serializedDeletionVector)
+          assert(error.getMessage.contains("do not contain JVM payload bytes"))
+        }
+    }
+  }
+
+  private def javaRoundTrip(options: DeltaFileReadOptions): DeltaFileReadOptions = {
+    val bytes = new ByteArrayOutputStream()
+    val output = new ObjectOutputStream(bytes)
+    try {
+      output.writeObject(options)
+    } finally {
+      output.close()
+    }
+
+    val input = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray))
+    try {
+      input.readObject().asInstanceOf[DeltaFileReadOptions]
+    } finally {
+      input.close()
+    }
+  }
+
+}

backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala

@@ -17,15 +17,18 @@
 package org.apache.gluten.delta
 
 import org.apache.gluten.delta.DeltaDeletionVectorScanInfo.RowIndexFilterType
+import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
 
 import org.apache.spark.SparkConf
 import org.apache.spark.paths.SparkPath
 import org.apache.spark.sql.QueryTest
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.delta.{DeltaLog, GlutenDeltaParquetFileFormat}
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
 import org.apache.spark.sql.delta.catalog.DeltaCatalog
 import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
 import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.tags.ExtendedSQLTest
@@ -37,7 +40,8 @@ import org.apache.hadoop.fs.Path
 class DeltaDeletionVectorScanInfoSuite
   extends QueryTest
   with SharedSparkSession
-  with DeltaSQLTestUtils {
+  with DeltaSQLTestUtils
+  with DeltaDeletionVectorDeferredReadTests {
 
   import testImplicits._
 
@@ -181,7 +185,61 @@ class DeltaDeletionVectorScanInfoSuite
     }
   }
 
-  private def partitionedFileWithMetadata(
+  override protected def loadDeletionVectorFile(tablePath: Path): TestDeletionVectorFile = {
+    val dataFile = DeltaLog
+      .forTable(spark, tablePath)
+      .update()
+      .allFiles
+      .collect()
+      .find(_.deletionVector != null)
+      .get
+    val descriptor = dataFile.deletionVector
+    TestDeletionVectorFile(
+      relativePath = dataFile.path,
+      fileSize = dataFile.size,
+      encodedDescriptor = descriptor.serializeToBase64(),
+      storageType = descriptor.storageType,
+      absolutePath = descriptor.absolutePath(tablePath).toString,
+      offset = descriptor.offset.get.toLong,
+      payloadSize = descriptor.sizeInBytes.toLong,
+      cardinality = descriptor.cardinality
+    )
+  }
+
+  override protected def deletionVectorMetadata(
+      encodedDescriptor: String): Map[String, Object] = {
+    Map(
+      GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> encodedDescriptor,
+      GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+    )
+  }
+
+  override protected def encodeDeletionVectorDescriptor(
+      descriptor: DeletionVectorDescriptor): String = descriptor.serializeToBase64()
+
+  override protected def normalizeDeletionVectorOptions(
+      partitionedFile: PartitionedFile,
+      tablePath: Path,
+      readTime: SQLMetric,
+      readBytes: SQLMetric,
+      readAttempts: SQLMetric): DeltaFileReadOptions = {
+    DeltaDeletionVectorScanInfo
+      .normalize(
+        Seq(partitionedFile),
+        tablePath,
+        Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts)))
+      .get
+      ._2
+      .head
+  }
+
+  override protected def normalizeDeletionVectorOptions(
+      partitionedFile: PartitionedFile,
+      tablePath: Path): DeltaFileReadOptions = {
+    DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), tablePath).get._2.head
+  }
+
+  override protected def partitionedFileWithMetadata(
       tablePath: String,
       relativeFilePath: String,
       fileSize: Long,