diff --git a/.gitignore b/.gitignore index 85c0cd6f05d..c6aeadc0b18 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,10 @@ hs_err_pid* # IDEA config .idea/ +# Eclipse config +.classpath +.project +.settings/ # vscode config .vscode # vscode scala 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 new file mode 100644 index 00000000000..597a5b079d8 --- /dev/null +++ b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala @@ -0,0 +1,238 @@ +/* + * 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.spark.sql.execution.benchmark + +import org.apache.gluten.delta.DeltaDeletionVectorScanInfo +import org.apache.gluten.extension.DeltaPostTransformRules + +import org.apache.spark.benchmark.Benchmark +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.delta.DeltaLog + +import org.apache.hadoop.fs.Path + +/** + * Benchmark for Delta Lake planning-time operations in Gluten. + * + * Measures two hot paths that our performance optimizations target: + * + * 1. '''DV Materialization''' (`DeltaDeletionVectorScanInfo.normalize`): resolves table paths, + * loads DV bitmaps from storage, and serializes them into split metadata. Our optimizations + * (caching table path, Hadoop conf, DV store across files) target this path. + * 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, + * batched attribute mapping) target this path. + * + * To run: + * {{{ + * bin/spark-submit --class org.apache.spark.sql.execution.benchmark.DeltaPlanningBenchmark \ + * --jars , + * }}} + * + * Or via Maven (from the backends-velox module): + * {{{ + * ./build/mvn test -pl backends-velox -Pspark-3.5 -Pbackends-velox -Pdelta -Pjava-17 \ + * -Dtest=none -DfailIfNoTests=false \ + * -Dsuites="org.apache.spark.sql.execution.benchmark.DeltaPlanningBenchmark" + * }}} + */ +object DeltaPlanningBenchmark extends SqlBasedBenchmark { + + override def getSparkSession: SparkSession = { + SparkSession + .builder() + .master("local[1]") + .appName("DeltaPlanningBenchmark") + .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") + .config( + "spark.sql.catalog.spark_catalog", + "org.apache.spark.sql.delta.catalog.DeltaCatalog") + .config("spark.plugins", "org.apache.gluten.GlutenPlugin") + .config("spark.memory.offHeap.enabled", "true") + .config("spark.memory.offHeap.size", "1024MB") + .config("spark.ui.enabled", "false") + .config("spark.default.parallelism", "1") + .getOrCreate() + } + + private val numFiles = + spark.sparkContext.conf.getInt("spark.gluten.benchmark.delta.numFiles", 100) + private val rowsPerFile = + spark.sparkContext.conf.getInt("spark.gluten.benchmark.delta.rowsPerFile", 10000) + private val benchmarkIters = + spark.sparkContext.conf.getInt("spark.gluten.benchmark.iterations", 5) + + override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { + runDvMaterializationBenchmark() + runPostTransformRulesBenchmark() + runNonDeltaRulesOverheadBenchmark() + } + + /** + * Benchmarks DeltaDeletionVectorScanInfo.normalize() -- the critical path that loads DVs from + * storage on the driver. Measures how caching table path + DV store reduces overhead. + */ + private def runDvMaterializationBenchmark(): Unit = { + val benchmark = new Benchmark( + s"DV Materialization (normalize) - $numFiles files", + numFiles.toLong, + minNumIters = benchmarkIters, + output = output) + + withDeltaTableWithDVs(numFiles, rowsPerFile) { + (path, partitionedFiles) => + benchmark.addCase(s"normalize() - $numFiles DV files", benchmarkIters) { + _ => + DeltaDeletionVectorScanInfo.normalize( + partitionColumnCount = 0, + partitionFiles = partitionedFiles) + } + + benchmark.run() + } + } + + /** + * Benchmarks DeltaPostTransformRules application on a Delta plan with DV columns. Measures the + * combined cost of DV stripping, input_file pushdown, and column mapping. + */ + private def runPostTransformRulesBenchmark(): Unit = { + val benchmark = new Benchmark( + "Post-transform rules (Delta plan)", + 1L, + minNumIters = benchmarkIters, + output = output) + + withDeltaTableWithDVs(numFiles = 10, rowsPerFile = 1000) { + (path, _) => + val df = spark.read.format("delta").load(path) + // Force planning to get the executed plan with DeltaScanTransformer + val plan = df.queryExecution.executedPlan + + benchmark.addCase("apply rules (Delta plan with DV)", benchmarkIters) { + _ => + val result = DeltaPostTransformRules.rules.foldLeft(plan) { + (p, rule) => rule(p) + } + // Prevent dead code elimination + assert(result != null) + } + + benchmark.run() + } + } + + /** + * Benchmarks post-transform rules on a non-Delta plan to verify zero overhead from the early-exit + * guard. This is the "control" case showing that non-Delta queries don't pay for Delta rule + * traversals. + */ + private def runNonDeltaRulesOverheadBenchmark(): Unit = { + val benchmark = new Benchmark( + "Post-transform rules (non-Delta plan)", + 1L, + minNumIters = benchmarkIters, + output = output) + + withTempPath { + p => + // Create a parquet table (not Delta) + val path = p.getCanonicalPath + spark + .range(0, 100000, 1, numPartitions = 10) + .selectExpr("id", "id * 2 as value", "cast(id as string) as name") + .write + .parquet(path) + + val df = spark.read.parquet(path) + val plan = df.queryExecution.executedPlan + + benchmark.addCase("apply rules (non-Delta parquet plan)", benchmarkIters) { + _ => + val result = DeltaPostTransformRules.rules.foldLeft(plan) { + (p, rule) => rule(p) + } + assert(result != null) + } + + benchmark.run() + } + } + + /** + * Creates a Delta table with deletion vectors and provides the partitioned files for direct DV + * materialization benchmarking. + */ + private def withDeltaTableWithDVs(numFiles: Int, rowsPerFile: Int)( + f: (String, Seq[org.apache.spark.sql.execution.datasources.PartitionedFile]) => Unit + ): Unit = { + withTempPath { + p => + val path = p.getCanonicalPath + val totalRows = numFiles.toLong * rowsPerFile + + // Write data across multiple files + spark + .range(0, totalRows, 1, numPartitions = numFiles) + .selectExpr("id", "id * 2 as value") + .write + .format("delta") + .save(path) + + // Enable DVs and delete some rows to create DV entries + spark.sql(s"""ALTER TABLE delta.`$path` + SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)""") + // Delete ~10% of rows to generate DVs on most files + spark.sql(s"DELETE FROM delta.`$path` WHERE id % 10 = 0") + + // Extract partitioned files with DV metadata + val deltaLog = DeltaLog.forTable(spark, new Path(path)) + val snapshot = deltaLog.update() + val allFiles = snapshot.allFiles.collect() + + import org.apache.spark.paths.SparkPath + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.delta.GlutenDeltaParquetFileFormat + import org.apache.spark.sql.execution.datasources.PartitionedFile + + val partitionedFiles = allFiles.map { + dataFile => + val metadata: Map[String, Object] = + if (dataFile.deletionVector != null) { + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + } else { + Map.empty[String, Object] + } + PartitionedFile( + partitionValues = InternalRow.empty, + filePath = SparkPath.fromPath(new Path(path, dataFile.path)), + start = 0L, + length = dataFile.size, + fileSize = dataFile.size, + otherConstantMetadataColumnValues = metadata + ) + }.toSeq + + f(path, partitionedFiles) + } + } +} 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 263d0ffb35a..812f9b9ec36 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 @@ -24,11 +24,13 @@ import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat 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.HadoopFileSystemDVStore +import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore, HadoopFileSystemDVStore} import org.apache.spark.sql.execution.datasources.PartitionedFile +import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path +import java.io.DataInputStream import java.util.{Map => JMap} import scala.collection.JavaConverters._ @@ -61,10 +63,26 @@ object DeltaDeletionVectorScanInfo { * Materializes per-file Delta DV read options for a split, alongside each file's metadata with * the DV bookkeeping keys stripped. Returns None when no file in the split carries a deletion * vector, so callers can keep the generic split representation. + * + * Performance: resolves the table path once (using the first file) and reuses a single Hadoop + * Configuration instance across all files in the partition to avoid redundant filesystem I/O and + * object allocation. */ def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { - val scanInfos = extractAll(activeSparkSession, partitionColumnCount, partitionFiles) + if (partitionFiles.isEmpty) { + return None + } + val spark = activeSparkSession + // Create a single Hadoop Configuration for the entire partition. + val hadoopConf = spark.sessionState.newHadoopConf() + // Resolve table path once using the first file -- all files in a Delta table share the same + // root, so this avoids N-1 redundant filesystem existence checks. + val cachedTablePath = resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head) + + val scanInfos = partitionFiles.map { + file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath) + } if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) { Some( ( @@ -75,21 +93,25 @@ object DeltaDeletionVectorScanInfo { } } + /** Public entry point for extracting DV info from a single file (used by tests). */ def extract( spark: SparkSession, partitionColumnCount: Int, file: PartitionedFile): PartitionFileScanInfo = { - val metadata = otherMetadataColumns(file) - val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) - val dvInfo = extractDeletionVectorInfo(spark, partitionColumnCount, file, metadata) - PartitionFileScanInfo(normalizedMetadata, dvInfo) + val hadoopConf = spark.sessionState.newHadoopConf() + val tablePath = resolveTablePath(hadoopConf, partitionColumnCount, file) + extract(partitionColumnCount, file, hadoopConf, tablePath) } - def extractAll( - spark: SparkSession, + private def extract( partitionColumnCount: Int, - files: Seq[PartitionedFile]): Seq[PartitionFileScanInfo] = { - files.map(extract(spark, partitionColumnCount, _)) + file: PartitionedFile, + hadoopConf: Configuration, + tablePath: Path): PartitionFileScanInfo = { + val metadata = otherMetadataColumns(file) + val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) + val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath) + PartitionFileScanInfo(normalizedMetadata, dvInfo) } private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo): DeltaFileReadOptions = { @@ -119,10 +141,9 @@ object DeltaDeletionVectorScanInfo { } private def extractDeletionVectorInfo( - spark: SparkSession, - partitionColumnCount: Int, - file: PartitionedFile, - metadata: Map[String, Object]): DeletionVectorInfo = { + metadata: Map[String, Object], + hadoopConf: Configuration, + tablePath: Path): DeletionVectorInfo = { val descriptorValue = metadata.get(RowIndexFilterIdEncoded) val filterTypeValue = metadata.get(RowIndexFilterTypeKey) @@ -131,7 +152,7 @@ object DeltaDeletionVectorScanInfo { DeletionVectorInfo(false, KEEP_ALL, 0L, Array.emptyByteArray) case (Some(encodedDescriptor), Some(filterType)) => val descriptor = parseDescriptor(encodedDescriptor.toString) - val serializedPayload = serializePayload(spark, partitionColumnCount, file, descriptor) + val serializedPayload = serializePayload(hadoopConf, tablePath, descriptor) DeletionVectorInfo( true, parseRowIndexFilterType(filterType.toString), @@ -172,25 +193,63 @@ object DeltaDeletionVectorScanInfo { } } + /** + * 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. + */ private def serializePayload( - spark: SparkSession, - partitionColumnCount: Int, - file: PartitionedFile, + hadoopConf: Configuration, + tablePath: Path, descriptor: DeletionVectorDescriptor): Array[Byte] = { - val tablePath = resolveTablePath(spark, partitionColumnCount, file) if (tablePath == null) { throw new IllegalStateException( "Unable to resolve Delta table path while materializing deletion vector payload") } - val dvStore = new HadoopFileSystemDVStore(spark.sessionState.newHadoopConf()) - StoredBitmap - .create(descriptor, tablePath) - .load(dvStore) - .serializeAsByteArray(RoaringBitmapArrayFormat.Portable) + 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) + } + } + + /** + * Reads raw DV bytes directly from the DV file on disk. The file layout per entry is: [4 bytes + * BE] data_size, [N bytes] payload (Portable Roaring), [4 bytes BE] CRC32 checksum. + * `DeletionVectorStore.readRangeFromStream` handles all of this including checksum verification, + * and returns the raw payload bytes. + */ + private def readRawDvBytes( + hadoopConf: Configuration, + tablePath: Path, + descriptor: DeletionVectorDescriptor): Array[Byte] = { + val dvPath = descriptor.absolutePath(tablePath) + val fs = dvPath.getFileSystem(hadoopConf) + val stream = new DataInputStream(fs.open(dvPath)) + try { + val offset = descriptor.offset.getOrElse(0) + if (offset > 0) { + stream.skipBytes(offset) + } + DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes) + } finally { + stream.close() + } } private def resolveTablePath( - spark: SparkSession, + hadoopConf: org.apache.hadoop.conf.Configuration, partitionColumnCount: Int, file: PartitionedFile): Path = { val fileParent = new Path(unescapePathName(file.filePath.toString)).getParent @@ -198,21 +257,23 @@ object DeltaDeletionVectorScanInfo { for (_ <- 0 until partitionColumnCount) { tablePath = tablePath.getParent } - if (tablePath != null && isDeltaTablePath(spark, tablePath)) { + if (tablePath != null && isDeltaTablePath(hadoopConf, tablePath)) { return tablePath } var candidate = fileParent - while (candidate != null && !isDeltaTablePath(spark, candidate)) { + while (candidate != null && !isDeltaTablePath(hadoopConf, candidate)) { candidate = candidate.getParent } if (candidate != null) candidate else tablePath } - private def isDeltaTablePath(spark: SparkSession, tablePath: Path): Boolean = { + private def isDeltaTablePath( + hadoopConf: org.apache.hadoop.conf.Configuration, + tablePath: Path): Boolean = { val deltaLogPath = new Path(tablePath, "_delta_log") try { - deltaLogPath.getFileSystem(spark.sessionState.newHadoopConf()).exists(deltaLogPath) + deltaLogPath.getFileSystem(hadoopConf).exists(deltaLogPath) } catch { case NonFatal(_) => false } 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 cddc8849fc5..22ffb3c89a8 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 @@ -24,11 +24,13 @@ import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat 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.HadoopFileSystemDVStore +import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore, HadoopFileSystemDVStore} import org.apache.spark.sql.execution.datasources.PartitionedFile +import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path +import java.io.DataInputStream import java.util.{Map => JMap} import scala.collection.JavaConverters._ @@ -62,10 +64,23 @@ object DeltaDeletionVectorScanInfo { * Materializes per-file Delta DV read options for a split, alongside each file's metadata with * the DV bookkeeping keys stripped. Returns None when no file in the split carries a deletion * vector, so callers can keep the generic split representation. + * + * Performance: resolves the table path once (using the first file) and reuses a single Hadoop + * Configuration instance across all files in the partition to avoid redundant filesystem I/O and + * object allocation. */ def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = { - val scanInfos = extractAll(activeSparkSession, partitionColumnCount, partitionFiles) + if (partitionFiles.isEmpty) { + return None + } + val spark = activeSparkSession + val hadoopConf = spark.sessionState.newHadoopConf() + val cachedTablePath = resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head) + + val scanInfos = partitionFiles.map { + file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath) + } if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) { Some( ( @@ -76,21 +91,25 @@ object DeltaDeletionVectorScanInfo { } } + /** Public entry point for extracting DV info from a single file (used by tests). */ def extract( spark: SparkSession, partitionColumnCount: Int, file: PartitionedFile): PartitionFileScanInfo = { - val metadata = otherMetadataColumns(file) - val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) - val dvInfo = extractDeletionVectorInfo(spark, partitionColumnCount, file, metadata) - PartitionFileScanInfo(normalizedMetadata, dvInfo) + val hadoopConf = spark.sessionState.newHadoopConf() + val tablePath = resolveTablePath(hadoopConf, partitionColumnCount, file) + extract(partitionColumnCount, file, hadoopConf, tablePath) } - def extractAll( - spark: SparkSession, + private def extract( partitionColumnCount: Int, - files: Seq[PartitionedFile]): Seq[PartitionFileScanInfo] = { - files.map(extract(spark, partitionColumnCount, _)) + file: PartitionedFile, + hadoopConf: Configuration, + tablePath: Path): PartitionFileScanInfo = { + val metadata = otherMetadataColumns(file) + val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) + val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath) + PartitionFileScanInfo(normalizedMetadata, dvInfo) } private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo): DeltaFileReadOptions = { @@ -120,10 +139,9 @@ object DeltaDeletionVectorScanInfo { } private def extractDeletionVectorInfo( - spark: SparkSession, - partitionColumnCount: Int, - file: PartitionedFile, - metadata: Map[String, Object]): DeletionVectorInfo = { + metadata: Map[String, Object], + hadoopConf: Configuration, + tablePath: Path): DeletionVectorInfo = { val descriptorValue = metadata.get(RowIndexFilterIdEncoded) val filterTypeValue = metadata.get(RowIndexFilterTypeKey) @@ -132,7 +150,7 @@ object DeltaDeletionVectorScanInfo { DeletionVectorInfo(false, KEEP_ALL, 0L, Array.emptyByteArray) case (Some(encodedDescriptor), Some(filterType)) => val descriptor = parseDescriptor(encodedDescriptor.toString) - val serializedPayload = serializePayload(spark, partitionColumnCount, file, descriptor) + val serializedPayload = serializePayload(hadoopConf, tablePath, descriptor) DeletionVectorInfo( true, parseRowIndexFilterType(filterType.toString), @@ -154,22 +172,35 @@ object DeltaDeletionVectorScanInfo { } } - private def parseDescriptor(encodedDescriptor: String): DeletionVectorDescriptor = { + /** Cached reflective methods for parsing DV descriptors (Delta 4.0 API compatibility). */ + private lazy val descriptorParseMethods: Seq[java.lang.reflect.Method] = { val methods = Seq("deserializeFromBase64", "fromJson") - methods.iterator - .map { - methodName => - Try { - val method = DeletionVectorDescriptor.getClass.getMethod(methodName, classOf[String]) - method - .invoke(DeletionVectorDescriptor, encodedDescriptor) - .asInstanceOf[DeletionVectorDescriptor] - }.toOption - } - .collectFirst { case Some(descriptor) => descriptor } - .getOrElse { - throw new IllegalArgumentException("Unable to parse Delta deletion vector descriptor") + val found = methods.flatMap { + methodName => + Try(DeletionVectorDescriptor.getClass.getMethod(methodName, classOf[String])).toOption + } + if (found.isEmpty) { + throw new IllegalStateException( + "Unable to find DeletionVectorDescriptor parse method (tried: " + + methods.mkString(", ") + ")") + } + found + } + + private def parseDescriptor(encodedDescriptor: String): DeletionVectorDescriptor = { + var lastException: Throwable = null + for (method <- descriptorParseMethods) { + try { + return method + .invoke(DeletionVectorDescriptor, encodedDescriptor) + .asInstanceOf[DeletionVectorDescriptor] + } catch { + case NonFatal(e) => lastException = e } + } + throw new IllegalArgumentException( + "Unable to parse Delta deletion vector descriptor", + lastException) } private def parseRowIndexFilterType(filterType: String): RowIndexFilterType = { @@ -183,24 +214,46 @@ object DeltaDeletionVectorScanInfo { } private def serializePayload( - spark: SparkSession, - partitionColumnCount: Int, - file: PartitionedFile, + hadoopConf: Configuration, + tablePath: Path, descriptor: DeletionVectorDescriptor): Array[Byte] = { - val tablePath = resolveTablePath(spark, partitionColumnCount, file) if (tablePath == null) { throw new IllegalStateException( "Unable to resolve Delta table path while materializing deletion vector payload") } - val dvStore = new HadoopFileSystemDVStore(spark.sessionState.newHadoopConf()) - StoredBitmap - .create(descriptor, tablePath) - .load(dvStore) - .serializeAsByteArray(RoaringBitmapArrayFormat.Portable) + if (descriptor.storageType != "i") { + // On-disk DV: read raw bytes directly (already in Portable Roaring format). + readRawDvBytes(hadoopConf, tablePath, descriptor) + } else { + // Inline DV: bytes are in the descriptor metadata. + val dvStore = new HadoopFileSystemDVStore(hadoopConf) + StoredBitmap + .create(descriptor, tablePath) + .load(dvStore) + .serializeAsByteArray(RoaringBitmapArrayFormat.Portable) + } + } + + private def readRawDvBytes( + hadoopConf: Configuration, + tablePath: Path, + descriptor: DeletionVectorDescriptor): Array[Byte] = { + val dvPath = descriptor.absolutePath(tablePath) + val fs = dvPath.getFileSystem(hadoopConf) + val stream = new DataInputStream(fs.open(dvPath)) + try { + val offset = descriptor.offset.getOrElse(0) + if (offset > 0) { + stream.skipBytes(offset) + } + DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes) + } finally { + stream.close() + } } private def resolveTablePath( - spark: SparkSession, + hadoopConf: org.apache.hadoop.conf.Configuration, partitionColumnCount: Int, file: PartitionedFile): Path = { val fileParent = new Path(unescapePathName(file.filePath.toString)).getParent @@ -208,21 +261,23 @@ object DeltaDeletionVectorScanInfo { for (_ <- 0 until partitionColumnCount) { tablePath = tablePath.getParent } - if (tablePath != null && isDeltaTablePath(spark, tablePath)) { + if (tablePath != null && isDeltaTablePath(hadoopConf, tablePath)) { return tablePath } var candidate = fileParent - while (candidate != null && !isDeltaTablePath(spark, candidate)) { + while (candidate != null && !isDeltaTablePath(hadoopConf, candidate)) { candidate = candidate.getParent } if (candidate != null) candidate else tablePath } - private def isDeltaTablePath(spark: SparkSession, tablePath: Path): Boolean = { + private def isDeltaTablePath( + hadoopConf: org.apache.hadoop.conf.Configuration, + tablePath: Path): Boolean = { val deltaLogPath = new Path(tablePath, "_delta_log") try { - deltaLogPath.getFileSystem(spark.sessionState.newHadoopConf()).exists(deltaLogPath) + deltaLogPath.getFileSystem(hadoopConf).exists(deltaLogPath) } catch { case NonFatal(_) => false } 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 6ac644622dc..86667d988ff 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 @@ -80,7 +80,7 @@ case class DeltaScanTransformer( // fields stay logical vs. become physical, and the longer-term cleanup direction (do all // physical translation at substrait emission time so this override and the alias-back // ProjectExec both go away). - override def scanFilters: Seq[Expression] = relation.fileFormat match { + override lazy val scanFilters: Seq[Expression] = relation.fileFormat match { case d: DeltaParquetFileFormat if d.columnMappingMode != NoMapping => val physicalByExprId = output.collect { case ar: AttributeReference => ar.exprId -> ar }.toMap dataFilters.map(_.transformDown { diff --git a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala index d984faf75b7..fb694c70d9c 100644 --- a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala +++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala @@ -36,9 +36,23 @@ import scala.collection.mutable.ListBuffer object DeltaPostTransformRules { def rules: Seq[Rule[SparkPlan]] = RemoveTransitions :: - nativeDeletionVectorRule :: - pushDownInputFileExprRule :: - columnMappingRule :: Nil + deltaSpecificRules :: + Nil + + /** + * Combines the three Delta-specific post-transform rules into a single rule that first checks + * whether the plan contains any DeltaScanTransformer. If not, the plan is returned unchanged, + * eliminating the overhead of multiple full tree traversals for non-Delta queries. + */ + private val deltaSpecificRules: Rule[SparkPlan] = (plan: SparkPlan) => { + if (!plan.exists(_.isInstanceOf[DeltaScanTransformer])) { + plan + } else { + val afterDv = nativeDeletionVectorRule(plan) + val afterPushDown = pushDownInputFileExprRule(afterDv) + columnMappingRule(afterPushDown) + } + } private val deletionVectorDeletedRowColumnName = "__delta_internal_is_row_deleted" private val deletionVectorRowIndexColumnName = "__delta_internal_row_index" @@ -164,10 +178,19 @@ object DeltaPostTransformRules { } } + /** + * Checks whether a plan subtree contains a DeltaScanTransformer. Uses a shallow check (direct + * child or grandchild) rather than a full subtree traversal, which is safe because transformUp + * processes bottom-up and the DV-related Filter/Project nodes sit directly above the scan in + * Delta's injected plan shape. + */ private def containsNativeDeltaScan(plan: SparkPlan): Boolean = { - plan.exists { + plan match { case _: DeltaScanTransformer => true - case _ => false + case _ => plan.children.exists { + case _: DeltaScanTransformer => true + case child => child.children.exists(_.isInstanceOf[DeltaScanTransformer]) + } } } @@ -261,12 +284,12 @@ object DeltaPostTransformRules { } } + private val inputFileRelatedNames: Set[String] = + Set(InputFileName(), InputFileBlockStart(), InputFileBlockLength()).map(_.prettyName) + private def isInputFileRelatedAttribute(attr: Attribute): Boolean = { attr match { - case AttributeReference(name, _, _, _) => - Seq(InputFileName(), InputFileBlockStart(), InputFileBlockLength()) - .map(_.prettyName) - .contains(name) + case AttributeReference(name, _, _, _) => inputFileRelatedNames.contains(name) case _ => false } } @@ -405,18 +428,31 @@ object DeltaPostTransformRules { case class ColumnMapping(logicalName: String, logicalType: DataType, physicalAttr: Attribute) val columnMappings = ListBuffer.empty[ColumnMapping] val seenNames = mutable.Set.empty[String] + + // Batch: collect all data attributes that need physical name mapping, call + // createPhysicalAttributes once (instead of per-column), then build a lookup map. + val dataAttrsNeedingMapping = plan.output.filter { + attr => + !plan.isMetadataColumn(attr) && + !isInputFileRelatedAttribute(attr) && + !isPartitionCol(attr.name) + } + val physicalDataAttrs = if (dataAttrsNeedingMapping.nonEmpty) { + DeltaColumnMapping.createPhysicalAttributes( + dataAttrsNeedingMapping, + fmt.referenceSchema, + fmt.columnMappingMode) + } else { + Seq.empty + } + val physicalByExprId = dataAttrsNeedingMapping.zip(physicalDataAttrs) + .map { + case (logical, physical) => + logical.exprId -> physical + }.toMap + def mapAttribute(attr: Attribute) = { - val newAttr = if (plan.isMetadataColumn(attr)) { - attr - } else if (isInputFileRelatedAttribute(attr)) { - attr - } else if (isPartitionCol(attr.name)) { - attr - } else { - DeltaColumnMapping - .createPhysicalAttributes(Seq(attr), fmt.referenceSchema, fmt.columnMappingMode) - .head - } + val newAttr = physicalByExprId.getOrElse(attr.exprId, attr) if (seenNames.add(attr.name)) { columnMappings += ColumnMapping(attr.name, attr.dataType, newAttr) } diff --git a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala index c138bed1cdf..9f8e994f8d0 100644 --- a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala +++ b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala @@ -16,6 +16,8 @@ */ package org.apache.gluten.execution +import org.apache.gluten.extension.DeltaPostTransformRules + import org.apache.spark.SparkConf import org.apache.spark.sql.Row import org.apache.spark.sql.types._ @@ -794,4 +796,60 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { checkAnswer(df, Row(0, null) :: Row(101, Seq(Row("a", 1), null)) :: Nil) } } + + test("post-transform rules are no-op on non-Delta plans") { + withTempPath { + p => + val path = p.getCanonicalPath + spark.range(100).selectExpr("id", "id * 2 as value").write.parquet(path) + val df = spark.read.parquet(path) + val plan = df.queryExecution.executedPlan + + // Apply only the Delta-specific rules (skip RemoveTransitions which is generic) + val deltaRules = DeltaPostTransformRules.rules.tail + val transformed = deltaRules.foldLeft(plan)((p, rule) => rule(p)) + // No DeltaScanTransformer in the plan, so rules should return the same object (early-exit) + assert(transformed eq plan, "Delta rules should return the exact same plan instance") + } + } + + test("Delta scan is offloaded to DeltaScanTransformer") { + withTempPath { + p => + import testImplicits._ + val path = p.getCanonicalPath + Seq(1, 2, 3, 4, 5).toDF("id").coalesce(1).write.format("delta").save(path) + val df = spark.read.format("delta").load(path) + val plan = df.queryExecution.executedPlan + + // Delta scan should be offloaded to DeltaScanTransformer + val deltaScans = plan.collect { case s: DeltaScanTransformer => s } + assert(deltaScans.nonEmpty, "Delta plan should contain DeltaScanTransformer") + } + } + + test("scanFilters returns consistent results on repeated access") { + withTempPath { + p => + import testImplicits._ + val path = p.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c")).toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + val df = spark.read.format("delta").load(path).where("id > 1") + val plan = df.queryExecution.executedPlan + val scans = plan.collect { case s: DeltaScanTransformer => s } + + assert(scans.nonEmpty, "Delta plan should contain DeltaScanTransformer") + val scan = scans.head + // scanFilters is now a lazy val; repeated calls should return the same instance + val first = scan.scanFilters + val second = scan.scanFilters + val third = scan.scanFilters + assert(first eq second, "scanFilters should return the same cached instance") + assert(second eq third, "scanFilters should return the same cached instance") + } + } } 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 f79486947a8..a95f676951d 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 @@ -16,7 +16,7 @@ */ package org.apache.gluten.substrait.rel; -import com.google.protobuf.ByteString; +import com.google.protobuf.UnsafeByteOperations; import io.substrait.proto.ReadRel; import java.io.Serializable; @@ -53,7 +53,8 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui if (options.hasDeletionVector()) { deltaBuilder .setDeletionVectorCardinality(options.deletionVectorCardinality()) - .setSerializedDeletionVector(ByteString.copyFrom(options.serializedDeletionVector())); + .setSerializedDeletionVector( + UnsafeByteOperations.unsafeWrap(options.serializedDeletionVector())); } fileBuilder.setDelta(deltaBuilder.build()); diff --git a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java index 8a79351ad2c..dfea5dd7531 100644 --- a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java +++ b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java @@ -96,6 +96,9 @@ public enum ReadFileFormat { /** * Copies an existing node, replacing its per-file extra metadata. Lets data-lake subclasses * decorate a generically built node without re-deriving the file listing. + * + *

Note: performs a shallow list copy (element references are shared, not deep-copied). This is + * safe because callers supply freshly built maps and the original node is discarded immediately. */ protected LocalFilesNode(LocalFilesNode other, List> otherMetadataColumns) { this.index = other.index;