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..f2d3ffd2c28 --- /dev/null +++ b/backends-velox/src-delta/test/scala/org/apache/gluten/delta/DeltaDeletionVectorDeferredReadTests.scala @@ -0,0 +1,258 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.delta + +import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeltaFileReadOptions, 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) + } + } + + 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..6b44941b45d 100644 --- a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala +++ b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala @@ -58,11 +58,18 @@ class DeltaDeletionVectorHandoffSuite val df = spark.read.format("delta").load(path) val executedPlan = df.queryExecution.executedPlan - assert(executedPlan.collect { case _: DeltaScanTransformer => true }.nonEmpty) + val nativeScans = executedPlan.collect { case scan: DeltaScanTransformer => scan } + assert(nativeScans.nonEmpty) val planText = executedPlan.toString() assert(!planText.contains("__delta_internal_is_row_deleted")) assert(!planText.contains("__delta_internal_row_index")) checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 1L) + assert(metrics("dvPayloadReadBytes").value > 0L) + assert(metrics("dvPayloadReadTime").value > 0L) } } } diff --git a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala index 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..159e848df07 100644 --- a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala +++ b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala @@ -88,11 +88,18 @@ class DeltaDeletionVectorHandoffSuite val df = spark.read.format("delta").load(path) val executedPlan = df.queryExecution.executedPlan - assert(executedPlan.collect { case _: DeltaScanTransformer => true }.nonEmpty) + val nativeScans = executedPlan.collect { case scan: DeltaScanTransformer => scan } + assert(nativeScans.nonEmpty) val planText = executedPlan.toString() assert(!planText.contains("__delta_internal_is_row_deleted")) assert(!planText.contains("__delta_internal_row_index")) checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 1L) + assert(metrics("dvPayloadReadBytes").value > 0L) + assert(metrics("dvPayloadReadTime").value > 0L) } } } diff --git a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala index 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..8d65a48c10c 100644 --- a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -18,7 +18,7 @@ package org.apache.gluten.delta import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.rel.DeltaLocalFilesNode -import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions +import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayload, DeltaFileReadOptions, SerializedDeletionVectorPayload} import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat @@ -26,10 +26,12 @@ import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArrayFormat, StoredBitmap} import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore, HadoopFileSystemDVStore} import org.apache.spark.sql.execution.datasources.PartitionedFile +import org.apache.spark.util.SerializableConfiguration import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path +import java.io.DataInputStream import java.util.{Map => JMap} import scala.collection.JavaConverters._ @@ -47,7 +49,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 +69,33 @@ object DeltaDeletionVectorScanInfo { * the DV bookkeeping keys stripped. Returns None when no file in the split carries a deletion * vector, so callers can keep the generic split representation. * - * `tablePath` is the Delta table root, supplied by the caller from `TahoeFileIndex.path`, and is - * used to resolve on-disk DV locations. A single Hadoop Configuration is reused across all files - * in the partition. + * `tablePath` is the authoritative Delta table root supplied by `TahoeFileIndex.path`. On-disk DV + * descriptors retain a shared serializable Hadoop configuration but do not open their sidecar + * until executor-side split serialization. Inline DVs remain eager because their bytes are + * already present in Delta metadata. */ def normalize( partitionFiles: Seq[PartitionedFile], tablePath: Path) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { + normalize(partitionFiles, tablePath, None) + } + + def normalize( + partitionFiles: Seq[PartitionedFile], + tablePath: Path, + readMetrics: Option[DeletionVectorReadMetrics]) + : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { if (partitionFiles.isEmpty) { return None } val spark = activeSparkSession - // Create a single Hadoop Configuration for the entire partition. val hadoopConf = spark.sessionState.newHadoopConf() + val serializableHadoopConf = new SerializableConfiguration(hadoopConf) - val scanInfos = partitionFiles.map(file => extract(file, hadoopConf, tablePath)) + val scanInfos = partitionFiles.map { + file => extract(file, hadoopConf, serializableHadoopConf, tablePath, readMetrics) + } if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) { Some( ( @@ -95,16 +112,24 @@ object DeltaDeletionVectorScanInfo { file: PartitionedFile, tablePath: Path): PartitionFileScanInfo = { val hadoopConf = spark.sessionState.newHadoopConf() - extract(file, hadoopConf, tablePath) + val serializableHadoopConf = new SerializableConfiguration(hadoopConf) + extract(file, hadoopConf, serializableHadoopConf, tablePath, None) } private def extract( file: PartitionedFile, hadoopConf: Configuration, - tablePath: Path): PartitionFileScanInfo = { + serializableHadoopConf: SerializableConfiguration, + tablePath: Path, + readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo = { val metadata = otherMetadataColumns(file) val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) - val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath) + val dvInfo = extractDeletionVectorInfo( + metadata, + hadoopConf, + serializableHadoopConf, + tablePath, + readMetrics) PartitionFileScanInfo(normalizedMetadata, dvInfo) } @@ -113,7 +138,7 @@ object DeltaDeletionVectorScanInfo { toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), dvInfo.hasDeletionVector, dvInfo.cardinality, - dvInfo.serializedDeletionVector) + dvInfo.deletionVectorPayload) } private def toSubstraitRowIndexFilterType( @@ -137,21 +162,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 +223,43 @@ object DeltaDeletionVectorScanInfo { } } + /** Selects a deferred source for on-disk DVs and eager bytes for inline DVs. */ + private def deletionVectorPayload( + hadoopConf: Configuration, + serializableHadoopConf: SerializableConfiguration, + tablePath: Path, + descriptor: DeletionVectorDescriptor, + readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorPayload = { + if (tablePath == null) { + throw new IllegalStateException( + "Unable to resolve Delta table path while preparing deletion vector payload") + } + if (descriptor.storageType != "i") { + val dvPath = descriptor.absolutePath(tablePath) + new OnDiskDeletionVectorPayload( + serializableHadoopConf, + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes, + readMetrics) + } else { + new 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 +267,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 +282,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..4806ddb0a72 100644 --- a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -18,7 +18,7 @@ package org.apache.gluten.delta import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.rel.DeltaLocalFilesNode -import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions +import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayload, DeltaFileReadOptions, SerializedDeletionVectorPayload} import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat @@ -26,10 +26,12 @@ import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArrayFormat, StoredBitmap} import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore, HadoopFileSystemDVStore} import org.apache.spark.sql.execution.datasources.PartitionedFile +import org.apache.spark.util.SerializableConfiguration import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path +import java.io.DataInputStream import java.util.{Map => JMap} import scala.collection.JavaConverters._ @@ -48,7 +50,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 +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 val hadoopConf = spark.sessionState.newHadoopConf() + val serializableHadoopConf = new SerializableConfiguration(hadoopConf) - val scanInfos = partitionFiles.map(file => extract(file, hadoopConf, tablePath)) + val scanInfos = partitionFiles.map { + file => extract(file, hadoopConf, serializableHadoopConf, tablePath, readMetrics) + } if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) { Some( ( @@ -95,16 +113,24 @@ object DeltaDeletionVectorScanInfo { file: PartitionedFile, tablePath: Path): PartitionFileScanInfo = { val hadoopConf = spark.sessionState.newHadoopConf() - extract(file, hadoopConf, tablePath) + val serializableHadoopConf = new SerializableConfiguration(hadoopConf) + extract(file, hadoopConf, serializableHadoopConf, tablePath, None) } private def extract( file: PartitionedFile, hadoopConf: Configuration, - tablePath: Path): PartitionFileScanInfo = { + serializableHadoopConf: SerializableConfiguration, + tablePath: Path, + readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo = { val metadata = otherMetadataColumns(file) val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) - val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath) + val dvInfo = extractDeletionVectorInfo( + metadata, + hadoopConf, + serializableHadoopConf, + tablePath, + readMetrics) PartitionFileScanInfo(normalizedMetadata, dvInfo) } @@ -113,7 +139,7 @@ object DeltaDeletionVectorScanInfo { toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), dvInfo.hasDeletionVector, dvInfo.cardinality, - dvInfo.serializedDeletionVector) + dvInfo.deletionVectorPayload) } private def toSubstraitRowIndexFilterType( @@ -137,21 +163,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 +246,120 @@ object DeltaDeletionVectorScanInfo { } } - private def serializePayload( + /** Selects a deferred source for on-disk DVs and eager bytes for inline DVs. */ + private def deletionVectorPayload( hadoopConf: Configuration, + serializableHadoopConf: SerializableConfiguration, tablePath: Path, - descriptor: DeletionVectorDescriptor): Array[Byte] = { + descriptor: DeletionVectorDescriptor, + readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorPayload = { if (tablePath == null) { throw new IllegalStateException( - "Unable to resolve Delta table path while materializing deletion vector payload") + "Unable to resolve Delta table path while preparing deletion vector payload") } if (descriptor.storageType != "i") { - // On-disk DV: read raw bytes directly (already in Portable Roaring format). - readRawDvBytes(hadoopConf, tablePath, descriptor) + val dvPath = descriptor.absolutePath(tablePath) + new OnDiskDeletionVectorPayload( + serializableHadoopConf, + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes, + readMetrics) } else { - // Inline DV: bytes are in the descriptor metadata. - val dvStore = new HadoopFileSystemDVStore(hadoopConf) - StoredBitmap - .create(descriptor, tablePath) - .load(dvStore) - .serializeAsByteArray(RoaringBitmapArrayFormat.Portable) + new 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..04be0b268d5 100644 --- a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java +++ b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java @@ -79,24 +79,76 @@ public 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; + } + } + public static class DeltaFileReadOptions implements Serializable { private static final long serialVersionUID = 1L; private final RowIndexFilterType rowIndexFilterType; private final boolean hasDeletionVector; private final long deletionVectorCardinality; - private final byte[] serializedDeletionVector; + private final DeletionVectorPayload deletionVectorPayload; public DeltaFileReadOptions( RowIndexFilterType rowIndexFilterType, boolean hasDeletionVector, long deletionVectorCardinality, byte[] serializedDeletionVector) { + this( + rowIndexFilterType, + hasDeletionVector, + deletionVectorCardinality, + new 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; } public RowIndexFilterType rowIndexFilterType() { @@ -111,8 +163,19 @@ 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; + return deletionVectorPayload.materialize(); + } + + public boolean isDeletionVectorPayloadMaterialized() { + return deletionVectorPayload.isMaterialized(); } } } diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala index 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("")}"