This is an automated email from the ASF dual-hosted git repository.
zhztheplayer 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 21cfeddd06 [VL] Defer Delta deletion vector reads to executors (#12836)
21cfeddd06 is described below
commit 21cfeddd060d7547b9dc7a79a9ab985a42fd18de
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Mon Aug 31 05:58:17 2026 +0300
[VL] Defer Delta deletion vector reads to executors (#12836)
---
.../util/delta-spark-ut/known-failures.txt | 1 -
.../DeltaDeletionVectorDeferredReadTests.scala | 258 +++++++++++++++++++++
.../delta/DeltaDeletionVectorScanInfoSuite.scala | 62 ++++-
.../delta/DeltaDeletionVectorHandoffSuite.scala | 9 +-
.../benchmark/DeltaPlanningBenchmark.scala | 30 ++-
.../delta/DeltaDeletionVectorScanInfoSuite.scala | 62 ++++-
.../delta/DeltaDeletionVectorHandoffSuite.scala | 9 +-
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 8 +
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 8 +
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 190 +++++++++++----
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 179 +++++++++++---
.../gluten/delta/TaskAccumulatorRegistry.java} | 29 +--
.../gluten/delta/DeletionVectorReadMetrics.scala | 61 +++++
.../gluten/execution/DeltaScanTransformer.scala | 39 +++-
.../gluten/substrait/rel/DeltaLocalFilesNode.java | 71 +++++-
.../execution/FileSourceScanExecTransformer.scala | 14 +-
16 files changed, 906 insertions(+), 124 deletions(-)
diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt
b/.github/workflows/util/delta-spark-ut/known-failures.txt
index ba9f191a56..88a825490f 100644
--- a/.github/workflows/util/delta-spark-ut/known-failures.txt
+++ b/.github/workflows/util/delta-spark-ut/known-failures.txt
@@ -92,7 +92,6 @@
org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#SC-86916: invalidate
org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#SC-86916:
read/write Delta paths using DataFrame should pick up Hadoop file system options
org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#all operations
should propagate Hadoop file system options
org.apache.spark.sql.delta.DeltaDataFrameHadoopOptionsSuite#operations without
Hadoop options should fail for fake:// filesystem
-org.apache.spark.sql.delta.DeltaFastDropFeatureSuite#Vacuum does not delete
deletion vector files.generateDVTombstones: false
org.apache.spark.sql.delta.DeltaGenerateSymlinkManifestSuite#incremental
manifest: failure to generate manifest throws exception
org.apache.spark.sql.delta.DeltaGenerateSymlinkManifestSuite#special partition
column values
org.apache.spark.sql.delta.DeltaHistoryManagerSuite#data skipping still works
with time travel
diff --git
a/backends-velox/src-delta/test/scala/org/apache/gluten/delta/DeltaDeletionVectorDeferredReadTests.scala
b/backends-velox/src-delta/test/scala/org/apache/gluten/delta/DeltaDeletionVectorDeferredReadTests.scala
new file mode 100644
index 0000000000..aa53e80c73
--- /dev/null
+++
b/backends-velox/src-delta/test/scala/org/apache/gluten/delta/DeltaDeletionVectorDeferredReadTests.scala
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.delta
+
+import
org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeltaFileReadOptions,
InMemoryDeletionVectorPayload}
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray,
RoaringBitmapArrayFormat}
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.test.SharedSparkSession
+
+import org.apache.hadoop.fs.Path
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream,
ObjectInputStream, ObjectOutputStream}
+import java.util.concurrent.{CountDownLatch, Executors}
+
+import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.concurrent.duration._
+
+final private[delta] case class TestDeletionVectorFile(
+ relativePath: String,
+ fileSize: Long,
+ encodedDescriptor: String,
+ storageType: String,
+ absolutePath: String,
+ offset: Long,
+ payloadSize: Long,
+ cardinality: Long)
+
+/** Shared executor-side DV payload tests for the Delta 3.3 and Delta 4.0
source profiles. */
+trait DeltaDeletionVectorDeferredReadTests {
+ self: QueryTest with SharedSparkSession =>
+
+ import testImplicits._
+
+ protected def loadDeletionVectorFile(tablePath: Path): TestDeletionVectorFile
+
+ protected def deletionVectorMetadata(encodedDescriptor: String): Map[String,
Object]
+
+ protected def encodeDeletionVectorDescriptor(descriptor:
DeletionVectorDescriptor): String
+
+ protected def partitionedFileWithMetadata(
+ tablePath: String,
+ relativeFilePath: String,
+ fileSize: Long,
+ metadata: Map[String, Object]): PartitionedFile
+
+ protected def normalizeDeletionVectorOptions(
+ partitionedFile: PartitionedFile,
+ tablePath: Path,
+ readTime: SQLMetric,
+ readBytes: SQLMetric,
+ readAttempts: SQLMetric): DeltaFileReadOptions
+
+ protected def normalizeDeletionVectorOptions(
+ partitionedFile: PartitionedFile,
+ tablePath: Path): DeltaFileReadOptions
+
+ test("eager DV payload owns its input bytes") {
+ val input = Array[Byte](1, 2, 3)
+ val payload = new InMemoryDeletionVectorPayload(input)
+
+ input(0) = 9
+
+ assert(payload.materialize().sameElements(Array[Byte](1, 2, 3)))
+ }
+
+ test("defers on-disk DV reads through serialization and coalesces concurrent
materialization") {
+ withTempDir {
+ tempDir =>
+ val tablePath = new Path(tempDir.getCanonicalPath, "table")
+ val unrelatedPath = new Path(tempDir.getCanonicalPath, "unrelated")
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+ .toDF("id", "value")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(tablePath.toString)
+
+ spark.sql(
+ s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)")
+
+ val dataFile = loadDeletionVectorFile(tablePath)
+ assert(dataFile.storageType == "u")
+ val partitionedFile = partitionedFileWithMetadata(
+ unrelatedPath.toString,
+ dataFile.relativePath,
+ dataFile.fileSize,
+ deletionVectorMetadata(dataFile.encodedDescriptor)
+ )
+
+ val readTime = SQLMetrics.createNanoTimingMetric(spark.sparkContext,
"DV read time")
+ val readBytes = SQLMetrics.createSizeMetric(spark.sparkContext, "DV
read bytes")
+ val readAttempts = SQLMetrics.createMetric(spark.sparkContext, "DV
read attempts")
+ val options = normalizeDeletionVectorOptions(
+ partitionedFile,
+ tablePath,
+ readTime,
+ readBytes,
+ readAttempts)
+ assert(!options.isDeletionVectorPayloadMaterialized)
+
+ val executorCopy = javaRoundTrip(options)
+ assert(!executorCopy.isDeletionVectorPayloadMaterialized)
+ assert(executorCopy.serializedDeletionVector.nonEmpty)
+ assert(executorCopy.isDeletionVectorPayloadMaterialized)
+ assert(!options.isDeletionVectorPayloadMaterialized)
+
+ val start = new CountDownLatch(1)
+ val pool = Executors.newFixedThreadPool(8)
+ implicit val executionContext: ExecutionContext =
+ ExecutionContext.fromExecutorService(pool)
+ val reads = (1 to 16).map {
+ _ =>
+ Future {
+ start.await()
+ options.serializedDeletionVector
+ }
+ }
+ start.countDown()
+ val payloads =
+ try {
+ Await.result(Future.sequence(reads), 30.seconds)
+ } finally {
+ pool.shutdownNow()
+ }
+
+ assert(payloads.head.nonEmpty)
+ assert(payloads.forall(_ eq payloads.head))
+ assert(options.isDeletionVectorPayloadMaterialized)
+ assert(readAttempts.value == 1L)
+ assert(readBytes.value == payloads.head.length.toLong)
+ assert(readTime.value > 0L)
+ }
+ }
+
+ test("keeps inline DV payloads eager without filesystem access") {
+ val bitmap = new RoaringBitmapArray()
+ bitmap.add(3L)
+ bitmap.add(7L)
+ val expectedPayload =
bitmap.serializeAsByteArray(RoaringBitmapArrayFormat.Portable)
+ val descriptor = DeletionVectorDescriptor.inlineInLog(expectedPayload,
cardinality = 2L)
+ val tablePath = new Path("unsupported-inline-dv-test://authority/table")
+ val partitionedFile = partitionedFileWithMetadata(
+ tablePath.toString,
+ "data.parquet",
+ fileSize = 0L,
+ metadata =
deletionVectorMetadata(encodeDeletionVectorDescriptor(descriptor)))
+
+ val readTime = SQLMetrics.createNanoTimingMetric(spark.sparkContext, "DV
read time")
+ val readBytes = SQLMetrics.createSizeMetric(spark.sparkContext, "DV read
bytes")
+ val readAttempts = SQLMetrics.createMetric(spark.sparkContext, "DV read
attempts")
+ val options = normalizeDeletionVectorOptions(
+ partitionedFile,
+ tablePath,
+ readTime,
+ readBytes,
+ readAttempts)
+
+ assert(options.isDeletionVectorPayloadMaterialized)
+ assert(options.serializedDeletionVector.sameElements(expectedPayload))
+ assert(readAttempts.value == 0L)
+ assert(readBytes.value == 0L)
+ assert(readTime.value == 0L)
+ }
+
+ test("does not cache failed deferred DV reads") {
+ withTempDir {
+ tempDir =>
+ val tablePath = new Path(tempDir.getCanonicalPath, "table")
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+ .toDF("id", "value")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(tablePath.toString)
+
+ spark.sql(
+ s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)")
+
+ val dataFile = loadDeletionVectorFile(tablePath)
+ val partitionedFile = partitionedFileWithMetadata(
+ tablePath.toString,
+ dataFile.relativePath,
+ dataFile.fileSize,
+ deletionVectorMetadata(dataFile.encodedDescriptor)
+ )
+
+ val readTime = SQLMetrics.createNanoTimingMetric(spark.sparkContext,
"DV read time")
+ val readBytes = SQLMetrics.createSizeMetric(spark.sparkContext, "DV
read bytes")
+ val readAttempts = SQLMetrics.createMetric(spark.sparkContext, "DV
read attempts")
+ val options = normalizeDeletionVectorOptions(
+ partitionedFile,
+ tablePath,
+ readTime,
+ readBytes,
+ readAttempts)
+
+ val dvPath = new Path(dataFile.absolutePath)
+ val backupPath = new Path(dvPath.toString + ".retry-test-backup")
+ val fs = dvPath.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fs.rename(dvPath, backupPath))
+ try {
+ intercept[Exception] {
+ options.serializedDeletionVector
+ }
+ assert(!options.isDeletionVectorPayloadMaterialized)
+ assert(readAttempts.value == 1L)
+ assert(readBytes.value == 0L)
+ } finally {
+ assert(fs.rename(backupPath, dvPath))
+ }
+
+ val payload = options.serializedDeletionVector
+ assert(payload.nonEmpty)
+ assert(options.isDeletionVectorPayloadMaterialized)
+ assert(readAttempts.value == 2L)
+ assert(readBytes.value == payload.length.toLong)
+ assert(readTime.value > 0L)
+ }
+ }
+
+ private def javaRoundTrip(options: DeltaFileReadOptions):
DeltaFileReadOptions = {
+ val bytes = new ByteArrayOutputStream()
+ val output = new ObjectOutputStream(bytes)
+ try {
+ output.writeObject(options)
+ } finally {
+ output.close()
+ }
+
+ val input = new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray))
+ try {
+ input.readObject().asInstanceOf[DeltaFileReadOptions]
+ } finally {
+ input.close()
+ }
+ }
+
+}
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 94c4bd2193..448bde4f84 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
@@ -17,15 +17,18 @@
package org.apache.gluten.delta
import org.apache.gluten.delta.DeltaDeletionVectorScanInfo.RowIndexFilterType
+import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
import org.apache.spark.SparkConf
import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.delta.{DeltaLog, GlutenDeltaParquetFileFormat}
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.delta.catalog.DeltaCatalog
import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.tags.ExtendedSQLTest
@@ -37,7 +40,8 @@ import org.apache.hadoop.fs.Path
class DeltaDeletionVectorScanInfoSuite
extends QueryTest
with SharedSparkSession
- with DeltaSQLTestUtils {
+ with DeltaSQLTestUtils
+ with DeltaDeletionVectorDeferredReadTests {
import testImplicits._
@@ -181,7 +185,61 @@ class DeltaDeletionVectorScanInfoSuite
}
}
- private def partitionedFileWithMetadata(
+ override protected def loadDeletionVectorFile(tablePath: Path):
TestDeletionVectorFile = {
+ val dataFile = DeltaLog
+ .forTable(spark, tablePath)
+ .update()
+ .allFiles
+ .collect()
+ .find(_.deletionVector != null)
+ .get
+ val descriptor = dataFile.deletionVector
+ TestDeletionVectorFile(
+ relativePath = dataFile.path,
+ fileSize = dataFile.size,
+ encodedDescriptor = descriptor.serializeToBase64(),
+ storageType = descriptor.storageType,
+ absolutePath = descriptor.absolutePath(tablePath).toString,
+ offset = descriptor.offset.get.toLong,
+ payloadSize = descriptor.sizeInBytes.toLong,
+ cardinality = descriptor.cardinality
+ )
+ }
+
+ override protected def deletionVectorMetadata(
+ encodedDescriptor: String): Map[String, Object] = {
+ Map(
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
encodedDescriptor,
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+ )
+ }
+
+ override protected def encodeDeletionVectorDescriptor(
+ descriptor: DeletionVectorDescriptor): String =
descriptor.serializeToBase64()
+
+ override protected def normalizeDeletionVectorOptions(
+ partitionedFile: PartitionedFile,
+ tablePath: Path,
+ readTime: SQLMetric,
+ readBytes: SQLMetric,
+ readAttempts: SQLMetric): DeltaFileReadOptions = {
+ DeltaDeletionVectorScanInfo
+ .normalize(
+ Seq(partitionedFile),
+ tablePath,
+ Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts)))
+ .get
+ ._2
+ .head
+ }
+
+ override protected def normalizeDeletionVectorOptions(
+ partitionedFile: PartitionedFile,
+ tablePath: Path): DeltaFileReadOptions = {
+ DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile),
tablePath).get._2.head
+ }
+
+ override protected def partitionedFileWithMetadata(
tablePath: String,
relativeFilePath: String,
fileSize: Long,
diff --git
a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
index f5510a9525..6b44941b45 100644
---
a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
+++
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
@@ -58,11 +58,18 @@ class DeltaDeletionVectorHandoffSuite
val df = spark.read.format("delta").load(path)
val executedPlan = df.queryExecution.executedPlan
- assert(executedPlan.collect { case _: DeltaScanTransformer => true
}.nonEmpty)
+ val nativeScans = executedPlan.collect { case scan:
DeltaScanTransformer => scan }
+ assert(nativeScans.nonEmpty)
val planText = executedPlan.toString()
assert(!planText.contains("__delta_internal_is_row_deleted"))
assert(!planText.contains("__delta_internal_row_index"))
checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
+
+ val metrics = nativeScans.head.metrics
+ assert(metrics("dvDescriptorCount").value == 1L)
+ assert(metrics("dvPayloadReadAttempts").value == 1L)
+ assert(metrics("dvPayloadReadBytes").value > 0L)
+ assert(metrics("dvPayloadReadTime").value > 0L)
}
}
}
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 68a2332f76..41db7b57ab 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`):
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.
+ * 1. '''DV descriptor handoff''' (`DeltaDeletionVectorScanInfo.normalize`):
parses descriptors
+ * and creates executor-materialized payload sources without loading
on-disk DV bytes on the
+ * driver.
* 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,
@@ -78,29 +78,41 @@ object DeltaPlanningBenchmark extends SqlBasedBenchmark {
spark.sparkContext.conf.getInt("spark.gluten.benchmark.iterations", 5)
override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
- runDvMaterializationBenchmark()
+ runDvDescriptorHandoffBenchmark()
runPostTransformRulesBenchmark()
runNonDeltaRulesOverheadBenchmark()
}
/**
- * Benchmarks DeltaDeletionVectorScanInfo.normalize() -- the critical path
that loads DVs from
- * storage on the driver. Measures how reusing the DV store across files
reduces overhead.
+ * Benchmarks DeltaDeletionVectorScanInfo.normalize() -- the planning path
that constructs
+ * executor-deferred DV descriptors. This deliberately does not access
serialized payload bytes,
+ * which would model executor work rather than driver planning.
*/
- private def runDvMaterializationBenchmark(): Unit = {
+ private def runDvDescriptorHandoffBenchmark(): Unit = {
val benchmark = new Benchmark(
- s"DV Materialization (normalize) - $numFiles files",
+ s"DV Descriptor Handoff (normalize) - $numFiles files",
numFiles.toLong,
minNumIters = benchmarkIters,
output = output)
withDeltaTableWithDVs(numFiles, rowsPerFile) {
(path, partitionedFiles) =>
+ var latestResult =
+ DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new
Path(path))
+ assert(
+ latestResult.exists(
+ _._2.forall(options =>
!options.isDeletionVectorPayloadMaterialized)))
+
benchmark.addCase(s"normalize() - $numFiles DV files", benchmarkIters)
{
- _ => DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new
Path(path))
+ _ =>
+ latestResult =
+ DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new
Path(path))
}
benchmark.run()
+ assert(
+ latestResult.exists(
+ _._2.forall(options =>
!options.isDeletionVectorPayloadMaterialized)))
}
}
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 94c4bd2193..448bde4f84 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
@@ -17,15 +17,18 @@
package org.apache.gluten.delta
import org.apache.gluten.delta.DeltaDeletionVectorScanInfo.RowIndexFilterType
+import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
import org.apache.spark.SparkConf
import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.delta.{DeltaLog, GlutenDeltaParquetFileFormat}
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.delta.catalog.DeltaCatalog
import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.tags.ExtendedSQLTest
@@ -37,7 +40,8 @@ import org.apache.hadoop.fs.Path
class DeltaDeletionVectorScanInfoSuite
extends QueryTest
with SharedSparkSession
- with DeltaSQLTestUtils {
+ with DeltaSQLTestUtils
+ with DeltaDeletionVectorDeferredReadTests {
import testImplicits._
@@ -181,7 +185,61 @@ class DeltaDeletionVectorScanInfoSuite
}
}
- private def partitionedFileWithMetadata(
+ override protected def loadDeletionVectorFile(tablePath: Path):
TestDeletionVectorFile = {
+ val dataFile = DeltaLog
+ .forTable(spark, tablePath)
+ .update()
+ .allFiles
+ .collect()
+ .find(_.deletionVector != null)
+ .get
+ val descriptor = dataFile.deletionVector
+ TestDeletionVectorFile(
+ relativePath = dataFile.path,
+ fileSize = dataFile.size,
+ encodedDescriptor = descriptor.serializeToBase64(),
+ storageType = descriptor.storageType,
+ absolutePath = descriptor.absolutePath(tablePath).toString,
+ offset = descriptor.offset.get.toLong,
+ payloadSize = descriptor.sizeInBytes.toLong,
+ cardinality = descriptor.cardinality
+ )
+ }
+
+ override protected def deletionVectorMetadata(
+ encodedDescriptor: String): Map[String, Object] = {
+ Map(
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
encodedDescriptor,
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED"
+ )
+ }
+
+ override protected def encodeDeletionVectorDescriptor(
+ descriptor: DeletionVectorDescriptor): String =
descriptor.serializeToBase64()
+
+ override protected def normalizeDeletionVectorOptions(
+ partitionedFile: PartitionedFile,
+ tablePath: Path,
+ readTime: SQLMetric,
+ readBytes: SQLMetric,
+ readAttempts: SQLMetric): DeltaFileReadOptions = {
+ DeltaDeletionVectorScanInfo
+ .normalize(
+ Seq(partitionedFile),
+ tablePath,
+ Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts)))
+ .get
+ ._2
+ .head
+ }
+
+ override protected def normalizeDeletionVectorOptions(
+ partitionedFile: PartitionedFile,
+ tablePath: Path): DeltaFileReadOptions = {
+ DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile),
tablePath).get._2.head
+ }
+
+ override protected def partitionedFileWithMetadata(
tablePath: String,
relativeFilePath: String,
fileSize: Long,
diff --git
a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
index dda547b015..159e848df0 100644
---
a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
+++
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala
@@ -88,11 +88,18 @@ class DeltaDeletionVectorHandoffSuite
val df = spark.read.format("delta").load(path)
val executedPlan = df.queryExecution.executedPlan
- assert(executedPlan.collect { case _: DeltaScanTransformer => true
}.nonEmpty)
+ val nativeScans = executedPlan.collect { case scan:
DeltaScanTransformer => scan }
+ assert(nativeScans.nonEmpty)
val planText = executedPlan.toString()
assert(!planText.contains("__delta_internal_is_row_deleted"))
assert(!planText.contains("__delta_internal_row_index"))
checkAnswer(df, Seq((1, "a"), (2, "b")).toDF())
+
+ val metrics = nativeScans.head.metrics
+ assert(metrics("dvDescriptorCount").value == 1L)
+ assert(metrics("dvPayloadReadAttempts").value == 1L)
+ assert(metrics("dvPayloadReadBytes").value > 0L)
+ assert(metrics("dvPayloadReadTime").value > 0L)
}
}
}
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 903a206663..084f3d3319 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
@@ -29,5 +29,13 @@ object DeltaDeletionVectorScanInfo {
def normalize(
partitionFiles: Seq[PartitionedFile],
tablePath: Path)
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ normalize(partitionFiles, tablePath, None)
+ }
+
+ def normalize(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics])
: 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 903a206663..084f3d3319 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
@@ -29,5 +29,13 @@ object DeltaDeletionVectorScanInfo {
def normalize(
partitionFiles: Seq[PartitionedFile],
tablePath: Path)
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ normalize(partitionFiles, tablePath, None)
+ }
+
+ def normalize(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics])
: 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 e5c4d8590b..71a2cd265a 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
@@ -18,7 +18,7 @@ package org.apache.gluten.delta
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.DeltaLocalFilesNode
-import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
+import
org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayload,
DeltaFileReadOptions, InMemoryDeletionVectorPayload}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.delta.DeltaParquetFileFormat
@@ -26,10 +26,12 @@ import
org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArrayFormat,
StoredBitmap}
import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore,
HadoopFileSystemDVStore}
import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.util.SerializableConfiguration
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._
@@ -47,7 +49,9 @@ object DeltaDeletionVectorScanInfo {
hasDeletionVector: Boolean,
rowIndexFilterType: RowIndexFilterType,
cardinality: Long,
- serializedDeletionVector: Array[Byte])
+ deletionVectorPayload: DeletionVectorPayload) {
+ def serializedDeletionVector: Array[Byte] =
deletionVectorPayload.materialize()
+ }
final case class PartitionFileScanInfo(
normalizedOtherMetadataColumns: Map[String, Object],
@@ -63,22 +67,33 @@ 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.
*
- * `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.
+ * `tablePath` is the authoritative Delta table root supplied by
`TahoeFileIndex.path`. On-disk DV
+ * descriptors retain a shared serializable Hadoop configuration but do not
open their sidecar
+ * until executor-side split serialization. Inline DVs remain eager because
their bytes are
+ * already present in Delta metadata.
*/
def normalize(
partitionFiles: Seq[PartitionedFile],
tablePath: Path)
: Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ normalize(partitionFiles, tablePath, None)
+ }
+
+ def normalize(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
if (partitionFiles.isEmpty) {
return None
}
val spark = activeSparkSession
- // Create a single Hadoop Configuration for the entire partition.
val hadoopConf = spark.sessionState.newHadoopConf()
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
- val scanInfos = partitionFiles.map(file => extract(file, hadoopConf,
tablePath))
+ val scanInfos = partitionFiles.map {
+ file => extract(file, hadoopConf, serializableHadoopConf, tablePath,
readMetrics)
+ }
if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) {
Some(
(
@@ -95,16 +110,24 @@ object DeltaDeletionVectorScanInfo {
file: PartitionedFile,
tablePath: Path): PartitionFileScanInfo = {
val hadoopConf = spark.sessionState.newHadoopConf()
- extract(file, hadoopConf, tablePath)
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
+ extract(file, hadoopConf, serializableHadoopConf, tablePath, None)
}
private def extract(
file: PartitionedFile,
hadoopConf: Configuration,
- tablePath: Path): PartitionFileScanInfo = {
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo =
{
val metadata = otherMetadataColumns(file)
val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded,
RowIndexFilterTypeKey)
- val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath)
+ val dvInfo = extractDeletionVectorInfo(
+ metadata,
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ readMetrics)
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
@@ -113,7 +136,7 @@ object DeltaDeletionVectorScanInfo {
toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType),
dvInfo.hasDeletionVector,
dvInfo.cardinality,
- dvInfo.serializedDeletionVector)
+ dvInfo.deletionVectorPayload)
}
private def toSubstraitRowIndexFilterType(
@@ -137,21 +160,32 @@ object DeltaDeletionVectorScanInfo {
private def extractDeletionVectorInfo(
metadata: Map[String, Object],
hadoopConf: Configuration,
- tablePath: Path): DeletionVectorInfo = {
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorInfo = {
val descriptorValue = metadata.get(RowIndexFilterIdEncoded)
val filterTypeValue = metadata.get(RowIndexFilterTypeKey)
(descriptorValue, filterTypeValue) match {
case (None, None) =>
- DeletionVectorInfo(false, KEEP_ALL, 0L, Array.emptyByteArray)
+ DeletionVectorInfo(
+ false,
+ KEEP_ALL,
+ 0L,
+ new InMemoryDeletionVectorPayload(Array.emptyByteArray))
case (Some(encodedDescriptor), Some(filterType)) =>
val descriptor = parseDescriptor(encodedDescriptor.toString)
- val serializedPayload = serializePayload(hadoopConf, tablePath,
descriptor)
+ val payload = deletionVectorPayload(
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ descriptor,
+ readMetrics)
DeletionVectorInfo(
true,
parseRowIndexFilterType(filterType.toString),
descriptor.cardinality,
- serializedPayload)
+ payload)
case _ =>
throw new IllegalStateException(
s"Both $RowIndexFilterIdEncoded and $RowIndexFilterTypeKey must
either be present or absent")
@@ -187,17 +221,43 @@ object DeltaDeletionVectorScanInfo {
}
}
+ /** Selects a deferred source for on-disk DVs and eager bytes for inline
DVs. */
+ private def deletionVectorPayload(
+ hadoopConf: Configuration,
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ descriptor: DeletionVectorDescriptor,
+ readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorPayload =
{
+ if (tablePath == null) {
+ throw new IllegalStateException(
+ "Unable to resolve Delta table path while preparing deletion vector
payload")
+ }
+ if (descriptor.storageType != "i") {
+ val dvPath = descriptor.absolutePath(tablePath)
+ new OnDiskDeletionVectorPayload(
+ serializableHadoopConf,
+ dvPath.toString,
+ requiredOffset(descriptor),
+ descriptor.sizeInBytes,
+ readMetrics)
+ } else {
+ new InMemoryDeletionVectorPayload(serializeInlinePayload(hadoopConf,
tablePath, descriptor))
+ }
+ }
+
+ private def requiredOffset(descriptor: DeletionVectorDescriptor): Long = {
+ descriptor.offset
+ .map(_.toLong)
+ .getOrElse {
+ throw new IllegalStateException(
+ s"On-disk Delta deletion vector '${descriptor.storageType}' is
missing its offset")
+ }
+ }
+
/**
- * Reads the DV payload bytes for the native engine. For on-disk DVs, reads
the raw bytes directly
- * from the DV file using Delta's `DeletionVectorStore.readRangeFromStream`,
which includes
- * checksum verification. The on-disk format is already Portable Roaring
Bitmap Array (the format
- * the native Velox side expects), so this skips the expensive
- * deserialize-into-Java-Roaring-objects + re-serialize round-trip.
- *
- * Falls back to the standard load+serialize path for inline DVs (small
payloads embedded in Delta
- * metadata) which don't have a file to read from.
+ * Decodes an inline DV already embedded in Delta metadata into Velox's
portable bitmap format.
*/
- private def serializePayload(
+ private def serializeInlinePayload(
hadoopConf: Configuration,
tablePath: Path,
descriptor: DeletionVectorDescriptor): Array[Byte] = {
@@ -205,17 +265,11 @@ object DeltaDeletionVectorScanInfo {
throw new IllegalStateException(
"Unable to resolve Delta table path while materializing deletion
vector payload")
}
- if (descriptor.storageType != "i") {
- // On-disk DV (storageType "u" for UUID or "p" for path): read raw bytes
directly.
- readRawDvBytes(hadoopConf, tablePath, descriptor)
- } else {
- // Inline DV (storageType "i"): bytes are in the descriptor metadata.
- val dvStore = new HadoopFileSystemDVStore(hadoopConf)
- StoredBitmap
- .create(descriptor, tablePath)
- .load(dvStore)
- .serializeAsByteArray(RoaringBitmapArrayFormat.Portable)
- }
+ val dvStore = new HadoopFileSystemDVStore(hadoopConf)
+ StoredBitmap
+ .create(descriptor, tablePath)
+ .load(dvStore)
+ .serializeAsByteArray(RoaringBitmapArrayFormat.Portable)
}
/**
@@ -226,25 +280,67 @@ object DeltaDeletionVectorScanInfo {
*/
private def readRawDvBytes(
hadoopConf: Configuration,
- tablePath: Path,
- descriptor: DeletionVectorDescriptor): Array[Byte] = {
- val dvPath = descriptor.absolutePath(tablePath)
+ dvPath: Path,
+ offset: Long,
+ sizeInBytes: Int): Array[Byte] = {
val fs = dvPath.getFileSystem(hadoopConf)
// 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)
+ // then fail the CRC check in `readRangeFromStream`.
+ val fileStream = fs.open(dvPath)
try {
- val offset = descriptor.offset.getOrElse(0)
- if (offset > 0) {
- stream.seek(offset.toLong)
- }
- DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes)
+ fileStream.seek(offset)
+ DeletionVectorStore.readRangeFromStream(new DataInputStream(fileStream),
sizeInBytes)
} finally {
- stream.close()
+ fileStream.close()
+ }
+ }
+
+ /**
+ * Executor-side on-disk payload source. Successful materialization is
memoized for repeated split
+ * serialization; failed reads remain retryable.
+ */
+ @SerialVersionUID(1L)
+ final private class OnDiskDeletionVectorPayload(
+ serializableHadoopConf: SerializableConfiguration,
+ absolutePath: String,
+ offset: Long,
+ sizeInBytes: Int,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ extends DeletionVectorPayload {
+ require(offset >= 0, s"Deletion vector offset must be non-negative:
$offset")
+ require(sizeInBytes >= 0, s"Deletion vector size must be non-negative:
$sizeInBytes")
+
+ @transient @volatile private var cachedPayload: Array[Byte] = _
+
+ override def materialize(): Array[Byte] = {
+ var payload = cachedPayload
+ if (payload == null) {
+ this.synchronized {
+ payload = cachedPayload
+ if (payload == null) {
+ val startedAt = System.nanoTime()
+ readMetrics.foreach(_.registerForCurrentTask())
+ readMetrics.foreach(_.readAttempts.add(1L))
+ try {
+ payload = readRawDvBytes(
+ serializableHadoopConf.value,
+ new Path(absolutePath),
+ offset,
+ sizeInBytes)
+ readMetrics.foreach(_.readBytes.add(payload.length.toLong))
+ cachedPayload = payload
+ } finally {
+ readMetrics.foreach(_.readTimeNanos.add(System.nanoTime() -
startedAt))
+ }
+ }
+ }
+ }
+ payload
}
+
+ override def isMaterialized(): Boolean = cachedPayload != null
}
}
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 11530d665a..e27d10774f 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
@@ -18,7 +18,7 @@ package org.apache.gluten.delta
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.DeltaLocalFilesNode
-import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
+import
org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayload,
DeltaFileReadOptions, InMemoryDeletionVectorPayload}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.delta.DeltaParquetFileFormat
@@ -26,10 +26,12 @@ import
org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArrayFormat,
StoredBitmap}
import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore,
HadoopFileSystemDVStore}
import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.util.SerializableConfiguration
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._
@@ -48,7 +50,9 @@ object DeltaDeletionVectorScanInfo {
hasDeletionVector: Boolean,
rowIndexFilterType: RowIndexFilterType,
cardinality: Long,
- serializedDeletionVector: Array[Byte])
+ deletionVectorPayload: DeletionVectorPayload) {
+ def serializedDeletionVector: Array[Byte] =
deletionVectorPayload.materialize()
+ }
final case class PartitionFileScanInfo(
normalizedOtherMetadataColumns: Map[String, Object],
@@ -64,21 +68,33 @@ 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.
*
- * `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.
+ * `tablePath` is the authoritative Delta table root supplied by
`TahoeFileIndex.path`. On-disk DV
+ * descriptors retain a shared serializable Hadoop configuration but do not
open their sidecar
+ * until executor-side split serialization. Inline DVs remain eager because
their bytes are
+ * already present in Delta metadata.
*/
def normalize(
partitionFiles: Seq[PartitionedFile],
tablePath: Path)
: Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ normalize(partitionFiles, tablePath, None)
+ }
+
+ def normalize(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
if (partitionFiles.isEmpty) {
return None
}
val spark = activeSparkSession
val hadoopConf = spark.sessionState.newHadoopConf()
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
- val scanInfos = partitionFiles.map(file => extract(file, hadoopConf,
tablePath))
+ val scanInfos = partitionFiles.map {
+ file => extract(file, hadoopConf, serializableHadoopConf, tablePath,
readMetrics)
+ }
if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) {
Some(
(
@@ -95,16 +111,24 @@ object DeltaDeletionVectorScanInfo {
file: PartitionedFile,
tablePath: Path): PartitionFileScanInfo = {
val hadoopConf = spark.sessionState.newHadoopConf()
- extract(file, hadoopConf, tablePath)
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
+ extract(file, hadoopConf, serializableHadoopConf, tablePath, None)
}
private def extract(
file: PartitionedFile,
hadoopConf: Configuration,
- tablePath: Path): PartitionFileScanInfo = {
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo =
{
val metadata = otherMetadataColumns(file)
val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded,
RowIndexFilterTypeKey)
- val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath)
+ val dvInfo = extractDeletionVectorInfo(
+ metadata,
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ readMetrics)
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
@@ -113,7 +137,7 @@ object DeltaDeletionVectorScanInfo {
toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType),
dvInfo.hasDeletionVector,
dvInfo.cardinality,
- dvInfo.serializedDeletionVector)
+ dvInfo.deletionVectorPayload)
}
private def toSubstraitRowIndexFilterType(
@@ -137,21 +161,32 @@ object DeltaDeletionVectorScanInfo {
private def extractDeletionVectorInfo(
metadata: Map[String, Object],
hadoopConf: Configuration,
- tablePath: Path): DeletionVectorInfo = {
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorInfo = {
val descriptorValue = metadata.get(RowIndexFilterIdEncoded)
val filterTypeValue = metadata.get(RowIndexFilterTypeKey)
(descriptorValue, filterTypeValue) match {
case (None, None) =>
- DeletionVectorInfo(false, KEEP_ALL, 0L, Array.emptyByteArray)
+ DeletionVectorInfo(
+ false,
+ KEEP_ALL,
+ 0L,
+ new InMemoryDeletionVectorPayload(Array.emptyByteArray))
case (Some(encodedDescriptor), Some(filterType)) =>
val descriptor = parseDescriptor(encodedDescriptor.toString)
- val serializedPayload = serializePayload(hadoopConf, tablePath,
descriptor)
+ val payload = deletionVectorPayload(
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ descriptor,
+ readMetrics)
DeletionVectorInfo(
true,
parseRowIndexFilterType(filterType.toString),
descriptor.cardinality,
- serializedPayload)
+ payload)
case _ =>
throw new IllegalStateException(
s"Both $RowIndexFilterIdEncoded and $RowIndexFilterTypeKey must
either be present or absent")
@@ -209,48 +244,120 @@ object DeltaDeletionVectorScanInfo {
}
}
- private def serializePayload(
+ /** Selects a deferred source for on-disk DVs and eager bytes for inline
DVs. */
+ private def deletionVectorPayload(
hadoopConf: Configuration,
+ serializableHadoopConf: SerializableConfiguration,
tablePath: Path,
- descriptor: DeletionVectorDescriptor): Array[Byte] = {
+ descriptor: DeletionVectorDescriptor,
+ readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorPayload =
{
if (tablePath == null) {
throw new IllegalStateException(
- "Unable to resolve Delta table path while materializing deletion
vector payload")
+ "Unable to resolve Delta table path while preparing deletion vector
payload")
}
if (descriptor.storageType != "i") {
- // On-disk DV: read raw bytes directly (already in Portable Roaring
format).
- readRawDvBytes(hadoopConf, tablePath, descriptor)
+ val dvPath = descriptor.absolutePath(tablePath)
+ new OnDiskDeletionVectorPayload(
+ serializableHadoopConf,
+ dvPath.toString,
+ requiredOffset(descriptor),
+ descriptor.sizeInBytes,
+ readMetrics)
} else {
- // Inline DV: bytes are in the descriptor metadata.
- val dvStore = new HadoopFileSystemDVStore(hadoopConf)
- StoredBitmap
- .create(descriptor, tablePath)
- .load(dvStore)
- .serializeAsByteArray(RoaringBitmapArrayFormat.Portable)
+ new InMemoryDeletionVectorPayload(serializeInlinePayload(hadoopConf,
tablePath, descriptor))
}
}
- private def readRawDvBytes(
+ private def requiredOffset(descriptor: DeletionVectorDescriptor): Long = {
+ descriptor.offset
+ .map(_.toLong)
+ .getOrElse {
+ throw new IllegalStateException(
+ s"On-disk Delta deletion vector '${descriptor.storageType}' is
missing its offset")
+ }
+ }
+
+ /**
+ * Decodes an inline DV already embedded in Delta metadata into Velox's
portable bitmap format.
+ */
+ private def serializeInlinePayload(
hadoopConf: Configuration,
tablePath: Path,
descriptor: DeletionVectorDescriptor): Array[Byte] = {
- val dvPath = descriptor.absolutePath(tablePath)
+ if (tablePath == null) {
+ throw new IllegalStateException(
+ "Unable to resolve Delta table path while materializing deletion
vector payload")
+ }
+ val dvStore = new HadoopFileSystemDVStore(hadoopConf)
+ StoredBitmap
+ .create(descriptor, tablePath)
+ .load(dvStore)
+ .serializeAsByteArray(RoaringBitmapArrayFormat.Portable)
+ }
+
+ private def readRawDvBytes(
+ hadoopConf: Configuration,
+ dvPath: Path,
+ offset: Long,
+ sizeInBytes: Int): Array[Byte] = {
val fs = dvPath.getFileSystem(hadoopConf)
// 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)
+ // then fail the CRC check in `readRangeFromStream`.
+ val fileStream = fs.open(dvPath)
try {
- val offset = descriptor.offset.getOrElse(0)
- if (offset > 0) {
- stream.seek(offset.toLong)
- }
- DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes)
+ fileStream.seek(offset)
+ DeletionVectorStore.readRangeFromStream(new DataInputStream(fileStream),
sizeInBytes)
} finally {
- stream.close()
+ fileStream.close()
+ }
+ }
+
+ /**
+ * Executor-side on-disk payload source. Successful materialization is
memoized for repeated split
+ * serialization; failed reads remain retryable.
+ */
+ @SerialVersionUID(1L)
+ final private class OnDiskDeletionVectorPayload(
+ serializableHadoopConf: SerializableConfiguration,
+ absolutePath: String,
+ offset: Long,
+ sizeInBytes: Int,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ extends DeletionVectorPayload {
+ require(offset >= 0, s"Deletion vector offset must be non-negative:
$offset")
+ require(sizeInBytes >= 0, s"Deletion vector size must be non-negative:
$sizeInBytes")
+
+ @transient @volatile private var cachedPayload: Array[Byte] = _
+
+ override def materialize(): Array[Byte] = {
+ var payload = cachedPayload
+ if (payload == null) {
+ this.synchronized {
+ payload = cachedPayload
+ if (payload == null) {
+ val startedAt = System.nanoTime()
+ readMetrics.foreach(_.registerForCurrentTask())
+ readMetrics.foreach(_.readAttempts.add(1L))
+ try {
+ payload = readRawDvBytes(
+ serializableHadoopConf.value,
+ new Path(absolutePath),
+ offset,
+ sizeInBytes)
+ readMetrics.foreach(_.readBytes.add(payload.length.toLong))
+ cachedPayload = payload
+ } finally {
+ readMetrics.foreach(_.readTimeNanos.add(System.nanoTime() -
startedAt))
+ }
+ }
+ }
+ }
+ payload
}
+
+ override def isMaterialized(): Boolean = cachedPayload != null
}
}
diff --git
a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
b/gluten-delta/src/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java
similarity index 57%
copy from
gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
copy to
gluten-delta/src/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java
index 903a206663..e6e31a84a8 100644
---
a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
+++
b/gluten-delta/src/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java
@@ -14,20 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.gluten.delta
+package org.apache.gluten.delta;
-import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
+import org.apache.spark.TaskContext;
+import org.apache.spark.util.AccumulatorV2;
-import org.apache.spark.sql.execution.datasources.PartitionedFile
+/** Registers accumulators that Spark deserialized before installing the task
context. */
+final class TaskAccumulatorRegistry {
+ private TaskAccumulatorRegistry() {}
-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(
- partitionFiles: Seq[PartitionedFile],
- tablePath: Path)
- : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
+ static boolean registerForCurrentTask(AccumulatorV2<?, ?>... accumulators) {
+ TaskContext taskContext = TaskContext.get();
+ if (taskContext == null) {
+ return false;
+ }
+ for (AccumulatorV2<?, ?> accumulator : accumulators) {
+ taskContext.registerAccumulator(accumulator);
+ }
+ return true;
+ }
}
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/delta/DeletionVectorReadMetrics.scala
b/gluten-delta/src/main/scala/org/apache/gluten/delta/DeletionVectorReadMetrics.scala
new file mode 100644
index 0000000000..91c24de508
--- /dev/null
+++
b/gluten-delta/src/main/scala/org/apache/gluten/delta/DeletionVectorReadMetrics.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.delta
+
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.execution.metric.SQLMetric
+
+import java.io.ObjectInputStream
+
+/** Metrics updated while an executor materializes an on-disk deletion-vector
payload. */
+final case class DeletionVectorReadMetrics(
+ readTimeNanos: SQLMetric,
+ readBytes: SQLMetric,
+ readAttempts: SQLMetric) {
+
+ @transient @volatile private var registeredInTask = false
+
+ /**
+ * Spark can deserialize an input partition before installing `TaskContext`,
so accumulators
+ * nested in that partition cannot register from `AccumulatorV2.readObject`.
Register them when
+ * deferred I/O first runs inside the task instead. The shared metrics
object makes this
+ * once-per-task even when a partition contains multiple deletion vectors.
+ */
+ def registerForCurrentTask(): Unit = {
+ if (!registeredInTask && TaskContext.get() != null) {
+ this.synchronized {
+ if (!registeredInTask) {
+ registeredInTask = TaskAccumulatorRegistry.registerForCurrentTask(
+ readTimeNanos,
+ readBytes,
+ readAttempts)
+ }
+ }
+ }
+ }
+
+ /**
+ * `defaultReadObject` deserializes the nested SQL metrics first. Spark's
+ * `AccumulatorV2.readObject` registers each one when a task context exists,
so mirror that state
+ * here to avoid registering them a second time. Without a task context the
metrics remain
+ * unregistered and `registerForCurrentTask` handles them when
materialization begins.
+ */
+ private def readObject(input: ObjectInputStream): Unit = {
+ input.defaultReadObject()
+ registeredInTask = TaskContext.get() != null
+ }
+}
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 50073cd197..3d80c582d7 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
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.execution
-import org.apache.gluten.delta.DeltaDeletionVectorScanInfo
+import org.apache.gluten.delta.{DeletionVectorReadMetrics,
DeltaDeletionVectorScanInfo}
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.{DeltaLocalFilesBuilder,
LocalFilesNode, SplitInfo}
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
@@ -30,6 +30,7 @@ import org.apache.spark.sql.delta.{DeltaParquetFileFormat,
NoMapping}
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.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.collection.BitSet
@@ -62,6 +63,27 @@ case class DeltaScanTransformer(
override lazy val fileFormat: ReadFileFormat =
ReadFileFormat.ParquetReadFormat
+ override protected def additionalScanMetrics: Map[String, SQLMetric] = Map(
+ "dvDescriptorPreparationTime" ->
+ SQLMetrics.createNanoTimingMetric(
+ sparkContext,
+ "Delta deletion vector descriptor preparation time"),
+ "dvDescriptorCount" ->
+ SQLMetrics.createMetric(sparkContext, "Delta deletion vector descriptor
count"),
+ "dvPayloadReadTime" ->
+ SQLMetrics.createNanoTimingMetric(sparkContext, "Delta deletion vector
payload read time"),
+ "dvPayloadReadBytes" ->
+ SQLMetrics.createSizeMetric(sparkContext, "Delta deletion vector payload
bytes read"),
+ "dvPayloadReadAttempts" ->
+ SQLMetrics.createMetric(sparkContext, "Delta deletion vector payload
read attempts")
+ )
+
+ @transient private lazy val deletionVectorReadMetrics =
+ DeletionVectorReadMetrics(
+ metrics("dvPayloadReadTime"),
+ metrics("dvPayloadReadBytes"),
+ metrics("dvPayloadReadAttempts"))
+
// Delta CDF over a deletion-vector-enabled table needs DV-aware, row-level
reconciliation that
// the native scan path does not do yet: it would surface rows that are
still live (not covered
// by the DV) as CDF `delete` change rows. Fall back to Spark for both CDF
scan sides -- the add
@@ -129,10 +151,21 @@ case class DeltaScanTransformer(
val tableRootPath = tahoe.path
splitInfos.zip(partitions).map {
case (localFiles: LocalFilesNode, (filePartition: FilePartition, _))
=>
- DeltaDeletionVectorScanInfo
- .normalize(filePartition.files.toSeq, tableRootPath)
+ val startedAt = System.nanoTime()
+ val normalized =
+ try {
+ DeltaDeletionVectorScanInfo.normalize(
+ filePartition.files.toSeq,
+ tableRootPath,
+ Some(deletionVectorReadMetrics))
+ } finally {
+ metrics("dvDescriptorPreparationTime").add(System.nanoTime() -
startedAt)
+ }
+ normalized
.map {
case (otherMetadataColumns, deltaReadOptions) =>
+ metrics("dvDescriptorCount")
+ .add(deltaReadOptions.count(_.hasDeletionVector()).toLong)
DeltaLocalFilesBuilder.makeDeltaLocalFiles(
localFiles,
otherMetadataColumns.asJava,
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java
index a95f676951..ae69b8723f 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java
@@ -79,24 +79,76 @@ public class DeltaLocalFilesNode extends LocalFilesNode {
IF_NOT_CONTAINED
}
+ /**
+ * Serializable source for a deletion-vector payload.
+ *
+ * <p>The source travels inside a Spark input partition. Implementations may
therefore defer
+ * remote I/O until {@link #materialize()} is called while the split is
converted to protobuf on
+ * an executor. The returned byte array must not be modified: protobuf wraps
it without copying.
+ */
+ public interface DeletionVectorPayload extends Serializable {
+ byte[] materialize();
+
+ /** Returns whether the payload bytes are already resident in this object.
*/
+ boolean isMaterialized();
+ }
+
+ /** A payload source for inline DVs whose bytes are already present in Delta
metadata. */
+ public static final class InMemoryDeletionVectorPayload implements
DeletionVectorPayload {
+ private static final long serialVersionUID = 1L;
+
+ private final byte[] payload;
+
+ public InMemoryDeletionVectorPayload(byte[] payload) {
+ this.payload = payload == null ? new byte[0] : payload.clone();
+ }
+
+ @Override
+ public byte[] materialize() {
+ return payload;
+ }
+
+ @Override
+ public boolean isMaterialized() {
+ return true;
+ }
+ }
+
public static class DeltaFileReadOptions implements Serializable {
private static final long serialVersionUID = 1L;
private final RowIndexFilterType rowIndexFilterType;
private final boolean hasDeletionVector;
private final long deletionVectorCardinality;
- private final byte[] serializedDeletionVector;
+ private final DeletionVectorPayload deletionVectorPayload;
public DeltaFileReadOptions(
RowIndexFilterType rowIndexFilterType,
boolean hasDeletionVector,
long deletionVectorCardinality,
byte[] serializedDeletionVector) {
+ this(
+ rowIndexFilterType,
+ hasDeletionVector,
+ deletionVectorCardinality,
+ new InMemoryDeletionVectorPayload(serializedDeletionVector));
+ }
+
+ public DeltaFileReadOptions(
+ RowIndexFilterType rowIndexFilterType,
+ boolean hasDeletionVector,
+ long deletionVectorCardinality,
+ DeletionVectorPayload deletionVectorPayload) {
+ if (rowIndexFilterType == null) {
+ throw new IllegalArgumentException("rowIndexFilterType must not be
null");
+ }
+ if (deletionVectorPayload == null) {
+ throw new IllegalArgumentException("deletionVectorPayload must not be
null");
+ }
this.rowIndexFilterType = rowIndexFilterType;
this.hasDeletionVector = hasDeletionVector;
this.deletionVectorCardinality = deletionVectorCardinality;
- this.serializedDeletionVector =
- serializedDeletionVector == null ? new byte[0] :
serializedDeletionVector;
+ this.deletionVectorPayload = deletionVectorPayload;
}
public RowIndexFilterType rowIndexFilterType() {
@@ -111,8 +163,19 @@ public class DeltaLocalFilesNode extends LocalFilesNode {
return deletionVectorCardinality;
}
+ /**
+ * Materializes and returns the serialized deletion-vector bytes.
+ *
+ * <p>For an on-disk deletion vector this may perform blocking filesystem
I/O and is intended to
+ * run during executor-side split-to-protobuf conversion. The returned
array must not be
+ * modified because protobuf wraps it without copying.
+ */
public byte[] serializedDeletionVector() {
- return serializedDeletionVector;
+ return deletionVectorPayload.materialize();
+ }
+
+ public boolean isDeletionVectorPayloadMaterialized() {
+ return deletionVectorPayload.isMaterialized();
}
}
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
index 83023f027a..088c5f2d6d 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
@@ -116,15 +116,19 @@ abstract class FileSourceScanExecTransformerBase(
disableBucketedScan)
with DatasourceScanTransformer {
- // Executor-side metrics only (excludes driverMetricsAlias).
- @transient private lazy val executorSideScanMetrics: Map[String, SQLMetric] =
+ /** Format-specific metrics that should be displayed with the native file
scan. */
+ protected def additionalScanMetrics: Map[String, SQLMetric] = Map.empty
+
+ // Metrics attached to the native file scan. Format-specific additions may
be updated on the
+ // driver or executors; driver-only aliases are excluded.
+ @transient private lazy val nativeScanMetrics: Map[String, SQLMetric] =
BackendsApiManager.getMetricsApiInstance
.genFileSourceScanTransformerMetrics(sparkContext)
- .filter(m => !driverMetricsAlias.contains(m._1))
+ .filter(m => !driverMetricsAlias.contains(m._1)) ++ additionalScanMetrics
// Note: "metrics" is made transient to avoid sending driver-side metrics to
tasks.
@transient override lazy val metrics: Map[String, SQLMetric] =
- executorSideScanMetrics ++ driverMetricsAlias
+ nativeScanMetrics ++ driverMetricsAlias
override def scanFilters: Seq[Expression] = dataFilters
@@ -189,7 +193,7 @@ abstract class FileSourceScanExecTransformerBase(
override def metricsUpdater(): MetricsUpdater =
BackendsApiManager.getMetricsApiInstance
- .genFileSourceScanTransformerMetricsUpdater(executorSideScanMetrics)
+ .genFileSourceScanTransformerMetricsUpdater(nativeScanMetrics)
override val nodeName: String = {
s"${getClass.getSimpleName} $relation
${tableIdentifier.map(_.unquotedString).getOrElse("")}"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
