diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt b/.github/workflows/util/delta-spark-ut/known-failures.txt index 14dfbfc22d3..bf387b30b57 100644 --- a/.github/workflows/util/delta-spark-ut/known-failures.txt +++ b/.github/workflows/util/delta-spark-ut/known-failures.txt @@ -100,7 +100,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 00000000000..69c356fcef0 --- /dev/null +++ b/backends-velox/src-delta/test/scala/org/apache/gluten/delta/DeltaDeletionVectorDeferredReadTests.scala @@ -0,0 +1,296 @@ +/* + * 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.config.GlutenConfig +import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeltaFileReadOptions, SerializedDeletionVectorPayload} + +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 SerializedDeletionVectorPayload(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) + } + } + + test("passes authoritative on-disk DV descriptors to native without JVM 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) + ) + + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + val options = normalizeDeletionVectorOptions(partitionedFile, tablePath) + assert(options.hasNativeDeletionVectorDescriptor) + assert(!options.isDeletionVectorPayloadMaterialized) + val descriptor = options.nativeDeletionVectorDescriptor + assert(descriptor.absolutePath == dataFile.absolutePath) + assert(descriptor.offset == dataFile.offset) + assert(descriptor.payloadSize == dataFile.payloadSize) + val error = intercept[IllegalStateException](options.serializedDeletionVector) + assert(error.getMessage.contains("do not contain JVM payload bytes")) + } + } + } + + 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 94c4bd2193c..448bde4f84d 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 f5510a95255..c32952efde9 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 @@ -16,6 +16,7 @@ */ package org.apache.spark.sql.delta +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.execution.DeltaScanTransformer import org.apache.spark.sql.QueryTest @@ -58,11 +59,49 @@ 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) + } + } + + test("Spark 3.5 native Delta DV descriptor filters rows without JVM payload reads") { + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + withTempDir { + tempDir => + val path = tempDir.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + + val df = spark.read.format("delta").load(path) + val nativeScans = df.queryExecution.executedPlan.collect { + case scan: DeltaScanTransformer => scan + } + assert(nativeScans.nonEmpty) + checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 0L) + assert(metrics("dvPayloadReadBytes").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 68a2332f76b..41db7b57ab2 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 94c4bd2193c..448bde4f84d 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 dda547b015f..924a60094fe 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 @@ -16,6 +16,7 @@ */ package org.apache.spark.sql.delta +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.execution.DeltaScanTransformer import org.apache.spark.sql.QueryTest @@ -88,11 +89,49 @@ 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) + } + } + + test("Spark 4 native Delta DV descriptor filters rows without JVM payload reads") { + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + withTempDir { + tempDir => + val path = tempDir.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + + val df = spark.read.format("delta").load(path) + val nativeScans = df.queryExecution.executedPlan.collect { + case scan: DeltaScanTransformer => scan + } + assert(nativeScans.nonEmpty) + checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 0L) + assert(metrics("dvPayloadReadBytes").value == 0L) + } } } } diff --git a/cpp/velox/compute/VeloxPlanConverter.cc b/cpp/velox/compute/VeloxPlanConverter.cc index 82c1f290368..703486684e6 100644 --- a/cpp/velox/compute/VeloxPlanConverter.cc +++ b/cpp/velox/compute/VeloxPlanConverter.cc @@ -119,13 +119,29 @@ std::shared_ptr parseDeltaSplitInfo( return deltaSplitInfo; } + const auto cardinality = static_cast(deltaReadOptions.deletion_vector_cardinality()); + if (deltaReadOptions.has_deletion_vector_descriptor()) { + const auto& descriptor = deltaReadOptions.deletion_vector_descriptor(); + VELOX_USER_CHECK( + deltaReadOptions.serialized_deletion_vector().empty(), + "Delta split has both a serialized deletion vector and an on-disk descriptor"); + VELOX_USER_CHECK(!descriptor.absolute_path().empty(), "Delta deletion vector path is empty"); + VELOX_USER_CHECK_GT(descriptor.payload_size(), 0, "Delta deletion vector payload size must be positive"); + VELOX_USER_CHECK_LE( + descriptor.payload_size(), + std::numeric_limits::max(), + "Delta deletion vector payload is too large for the stored format"); + deltaSplitInfo->deletionVectors.emplace_back(delta::DeltaDeletionVectorDescriptor::onDisk( + cardinality, descriptor.absolute_path(), descriptor.offset(), descriptor.payload_size())); + return deltaSplitInfo; + } + const auto& serializedPayload = deltaReadOptions.serialized_deletion_vector(); VELOX_USER_CHECK(!serializedPayload.empty(), "Delta split has a deletion vector without a serialized payload"); VELOX_USER_CHECK_LE( serializedPayload.size(), static_cast(std::numeric_limits::max()), "Delta deletion vector serialized payload is too large"); - const auto cardinality = static_cast(deltaReadOptions.deletion_vector_cardinality()); auto payload = std::make_shared(serializedPayload); const SplitPayloadBufferView payloadView{ reinterpret_cast(payload->data()), static_cast(payload->size())}; diff --git a/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp b/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp index baf6c06617c..85135c78cc8 100644 --- a/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp +++ b/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp @@ -18,7 +18,9 @@ #include "compute/delta/DeltaDeletionVectorReader.h" #include +#include #include "velox/common/base/BitUtil.h" +#include "velox/common/base/Crc.h" #include "velox/common/base/Exceptions.h" namespace gluten::delta { @@ -28,6 +30,7 @@ namespace { constexpr uint64_t kDeltaBitmapArrayMagicBytes = 4; constexpr uint64_t kDeltaNativeBitmapArrayLengthBytes = 4; constexpr uint64_t kDeltaStoredPayloadLengthBytes = 4; +constexpr uint64_t kDeltaStoredChecksumBytes = 4; constexpr uint32_t kDeltaPortableBitmapArrayMagicNumber = 1681511377; constexpr uint32_t kDeltaNativeBitmapArrayMagicNumber = 1681511376; @@ -37,6 +40,51 @@ uint32_t readUint32LittleEndian(const char* data) { (static_cast(bytes[2]) << 16) | (static_cast(bytes[3]) << 24); } +uint32_t readUint32BigEndian(const char* data) { + const auto* bytes = reinterpret_cast(data); + return (static_cast(bytes[0]) << 24) | (static_cast(bytes[1]) << 16) | + (static_cast(bytes[2]) << 8) | static_cast(bytes[3]); +} + +std::string_view +extractStoredPayload(std::string_view storedRange, uint64_t expectedPayloadSize, const std::string& debugName) { + VELOX_USER_CHECK_LE( + expectedPayloadSize, + std::numeric_limits::max(), + "Deletion vector payload is too large for Delta stored format: {}", + debugName); + const auto expectedRangeSize = kDeltaStoredPayloadLengthBytes + expectedPayloadSize + kDeltaStoredChecksumBytes; + VELOX_USER_CHECK_EQ( + storedRange.size(), + expectedRangeSize, + "Deletion vector range size mismatch for {}: expected {}, got {}", + debugName, + expectedRangeSize, + storedRange.size()); + + const auto storedPayloadSize = readUint32BigEndian(storedRange.data()); + VELOX_USER_CHECK_EQ( + storedPayloadSize, + expectedPayloadSize, + "Deletion vector payload size mismatch for {}: expected {}, got {}", + debugName, + expectedPayloadSize, + storedPayloadSize); + + const auto payload = storedRange.substr(kDeltaStoredPayloadLengthBytes, expectedPayloadSize); + const auto storedChecksum = readUint32BigEndian(payload.data() + payload.size()); + bits::Crc32 crc; + crc.process_bytes(payload.data(), payload.size()); + VELOX_USER_CHECK_EQ( + crc.checksum(), + storedChecksum, + "Deletion vector checksum mismatch for {}: expected {}, got {}", + debugName, + storedChecksum, + crc.checksum()); + return payload; +} + roaring::Roaring64Map deserializeDeltaBitmapArray(std::string_view serializedPayload, const std::string& dvPath) { VELOX_USER_CHECK_GE( serializedPayload.size(), @@ -140,6 +188,19 @@ void DeltaDeletionVectorReader::loadSerializedDeletionVector( } } +void DeltaDeletionVectorReader::loadStoredDeletionVector( + std::string_view storedRange, + uint64_t expectedPayloadSize, + const std::string& debugName, + std::optional expectedCardinality) { + try { + loadSerializedDeletionVectorInternal( + extractStoredPayload(storedRange, expectedPayloadSize, debugName), debugName, expectedCardinality); + } catch (const std::exception& e) { + VELOX_USER_FAIL("Failed to load deletion vector from {}: {}", debugName, e.what()); + } +} + bool DeltaDeletionVectorReader::isRowDeleted(uint64_t rowPosition) { if (!deletionBitmap_.has_value()) { return false; diff --git a/cpp/velox/compute/delta/DeltaDeletionVectorReader.h b/cpp/velox/compute/delta/DeltaDeletionVectorReader.h index d98e921223c..a317a1da73c 100644 --- a/cpp/velox/compute/delta/DeltaDeletionVectorReader.h +++ b/cpp/velox/compute/delta/DeltaDeletionVectorReader.h @@ -33,9 +33,10 @@ using namespace facebook::velox; /// Reads and manages Delta Lake deletion vectors for filtering deleted rows /// during table scans. /// -/// The JVM Delta side materializes the deletion vector and hands the serialized -/// bitmap payload to native. This reader only deserializes that payload and -/// applies row filtering during scan. +/// The bitmap can arrive as an already materialized JVM payload or as a stored +/// file range loaded by DeltaSplitReader through Velox buffered input. This +/// class validates the Delta envelope, deserializes the bitmap, and applies row +/// filtering during scan; it deliberately owns no filesystem client. /// /// Usage example: /// @code @@ -58,6 +59,14 @@ class DeltaDeletionVectorReader { std::string_view serializedPayload, std::optional expectedCardinality = std::nullopt); + /// Loads a complete on-disk Delta DV range: + /// [4-byte big-endian payload size][payload][4-byte big-endian CRC32]. + void loadStoredDeletionVector( + std::string_view storedRange, + uint64_t expectedPayloadSize, + const std::string& debugName, + std::optional expectedCardinality = std::nullopt); + /// Checks if a specific row position is marked as deleted. /// Note: This method is not const because it may update internal caching /// state. diff --git a/cpp/velox/compute/delta/DeltaSplit.h b/cpp/velox/compute/delta/DeltaSplit.h index 3f9d9dd3470..656b67a3e24 100644 --- a/cpp/velox/compute/delta/DeltaSplit.h +++ b/cpp/velox/compute/delta/DeltaSplit.h @@ -37,18 +37,34 @@ enum class DeltaRowIndexFilterType { }; struct DeltaDeletionVectorDescriptor { + struct FileRange { + std::string absolutePath; + uint64_t offset; + uint64_t payloadSize; + }; + std::optional cardinality; std::optional serializedPayloadView; + std::optional fileRange; static DeltaDeletionVectorDescriptor serialized( std::optional cardinality = std::nullopt, std::optional serializedPayloadView = std::nullopt) { - return {cardinality, serializedPayloadView}; + return {cardinality, serializedPayloadView, std::nullopt}; + } + + static DeltaDeletionVectorDescriptor + onDisk(std::optional cardinality, std::string absolutePath, uint64_t offset, uint64_t payloadSize) { + return {cardinality, std::nullopt, FileRange{std::move(absolutePath), offset, payloadSize}}; } bool hasMaterializedPayload() const { return serializedPayloadView.has_value(); } + + bool hasFileRange() const { + return fileRange.has_value(); + } }; /// File-level statistics for a Delta data file. diff --git a/cpp/velox/compute/delta/DeltaSplitReader.cpp b/cpp/velox/compute/delta/DeltaSplitReader.cpp index f4060be967b..c7b3e5e919c 100644 --- a/cpp/velox/compute/delta/DeltaSplitReader.cpp +++ b/cpp/velox/compute/delta/DeltaSplitReader.cpp @@ -17,9 +17,13 @@ #include "compute/delta/DeltaSplitReader.h" +#include +#include #include #include "compute/delta/DeltaSplit.h" +#include "velox/common/base/RuntimeMetrics.h" +#include "velox/connectors/hive/BufferedInputBuilder.h" #include "velox/connectors/hive/HiveConfig.h" #include "velox/dwio/common/BufferUtil.h" @@ -118,15 +122,86 @@ void DeltaSplitReader::prepareSplit( validateStatisticsForDeletionVectors(*deltaSplit->statistics, descriptor); } - VELOX_USER_CHECK( - descriptor.hasMaterializedPayload(), - "Delta deletion vector payload was not materialized on the JVM side for split {}", - hiveSplit_->filePath); - deletionVectorReader_ = std::make_unique(); - const auto& payloadView = descriptor.serializedPayloadView.value(); - deletionVectorReader_->loadSerializedDeletionVector( - std::string_view(reinterpret_cast(payloadView.data), payloadView.size), descriptor.cardinality); + if (descriptor.hasFileRange()) { + loadDeletionVectorFromFile(descriptor); + } else { + VELOX_USER_CHECK( + descriptor.hasMaterializedPayload(), + "Delta deletion vector has neither a JVM payload nor an on-disk descriptor for split {}", + hiveSplit_->filePath); + const auto& payloadView = descriptor.serializedPayloadView.value(); + deletionVectorReader_->loadSerializedDeletionVector( + std::string_view(reinterpret_cast(payloadView.data), payloadView.size), descriptor.cardinality); + } +} + +void DeltaSplitReader::loadDeletionVectorFromFile(const DeltaDeletionVectorDescriptor& descriptor) { + VELOX_USER_CHECK(descriptor.hasFileRange(), "Delta deletion vector file range is required"); + const auto& fileRange = descriptor.fileRange.value(); + constexpr uint64_t kStoredEnvelopeBytes = 8; + VELOX_USER_CHECK_LE( + fileRange.payloadSize, + std::numeric_limits::max(), + "Delta deletion vector payload is too large for the stored format: {}", + fileRange.absolutePath); + VELOX_USER_CHECK_LE( + fileRange.payloadSize, + std::numeric_limits::max() - kStoredEnvelopeBytes, + "Delta deletion vector payload size overflows its stored range for {}", + fileRange.absolutePath); + const auto storedRangeSize = fileRange.payloadSize + kStoredEnvelopeBytes; + VELOX_USER_CHECK_LE( + fileRange.offset, + std::numeric_limits::max() - storedRangeSize, + "Delta deletion vector offset overflows its stored range for {}", + fileRange.absolutePath); + + if (ioStats_) { + ioStats_->addCounter("deltaDeletionVectorReadAttempts", RuntimeCounter(1)); + } + const auto startedAt = std::chrono::steady_clock::now(); + auto recordReadTime = [&]() { + if (ioStats_) { + const auto elapsed = + std::chrono::duration_cast(std::chrono::steady_clock::now() - startedAt).count(); + ioStats_->addCounter("deltaDeletionVectorReadWallNanos", RuntimeCounter(elapsed, RuntimeCounter::Unit::kNanos)); + } + }; + + try { + const FileHandleKey fileHandleKey{ + .filename = fileRange.absolutePath, .tokenProvider = connectorQueryCtx_->fsTokenProvider()}; + auto fileHandle = fileHandleFactory_->generate(fileHandleKey); + VELOX_CHECK_NOT_NULL(fileHandle.get()); + const auto fileSize = fileHandle->file->size(); + VELOX_USER_CHECK_LE( + fileRange.offset + storedRangeSize, + fileSize, + "Delta deletion vector range [{}..{}) exceeds file size {} for {}", + fileRange.offset, + fileRange.offset + storedRangeSize, + fileSize, + fileRange.absolutePath); + + auto input = BufferedInputBuilder::getInstance()->create( + *fileHandle, baseReaderOpts_, connectorQueryCtx_, dataIoStats_, ioStats_, ioExecutor_); + auto stream = input->enqueue({fileRange.offset, storedRangeSize}); + input->load(LogType::FILE); + std::string storedRange(storedRangeSize, '\0'); + stream->readFully(storedRange.data(), storedRange.size()); + deletionVectorReader_->loadStoredDeletionVector( + storedRange, fileRange.payloadSize, fileRange.absolutePath, descriptor.cardinality); + + if (ioStats_) { + ioStats_->addCounter( + "deltaDeletionVectorReadBytes", RuntimeCounter(storedRangeSize, RuntimeCounter::Unit::kBytes)); + } + recordReadTime(); + } catch (...) { + recordReadTime(); + throw; + } } uint64_t DeltaSplitReader::next(uint64_t size, VectorPtr& output) { diff --git a/cpp/velox/compute/delta/DeltaSplitReader.h b/cpp/velox/compute/delta/DeltaSplitReader.h index 88fad258285..404918541e9 100644 --- a/cpp/velox/compute/delta/DeltaSplitReader.h +++ b/cpp/velox/compute/delta/DeltaSplitReader.h @@ -93,6 +93,8 @@ class DeltaSplitReader : public DeltaSplitReaderBase { /// Also validates that cardinality doesn't exceed numRecords. void validateStatisticsForDeletionVectors(const DeltaFileStatistics& stats, const DeltaDeletionVectorDescriptor& dv); + void loadDeletionVectorFromFile(const DeltaDeletionVectorDescriptor& descriptor); + // Delta deletion vectors use file-global row positions, not split-relative // row numbers. uint64_t baseReadRowNumber_; diff --git a/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp b/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp index df4509b6e88..7646228ea3e 100644 --- a/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp +++ b/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp @@ -17,6 +17,7 @@ #include "compute/delta/DeltaDeletionVectorReader.h" #include "compute/delta/RoaringBitmapArray.h" +#include "velox/common/base/Crc.h" #include "velox/common/base/tests/GTestUtils.h" #include @@ -46,6 +47,23 @@ class DeltaDeletionVectorReaderTest : public ::testing::Test { return std::string(buffer->as(), serializedSize); } + std::string createStoredRange(const std::string& payload) { + auto appendBigEndian32 = [](std::string& out, uint32_t value) { + out.push_back(static_cast((value >> 24) & 0xff)); + out.push_back(static_cast((value >> 16) & 0xff)); + out.push_back(static_cast((value >> 8) & 0xff)); + out.push_back(static_cast(value & 0xff)); + }; + std::string storedRange; + storedRange.reserve(payload.size() + 8); + appendBigEndian32(storedRange, payload.size()); + storedRange.append(payload); + bits::Crc32 crc; + crc.process_bytes(payload.data(), payload.size()); + appendBigEndian32(storedRange, crc.checksum()); + return storedRange; + } + std::shared_ptr pool_; }; @@ -63,6 +81,32 @@ TEST_F(DeltaDeletionVectorReaderTest, LoadSerializedPayload) { EXPECT_FALSE(reader.isRowDeleted(20)); } +TEST_F(DeltaDeletionVectorReaderTest, LoadStoredRange) { + const auto payload = createSerializedPayload({2, 7, 12}); + const auto storedRange = createStoredRange(payload); + + DeltaDeletionVectorReader reader; + reader.loadStoredDeletionVector(storedRange, payload.size(), "dv.bin@17", 3); + + EXPECT_TRUE(reader.isRowDeleted(2)); + EXPECT_TRUE(reader.isRowDeleted(7)); + EXPECT_TRUE(reader.isRowDeleted(12)); + EXPECT_FALSE(reader.isRowDeleted(8)); +} + +TEST_F(DeltaDeletionVectorReaderTest, StoredRangeRejectsLengthAndChecksumMismatch) { + const auto payload = createSerializedPayload({1, 4}); + const auto storedRange = createStoredRange(payload); + + DeltaDeletionVectorReader reader; + VELOX_ASSERT_THROW( + reader.loadStoredDeletionVector(storedRange, payload.size() + 1, "dv.bin", 2), "range size mismatch"); + + auto corrupted = storedRange; + corrupted[4] ^= 1; + VELOX_ASSERT_THROW(reader.loadStoredDeletionVector(corrupted, payload.size(), "dv.bin", 2), "checksum mismatch"); +} + TEST_F(DeltaDeletionVectorReaderTest, LoadPortablePayload) { // Captured from a Delta 3.3.2 table after `DELETE WHERE id < 10`. const std::vector payloadBytes = {0xd1, 0xd3, 0x39, 0x64, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, diff --git a/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp b/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp index 13482ccec3e..cc2f9cca190 100644 --- a/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp +++ b/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp @@ -34,6 +34,17 @@ TEST(DeltaSplitTest, DescriptorCarriesPayloadView) { EXPECT_TRUE(descriptor.hasMaterializedPayload()); } +TEST(DeltaSplitTest, DescriptorCarriesOnDiskRange) { + auto descriptor = DeltaDeletionVectorDescriptor::onDisk(3, "s3://bucket/dv.bin", 17, 91); + + EXPECT_FALSE(descriptor.hasMaterializedPayload()); + ASSERT_TRUE(descriptor.hasFileRange()); + EXPECT_EQ(descriptor.fileRange->absolutePath, "s3://bucket/dv.bin"); + EXPECT_EQ(descriptor.fileRange->offset, 17); + EXPECT_EQ(descriptor.fileRange->payloadSize, 91); + EXPECT_EQ(descriptor.cardinality, 3); +} + TEST(DeltaSplitTest, SplitCarriesDeletionVectorDescriptor) { const std::string payload = "serialized"; gluten::SplitPayloadBufferView payloadView{ diff --git a/docs/Configuration.md b/docs/Configuration.md index 3926053a176..42b2bd8308e 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -55,6 +55,7 @@ nav_order: 15 | spark.gluten.sql.columnar.coalesce | 🔄 Dynamic | true | Enable or disable columnar coalesce. | | spark.gluten.sql.columnar.collectLimit | 🔄 Dynamic | true | Enable or disable columnar collectLimit. | | spark.gluten.sql.columnar.collectTail | 🔄 Dynamic | true | Enable or disable columnar collectTail. | +| spark.gluten.sql.columnar.delta.dv.nativeRangeRead.enabled | 🔄 Dynamic | false | When true, pass each on-disk Delta deletion vector's absolute path, offset, and size to Velox instead of materializing its payload on the executor JVM. Velox loads the range through its file-handle and buffered-input path during split preparation. Inline deletion vectors remain JVM-decoded. | | spark.gluten.sql.columnar.enableNestedColumnPruningInHiveTableScan | 🔄 Dynamic | true | Enable or disable nested column pruning in hivetablescan. | | spark.gluten.sql.columnar.enableVanillaVectorizedReaders | ⚓ Static | true | Enable or disable vanilla vectorized scan. | | spark.gluten.sql.columnar.executor.libpath | 🔄 Dynamic || The gluten executor library path. | @@ -167,4 +168,3 @@ nav_order: 15 | spark.gluten.sql.columnar.hybridExecution.enabled | ⚓ Static | false | Enable CPU/GPU hybrid execution. At runtime, the execution will be scheduled to target nodes based on the selected execution mode. | | spark.gluten.sql.columnar.hybridExecution.gpuResource.amountPerTask | ⚓ Static | 0.1 | The GPU resource amount per task. This is used to limit GPU tasks to target nodes. | | spark.gluten.sql.columnar.hybridExecution.gpuResource.name | ⚓ Static | gpu | The GPU resource name (Spark custom resource). This must match the resource name configured via spark..resource..* for GPU-stage scheduling to take effect. | - 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 903a2066639..084f3d3319f 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 903a2066639..084f3d3319f 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 e5c4d8590b1..daa24113fe7 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 @@ -16,9 +16,10 @@ */ package org.apache.gluten.delta +import org.apache.gluten.config.GlutenConfig 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, NativeDeletionVectorDescriptor, SerializedDeletionVectorPayload} import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat @@ -26,10 +27,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 +50,11 @@ object DeltaDeletionVectorScanInfo { hasDeletionVector: Boolean, rowIndexFilterType: RowIndexFilterType, cardinality: Long, - serializedDeletionVector: Array[Byte]) + deletionVectorPayload: DeletionVectorPayload) { + def serializedDeletionVector: Array[Byte] = deletionVectorPayload.materialize() + + def isPayloadMaterialized: Boolean = deletionVectorPayload.isMaterialized() + } final case class PartitionFileScanInfo( normalizedOtherMetadataColumns: Map[String, Object], @@ -63,22 +70,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,25 +113,42 @@ 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) } private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo): DeltaFileReadOptions = { - new DeltaFileReadOptions( - toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), - dvInfo.hasDeletionVector, - dvInfo.cardinality, - dvInfo.serializedDeletionVector) + dvInfo.deletionVectorPayload match { + case descriptor: NativeDeletionVectorDescriptor => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + descriptor) + case payload => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + payload) + } } private def toSubstraitRowIndexFilterType( @@ -137,21 +172,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 SerializedDeletionVectorPayload(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 +233,52 @@ 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" && + GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + val dvPath = descriptor.absolutePath(tablePath) + new NativeDeletionVectorDescriptor( + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes.toLong) + } else if (descriptor.storageType != "i") { + val dvPath = descriptor.absolutePath(tablePath) + new OnDiskDeletionVectorPayload( + serializableHadoopConf, + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes, + readMetrics) + } else { + new SerializedDeletionVectorPayload(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 +286,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 +301,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 11530d665a3..02e01a7169f 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 @@ -16,9 +16,10 @@ */ package org.apache.gluten.delta +import org.apache.gluten.config.GlutenConfig 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, NativeDeletionVectorDescriptor, SerializedDeletionVectorPayload} import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat @@ -26,10 +27,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 +51,11 @@ object DeltaDeletionVectorScanInfo { hasDeletionVector: Boolean, rowIndexFilterType: RowIndexFilterType, cardinality: Long, - serializedDeletionVector: Array[Byte]) + deletionVectorPayload: DeletionVectorPayload) { + def serializedDeletionVector: Array[Byte] = deletionVectorPayload.materialize() + + def isPayloadMaterialized: Boolean = deletionVectorPayload.isMaterialized() + } final case class PartitionFileScanInfo( normalizedOtherMetadataColumns: Map[String, Object], @@ -64,21 +71,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,25 +114,42 @@ 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) } private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo): DeltaFileReadOptions = { - new DeltaFileReadOptions( - toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), - dvInfo.hasDeletionVector, - dvInfo.cardinality, - dvInfo.serializedDeletionVector) + dvInfo.deletionVectorPayload match { + case descriptor: NativeDeletionVectorDescriptor => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + descriptor) + case payload => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + payload) + } } private def toSubstraitRowIndexFilterType( @@ -137,21 +173,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 SerializedDeletionVectorPayload(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 +256,129 @@ 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) + if ( + descriptor.storageType != "i" && + GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + val dvPath = descriptor.absolutePath(tablePath) + new NativeDeletionVectorDescriptor( + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes.toLong) + } else if (descriptor.storageType != "i") { + 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 SerializedDeletionVectorPayload(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/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java b/gluten-delta/src/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java new file mode 100644 index 00000000000..e6e31a84a84 --- /dev/null +++ b/gluten-delta/src/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java @@ -0,0 +1,36 @@ +/* + * 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.util.AccumulatorV2; + +/** Registers accumulators that Spark deserialized before installing the task context. */ +final class TaskAccumulatorRegistry { + private TaskAccumulatorRegistry() {} + + 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 00000000000..91c24de5083 --- /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 50073cd1973..3d80c582d77 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 a95f676951d..4d0b35d1b83 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 @@ -51,10 +51,18 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui .setHasDeletionVector(options.hasDeletionVector()); if (options.hasDeletionVector()) { - deltaBuilder - .setDeletionVectorCardinality(options.deletionVectorCardinality()) - .setSerializedDeletionVector( - UnsafeByteOperations.unsafeWrap(options.serializedDeletionVector())); + deltaBuilder.setDeletionVectorCardinality(options.deletionVectorCardinality()); + if (options.hasNativeDeletionVectorDescriptor()) { + NativeDeletionVectorDescriptor descriptor = options.nativeDeletionVectorDescriptor(); + deltaBuilder.setDeletionVectorDescriptor( + ReadRel.LocalFiles.FileOrFiles.DeltaReadOptions.DeletionVectorDescriptor.newBuilder() + .setAbsolutePath(descriptor.absolutePath()) + .setOffset(descriptor.offset()) + .setPayloadSize(descriptor.payloadSize())); + } else { + deltaBuilder.setSerializedDeletionVector( + UnsafeByteOperations.unsafeWrap(options.serializedDeletionVector())); + } } fileBuilder.setDelta(deltaBuilder.build()); @@ -79,24 +87,147 @@ public enum RowIndexFilterType { IF_NOT_CONTAINED } + /** + * Serializable source for a deletion-vector payload. + * + *

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 SerializedDeletionVectorPayload implements DeletionVectorPayload { + private static final long serialVersionUID = 1L; + + private final byte[] payload; + + public SerializedDeletionVectorPayload(byte[] payload) { + this.payload = payload == null ? new byte[0] : payload.clone(); + } + + @Override + public byte[] materialize() { + return payload; + } + + @Override + public boolean isMaterialized() { + return true; + } + } + + /** Immutable executor-native source for an on-disk deletion vector. */ + public static final class NativeDeletionVectorDescriptor implements DeletionVectorPayload { + private static final long serialVersionUID = 1L; + + private final String absolutePath; + private final long offset; + private final long payloadSize; + + public NativeDeletionVectorDescriptor(String absolutePath, long offset, long payloadSize) { + if (absolutePath == null || absolutePath.isEmpty()) { + throw new IllegalArgumentException("absolutePath must not be empty"); + } + if (offset < 0) { + throw new IllegalArgumentException("offset must be non-negative"); + } + if (payloadSize <= 0) { + throw new IllegalArgumentException("payloadSize must be positive"); + } + this.absolutePath = absolutePath; + this.offset = offset; + this.payloadSize = payloadSize; + } + + public String absolutePath() { + return absolutePath; + } + + public long offset() { + return offset; + } + + public long payloadSize() { + return payloadSize; + } + + @Override + public byte[] materialize() { + throw new IllegalStateException( + "Native deletion vector descriptors do not contain JVM payload bytes"); + } + + @Override + public boolean isMaterialized() { + return false; + } + } + 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; + private final NativeDeletionVectorDescriptor nativeDeletionVectorDescriptor; public DeltaFileReadOptions( RowIndexFilterType rowIndexFilterType, boolean hasDeletionVector, long deletionVectorCardinality, byte[] serializedDeletionVector) { + this( + rowIndexFilterType, + hasDeletionVector, + deletionVectorCardinality, + new SerializedDeletionVectorPayload(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; + this.nativeDeletionVectorDescriptor = null; + } + + public DeltaFileReadOptions( + RowIndexFilterType rowIndexFilterType, + boolean hasDeletionVector, + long deletionVectorCardinality, + NativeDeletionVectorDescriptor nativeDeletionVectorDescriptor) { + if (rowIndexFilterType == null) { + throw new IllegalArgumentException("rowIndexFilterType must not be null"); + } + if (!hasDeletionVector) { + throw new IllegalArgumentException( + "A native deletion vector descriptor requires hasDeletionVector=true"); + } + if (nativeDeletionVectorDescriptor == null) { + throw new IllegalArgumentException("nativeDeletionVectorDescriptor must not be null"); + } + this.rowIndexFilterType = rowIndexFilterType; + this.hasDeletionVector = hasDeletionVector; + this.deletionVectorCardinality = deletionVectorCardinality; + this.deletionVectorPayload = null; + this.nativeDeletionVectorDescriptor = nativeDeletionVectorDescriptor; } public RowIndexFilterType rowIndexFilterType() { @@ -111,8 +242,31 @@ public long deletionVectorCardinality() { return deletionVectorCardinality; } + /** + * Materializes and returns the serialized deletion-vector bytes. + * + *

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; + if (nativeDeletionVectorDescriptor != null) { + throw new IllegalStateException( + "Native deletion vector descriptors do not contain JVM payload bytes"); + } + return deletionVectorPayload.materialize(); + } + + public boolean isDeletionVectorPayloadMaterialized() { + return deletionVectorPayload != null && deletionVectorPayload.isMaterialized(); + } + + public boolean hasNativeDeletionVectorDescriptor() { + return nativeDeletionVectorDescriptor != null; + } + + public NativeDeletionVectorDescriptor nativeDeletionVectorDescriptor() { + return nativeDeletionVectorDescriptor; } } } diff --git a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto index ad74b2b86da..0a30e64aa3b 100644 --- a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto +++ b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto @@ -210,6 +210,15 @@ message ReadRel { bool has_deletion_vector = 2; uint64 deletion_vector_cardinality = 3; bytes serialized_deletion_vector = 4; + message DeletionVectorDescriptor { + // Authoritative absolute URI resolved by Delta on the JVM. + string absolute_path = 1; + // Offset of the 4-byte stored-payload length prefix. + uint64 offset = 2; + // Bitmap payload size, excluding the length prefix and CRC32. + uint64 payload_size = 3; + } + DeletionVectorDescriptor deletion_vector_descriptor = 5; } // File reading options diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 21a60b57bf3..fc08820253e 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala @@ -74,6 +74,9 @@ class GlutenConfig(conf: SQLConf) extends GlutenCoreConfig(conf) { def enableColumnarFileScan: Boolean = getConf(COLUMNAR_FILESCAN_ENABLED) + def enableNativeDeltaDeletionVectorPayloadRead: Boolean = + getConf(DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED) + def enableColumnarHiveTableScan: Boolean = getConf(COLUMNAR_HIVETABLESCAN_ENABLED) def enableColumnarHiveTableScanNestedColumnPruning: Boolean = @@ -880,6 +883,16 @@ object GlutenConfig extends ConfigRegistry { .booleanConf .createWithDefault(true) + val DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED = + buildConf("spark.gluten.sql.columnar.delta.dv.nativeRangeRead.enabled") + .doc( + "When true, pass each on-disk Delta deletion vector's absolute path, offset, and size " + + "to Velox instead of materializing its payload on the executor JVM. Velox loads the " + + "range through its file-handle and buffered-input path during split preparation. " + + "Inline deletion vectors remain JVM-decoded.") + .booleanConf + .createWithDefault(false) + val COLUMNAR_HIVETABLESCAN_ENABLED = buildConf("spark.gluten.sql.columnar.hivetablescan") .doc("Enable or disable columnar hivetablescan.") 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 83023f027ac..088c5f2d6d3 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("")}" diff --git a/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala b/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala index 085b39973e8..1a3cd78b34f 100644 --- a/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala +++ b/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala @@ -48,6 +48,21 @@ class GlutenRuntimeConfigSuite extends GlutenQueryTest with SharedSparkSession { } } + test("native Delta deletion vector payload reads are opt-in and configurable") { + val conf = SparkSession.active.conf + val key = GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key + val original = conf.get(key) + try { + assert(!GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead) + conf.set(key, true) + assert(GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead) + conf.set(key, false) + assert(!GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead) + } finally { + conf.set(key, original) + } + } + test("Memory manager capacity ratio config validation") { assert(GlutenConfig.MEMORY_MANAGER_CAPACITY_RATIO.defaultValue.get == 0.75)