[GLUTEN][VL] Use authoritative TahoeFileIndex table root for Delta deletion vectors

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

@@ -82,7 +82,7 @@ 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)
@@ -106,7 +106,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 +131,56 @@ 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)
+    }
+  }
+
   private def partitionedFileWithMetadata(
       tablePath: String,
       relativeFilePath: String,

backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala

@@ -97,10 +97,7 @@ object DeltaPlanningBenchmark extends SqlBasedBenchmark {
     withDeltaTableWithDVs(numFiles, rowsPerFile) {
       (path, partitionedFiles) =>
         benchmark.addCase(s"normalize() - $numFiles DV files", benchmarkIters) {
-          _ =>
-            DeltaDeletionVectorScanInfo.normalize(
-              partitionColumnCount = 0,
-              partitionFiles = partitionedFiles)
+          _ => DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new Path(path))
         }
 
         benchmark.run()

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

@@ -82,7 +82,7 @@ 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)
@@ -106,7 +106,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 +131,56 @@ 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)
+    }
+  }
+
   private def partitionedFileWithMetadata(
       tablePath: String,
       relativeFilePath: String,

gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala

@@ -20,10 +20,14 @@ import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
 
 import org.apache.spark.sql.execution.datasources.PartitionedFile
 
+import org.apache.hadoop.fs.Path
+
 import java.util.{Map => JMap}
 
 /** Reading deletion vectors natively requires Delta 3.3+, so there is nothing to materialize. */
 object DeltaDeletionVectorScanInfo {
-  def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile])
+  def normalize(
+      partitionFiles: Seq[PartitionedFile],
+      tablePath: Path)
       : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
 }

gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala

@@ -20,10 +20,14 @@ import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
 
 import org.apache.spark.sql.execution.datasources.PartitionedFile
 
+import org.apache.hadoop.fs.Path
+
 import java.util.{Map => JMap}
 
 /** Reading deletion vectors natively requires Delta 3.3+, so there is nothing to materialize. */
 object DeltaDeletionVectorScanInfo {
-  def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile])
+  def normalize(
+      partitionFiles: Seq[PartitionedFile],
+      tablePath: Path)
       : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
 }

gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala

@@ -64,25 +64,22 @@ object DeltaDeletionVectorScanInfo {
    * the DV bookkeeping keys stripped. Returns None when no file in the split carries a deletion
    * vector, so callers can keep the generic split representation.
    *
-   * Performance: resolves the table path once (using the first file) and reuses a single Hadoop
-   * Configuration instance across all files in the partition to avoid redundant filesystem I/O and
-   * object allocation.
+   * `tablePath` is the Delta table root, supplied by the caller from `TahoeFileIndex.path`, and is
+   * used to resolve on-disk DV locations. A single Hadoop Configuration is reused across all files
+   * in the partition.
    */
-  def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile])
+  def normalize(
+      partitionFiles: Seq[PartitionedFile],
+      tablePath: Path)
       : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
     if (partitionFiles.isEmpty) {
       return None
     }
     val spark = activeSparkSession
     // Create a single Hadoop Configuration for the entire partition.
     val hadoopConf = spark.sessionState.newHadoopConf()
-    // Resolve table path once using the first file -- all files in a Delta table share the same
-    // root, so this avoids N-1 redundant filesystem existence checks.
-    val cachedTablePath = resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head)
 
-    val scanInfos = partitionFiles.map {
-      file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath)
-    }
+    val scanInfos = partitionFiles.map(file => extract(file, hadoopConf, tablePath))
     if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) {
       Some(
         (
@@ -96,15 +93,13 @@ object DeltaDeletionVectorScanInfo {
   /** Public entry point for extracting DV info from a single file (used by tests). */
   def extract(
       spark: SparkSession,
-      partitionColumnCount: Int,
-      file: PartitionedFile): PartitionFileScanInfo = {
+      file: PartitionedFile,
+      tablePath: Path): PartitionFileScanInfo = {
     val hadoopConf = spark.sessionState.newHadoopConf()
-    val tablePath = resolveTablePath(hadoopConf, partitionColumnCount, file)
-    extract(partitionColumnCount, file, hadoopConf, tablePath)
+    extract(file, hadoopConf, tablePath)
   }
 
   private def extract(
-      partitionColumnCount: Int,
       file: PartitionedFile,
       hadoopConf: Configuration,
       tablePath: Path): PartitionFileScanInfo = {
@@ -248,60 +243,4 @@ object DeltaDeletionVectorScanInfo {
     }
   }
 
-  private def resolveTablePath(
-      hadoopConf: org.apache.hadoop.conf.Configuration,
-      partitionColumnCount: Int,
-      file: PartitionedFile): Path = {
-    val fileParent = new Path(unescapePathName(file.filePath.toString)).getParent
-    var tablePath = fileParent
-    for (_ <- 0 until partitionColumnCount) {
-      tablePath = tablePath.getParent
-    }
-    if (tablePath != null && isDeltaTablePath(hadoopConf, tablePath)) {
-      return tablePath
-    }
-
-    var candidate = fileParent
-    while (candidate != null && !isDeltaTablePath(hadoopConf, candidate)) {
-      candidate = candidate.getParent
-    }
-    if (candidate != null) candidate else tablePath
-  }
-
-  private def isDeltaTablePath(
-      hadoopConf: org.apache.hadoop.conf.Configuration,
-      tablePath: Path): Boolean = {
-    val deltaLogPath = new Path(tablePath, "_delta_log")
-    try {
-      deltaLogPath.getFileSystem(hadoopConf).exists(deltaLogPath)
-    } catch {
-      case NonFatal(_) => false
-    }
-  }
-
-  private def unescapePathName(path: String): String = {
-    if (path == null || path.indexOf('%') < 0) {
-      path
-    } else {
-      val builder = new StringBuilder(path.length)
-      var index = 0
-      while (index < path.length) {
-        if (path.charAt(index) == '%' && index + 2 < path.length) {
-          val high = Character.digit(path.charAt(index + 1), 16)
-          val low = Character.digit(path.charAt(index + 2), 16)
-          if (high >= 0 && low >= 0) {
-            builder.append(((high << 4) | low).toChar)
-            index += 3
-          } else {
-            builder.append(path.charAt(index))
-            index += 1
-          }
-        } else {
-          builder.append(path.charAt(index))
-          index += 1
-        }
-      }
-      builder.toString()
-    }
-  }
 }