This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 33f3b21726 [GLUTEN][VL] Use authoritative TahoeFileIndex table root
for Delta deletion vectors (#12612)
33f3b21726 is described below
commit 33f3b21726386b39ba4d823f4069907301e4857e
Author: Ismaël Mejía <[email protected]>
AuthorDate: Fri Aug 21 11:46:53 2026 +0200
[GLUTEN][VL] Use authoritative TahoeFileIndex table root for Delta deletion
vectors (#12612)
---
.../util/delta-spark-ut/known-failures.txt | 2 -
.../delta/DeltaDeletionVectorScanInfoSuite.scala | 50 +++++++++++-
.../benchmark/DeltaPlanningBenchmark.scala | 13 ++-
.../delta/DeltaDeletionVectorScanInfoSuite.scala | 50 +++++++++++-
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 6 +-
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 6 +-
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 91 ++++-----------------
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 89 ++++-----------------
.../gluten/execution/DeltaScanTransformer.scala | 39 +++++----
.../org/apache/gluten/execution/DeltaSuite.scala | 93 ++++++++++++++++++++++
10 files changed, 260 insertions(+), 179 deletions(-)
diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt
b/.github/workflows/util/delta-spark-ut/known-failures.txt
index 0b68094b2f..085ed403a8 100644
--- a/.github/workflows/util/delta-spark-ut/known-failures.txt
+++ b/.github/workflows/util/delta-spark-ut/known-failures.txt
@@ -59,8 +59,6 @@ org.apache.spark.sql.delta.CloneTableSQLSuite#shallow clone
across file systems
org.apache.spark.sql.delta.CloneTableSQLWithCatalogOwnedBatch100Suite#shallow
clone across file systems
org.apache.spark.sql.delta.CloneTableSQLWithCatalogOwnedBatch1Suite#shallow
clone across file systems
org.apache.spark.sql.delta.CloneTableSQLWithCatalogOwnedBatch2Suite#shallow
clone across file systems
-org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#Cloning table
with persistent DVs and absolute parquet paths
-org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#Shallow clone
round-trip with DVs
org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#shallow clone
across file systems
org.apache.spark.sql.delta.CloneTableScalaSuite#shallow clone across file
systems
org.apache.spark.sql.delta.ConvertToDeltaSQLSuite#external tables use correct
path scheme
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 cbb70817ba..94c4bd2193 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 597a5b079d..68a2332f76 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
@@ -30,9 +30,9 @@ import org.apache.hadoop.fs.Path
*
* Measures two hot paths that our performance optimizations target:
*
- * 1. '''DV Materialization''' (`DeltaDeletionVectorScanInfo.normalize`):
resolves table paths,
- * loads DV bitmaps from storage, and serializes them into split
metadata. Our optimizations
- * (caching table path, Hadoop conf, DV store across files) target this
path.
+ * 1. '''DV Materialization''' (`DeltaDeletionVectorScanInfo.normalize`):
loads DV bitmaps from
+ * storage and serializes them into split metadata. Our optimizations
(reusing the Hadoop conf
+ * and DV store across files) target this path.
* 2. '''Post-transform rule application'''
(`DeltaPostTransformRules.rules`): traverses the
* physical plan to strip DV synthetic columns, push down
input_file_name, and apply column
* mapping. Our optimizations (early-exit guard, shallow child check,
pre-computed names,
@@ -85,7 +85,7 @@ object DeltaPlanningBenchmark extends SqlBasedBenchmark {
/**
* Benchmarks DeltaDeletionVectorScanInfo.normalize() -- the critical path
that loads DVs from
- * storage on the driver. Measures how caching table path + DV store reduces
overhead.
+ * storage on the driver. Measures how reusing the DV store across files
reduces overhead.
*/
private def runDvMaterializationBenchmark(): Unit = {
val benchmark = new Benchmark(
@@ -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 cbb70817ba..94c4bd2193 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 af8e8df7da..903a206663 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 af8e8df7da..903a206663 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 812f9b9ec3..e5c4d8590b 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
@@ -30,7 +30,6 @@ import
org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
-import java.io.DataInputStream
import java.util.{Map => JMap}
import scala.collection.JavaConverters._
@@ -64,11 +63,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 +77,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 +92,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 = {
@@ -236,11 +230,16 @@ object DeltaDeletionVectorScanInfo {
descriptor: DeletionVectorDescriptor): Array[Byte] = {
val dvPath = descriptor.absolutePath(tablePath)
val fs = dvPath.getFileSystem(hadoopConf)
- val stream = new DataInputStream(fs.open(dvPath))
+ // Positioned absolute seek, matching Delta's own
`HadoopFileSystemDVStore.read`. `seek` is a
+ // single positioned reposition (a ranged read on object stores), whereas
`DataInputStream.
+ // skipBytes` is best-effort -- it can skip fewer bytes than requested
without error, which would
+ // then fail the CRC check in `readRangeFromStream`. `FSDataInputStream`
is a `DataInputStream`,
+ // so it is passed through directly.
+ val stream = fs.open(dvPath)
try {
val offset = descriptor.offset.getOrElse(0)
if (offset > 0) {
- stream.skipBytes(offset)
+ stream.seek(offset.toLong)
}
DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes)
} finally {
@@ -248,60 +247,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 22ffb3c89a..11530d665a 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
@@ -30,7 +30,6 @@ import
org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
-import java.io.DataInputStream
import java.util.{Map => JMap}
import scala.collection.JavaConverters._
@@ -65,22 +64,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 +92,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 = {
@@ -240,11 +236,16 @@ object DeltaDeletionVectorScanInfo {
descriptor: DeletionVectorDescriptor): Array[Byte] = {
val dvPath = descriptor.absolutePath(tablePath)
val fs = dvPath.getFileSystem(hadoopConf)
- val stream = new DataInputStream(fs.open(dvPath))
+ // Positioned absolute seek, matching Delta's own
`HadoopFileSystemDVStore.read`. `seek` is a
+ // single positioned reposition (a ranged read on object stores), whereas
`DataInputStream.
+ // skipBytes` is best-effort -- it can skip fewer bytes than requested
without error, which would
+ // then fail the CRC check in `readRangeFromStream`. `FSDataInputStream`
is a `DataInputStream`,
+ // so it is passed through directly.
+ val stream = fs.open(dvPath)
try {
val offset = descriptor.offset.getOrElse(0)
if (offset > 0) {
- stream.skipBytes(offset)
+ stream.seek(offset.toLong)
}
DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes)
} finally {
@@ -252,60 +253,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 c2cb6604db..50073cd197 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,7 +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.{CdcAddFileIndex, TahoeRemoveFileIndex}
+import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeFileIndex,
TahoeRemoveFileIndex}
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.datasources.{FilePartition,
HadoopFsRelation}
import org.apache.spark.sql.types.StructType
@@ -120,20 +120,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 138d318cff..d5fd0bfb7e 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,13 @@ import org.apache.gluten.extension.DeltaPostTransformRules
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.TableIdentifier
+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 {
@@ -628,6 +632,95 @@ abstract class DeltaSuite extends
WholeStageTransformerSuite {
}
}
+ testWithMinSparkVersion("deletion vector on partitioned table", "3.4") {
+ withTempPath {
+ p =>
+ import testImplicits._
+ val path = p.getCanonicalPath
+ // End-to-end DV read over a partitioned table: data files live under
partition subdirs
+ // (region=.../...) while the DELETE writes table-root-relative ("u")
UUID deletion vectors.
+ // This exercises the full native DV path -- resolving each DV against
the table root
+ // (TahoeFileIndex.path) and applying it -- and asserts correct
results. The root
+ // discrimination itself is unit-tested in
DeltaDeletionVectorScanInfoSuite ("normalize
+ // materializes DV read options using the supplied table path"), which
points a
+ // PartitionedFile at an unrelated directory.
+ 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("deletion vector on shallow-cloned table", "3.4") {
+ withTable("dv_clone_source", "dv_clone_target") {
+ import testImplicits._
+ // Shallow clone is the case the old data-file walk-up got wrong. The
clone's AddFile paths
+ // point ABSOLUTE into the source table, while its _delta_log (and the
DV written by a DELETE
+ // on the clone) live under the clone root. Walking up from a data file
therefore lands on the
+ // SOURCE table's _delta_log and resolves the wrong root, so the
clone-root-relative "u" DV
+ // cannot be found. Sourcing the root from TahoeFileIndex.path fixes
this. This pins the
+ // DeltaScanTransformer Tahoe arm on Spark 3.5 (the
CloneTableScalaDeletionVectorSuite shards
+ // only run on delta40).
+ spark.sql(
+ "CREATE TABLE dv_clone_source (id INT, region STRING) USING delta " +
+ "TBLPROPERTIES ('delta.enableDeletionVectors' = true)")
+ Seq((1, "a"), (2, "a"), (3, "b"), (4, "b"), (5, "a"), (6, "b"))
+ .toDF("id", "region")
+ .write
+ .format("delta")
+ .mode("append")
+ .saveAsTable("dv_clone_source")
+ spark.sql("CREATE TABLE dv_clone_target SHALLOW CLONE dv_clone_source")
+ // DELETE on the clone writes a clone-root-relative UUID ("u") DV; the
data files stay
+ // absolute into the source.
+ spark.sql("DELETE FROM dv_clone_target WHERE id IN (2, 3, 6)")
+ val targetLocation =
+
spark.sessionState.catalog.getTableMetadata(TableIdentifier("dv_clone_target")).location
+ val deletionVectors = DeltaLog
+ .forTable(spark, new Path(targetLocation))
+ .update()
+ .allFiles
+ .collect()
+ .flatMap(file => Option(file.deletionVector))
+ assert(deletionVectors.nonEmpty, "DELETE on the clone should produce
deletion vectors")
+ assert(
+ deletionVectors.exists(_.storageType == "u"),
+ "DELETE on the clone should produce a clone-root-relative UUID
deletion vector")
+ val df = spark.table("dv_clone_target")
+ if (SparkVersionUtil.gteSpark35) {
+ assert(
+ df.queryExecution.executedPlan
+ .collect { case _: DeltaScanTransformer => true }
+ .nonEmpty)
+ }
+ // The clone's DELETE must not affect the source table.
+ checkAnswer(
+ spark.table("dv_clone_source"),
+ Seq((1, "a"), (2, "a"), (3, "b"), (4, "b"), (5, "a"), (6,
"b")).toDF("id", "region"))
+ checkAnswer(df, Seq((1, "a"), (4, "b"), (5, "a")).toDF("id", "region"))
+ }
+ }
+
test("delta: push down input_file_name expression") {
withTable("source_table") {
withTable("target_table") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
