This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new c5cb134c79 [GLUTEN][VL] Optimize Delta Lake DV materialization and
plan rule performance (#12390)
c5cb134c79 is described below
commit c5cb134c792008844084c1b68b40ee4db0407d81
Author: Ismaël Mejía <[email protected]>
AuthorDate: Fri Jul 10 10:06:48 2026 +0200
[GLUTEN][VL] Optimize Delta Lake DV materialization and plan rule
performance (#12390)
* [GLUTEN][VL] Eliminate per-file I/O and allocation in DV materialization
Cache the resolved table path and Hadoop Configuration across all files
in a partition during normalize(). Previously, each file triggered
independent filesystem exists() checks (to find the _delta_log
directory) and allocated a fresh Hadoop Configuration clone. For a
partition with N files on object storage, this produced N+ redundant
HTTP HEAD requests on the driver critical path.
For on-disk DVs, read the raw bytes directly from the DV file using
Delta's DeletionVectorStore.readRangeFromStream (which includes
checksum verification) instead of going through StoredBitmap.load()
+ serializeAsByteArray(). The on-disk format is already Portable
Roaring Bitmap Array -- the same format the native Velox side expects
-- so this eliminates the expensive deserialize-into-Java-Roaring-
objects + re-serialize round-trip per file.
Changes:
- Resolve table path once using the first file, reuse for all others
- Create one Hadoop Configuration per normalize() call
- Read raw DV bytes directly for on-disk DVs (skip deser+reser)
- Fall back to load+serialize for inline DVs (small, in-metadata)
- (delta40) Cache the reflective method lookup for parseDescriptor
Assisted-by: GitHub Copilot:claude-opus-4.6
* [GLUTEN][VL] Optimize Delta post-transform rules for non-Delta queries
Reduce plan traversal overhead from 5 full passes to effectively 1 for
Delta queries and 0 for non-Delta queries:
- Add early-exit guard: check plan.exists(DeltaScanTransformer) once and
skip all Delta-specific rules if no Delta scan is present. This
eliminates all overhead for non-Delta queries.
- Replace quadratic containsNativeDeltaScan (full subtree .exists() per
Filter/Project node) with a shallow 2-level child check that is O(1),
safe because transformUp processes bottom-up.
- Pre-compute inputFileRelatedNames as a static Set[String] instead of
allocating 3 Expression objects + 2 Seqs per call per column.
- Batch createPhysicalAttributes: single call with full attribute list
instead of per-column invocation that walks the reference schema N
times for a table with N columns.
- Fuse nativeDeletionVectorRule, pushDownInputFileExprRule, and
columnMappingRule into a single registered rule to reduce the number
of injected post-transforms from 4 to 2.
Assisted-by: GitHub Copilot:claude-opus-4.6
* [GLUTEN][VL] Reduce allocation in Delta scan and DV serialization
- DeltaScanTransformer.scanFilters: change from def to lazy val to
avoid rebuilding the physicalByExprId map and re-traversing filter
expression trees on every call (invoked 3+ times per scan node).
- DeltaLocalFilesNode: use UnsafeByteOperations.unsafeWrap() instead of
ByteString.copyFrom() for the DV byte array. This is a zero-copy
wrap since the byte[] lifetime is guaranteed by DeltaFileReadOptions,
eliminating an O(DV_size) memcpy per file on the driver.
- LocalFilesNode: improve documentation on the copy constructor noting
that the original is discarded after construction.
Assisted-by: GitHub Copilot:claude-opus-4.6
* [GLUTEN][VL] Add DeltaPlanningBenchmark for JVM-side planning perf
Adds a Spark Benchmark that measures the two hot paths optimized in
this patch series:
1. DV Materialization (DeltaDeletionVectorScanInfo.normalize):
Creates a Delta table with N DV-bearing files and times the
normalize() call that resolves table paths, loads DV bitmaps, and
builds split metadata. Directly measures the impact of caching
table path + Hadoop conf + DV store ("Eliminate per-file I/O" commit).
2. Post-transform rule application (DeltaPostTransformRules.rules):
Applies the Delta post-transform rules to a plan containing
DeltaScanTransformer nodes. Measures rule traversal overhead
including the early-exit guard, shallow containsNativeDeltaScan,
pre-computed names, and batched attribute mapping ("Optimize Delta
post-transform rules" commit).
3. Non-Delta plan overhead (control):
Applies the same rules to a plain parquet plan to verify the
early-exit guard produces zero overhead for non-Delta queries.
Configurable via spark.gluten.benchmark.delta.numFiles (default 100)
and spark.gluten.benchmark.delta.rowsPerFile (default 10000).
Measured results (local filesystem, 100 DV-bearing files):
Benchmark Before After Speedup
------- ------ ----- -------
DV Materialization (100 files) 22 ms 7 ms 3.3x
Post-transform rules (Delta) 37 us 20 us 1.8x
Post-transform rules (parquet) 4908 ns 220 ns 22.3x
Call count reduction for 100 DV-bearing files:
Operation Before After Eliminated
--------- ------ ----- ----------
FileSystem.exists() (HEAD reqs) 100-300 1 99-299
newHadoopConf() (deep clone) 100-300 1 99-299
new HadoopFileSystemDVStore() 100 1 99
Plan tree traversals (non-Delta) 5 0 5
Plan tree traversals (Delta) 5 1 4
containsNativeDeltaScan subtree O(n^2) O(1) --
createPhysicalAttributes calls N cols 1 N-1
Projected DV materialization time by storage backend (100 files):
Storage exists() latency Before After Speedup
------- ---------------- ------ ----- -------
Local FS ~67 us/call 22 ms 7 ms 3.3x
ABFS 20-80 ms/call 2-24 sec 1.0-1.1 s 2-22x
GCS 30-100 ms/call 3-30 sec 1.0-1.1 s 3-27x
S3 50-150 ms/call 5-45 sec 1.1-1.2 s 5-38x
After = 1 exists() call + 100 DV loads (~10 ms each on object stores)
Before = 100-300 exists() calls + 100 DV loads
Remote object storage impact analysis:
The dominant cost in DV materialization is resolveTablePath(), which
calls FileSystem.exists() to locate the _delta_log directory. On local
FS this is ~67us per call; on object stores each exists() is an HTTP
HEAD request with the latencies shown above.
Before this patch, resolveTablePath() was called per-file, plus
isDeltaTablePath() could walk up 1-3 parent directories per file.
After: a single exists() call resolves the table path for all files.
The DV bitmap load (StoredBitmap.load) remains per-file but benefits
from connection pooling via the shared HadoopFileSystemDVStore (the
FS instance is reused across all files since the same Configuration
object hits Hadoop's FileSystem cache).
Assisted-by: GitHub Copilot:claude-opus-4.6
* [MINOR] Add Eclipse project files to .gitignore
Assisted-by: GitHub Copilot:claude-opus-4.6
* Address review comments: defensive guards, test improvements, fallback
parse
- Fix misleading 'direct list reference transfer' comment in LocalFilesNode
to accurately describe the shallow list copy behavior.
- Add empty partitionFiles guard in normalize() for both delta33 and delta40
to prevent NoSuchElementException on empty input.
- Strengthen test assertions: use 'eq' identity check for early-exit guard,
rename test to match actual behavior, replace silent 'if' with assert.
- Fix --jars doc syntax to use comma-separated format in benchmark.
- Remove orphaned parseDescriptor Scaladoc block.
- Cache all available parse methods and try in order, preserving fallback
semantics while avoiding per-call getMethod overhead.
Assisted-by: GitHub Copilot:claude-opus-4.6
---
.gitignore | 4 +
.../benchmark/DeltaPlanningBenchmark.scala | 238 +++++++++++++++++++++
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 119 ++++++++---
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 141 ++++++++----
.../gluten/execution/DeltaScanTransformer.scala | 2 +-
.../gluten/extension/DeltaPostTransformRules.scala | 76 +++++--
.../org/apache/gluten/execution/DeltaSuite.scala | 58 +++++
.../gluten/substrait/rel/DeltaLocalFilesNode.java | 5 +-
.../gluten/substrait/rel/LocalFilesNode.java | 3 +
9 files changed, 551 insertions(+), 95 deletions(-)
diff --git a/.gitignore b/.gitignore
index 85c0cd6f05..c6aeadc0b1 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 0000000000..597a5b079d
--- /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 <spark-core-test-jar>,<gluten-backends-velox-jar>
+ * }}}
+ *
+ * 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 263d0ffb35..812f9b9ec3 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 cddc8849fc..22ffb3c89a 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 6ac644622d..86667d988f 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 d984faf75b..fb694c70d9 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 c138bed1cd..9f8e994f8d 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 f79486947a..a95f676951 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 @@ public class DeltaLocalFilesNode extends LocalFilesNode {
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 8a79351ad2..dfea5dd753 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 class LocalFilesNode implements SplitInfo {
/**
* 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.
+ *
+ * <p>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<Map<String, Object>>
otherMetadataColumns) {
this.index = other.index;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
