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..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 @@ -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, 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 cbb70817ba4..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 @@ -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, 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..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 @@ -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 } 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..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 @@ -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 } 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..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,11 +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: 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 @@ -76,13 +78,8 @@ 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) - 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() - } - } } 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..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,22 +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: 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 val hadoopConf = spark.sessionState.newHadoopConf() - 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( ( @@ -94,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 = { @@ -252,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 86667d988ff..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 @@ -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 @@ -98,20 +99,29 @@ case class DeltaScanTransformer( override def getSplitInfosFromPartitions( partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = { val splitInfos = super.getSplitInfosFromPartitions(partitions) - val partitionColumnCount = getPartitionSchema.fields.length - splitInfos.zip(partitions).map { - case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) => - DeltaDeletionVectorScanInfo - .normalize(partitionColumnCount, filePartition.files.toSeq) - .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..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 { @@ -440,6 +443,41 @@ 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 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( + 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") {