From 2de429f8a61a4aca41d254196318387343022d0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 21 Jul 2026 11:52:09 +0200 Subject: [PATCH 1/3] [GLUTEN][VL] Pass Delta table root from TahoeFileIndex to DV materialization DeltaScanTransformer already knows the Delta table root via relation.location (a TahoeFileIndex, which PreparedDeltaFileIndex also extends). Thread that path into DeltaDeletionVectorScanInfo.normalize so it no longer re-derives the root from a file path via _delta_log existence probing -- one FileSystem.exists() (an HTTP HEAD on object stores) per partition. normalize gains an optional tablePath parameter; when absent it falls back to the previous resolveTablePath heuristic, so non-TahoeFileIndex locations (e.g. DeltaParquetFileFormat scans without a Tahoe index) and the public single-file extract entry point are unchanged. Add a test to DeltaDeletionVectorScanInfoSuite (delta33 and delta40) asserting the supplied-path and derived-path branches materialize an identical DV payload. --- .../DeltaDeletionVectorScanInfoSuite.scala | 55 +++++++++++++++++++ .../DeltaDeletionVectorScanInfoSuite.scala | 55 +++++++++++++++++++ .../delta/DeltaDeletionVectorScanInfo.scala | 7 ++- .../delta/DeltaDeletionVectorScanInfo.scala | 7 ++- .../delta/DeltaDeletionVectorScanInfo.scala | 19 ++++--- .../delta/DeltaDeletionVectorScanInfo.scala | 17 ++++-- .../execution/DeltaScanTransformer.scala | 12 +++- 7 files changed, 157 insertions(+), 15 deletions(-) diff --git a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala index cbb70817ba4..047ac277a27 100644 --- a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala +++ b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala @@ -137,6 +137,61 @@ class DeltaDeletionVectorScanInfoSuite } } + test("normalize materializes the same DV whether the table path is supplied or derived") { + withTempDir { + tempDir => + val path = tempDir.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + + val dataFile = DeltaLog + .forTable(spark, new Path(path)) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + val partitionedFile = partitionedFileWithMetadata( + path, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + // Supplying the table root (as DeltaScanTransformer does via TahoeFileIndex.path) must + // yield exactly the same materialized DV as deriving it from the file path. This proves the + // new tablePath parameter is honored and the fast path stays correct. + val withSuppliedPath = + DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), Some(new Path(path))) + val withDerivedPath = + DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), None) + + assert(withSuppliedPath.isDefined, "normalize should materialize DV options") + assert(withDerivedPath.isDefined, "normalize should materialize DV options") + + val supplied = withSuppliedPath.get._2.head + val derived = withDerivedPath.get._2.head + assert(supplied.hasDeletionVector) + assert(supplied.deletionVectorCardinality == dataFile.deletionVector.cardinality) + assert(supplied.serializedDeletionVector.nonEmpty) + assert( + supplied.serializedDeletionVector.sameElements(derived.serializedDeletionVector), + "supplied-path and derived-path DV payloads must be identical") + } + } + private def partitionedFileWithMetadata( tablePath: String, relativeFilePath: String, diff --git a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala index cbb70817ba4..047ac277a27 100644 --- a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala +++ b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala @@ -137,6 +137,61 @@ class DeltaDeletionVectorScanInfoSuite } } + test("normalize materializes the same DV whether the table path is supplied or derived") { + withTempDir { + tempDir => + val path = tempDir.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + + val dataFile = DeltaLog + .forTable(spark, new Path(path)) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + val partitionedFile = partitionedFileWithMetadata( + path, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + // Supplying the table root (as DeltaScanTransformer does via TahoeFileIndex.path) must + // yield exactly the same materialized DV as deriving it from the file path. This proves the + // new tablePath parameter is honored and the fast path stays correct. + val withSuppliedPath = + DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), Some(new Path(path))) + val withDerivedPath = + DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), None) + + assert(withSuppliedPath.isDefined, "normalize should materialize DV options") + assert(withDerivedPath.isDefined, "normalize should materialize DV options") + + val supplied = withSuppliedPath.get._2.head + val derived = withDerivedPath.get._2.head + assert(supplied.hasDeletionVector) + assert(supplied.deletionVectorCardinality == dataFile.deletionVector.cardinality) + assert(supplied.serializedDeletionVector.nonEmpty) + assert( + supplied.serializedDeletionVector.sameElements(derived.serializedDeletionVector), + "supplied-path and derived-path DV payloads must be identical") + } + } + private def partitionedFileWithMetadata( tablePath: String, relativeFilePath: String, diff --git a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index af8e8df7da8..2b868c8611f 100644 --- a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -20,10 +20,15 @@ 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( + partitionColumnCount: Int, + partitionFiles: Seq[PartitionedFile], + tablePath: Option[Path] = None) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None } diff --git a/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index af8e8df7da8..2b868c8611f 100644 --- a/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -20,10 +20,15 @@ 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( + partitionColumnCount: Int, + partitionFiles: Seq[PartitionedFile], + tablePath: Option[Path] = None) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None } diff --git a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index 812f9b9ec36..15df9759e29 100644 --- a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -64,11 +64,15 @@ 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. + * Performance: reuses a single Hadoop Configuration instance across all files in the partition. + * The table root is taken from `tablePath` when the caller can supply it (e.g. from + * `TahoeFileIndex.path`); otherwise it is derived once from the first file, which requires a + * `_delta_log` existence probe. Passing `tablePath` avoids that filesystem round-trip. */ - def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) + def normalize( + partitionColumnCount: Int, + partitionFiles: Seq[PartitionedFile], + tablePath: Option[Path] = None) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { if (partitionFiles.isEmpty) { return None @@ -76,9 +80,10 @@ object DeltaDeletionVectorScanInfo { 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) + // Prefer the caller-supplied table root (TahoeFileIndex.path). Fall back to deriving it from + // the first file -- which probes the filesystem for _delta_log -- only when unavailable. + val cachedTablePath = + tablePath.getOrElse(resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head)) val scanInfos = partitionFiles.map { file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath) diff --git a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index 22ffb3c89a8..dc467471da4 100644 --- a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -65,18 +65,25 @@ 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. + * Performance: reuses a single Hadoop Configuration instance across all files in the partition. + * The table root is taken from `tablePath` when the caller can supply it (e.g. from + * `TahoeFileIndex.path`); otherwise it is derived once from the first file, which requires a + * `_delta_log` existence probe. Passing `tablePath` avoids that filesystem round-trip. */ - def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) + def normalize( + partitionColumnCount: Int, + partitionFiles: Seq[PartitionedFile], + tablePath: Option[Path] = None) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { if (partitionFiles.isEmpty) { return None } val spark = activeSparkSession val hadoopConf = spark.sessionState.newHadoopConf() - val cachedTablePath = resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head) + // Prefer the caller-supplied table root (TahoeFileIndex.path). Fall back to deriving it from + // the first file -- which probes the filesystem for _delta_log -- only when unavailable. + val cachedTablePath = + tablePath.getOrElse(resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head)) val scanInfos = partitionFiles.map { file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath) diff --git a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala index 86667d988ff..b78e57014a4 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala +++ b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala @@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, import org.apache.spark.sql.catalyst.plans.QueryPlan import org.apache.spark.sql.connector.read.streaming.SparkDataStream import org.apache.spark.sql.delta.{DeltaParquetFileFormat, NoMapping} +import org.apache.spark.sql.delta.files.TahoeFileIndex import org.apache.spark.sql.execution.FileSourceScanExec import org.apache.spark.sql.execution.datasources.{FilePartition, HadoopFsRelation} import org.apache.spark.sql.types.StructType @@ -99,10 +100,19 @@ case class DeltaScanTransformer( partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = { val splitInfos = super.getSplitInfosFromPartitions(partitions) val partitionColumnCount = getPartitionSchema.fields.length + // The Delta table root is already known from the scan's file index (TahoeFileIndex.path), so + // pass it down to DV materialization. This avoids re-deriving the root from a file path via + // `_delta_log` existence probing -- one FileSystem.exists() call (an HTTP HEAD on object + // stores) per partition. When the location is not a TahoeFileIndex we pass None and + // normalize() falls back to deriving the path from the files. + val tableRootPath = relation.location match { + case tahoe: TahoeFileIndex => Some(tahoe.path) + case _ => None + } splitInfos.zip(partitions).map { case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) => DeltaDeletionVectorScanInfo - .normalize(partitionColumnCount, filePartition.files.toSeq) + .normalize(partitionColumnCount, filePartition.files.toSeq, tableRootPath) .map { case (otherMetadataColumns, deltaReadOptions) => DeltaLocalFilesBuilder.makeDeltaLocalFiles( From f3cf6d72ed1f2ea212d327476e2e352dbf9d2f2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 21 Jul 2026 12:02:21 +0200 Subject: [PATCH 2/3] [GLUTEN][VL] Require TahoeFileIndex table root for DV; drop resolveTablePath fallback Now that DeltaScanTransformer passes the table root from TahoeFileIndex.path, make it the single source of truth for DV materialization and remove the previous file-path-derivation fallback. - DeltaDeletionVectorScanInfo.normalize takes a required tablePath: Path (no Option, no fallback); partitionColumnCount is dropped since only the walk-up used it. - Delete resolveTablePath / isDeltaTablePath / unescapePathName and their per-partition _delta_log FileSystem.exists() probing. - DeltaScanTransformer materializes DVs only when relation.location is a TahoeFileIndex (which also covers PreparedDeltaFileIndex); other locations carry no Delta DV metadata and keep the generic split. - Update the public single-file extract(spark, file, tablePath), the delta23/24 stubs, the benchmark, and the suites accordingly. Net ~160 fewer lines. The Hadoop-conf caching and raw on-disk DV byte reading optimizations are retained. --- .../DeltaDeletionVectorScanInfoSuite.scala | 33 +++----- .../benchmark/DeltaPlanningBenchmark.scala | 5 +- .../DeltaDeletionVectorScanInfoSuite.scala | 33 +++----- .../delta/DeltaDeletionVectorScanInfo.scala | 3 +- .../delta/DeltaDeletionVectorScanInfo.scala | 3 +- .../delta/DeltaDeletionVectorScanInfo.scala | 82 ++----------------- .../delta/DeltaDeletionVectorScanInfo.scala | 82 ++----------------- .../execution/DeltaScanTransformer.scala | 46 +++++------ .../apache/gluten/execution/DeltaSuite.scala | 25 ++++++ 9 files changed, 87 insertions(+), 225 deletions(-) diff --git a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala index 047ac277a27..ac8686e101a 100644 --- a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala +++ b/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,13 +131,13 @@ 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 the same DV whether the table path is supplied or derived") { + test("normalize materializes DV read options using the supplied table path") { withTempDir { tempDir => val path = tempDir.getCanonicalPath @@ -170,25 +170,12 @@ class DeltaDeletionVectorScanInfoSuite ) ) - // Supplying the table root (as DeltaScanTransformer does via TahoeFileIndex.path) must - // yield exactly the same materialized DV as deriving it from the file path. This proves the - // new tablePath parameter is honored and the fast path stays correct. - val withSuppliedPath = - DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), Some(new Path(path))) - val withDerivedPath = - DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), None) - - assert(withSuppliedPath.isDefined, "normalize should materialize DV options") - assert(withDerivedPath.isDefined, "normalize should materialize DV options") - - val supplied = withSuppliedPath.get._2.head - val derived = withDerivedPath.get._2.head - assert(supplied.hasDeletionVector) - assert(supplied.deletionVectorCardinality == dataFile.deletionVector.cardinality) - assert(supplied.serializedDeletionVector.nonEmpty) - assert( - supplied.serializedDeletionVector.sameElements(derived.serializedDeletionVector), - "supplied-path and derived-path DV payloads must be identical") + val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), new Path(path)) + 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) } } diff --git a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala index 597a5b079d8..c190397691b 100644 --- a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala +++ b/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() diff --git a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala index 047ac277a27..ac8686e101a 100644 --- a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala +++ b/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,13 +131,13 @@ 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 the same DV whether the table path is supplied or derived") { + test("normalize materializes DV read options using the supplied table path") { withTempDir { tempDir => val path = tempDir.getCanonicalPath @@ -170,25 +170,12 @@ class DeltaDeletionVectorScanInfoSuite ) ) - // Supplying the table root (as DeltaScanTransformer does via TahoeFileIndex.path) must - // yield exactly the same materialized DV as deriving it from the file path. This proves the - // new tablePath parameter is honored and the fast path stays correct. - val withSuppliedPath = - DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), Some(new Path(path))) - val withDerivedPath = - DeltaDeletionVectorScanInfo.normalize(0, Seq(partitionedFile), None) - - assert(withSuppliedPath.isDefined, "normalize should materialize DV options") - assert(withDerivedPath.isDefined, "normalize should materialize DV options") - - val supplied = withSuppliedPath.get._2.head - val derived = withDerivedPath.get._2.head - assert(supplied.hasDeletionVector) - assert(supplied.deletionVectorCardinality == dataFile.deletionVector.cardinality) - assert(supplied.serializedDeletionVector.nonEmpty) - assert( - supplied.serializedDeletionVector.sameElements(derived.serializedDeletionVector), - "supplied-path and derived-path DV payloads must be identical") + val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), new Path(path)) + 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) } } diff --git a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index 2b868c8611f..903a2066639 100644 --- a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -27,8 +27,7 @@ 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], - tablePath: Option[Path] = None) + tablePath: Path) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None } diff --git a/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index 2b868c8611f..903a2066639 100644 --- a/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -27,8 +27,7 @@ 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], - tablePath: Option[Path] = None) + tablePath: Path) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None } diff --git a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index 15df9759e29..7f36aaf8832 100644 --- a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -64,15 +64,13 @@ 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: reuses a single Hadoop Configuration instance across all files in the partition. - * The table root is taken from `tablePath` when the caller can supply it (e.g. from - * `TahoeFileIndex.path`); otherwise it is derived once from the first file, which requires a - * `_delta_log` existence probe. Passing `tablePath` avoids that filesystem round-trip. + * `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], - tablePath: Option[Path] = None) + tablePath: Path) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { if (partitionFiles.isEmpty) { return None @@ -80,14 +78,8 @@ object DeltaDeletionVectorScanInfo { val spark = activeSparkSession // Create a single Hadoop Configuration for the entire partition. val hadoopConf = spark.sessionState.newHadoopConf() - // Prefer the caller-supplied table root (TahoeFileIndex.path). Fall back to deriving it from - // the first file -- which probes the filesystem for _delta_log -- only when unavailable. - val cachedTablePath = - tablePath.getOrElse(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( ( @@ -101,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 = { @@ -253,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() - } - } } diff --git a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index dc467471da4..24f95841147 100644 --- a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -65,29 +65,21 @@ 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: reuses a single Hadoop Configuration instance across all files in the partition. - * The table root is taken from `tablePath` when the caller can supply it (e.g. from - * `TahoeFileIndex.path`); otherwise it is derived once from the first file, which requires a - * `_delta_log` existence probe. Passing `tablePath` avoids that filesystem round-trip. + * `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], - tablePath: Option[Path] = None) + tablePath: Path) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { if (partitionFiles.isEmpty) { return None } val spark = activeSparkSession val hadoopConf = spark.sessionState.newHadoopConf() - // Prefer the caller-supplied table root (TahoeFileIndex.path). Fall back to deriving it from - // the first file -- which probes the filesystem for _delta_log -- only when unavailable. - val cachedTablePath = - tablePath.getOrElse(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( ( @@ -101,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 = { @@ -259,60 +249,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() - } - } } diff --git a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala index b78e57014a4..8462c1fca5f 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala +++ b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala @@ -99,29 +99,29 @@ case class DeltaScanTransformer( override def getSplitInfosFromPartitions( partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = { val splitInfos = super.getSplitInfosFromPartitions(partitions) - val partitionColumnCount = getPartitionSchema.fields.length - // The Delta table root is already known from the scan's file index (TahoeFileIndex.path), so - // pass it down to DV materialization. This avoids re-deriving the root from a file path via - // `_delta_log` existence probing -- one FileSystem.exists() call (an HTTP HEAD on object - // stores) per partition. When the location is not a TahoeFileIndex we pass None and - // normalize() falls back to deriving the path from the files. - val tableRootPath = relation.location match { - case tahoe: TahoeFileIndex => Some(tahoe.path) - case _ => None - } - splitInfos.zip(partitions).map { - case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) => - DeltaDeletionVectorScanInfo - .normalize(partitionColumnCount, filePartition.files.toSeq, tableRootPath) - .map { - case (otherMetadataColumns, deltaReadOptions) => - DeltaLocalFilesBuilder.makeDeltaLocalFiles( - localFiles, - otherMetadataColumns.asJava, - deltaReadOptions.asJava): SplitInfo - } - .getOrElse(localFiles) - case (splitInfo, _) => splitInfo + // Deletion vectors only exist on Delta tables read through a TahoeFileIndex (which also covers + // PreparedDeltaFileIndex). Its `path` is the authoritative table root and is used to resolve + // per-file DV locations. Any other location cannot carry Delta DV metadata, so the generic + // split representation is returned unchanged. + relation.location match { + case tahoe: TahoeFileIndex => + val tableRootPath = tahoe.path + splitInfos.zip(partitions).map { + case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) => + DeltaDeletionVectorScanInfo + .normalize(filePartition.files.toSeq, tableRootPath) + .map { + case (otherMetadataColumns, deltaReadOptions) => + DeltaLocalFilesBuilder.makeDeltaLocalFiles( + localFiles, + otherMetadataColumns.asJava, + deltaReadOptions.asJava): SplitInfo + } + .getOrElse(localFiles) + case (splitInfo, _) => splitInfo + } + case _ => + splitInfos } } diff --git a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala index 9f8e994f8d0..e376d8e0336 100644 --- a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala +++ b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala @@ -440,6 +440,31 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { } } + testWithMinSparkVersion("deletion vector on partitioned table", "3.4") { + withTempPath { + p => + import testImplicits._ + val path = p.getCanonicalPath + // Partitioned so data files live under partition subdirs (region=.../...). The DV path is + // resolved from the table root (TahoeFileIndex.path) regardless of partition nesting; this + // guards the removal of the old partition-count-based table-path walk-up. + val data = + Seq((1, "a"), (2, "a"), (3, "b"), (4, "b"), (5, "a"), (6, "b")).toDF("id", "region") + data.write.format("delta").partitionBy("region").save(path) + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (2, 3, 6)") + val df = spark.read.format("delta").load(path) + if (SparkVersionUtil.gteSpark35) { + assert( + df.queryExecution.executedPlan + .collect { case _: DeltaScanTransformer => true } + .nonEmpty) + } + checkAnswer(df, Seq((1, "a"), (4, "b"), (5, "a")).toDF("id", "region")) + } + } + testWithMinSparkVersion("delta: push down input_file_name expression", "3.2") { withTable("source_table") { withTable("target_table") { From 10988b1381d2caf995eeef333dfa0a59d8df5e46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 23 Jul 2026 17:55:07 +0200 Subject: [PATCH 3/3] [GLUTEN][VL] Harden Delta DV table path tests --- .../delta/DeltaDeletionVectorScanInfoSuite.scala | 16 +++++++++------- .../delta/DeltaDeletionVectorScanInfoSuite.scala | 16 +++++++++------- .../org/apache/gluten/execution/DeltaSuite.scala | 13 +++++++++++++ 3 files changed, 31 insertions(+), 14 deletions(-) diff --git a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala index ac8686e101a..94c4bd2193c 100644 --- a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala +++ b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala @@ -140,27 +140,29 @@ class DeltaDeletionVectorScanInfoSuite test("normalize materializes DV read options using the supplied table path") { withTempDir { tempDir => - val path = tempDir.getCanonicalPath + 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(path) + .save(tablePath.toString) spark.sql( - s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") - spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + 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, new Path(path)) + .forTable(spark, tablePath) .update() .allFiles .collect() .find(_.deletionVector != null) .get + assert(dataFile.deletionVector.storageType == "u") val partitionedFile = partitionedFileWithMetadata( - path, + unrelatedPath.toString, dataFile.path, dataFile.size, Map( @@ -170,7 +172,7 @@ class DeltaDeletionVectorScanInfoSuite ) ) - val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), new Path(path)) + val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), tablePath) assert(result.isDefined, "normalize should materialize DV options") val opts = result.get._2.head assert(opts.hasDeletionVector) diff --git a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala index ac8686e101a..94c4bd2193c 100644 --- a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala +++ b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala @@ -140,27 +140,29 @@ class DeltaDeletionVectorScanInfoSuite test("normalize materializes DV read options using the supplied table path") { withTempDir { tempDir => - val path = tempDir.getCanonicalPath + 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(path) + .save(tablePath.toString) spark.sql( - s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") - spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + 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, new Path(path)) + .forTable(spark, tablePath) .update() .allFiles .collect() .find(_.deletionVector != null) .get + assert(dataFile.deletionVector.storageType == "u") val partitionedFile = partitionedFileWithMetadata( - path, + unrelatedPath.toString, dataFile.path, dataFile.size, Map( @@ -170,7 +172,7 @@ class DeltaDeletionVectorScanInfoSuite ) ) - val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), new Path(path)) + val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), tablePath) assert(result.isDefined, "normalize should materialize DV options") val opts = result.get._2.head assert(opts.hasDeletionVector) diff --git a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala index e376d8e0336..39eed5d86d4 100644 --- a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala +++ b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala @@ -20,9 +20,12 @@ import org.apache.gluten.extension.DeltaPostTransformRules import org.apache.spark.SparkConf import org.apache.spark.sql.Row +import org.apache.spark.sql.delta.DeltaLog import org.apache.spark.sql.types._ import org.apache.spark.util.SparkVersionUtil +import org.apache.hadoop.fs.Path + import scala.collection.JavaConverters._ abstract class DeltaSuite extends WholeStageTransformerSuite { @@ -454,6 +457,16 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { spark.sql( s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (2, 3, 6)") + val deletionVectors = DeltaLog + .forTable(spark, new Path(path)) + .update() + .allFiles + .collect() + .flatMap(file => Option(file.deletionVector)) + assert(deletionVectors.nonEmpty, "DELETE should produce deletion vectors") + assert( + deletionVectors.exists(_.storageType == "u"), + "DELETE should produce a table-root-relative UUID deletion vector") val df = spark.read.format("delta").load(path) if (SparkVersionUtil.gteSpark35) { assert(