[GLUTEN][VL] Move Delta deletion vector reads off the driver

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

@@ -16,7 +16,9 @@
  */
 package org.apache.gluten.delta
 
+import org.apache.gluten.config.GlutenConfig
 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
@@ -26,13 +28,20 @@ import org.apache.spark.sql.delta.{DeltaLog, GlutenDeltaParquetFileFormat}
 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.SQLMetrics
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.tags.ExtendedSQLTest
 
 import io.delta.sql.DeltaSparkSessionExtension
 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._
+
 @ExtendedSQLTest
 class DeltaDeletionVectorScanInfoSuite
   extends QueryTest
@@ -82,13 +91,15 @@ class DeltaDeletionVectorScanInfoSuite
           )
         )
 
-        val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile)
+        val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path))
         val dvInfo = scanInfo.deletionVectorInfo
 
         assert(dvInfo.hasDeletionVector)
         assert(dvInfo.rowIndexFilterType == RowIndexFilterType.IF_CONTAINED)
         assert(dvInfo.cardinality == dataFile.deletionVector.cardinality)
+        assert(!dvInfo.isPayloadMaterialized)
         assert(dvInfo.serializedDeletionVector.nonEmpty)
+        assert(dvInfo.isPayloadMaterialized)
         assert(scanInfo.normalizedOtherMetadataColumns == Map("kept_key" -> "kept_value"))
     }
   }
@@ -106,7 +117,7 @@ class DeltaDeletionVectorScanInfoSuite
           dataFile.size,
           Map("kept_key" -> "kept_value"))
 
-        val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile)
+        val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path))
         val dvInfo = scanInfo.deletionVectorInfo
 
         assert(!dvInfo.hasDeletionVector)
@@ -131,12 +142,287 @@ class DeltaDeletionVectorScanInfoSuite
           Map(GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"))
 
         val error = intercept[IllegalStateException] {
-          DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile)
+          DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path))
         }
         assert(error.getMessage.contains("must either be present or absent"))
     }
   }
 
+  test("normalize materializes DV read options using the supplied table path") {
+    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 = DeltaLog
+          .forTable(spark, tablePath)
+          .update()
+          .allFiles
+          .collect()
+          .find(_.deletionVector != null)
+          .get
+        assert(dataFile.deletionVector.storageType == "u")
+        val partitionedFile = partitionedFileWithMetadata(
+          unrelatedPath.toString,
+          dataFile.path,
+          dataFile.size,
+          Map(
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
+              dataFile.deletionVector.serializeToBase64(),
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+          )
+        )
+
+        val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), tablePath)
+        assert(result.isDefined, "normalize should materialize DV options")
+        val opts = result.get._2.head
+        assert(opts.hasDeletionVector)
+        assert(opts.deletionVectorCardinality == dataFile.deletionVector.cardinality)
+        assert(opts.serializedDeletionVector.nonEmpty)
+    }
+  }
+
+  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 = DeltaLog
+          .forTable(spark, tablePath)
+          .update()
+          .allFiles
+          .collect()
+          .find(_.deletionVector != null)
+          .get
+        assert(dataFile.deletionVector.storageType == "u")
+        val partitionedFile = partitionedFileWithMetadata(
+          unrelatedPath.toString,
+          dataFile.path,
+          dataFile.size,
+          Map(
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
+              dataFile.deletionVector.serializeToBase64(),
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+          )
+        )
+
+        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 = DeltaDeletionVectorScanInfo
+          .normalize(
+            Seq(partitionedFile),
+            tablePath,
+            Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts)))
+          .get
+          ._2
+          .head
+        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)
+
+        withSQLConf(
+          GlutenConfig.DELTA_DELETION_VECTOR_DEFER_PAYLOAD_READ_ENABLED.key -> "false") {
+          val eagerOptions = DeltaDeletionVectorScanInfo
+            .normalize(Seq(partitionedFile), tablePath)
+            .get
+            ._2
+            .head
+          assert(eagerOptions.isDeletionVectorPayloadMaterialized)
+          assert(eagerOptions.serializedDeletionVector.nonEmpty)
+        }
+    }
+  }
+
+  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 = DeltaLog
+          .forTable(spark, tablePath)
+          .update()
+          .allFiles
+          .collect()
+          .find(_.deletionVector != null)
+          .get
+        val partitionedFile = partitionedFileWithMetadata(
+          tablePath.toString,
+          dataFile.path,
+          dataFile.size,
+          Map(
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
+              dataFile.deletionVector.serializeToBase64(),
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+          )
+        )
+
+        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 = DeltaDeletionVectorScanInfo
+          .normalize(
+            Seq(partitionedFile),
+            tablePath,
+            Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts)))
+          .get
+          ._2
+          .head
+
+        val dvPath = dataFile.deletionVector.absolutePath(tablePath)
+        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 = DeltaLog
+          .forTable(spark, tablePath)
+          .update()
+          .allFiles
+          .collect()
+          .find(_.deletionVector != null)
+          .get
+        val partitionedFile = partitionedFileWithMetadata(
+          tablePath.toString,
+          dataFile.path,
+          dataFile.size,
+          Map(
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
+              dataFile.deletionVector.serializeToBase64(),
+            GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+          )
+        )
+
+        withSQLConf(
+          GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") {
+          val options = DeltaDeletionVectorScanInfo
+            .normalize(Seq(partitionedFile), tablePath)
+            .get
+            ._2
+            .head
+          assert(options.hasNativeDeletionVectorDescriptor)
+          assert(!options.isDeletionVectorPayloadMaterialized)
+          val descriptor = options.nativeDeletionVectorDescriptor
+          assert(
+            descriptor.absolutePath == dataFile.deletionVector.absolutePath(tablePath).toString)
+          assert(descriptor.offset == dataFile.deletionVector.offset.get.toLong)
+          assert(descriptor.payloadSize == dataFile.deletionVector.sizeInBytes.toLong)
+          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()
+    }
+  }
+
   private def partitionedFileWithMetadata(
       tablePath: String,
       relativeFilePath: String,