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 cbb70817ba4..81049b9923b 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 @@ -16,7 +16,9 @@ */ package org.apache.gluten.delta +import org.apache.gluten.config.GlutenConfig 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 @@ -26,6 +28,7 @@ import org.apache.spark.sql.delta.{DeltaLog, GlutenDeltaParquetFileFormat} 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.SQLMetrics import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf} import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.tags.ExtendedSQLTest @@ -33,6 +36,12 @@ import org.apache.spark.tags.ExtendedSQLTest import io.delta.sql.DeltaSparkSessionExtension 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._ + @ExtendedSQLTest class DeltaDeletionVectorScanInfoSuite extends QueryTest @@ -82,13 +91,15 @@ class DeltaDeletionVectorScanInfoSuite ) ) - val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile) + val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path)) val dvInfo = scanInfo.deletionVectorInfo assert(dvInfo.hasDeletionVector) assert(dvInfo.rowIndexFilterType == RowIndexFilterType.IF_CONTAINED) assert(dvInfo.cardinality == dataFile.deletionVector.cardinality) + assert(!dvInfo.isPayloadMaterialized) assert(dvInfo.serializedDeletionVector.nonEmpty) + assert(dvInfo.isPayloadMaterialized) assert(scanInfo.normalizedOtherMetadataColumns == Map("kept_key" -> "kept_value")) } } @@ -106,7 +117,7 @@ class DeltaDeletionVectorScanInfoSuite dataFile.size, Map("kept_key" -> "kept_value")) - val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile) + val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path)) val dvInfo = scanInfo.deletionVectorInfo assert(!dvInfo.hasDeletionVector) @@ -131,12 +142,287 @@ class DeltaDeletionVectorScanInfoSuite Map(GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED")) val error = intercept[IllegalStateException] { - DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile) + DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path)) } assert(error.getMessage.contains("must either be present or absent")) } } + test("normalize materializes DV read options using the supplied table path") { + 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 = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + assert(dataFile.deletionVector.storageType == "u") + val partitionedFile = partitionedFileWithMetadata( + unrelatedPath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), tablePath) + assert(result.isDefined, "normalize should materialize DV options") + val opts = result.get._2.head + assert(opts.hasDeletionVector) + assert(opts.deletionVectorCardinality == dataFile.deletionVector.cardinality) + assert(opts.serializedDeletionVector.nonEmpty) + } + } + + 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 = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + assert(dataFile.deletionVector.storageType == "u") + val partitionedFile = partitionedFileWithMetadata( + unrelatedPath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + 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 = DeltaDeletionVectorScanInfo + .normalize( + Seq(partitionedFile), + tablePath, + Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts))) + .get + ._2 + .head + 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) + + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_DEFER_PAYLOAD_READ_ENABLED.key -> "false") { + val eagerOptions = DeltaDeletionVectorScanInfo + .normalize(Seq(partitionedFile), tablePath) + .get + ._2 + .head + assert(eagerOptions.isDeletionVectorPayloadMaterialized) + assert(eagerOptions.serializedDeletionVector.nonEmpty) + } + } + } + + 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 = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + val partitionedFile = partitionedFileWithMetadata( + tablePath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + 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 = DeltaDeletionVectorScanInfo + .normalize( + Seq(partitionedFile), + tablePath, + Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts))) + .get + ._2 + .head + + val dvPath = dataFile.deletionVector.absolutePath(tablePath) + val backupPath = new Path(dvPath.toString + ".retry-test-backup") + val fs = dvPath.getFileSystem(spark.sessionState.newHadoopConf()) + assert(fs.rename(dvPath, backupPath)) + try { + intercept[Exception] { + options.serializedDeletionVector + } + assert(!options.isDeletionVectorPayloadMaterialized) + assert(readAttempts.value == 1L) + assert(readBytes.value == 0L) + } finally { + assert(fs.rename(backupPath, dvPath)) + } + + val payload = options.serializedDeletionVector + assert(payload.nonEmpty) + assert(options.isDeletionVectorPayloadMaterialized) + assert(readAttempts.value == 2L) + assert(readBytes.value == payload.length.toLong) + assert(readTime.value > 0L) + } + } + + test("passes authoritative on-disk DV descriptors to native without JVM reads") { + withTempDir { + tempDir => + val tablePath = new Path(tempDir.getCanonicalPath, "table") + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(tablePath.toString) + spark.sql( + s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)") + + val dataFile = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + val partitionedFile = partitionedFileWithMetadata( + tablePath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + val options = DeltaDeletionVectorScanInfo + .normalize(Seq(partitionedFile), tablePath) + .get + ._2 + .head + assert(options.hasNativeDeletionVectorDescriptor) + assert(!options.isDeletionVectorPayloadMaterialized) + val descriptor = options.nativeDeletionVectorDescriptor + assert( + descriptor.absolutePath == dataFile.deletionVector.absolutePath(tablePath).toString) + assert(descriptor.offset == dataFile.deletionVector.offset.get.toLong) + assert(descriptor.payloadSize == dataFile.deletionVector.sizeInBytes.toLong) + val error = intercept[IllegalStateException](options.serializedDeletionVector) + assert(error.getMessage.contains("do not contain JVM payload bytes")) + } + } + } + + private def javaRoundTrip(options: DeltaFileReadOptions): DeltaFileReadOptions = { + val bytes = new ByteArrayOutputStream() + val output = new ObjectOutputStream(bytes) + try { + output.writeObject(options) + } finally { + output.close() + } + + val input = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray)) + try { + input.readObject().asInstanceOf[DeltaFileReadOptions] + } finally { + input.close() + } + } + private def partitionedFileWithMetadata( tablePath: String, relativeFilePath: String, diff --git a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala index f5510a95255..c32952efde9 100644 --- a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala +++ b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala @@ -16,6 +16,7 @@ */ package org.apache.spark.sql.delta +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.execution.DeltaScanTransformer import org.apache.spark.sql.QueryTest @@ -58,11 +59,49 @@ class DeltaDeletionVectorHandoffSuite val df = spark.read.format("delta").load(path) val executedPlan = df.queryExecution.executedPlan - assert(executedPlan.collect { case _: DeltaScanTransformer => true }.nonEmpty) + val nativeScans = executedPlan.collect { case scan: DeltaScanTransformer => scan } + assert(nativeScans.nonEmpty) val planText = executedPlan.toString() assert(!planText.contains("__delta_internal_is_row_deleted")) assert(!planText.contains("__delta_internal_row_index")) checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 1L) + assert(metrics("dvPayloadReadBytes").value > 0L) + assert(metrics("dvPayloadReadTime").value > 0L) + } + } + + test("Spark 3.5 native Delta DV descriptor filters rows without JVM payload reads") { + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + withTempDir { + tempDir => + val path = tempDir.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + + val df = spark.read.format("delta").load(path) + val nativeScans = df.queryExecution.executedPlan.collect { + case scan: DeltaScanTransformer => scan + } + assert(nativeScans.nonEmpty) + checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 0L) + assert(metrics("dvPayloadReadBytes").value == 0L) + } } } } diff --git a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala index 597a5b079d8..41aa1bb203c 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`): 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. + * 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,18 +78,19 @@ 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 caching table path + DV store 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) @@ -98,9 +99,10 @@ object DeltaPlanningBenchmark extends SqlBasedBenchmark { (path, partitionedFiles) => benchmark.addCase(s"normalize() - $numFiles DV files", benchmarkIters) { _ => - DeltaDeletionVectorScanInfo.normalize( - partitionColumnCount = 0, - partitionFiles = partitionedFiles) + val result = + DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new Path(path)) + assert( + result.exists(_._2.forall(options => !options.isDeletionVectorPayloadMaterialized))) } benchmark.run() 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 cbb70817ba4..81049b9923b 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 @@ -16,7 +16,9 @@ */ package org.apache.gluten.delta +import org.apache.gluten.config.GlutenConfig 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 @@ -26,6 +28,7 @@ import org.apache.spark.sql.delta.{DeltaLog, GlutenDeltaParquetFileFormat} 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.SQLMetrics import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf} import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.tags.ExtendedSQLTest @@ -33,6 +36,12 @@ import org.apache.spark.tags.ExtendedSQLTest import io.delta.sql.DeltaSparkSessionExtension 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._ + @ExtendedSQLTest class DeltaDeletionVectorScanInfoSuite extends QueryTest @@ -82,13 +91,15 @@ class DeltaDeletionVectorScanInfoSuite ) ) - val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile) + val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path)) val dvInfo = scanInfo.deletionVectorInfo assert(dvInfo.hasDeletionVector) assert(dvInfo.rowIndexFilterType == RowIndexFilterType.IF_CONTAINED) assert(dvInfo.cardinality == dataFile.deletionVector.cardinality) + assert(!dvInfo.isPayloadMaterialized) assert(dvInfo.serializedDeletionVector.nonEmpty) + assert(dvInfo.isPayloadMaterialized) assert(scanInfo.normalizedOtherMetadataColumns == Map("kept_key" -> "kept_value")) } } @@ -106,7 +117,7 @@ class DeltaDeletionVectorScanInfoSuite dataFile.size, Map("kept_key" -> "kept_value")) - val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile) + val scanInfo = DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path)) val dvInfo = scanInfo.deletionVectorInfo assert(!dvInfo.hasDeletionVector) @@ -131,12 +142,287 @@ class DeltaDeletionVectorScanInfoSuite Map(GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED")) val error = intercept[IllegalStateException] { - DeltaDeletionVectorScanInfo.extract(spark, 0, partitionedFile) + DeltaDeletionVectorScanInfo.extract(spark, partitionedFile, new Path(path)) } assert(error.getMessage.contains("must either be present or absent")) } } + test("normalize materializes DV read options using the supplied table path") { + 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 = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + assert(dataFile.deletionVector.storageType == "u") + val partitionedFile = partitionedFileWithMetadata( + unrelatedPath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + val result = DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile), tablePath) + assert(result.isDefined, "normalize should materialize DV options") + val opts = result.get._2.head + assert(opts.hasDeletionVector) + assert(opts.deletionVectorCardinality == dataFile.deletionVector.cardinality) + assert(opts.serializedDeletionVector.nonEmpty) + } + } + + 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 = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + assert(dataFile.deletionVector.storageType == "u") + val partitionedFile = partitionedFileWithMetadata( + unrelatedPath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + 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 = DeltaDeletionVectorScanInfo + .normalize( + Seq(partitionedFile), + tablePath, + Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts))) + .get + ._2 + .head + 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) + + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_DEFER_PAYLOAD_READ_ENABLED.key -> "false") { + val eagerOptions = DeltaDeletionVectorScanInfo + .normalize(Seq(partitionedFile), tablePath) + .get + ._2 + .head + assert(eagerOptions.isDeletionVectorPayloadMaterialized) + assert(eagerOptions.serializedDeletionVector.nonEmpty) + } + } + } + + 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 = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + val partitionedFile = partitionedFileWithMetadata( + tablePath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + 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 = DeltaDeletionVectorScanInfo + .normalize( + Seq(partitionedFile), + tablePath, + Some(DeletionVectorReadMetrics(readTime, readBytes, readAttempts))) + .get + ._2 + .head + + val dvPath = dataFile.deletionVector.absolutePath(tablePath) + val backupPath = new Path(dvPath.toString + ".retry-test-backup") + val fs = dvPath.getFileSystem(spark.sessionState.newHadoopConf()) + assert(fs.rename(dvPath, backupPath)) + try { + intercept[Exception] { + options.serializedDeletionVector + } + assert(!options.isDeletionVectorPayloadMaterialized) + assert(readAttempts.value == 1L) + assert(readBytes.value == 0L) + } finally { + assert(fs.rename(backupPath, dvPath)) + } + + val payload = options.serializedDeletionVector + assert(payload.nonEmpty) + assert(options.isDeletionVectorPayloadMaterialized) + assert(readAttempts.value == 2L) + assert(readBytes.value == payload.length.toLong) + assert(readTime.value > 0L) + } + } + + test("passes authoritative on-disk DV descriptors to native without JVM reads") { + withTempDir { + tempDir => + val tablePath = new Path(tempDir.getCanonicalPath, "table") + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(tablePath.toString) + spark.sql( + s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id IN (3, 4)") + + val dataFile = DeltaLog + .forTable(spark, tablePath) + .update() + .allFiles + .collect() + .find(_.deletionVector != null) + .get + val partitionedFile = partitionedFileWithMetadata( + tablePath.toString, + dataFile.path, + dataFile.size, + Map( + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED -> + dataFile.deletionVector.serializeToBase64(), + GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE -> "IF_CONTAINED" + ) + ) + + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + val options = DeltaDeletionVectorScanInfo + .normalize(Seq(partitionedFile), tablePath) + .get + ._2 + .head + assert(options.hasNativeDeletionVectorDescriptor) + assert(!options.isDeletionVectorPayloadMaterialized) + val descriptor = options.nativeDeletionVectorDescriptor + assert( + descriptor.absolutePath == dataFile.deletionVector.absolutePath(tablePath).toString) + assert(descriptor.offset == dataFile.deletionVector.offset.get.toLong) + assert(descriptor.payloadSize == dataFile.deletionVector.sizeInBytes.toLong) + val error = intercept[IllegalStateException](options.serializedDeletionVector) + assert(error.getMessage.contains("do not contain JVM payload bytes")) + } + } + } + + private def javaRoundTrip(options: DeltaFileReadOptions): DeltaFileReadOptions = { + val bytes = new ByteArrayOutputStream() + val output = new ObjectOutputStream(bytes) + try { + output.writeObject(options) + } finally { + output.close() + } + + val input = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray)) + try { + input.readObject().asInstanceOf[DeltaFileReadOptions] + } finally { + input.close() + } + } + private def partitionedFileWithMetadata( tablePath: String, relativeFilePath: String, diff --git a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala index dda547b015f..924a60094fe 100644 --- a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala +++ b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala @@ -16,6 +16,7 @@ */ package org.apache.spark.sql.delta +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.execution.DeltaScanTransformer import org.apache.spark.sql.QueryTest @@ -88,11 +89,49 @@ class DeltaDeletionVectorHandoffSuite val df = spark.read.format("delta").load(path) val executedPlan = df.queryExecution.executedPlan - assert(executedPlan.collect { case _: DeltaScanTransformer => true }.nonEmpty) + val nativeScans = executedPlan.collect { case scan: DeltaScanTransformer => scan } + assert(nativeScans.nonEmpty) val planText = executedPlan.toString() assert(!planText.contains("__delta_internal_is_row_deleted")) assert(!planText.contains("__delta_internal_row_index")) checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 1L) + assert(metrics("dvPayloadReadBytes").value > 0L) + assert(metrics("dvPayloadReadTime").value > 0L) + } + } + + test("Spark 4 native Delta DV descriptor filters rows without JVM payload reads") { + withSQLConf( + GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key -> "true") { + withTempDir { + tempDir => + val path = tempDir.getCanonicalPath + Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")) + .toDF("id", "value") + .coalesce(1) + .write + .format("delta") + .save(path) + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (3, 4)") + + val df = spark.read.format("delta").load(path) + val nativeScans = df.queryExecution.executedPlan.collect { + case scan: DeltaScanTransformer => scan + } + assert(nativeScans.nonEmpty) + checkAnswer(df, Seq((1, "a"), (2, "b")).toDF()) + + val metrics = nativeScans.head.metrics + assert(metrics("dvDescriptorCount").value == 1L) + assert(metrics("dvPayloadReadAttempts").value == 0L) + assert(metrics("dvPayloadReadBytes").value == 0L) + } } } } diff --git a/cpp/velox/compute/VeloxPlanConverter.cc b/cpp/velox/compute/VeloxPlanConverter.cc index 82c1f290368..703486684e6 100644 --- a/cpp/velox/compute/VeloxPlanConverter.cc +++ b/cpp/velox/compute/VeloxPlanConverter.cc @@ -119,13 +119,29 @@ std::shared_ptr parseDeltaSplitInfo( return deltaSplitInfo; } + const auto cardinality = static_cast(deltaReadOptions.deletion_vector_cardinality()); + if (deltaReadOptions.has_deletion_vector_descriptor()) { + const auto& descriptor = deltaReadOptions.deletion_vector_descriptor(); + VELOX_USER_CHECK( + deltaReadOptions.serialized_deletion_vector().empty(), + "Delta split has both a serialized deletion vector and an on-disk descriptor"); + VELOX_USER_CHECK(!descriptor.absolute_path().empty(), "Delta deletion vector path is empty"); + VELOX_USER_CHECK_GT(descriptor.payload_size(), 0, "Delta deletion vector payload size must be positive"); + VELOX_USER_CHECK_LE( + descriptor.payload_size(), + std::numeric_limits::max(), + "Delta deletion vector payload is too large for the stored format"); + deltaSplitInfo->deletionVectors.emplace_back(delta::DeltaDeletionVectorDescriptor::onDisk( + cardinality, descriptor.absolute_path(), descriptor.offset(), descriptor.payload_size())); + return deltaSplitInfo; + } + const auto& serializedPayload = deltaReadOptions.serialized_deletion_vector(); VELOX_USER_CHECK(!serializedPayload.empty(), "Delta split has a deletion vector without a serialized payload"); VELOX_USER_CHECK_LE( serializedPayload.size(), static_cast(std::numeric_limits::max()), "Delta deletion vector serialized payload is too large"); - const auto cardinality = static_cast(deltaReadOptions.deletion_vector_cardinality()); auto payload = std::make_shared(serializedPayload); const SplitPayloadBufferView payloadView{ reinterpret_cast(payload->data()), static_cast(payload->size())}; diff --git a/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp b/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp index 5c00348273f..73ff3f3ba68 100644 --- a/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp +++ b/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp @@ -33,7 +33,9 @@ #include "compute/delta/DeltaDeletionVectorReader.h" #include +#include #include "velox/common/base/BitUtil.h" +#include "velox/common/base/Crc.h" #include "velox/common/base/Exceptions.h" namespace gluten::delta { @@ -43,6 +45,7 @@ namespace { constexpr uint64_t kDeltaBitmapArrayMagicBytes = 4; constexpr uint64_t kDeltaNativeBitmapArrayLengthBytes = 4; constexpr uint64_t kDeltaStoredPayloadLengthBytes = 4; +constexpr uint64_t kDeltaStoredChecksumBytes = 4; constexpr uint32_t kDeltaPortableBitmapArrayMagicNumber = 1681511377; constexpr uint32_t kDeltaNativeBitmapArrayMagicNumber = 1681511376; @@ -52,6 +55,51 @@ uint32_t readUint32LittleEndian(const char* data) { (static_cast(bytes[2]) << 16) | (static_cast(bytes[3]) << 24); } +uint32_t readUint32BigEndian(const char* data) { + const auto* bytes = reinterpret_cast(data); + return (static_cast(bytes[0]) << 24) | (static_cast(bytes[1]) << 16) | + (static_cast(bytes[2]) << 8) | static_cast(bytes[3]); +} + +std::string_view +extractStoredPayload(std::string_view storedRange, uint64_t expectedPayloadSize, const std::string& debugName) { + VELOX_USER_CHECK_LE( + expectedPayloadSize, + std::numeric_limits::max(), + "Deletion vector payload is too large for Delta stored format: {}", + debugName); + const auto expectedRangeSize = kDeltaStoredPayloadLengthBytes + expectedPayloadSize + kDeltaStoredChecksumBytes; + VELOX_USER_CHECK_EQ( + storedRange.size(), + expectedRangeSize, + "Deletion vector range size mismatch for {}: expected {}, got {}", + debugName, + expectedRangeSize, + storedRange.size()); + + const auto storedPayloadSize = readUint32BigEndian(storedRange.data()); + VELOX_USER_CHECK_EQ( + storedPayloadSize, + expectedPayloadSize, + "Deletion vector payload size mismatch for {}: expected {}, got {}", + debugName, + expectedPayloadSize, + storedPayloadSize); + + const auto payload = storedRange.substr(kDeltaStoredPayloadLengthBytes, expectedPayloadSize); + const auto storedChecksum = readUint32BigEndian(payload.data() + payload.size()); + bits::Crc32 crc; + crc.process_bytes(payload.data(), payload.size()); + VELOX_USER_CHECK_EQ( + crc.checksum(), + storedChecksum, + "Deletion vector checksum mismatch for {}: expected {}, got {}", + debugName, + storedChecksum, + crc.checksum()); + return payload; +} + roaring::Roaring64Map deserializeDeltaBitmapArray(std::string_view serializedPayload, const std::string& dvPath) { VELOX_USER_CHECK_GE( serializedPayload.size(), @@ -155,6 +203,19 @@ void DeltaDeletionVectorReader::loadSerializedDeletionVector( } } +void DeltaDeletionVectorReader::loadStoredDeletionVector( + std::string_view storedRange, + uint64_t expectedPayloadSize, + const std::string& debugName, + std::optional expectedCardinality) { + try { + loadSerializedDeletionVectorInternal( + extractStoredPayload(storedRange, expectedPayloadSize, debugName), debugName, expectedCardinality); + } catch (const std::exception& e) { + VELOX_USER_FAIL("Failed to load deletion vector from {}: {}", debugName, e.what()); + } +} + bool DeltaDeletionVectorReader::isRowDeleted(uint64_t rowPosition) { if (!deletionBitmap_.has_value()) { return false; diff --git a/cpp/velox/compute/delta/DeltaDeletionVectorReader.h b/cpp/velox/compute/delta/DeltaDeletionVectorReader.h index 7eac6ea660d..cfbb9ff9b57 100644 --- a/cpp/velox/compute/delta/DeltaDeletionVectorReader.h +++ b/cpp/velox/compute/delta/DeltaDeletionVectorReader.h @@ -48,9 +48,10 @@ using namespace facebook::velox; /// Reads and manages Delta Lake deletion vectors for filtering deleted rows /// during table scans. /// -/// The JVM Delta side materializes the deletion vector and hands the serialized -/// bitmap payload to native. This reader only deserializes that payload and -/// applies row filtering during scan. +/// The bitmap can arrive as an already materialized JVM payload or as a stored +/// file range loaded by DeltaSplitReader through Velox buffered input. This +/// class validates the Delta envelope, deserializes the bitmap, and applies row +/// filtering during scan; it deliberately owns no filesystem client. /// /// Usage example: /// @code @@ -73,6 +74,14 @@ class DeltaDeletionVectorReader { std::string_view serializedPayload, std::optional expectedCardinality = std::nullopt); + /// Loads a complete on-disk Delta DV range: + /// [4-byte big-endian payload size][payload][4-byte big-endian CRC32]. + void loadStoredDeletionVector( + std::string_view storedRange, + uint64_t expectedPayloadSize, + const std::string& debugName, + std::optional expectedCardinality = std::nullopt); + /// Checks if a specific row position is marked as deleted. /// Note: This method is not const because it may update internal caching /// state. diff --git a/cpp/velox/compute/delta/DeltaSplit.h b/cpp/velox/compute/delta/DeltaSplit.h index df05e62fc35..67c5248ba3b 100644 --- a/cpp/velox/compute/delta/DeltaSplit.h +++ b/cpp/velox/compute/delta/DeltaSplit.h @@ -52,18 +52,34 @@ enum class DeltaRowIndexFilterType { }; struct DeltaDeletionVectorDescriptor { + struct FileRange { + std::string absolutePath; + uint64_t offset; + uint64_t payloadSize; + }; + std::optional cardinality; std::optional serializedPayloadView; + std::optional fileRange; static DeltaDeletionVectorDescriptor serialized( std::optional cardinality = std::nullopt, std::optional serializedPayloadView = std::nullopt) { - return {cardinality, serializedPayloadView}; + return {cardinality, serializedPayloadView, std::nullopt}; + } + + static DeltaDeletionVectorDescriptor + onDisk(std::optional cardinality, std::string absolutePath, uint64_t offset, uint64_t payloadSize) { + return {cardinality, std::nullopt, FileRange{std::move(absolutePath), offset, payloadSize}}; } bool hasMaterializedPayload() const { return serializedPayloadView.has_value(); } + + bool hasFileRange() const { + return fileRange.has_value(); + } }; /// File-level statistics for a Delta data file. diff --git a/cpp/velox/compute/delta/DeltaSplitReader.cpp b/cpp/velox/compute/delta/DeltaSplitReader.cpp index 37e1b2ec276..dc7b68814b3 100644 --- a/cpp/velox/compute/delta/DeltaSplitReader.cpp +++ b/cpp/velox/compute/delta/DeltaSplitReader.cpp @@ -32,9 +32,13 @@ #include "compute/delta/DeltaSplitReader.h" +#include +#include #include #include "compute/delta/DeltaSplit.h" +#include "velox/common/base/RuntimeMetrics.h" +#include "velox/connectors/hive/BufferedInputBuilder.h" #include "velox/connectors/hive/HiveConfig.h" #include "velox/dwio/common/BufferUtil.h" @@ -133,15 +137,86 @@ void DeltaSplitReader::prepareSplit( validateStatisticsForDeletionVectors(*deltaSplit->statistics, descriptor); } - VELOX_USER_CHECK( - descriptor.hasMaterializedPayload(), - "Delta deletion vector payload was not materialized on the JVM side for split {}", - hiveSplit_->filePath); - deletionVectorReader_ = std::make_unique(); - const auto& payloadView = descriptor.serializedPayloadView.value(); - deletionVectorReader_->loadSerializedDeletionVector( - std::string_view(reinterpret_cast(payloadView.data), payloadView.size), descriptor.cardinality); + if (descriptor.hasFileRange()) { + loadDeletionVectorFromFile(descriptor); + } else { + VELOX_USER_CHECK( + descriptor.hasMaterializedPayload(), + "Delta deletion vector has neither a JVM payload nor an on-disk descriptor for split {}", + hiveSplit_->filePath); + const auto& payloadView = descriptor.serializedPayloadView.value(); + deletionVectorReader_->loadSerializedDeletionVector( + std::string_view(reinterpret_cast(payloadView.data), payloadView.size), descriptor.cardinality); + } +} + +void DeltaSplitReader::loadDeletionVectorFromFile(const DeltaDeletionVectorDescriptor& descriptor) { + VELOX_USER_CHECK(descriptor.hasFileRange(), "Delta deletion vector file range is required"); + const auto& fileRange = descriptor.fileRange.value(); + constexpr uint64_t kStoredEnvelopeBytes = 8; + VELOX_USER_CHECK_LE( + fileRange.payloadSize, + std::numeric_limits::max(), + "Delta deletion vector payload is too large for the stored format: {}", + fileRange.absolutePath); + VELOX_USER_CHECK_LE( + fileRange.payloadSize, + std::numeric_limits::max() - kStoredEnvelopeBytes, + "Delta deletion vector payload size overflows its stored range for {}", + fileRange.absolutePath); + const auto storedRangeSize = fileRange.payloadSize + kStoredEnvelopeBytes; + VELOX_USER_CHECK_LE( + fileRange.offset, + std::numeric_limits::max() - storedRangeSize, + "Delta deletion vector offset overflows its stored range for {}", + fileRange.absolutePath); + + if (ioStats_) { + ioStats_->addCounter("deltaDeletionVectorReadAttempts", RuntimeCounter(1)); + } + const auto startedAt = std::chrono::steady_clock::now(); + auto recordReadTime = [&]() { + if (ioStats_) { + const auto elapsed = + std::chrono::duration_cast(std::chrono::steady_clock::now() - startedAt).count(); + ioStats_->addCounter("deltaDeletionVectorReadWallNanos", RuntimeCounter(elapsed, RuntimeCounter::Unit::kNanos)); + } + }; + + try { + const FileHandleKey fileHandleKey{ + .filename = fileRange.absolutePath, .tokenProvider = connectorQueryCtx_->fsTokenProvider()}; + auto fileHandle = fileHandleFactory_->generate(fileHandleKey); + VELOX_CHECK_NOT_NULL(fileHandle.get()); + const auto fileSize = fileHandle->file->size(); + VELOX_USER_CHECK_LE( + fileRange.offset + storedRangeSize, + fileSize, + "Delta deletion vector range [{}..{}) exceeds file size {} for {}", + fileRange.offset, + fileRange.offset + storedRangeSize, + fileSize, + fileRange.absolutePath); + + auto input = BufferedInputBuilder::getInstance()->create( + *fileHandle, baseReaderOpts_, connectorQueryCtx_, dataIoStats_, ioStats_, ioExecutor_); + auto stream = input->enqueue({fileRange.offset, storedRangeSize}); + input->load(LogType::FILE); + std::string storedRange(storedRangeSize, '\0'); + stream->readFully(storedRange.data(), storedRange.size()); + deletionVectorReader_->loadStoredDeletionVector( + storedRange, fileRange.payloadSize, fileRange.absolutePath, descriptor.cardinality); + + if (ioStats_) { + ioStats_->addCounter( + "deltaDeletionVectorReadBytes", RuntimeCounter(storedRangeSize, RuntimeCounter::Unit::kBytes)); + } + recordReadTime(); + } catch (...) { + recordReadTime(); + throw; + } } uint64_t DeltaSplitReader::next(uint64_t size, VectorPtr& output) { diff --git a/cpp/velox/compute/delta/DeltaSplitReader.h b/cpp/velox/compute/delta/DeltaSplitReader.h index 67f17216cf5..cada3befef5 100644 --- a/cpp/velox/compute/delta/DeltaSplitReader.h +++ b/cpp/velox/compute/delta/DeltaSplitReader.h @@ -108,6 +108,8 @@ class DeltaSplitReader : public DeltaSplitReaderBase { /// Also validates that cardinality doesn't exceed numRecords. void validateStatisticsForDeletionVectors(const DeltaFileStatistics& stats, const DeltaDeletionVectorDescriptor& dv); + void loadDeletionVectorFromFile(const DeltaDeletionVectorDescriptor& descriptor); + // Delta deletion vectors use file-global row positions, not split-relative // row numbers. uint64_t baseReadRowNumber_; diff --git a/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp b/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp index 7d6e03c034f..0ed02bfb91f 100644 --- a/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp +++ b/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp @@ -32,6 +32,7 @@ #include "compute/delta/DeltaDeletionVectorReader.h" #include "compute/delta/RoaringBitmapArray.h" +#include "velox/common/base/Crc.h" #include "velox/common/base/tests/GTestUtils.h" #include @@ -61,6 +62,23 @@ class DeltaDeletionVectorReaderTest : public ::testing::Test { return std::string(buffer->as(), serializedSize); } + std::string createStoredRange(const std::string& payload) { + auto appendBigEndian32 = [](std::string& out, uint32_t value) { + out.push_back(static_cast((value >> 24) & 0xff)); + out.push_back(static_cast((value >> 16) & 0xff)); + out.push_back(static_cast((value >> 8) & 0xff)); + out.push_back(static_cast(value & 0xff)); + }; + std::string storedRange; + storedRange.reserve(payload.size() + 8); + appendBigEndian32(storedRange, payload.size()); + storedRange.append(payload); + bits::Crc32 crc; + crc.process_bytes(payload.data(), payload.size()); + appendBigEndian32(storedRange, crc.checksum()); + return storedRange; + } + std::shared_ptr pool_; }; @@ -78,6 +96,32 @@ TEST_F(DeltaDeletionVectorReaderTest, LoadSerializedPayload) { EXPECT_FALSE(reader.isRowDeleted(20)); } +TEST_F(DeltaDeletionVectorReaderTest, LoadStoredRange) { + const auto payload = createSerializedPayload({2, 7, 12}); + const auto storedRange = createStoredRange(payload); + + DeltaDeletionVectorReader reader; + reader.loadStoredDeletionVector(storedRange, payload.size(), "dv.bin@17", 3); + + EXPECT_TRUE(reader.isRowDeleted(2)); + EXPECT_TRUE(reader.isRowDeleted(7)); + EXPECT_TRUE(reader.isRowDeleted(12)); + EXPECT_FALSE(reader.isRowDeleted(8)); +} + +TEST_F(DeltaDeletionVectorReaderTest, StoredRangeRejectsLengthAndChecksumMismatch) { + const auto payload = createSerializedPayload({1, 4}); + const auto storedRange = createStoredRange(payload); + + DeltaDeletionVectorReader reader; + VELOX_ASSERT_THROW( + reader.loadStoredDeletionVector(storedRange, payload.size() + 1, "dv.bin", 2), "range size mismatch"); + + auto corrupted = storedRange; + corrupted[4] ^= 1; + VELOX_ASSERT_THROW(reader.loadStoredDeletionVector(corrupted, payload.size(), "dv.bin", 2), "checksum mismatch"); +} + TEST_F(DeltaDeletionVectorReaderTest, LoadPortablePayload) { // Captured from a Delta 3.3.2 table after `DELETE WHERE id < 10`. const std::vector payloadBytes = {0xd1, 0xd3, 0x39, 0x64, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, diff --git a/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp b/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp index c081b278e2c..bbe44a044cd 100644 --- a/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp +++ b/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp @@ -49,6 +49,17 @@ TEST(DeltaSplitTest, DescriptorCarriesPayloadView) { EXPECT_TRUE(descriptor.hasMaterializedPayload()); } +TEST(DeltaSplitTest, DescriptorCarriesOnDiskRange) { + auto descriptor = DeltaDeletionVectorDescriptor::onDisk(3, "s3://bucket/dv.bin", 17, 91); + + EXPECT_FALSE(descriptor.hasMaterializedPayload()); + ASSERT_TRUE(descriptor.hasFileRange()); + EXPECT_EQ(descriptor.fileRange->absolutePath, "s3://bucket/dv.bin"); + EXPECT_EQ(descriptor.fileRange->offset, 17); + EXPECT_EQ(descriptor.fileRange->payloadSize, 91); + EXPECT_EQ(descriptor.cardinality, 3); +} + TEST(DeltaSplitTest, SplitCarriesDeletionVectorDescriptor) { const std::string payload = "serialized"; gluten::SplitPayloadBufferView payloadView{ diff --git a/docs/Configuration.md b/docs/Configuration.md index 3926053a176..22453fb1df4 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -20,136 +20,138 @@ nav_order: 15 ## Gluten configurations -| Key | Modifiability | Default | Description | -|---------------------------------------------------------------------|---------------|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| spark.gluten.costModel | 🔄 Dynamic | legacy | The class name of user-defined cost model that will be used by Gluten's transition planner. If not specified, a legacy built-in cost model will be used. The legacy cost model exhaustively offloads computations, and helps transition planner choose columnar-to-columnar transition over others. | -| spark.gluten.enabled | 🔄 Dynamic | true | Whether to enable gluten. Default value is true. Just an experimental property. Recommend to enable/disable Gluten through the setting for spark.plugins. | -| spark.gluten.execution.resource.expired.time | 🔄 Dynamic | 86400 | Expired time of execution with resource relation has cached. | -| spark.gluten.expression.blacklist | 🔄 Dynamic | <undefined> | A black list of expression to skip transform, multiple values separated by commas. | -| spark.gluten.loadLibFromJar | 🔄 Dynamic | false | Whether to load shared libraries from jars. | -| spark.gluten.loadLibOS | 🔄 Dynamic | <undefined> | The shared library loader's OS name. | -| spark.gluten.loadLibOSVersion | 🔄 Dynamic | <undefined> | The shared library loader's OS version. | -| spark.gluten.memory.isolation | 🔄 Dynamic | false | Enable isolated memory mode. If true, Gluten controls the maximum off-heap memory can be used by each task to X, X = executor memory / max task slots. It's recommended to set true if Gluten serves concurrent queries within a single session, since not all memory Gluten allocated is guaranteed to be spillable. In the case, the feature should be enabled to avoid OOM. | -| spark.gluten.memory.overAcquiredMemoryRatio | 🔄 Dynamic | 0.3 | If larger than 0, Velox backend will try over-acquire this ratio of the total allocated memory as backup to avoid OOM. | -| spark.gluten.memory.reservationBlockSize | 🔄 Dynamic | 8MB | Block size of native reservation listener reserve memory from Spark. | -| spark.gluten.numTaskSlotsPerExecutor | 🔄 Dynamic | -1 | Must provide default value since non-execution operations (e.g. org.apache.spark.sql.Dataset#summary) doesn't propagate configurations using org.apache.spark.sql.execution.SQLExecution#withSQLConfPropagated | -| spark.gluten.shuffleWriter.bufferSize | 🔄 Dynamic | <undefined> | -| spark.gluten.soft-affinity.duplicateReading.maxCacheItems | 🔄 Dynamic | 10000 | Enable Soft Affinity duplicate reading detection | -| spark.gluten.soft-affinity.duplicateReadingDetect.enabled | 🔄 Dynamic | false | If true, Enable Soft Affinity duplicate reading detection | -| spark.gluten.soft-affinity.enabled | 🔄 Dynamic | false | Whether to enable Soft Affinity scheduling. | -| spark.gluten.soft-affinity.min.target-hosts | 🔄 Dynamic | 1 | For on HDFS, if there are already target hosts, and then prefer to use the original target hosts to schedule | -| spark.gluten.soft-affinity.replications.num | 🔄 Dynamic | 2 | Calculate the number of the replications for scheduling to the target executors per file | -| spark.gluten.sql.adaptive.costEvaluator.enabled | ⚓ Static | true | If true, use org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator as custom cost evaluator class, else follow the configuration spark.sql.adaptive.customCostEvaluatorClass. | -| spark.gluten.sql.ansiFallback.enabled | 🔄 Dynamic | true | When true (default), Gluten will fall back to Spark when ANSI mode is enabled. When false, Gluten will attempt to execute in ANSI mode. | -| spark.gluten.sql.cacheWholeStageTransformerContext | 🔄 Dynamic | false | When true, `WholeStageTransformer` will cache the `WholeStageTransformerContext` when executing. It is used to get substrait plan node and native plan string. | -| spark.gluten.sql.collapseGetJsonObject.enabled | 🔄 Dynamic | false | Collapse nested get_json_object functions as one for optimization. | -| spark.gluten.sql.columnar.appendData | 🔄 Dynamic | true | Enable or disable columnar v2 command append data. | -| spark.gluten.sql.columnar.arrowUdf | 🔄 Dynamic | true | Enable or disable columnar arrow udf. | -| spark.gluten.sql.columnar.batchscan | 🔄 Dynamic | true | Enable or disable columnar batchscan. | -| spark.gluten.sql.columnar.batchscan.maxInputPartitions | 🔄 Dynamic | 2147483647 | Maximum number of Spark task partitions for supported DataSource V2 batch scans. | -| spark.gluten.sql.columnar.broadcastExchange | 🔄 Dynamic | true | Enable or disable columnar broadcastExchange. | -| spark.gluten.sql.columnar.broadcastJoin | 🔄 Dynamic | true | Enable or disable columnar broadcastJoin. | -| spark.gluten.sql.columnar.broadcastNestedLoopJoin.enabled | 🔄 Dynamic | true | Enable or disable columnar broadcastNestedLoopJoin. | -| spark.gluten.sql.columnar.cartesianProduct.enabled | 🔄 Dynamic | true | Enable or disable columnar cartesianProduct. | -| spark.gluten.sql.columnar.cast.avg | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.coalesce | 🔄 Dynamic | true | Enable or disable columnar coalesce. | -| spark.gluten.sql.columnar.collectLimit | 🔄 Dynamic | true | Enable or disable columnar collectLimit. | -| spark.gluten.sql.columnar.collectTail | 🔄 Dynamic | true | Enable or disable columnar collectTail. | -| spark.gluten.sql.columnar.enableNestedColumnPruningInHiveTableScan | 🔄 Dynamic | true | Enable or disable nested column pruning in hivetablescan. | -| spark.gluten.sql.columnar.enableVanillaVectorizedReaders | ⚓ Static | true | Enable or disable vanilla vectorized scan. | -| spark.gluten.sql.columnar.executor.libpath | 🔄 Dynamic || The gluten executor library path. | -| spark.gluten.sql.columnar.expand | 🔄 Dynamic | true | Enable or disable columnar expand. | -| spark.gluten.sql.columnar.fallback.expressions.threshold | 🔄 Dynamic | 50 | Fall back filter/project if number of nested expressions reaches this threshold, considering Spark codegen can bring better performance for such case. | -| spark.gluten.sql.columnar.fallback.ignoreRowToColumnar | 🔄 Dynamic | true | When true, the fallback policy ignores the RowToColumnar when counting fallback number. | -| spark.gluten.sql.columnar.fallback.preferColumnar | 🔄 Dynamic | true | When true, the fallback policy prefers to use Gluten plan rather than vanilla Spark plan if the both of them contains ColumnarToRow and the vanilla Spark plan ColumnarToRow number is not smaller than Gluten plan. | -| spark.gluten.sql.columnar.filescan | 🔄 Dynamic | true | Enable or disable columnar filescan. | -| spark.gluten.sql.columnar.filter | 🔄 Dynamic | true | Enable or disable columnar filter. | -| spark.gluten.sql.columnar.force.hashagg | 🔄 Dynamic | true | Whether to force to use gluten's hash agg for replacing vanilla spark's sort agg. | -| spark.gluten.sql.columnar.forceShuffledHashJoin | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.generate | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.hashagg | 🔄 Dynamic | true | Enable or disable columnar hashagg. | -| spark.gluten.sql.columnar.hivetablescan | 🔄 Dynamic | true | Enable or disable columnar hivetablescan. | -| spark.gluten.sql.columnar.libname | 🔄 Dynamic | gluten | The gluten library name. | -| spark.gluten.sql.columnar.libpath | 🔄 Dynamic || The gluten library path. | -| spark.gluten.sql.columnar.limit | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.localTableScan | 🔄 Dynamic | false | Enable or disable native columnar execution of LocalTableScanExec. When true, Gluten attempts to replace LocalTableScanExec (a driver-side local collection) with a backend transformer that converts the rows into columnar batches natively. | -| spark.gluten.sql.columnar.maxBatchSize | 🔄 Dynamic | 4096 | -| spark.gluten.sql.columnar.overwriteByExpression | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite by expression. | -| spark.gluten.sql.columnar.overwritePartitionsDynamic | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite partitions dynamic. | -| spark.gluten.sql.columnar.parquet.write.blockSize | 🔄 Dynamic | 128MB | -| spark.gluten.sql.columnar.partial.generate | 🔄 Dynamic | true | Evaluates the non-offload-able HiveUDTF using vanilla Spark generator | -| spark.gluten.sql.columnar.partial.project | 🔄 Dynamic | true | Break up one project node into 2 phases when some of the expressions are non offload-able. Phase one is a regular offloaded project transformer that evaluates the offload-able expressions in native, phase two preserves the output from phase one and evaluates the remaining non-offload-able expressions using vanilla Spark projections | -| spark.gluten.sql.columnar.physicalJoinOptimizationLevel | 🔄 Dynamic | 12 | Fallback to row operators if there are several continuous joins. | -| spark.gluten.sql.columnar.physicalJoinOptimizationOutputSize | 🔄 Dynamic | 52 | Fallback to row operators if there are several continuous joins and matched output size. | -| spark.gluten.sql.columnar.physicalJoinOptimizeEnable | 🔄 Dynamic | false | Enable or disable columnar physicalJoinOptimize. | -| spark.gluten.sql.columnar.preferStreamingAggregate | 🔄 Dynamic | true | Velox backend supports `StreamingAggregate`. `StreamingAggregate` uses the less memory as it does not need to hold all groups in memory, so it could avoid spill. When true and the child output ordering satisfies the grouping key then Gluten will choose `StreamingAggregate` as the native operator. | -| spark.gluten.sql.columnar.project | 🔄 Dynamic | true | Enable or disable columnar project. | -| spark.gluten.sql.columnar.project.collapse | 🔄 Dynamic | true | Combines two columnar project operators into one and perform alias substitution | -| spark.gluten.sql.columnar.query.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether query will fall back by counting the number of ColumnarToRow & vanilla leaf node. | -| spark.gluten.sql.columnar.range | 🔄 Dynamic | true | Enable or disable columnar range. | -| spark.gluten.sql.columnar.replaceData | 🔄 Dynamic | true | Enable or disable columnar v2 command replace data. | -| spark.gluten.sql.columnar.scanOnly | 🔄 Dynamic | false | When enabled, only scan and the filter after scan will be offloaded to native. | -| spark.gluten.sql.columnar.shuffle | 🔄 Dynamic | true | Enable or disable columnar shuffle. | -| spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled | ⚓ Static | true | If enabled, fall back to ColumnarShuffleManager when celeborn service is unavailable.Otherwise, throw an exception. | -| spark.gluten.sql.columnar.shuffle.celeborn.useRssSort | 🔄 Dynamic | true | If true, use RSS sort implementation for Celeborn sort-based shuffle.If false, use Gluten's row-based sort implementation. Only valid when `spark.celeborn.client.spark.shuffle.writer` is set to `sort`. | -| spark.gluten.sql.columnar.shuffle.codec | 🔄 Dynamic | <undefined> | By default, the supported codecs are lz4 and zstd. When spark.gluten.sql.columnar.shuffle.codecBackend=qat,the supported codecs are gzip and zstd. | -| spark.gluten.sql.columnar.shuffle.codecBackend | 🔄 Dynamic | <undefined> | -| spark.gluten.sql.columnar.shuffle.compression.threshold | 🔄 Dynamic | 100 | If number of rows in a batch falls below this threshold, will copy all buffers into one buffer to compress. | -| spark.gluten.sql.columnar.shuffle.dictionary.enabled | 🔄 Dynamic | false | Enable dictionary in hash-based shuffle. | -| spark.gluten.sql.columnar.shuffle.merge.threshold | 🔄 Dynamic | 0.25 | -| spark.gluten.sql.columnar.shuffle.partitionBufferEvictThreshold | 🔄 Dynamic | -1 | For Velox hash shuffle writer, evict partition buffers larger than this threshold after splitting an input batch. Use non-positive value to disable this feature. | -| spark.gluten.sql.columnar.shuffle.readerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for shuffle reader reading input stream from local or remote. | -| spark.gluten.sql.columnar.shuffle.realloc.threshold | 🔄 Dynamic | 0.25 | -| spark.gluten.sql.columnar.shuffle.sort.columns.threshold | 🔄 Dynamic | 100000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of columns is greater than this threshold. | -| spark.gluten.sql.columnar.shuffle.sort.deserializerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for sort-based shuffle reader deserializing raw input to columnar batch. | -| spark.gluten.sql.columnar.shuffle.sort.partitions.threshold | 🔄 Dynamic | 4000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of partitions is greater than this threshold. | -| spark.gluten.sql.columnar.shuffle.typeAwareCompress.enabled | 🔄 Dynamic | false | Enable type-aware compression (e.g. FFor for 64-bit integers) in shuffle. Not compatible with dictionary encoding; if both are enabled, type-aware compression is automatically disabled. | -| spark.gluten.sql.columnar.shuffledHashJoin | 🔄 Dynamic | true | Enable or disable columnar shuffledHashJoin. | -| spark.gluten.sql.columnar.shuffledHashJoin.optimizeBuildSide | 🔄 Dynamic | true | Whether to allow Gluten to choose an optimal build side for shuffled hash join. | -| spark.gluten.sql.columnar.smallFileThreshold | 🔄 Dynamic | 0.5 | The total size threshold of small files in table scan.To avoid small files being placed into the same partition, Gluten will try to distribute small files into different partitions when the total size of small files is below this threshold. | -| spark.gluten.sql.columnar.sort | 🔄 Dynamic | true | Enable or disable columnar sort. | -| spark.gluten.sql.columnar.sortMergeJoin | 🔄 Dynamic | true | Enable or disable columnar sortMergeJoin. This should be set with preferSortMergeJoin=false. | -| spark.gluten.sql.columnar.tableCache | ⚓ Static | true | Enable or disable columnar table cache. | -| spark.gluten.sql.columnar.tableCache.partitionStats.enabled | 🔄 Dynamic | false | When true, the Velox columnar cache serializer computes per-partition min/max/null/row-count stats and embeds them in the cached payload so that the Spark optimizer can prune whole partitions on equality / range predicates. When false (default), the serializer still writes the V3 per-column payload with empty stats so projected cache reads can lazily materialize only requested columns, while partition pruning is disabled. | -| spark.gluten.sql.columnar.takeOrderedAndProject | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.union | 🔄 Dynamic | true | Enable or disable columnar union. | -| spark.gluten.sql.columnar.wholeStage.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether whole stage will fall back in AQE supported case by counting the number of ColumnarToRow & vanilla leaf node. | -| spark.gluten.sql.columnar.window | 🔄 Dynamic | true | Enable or disable columnar window. | -| spark.gluten.sql.columnar.window.group.limit | 🔄 Dynamic | true | Enable or disable columnar window group limit. | -| spark.gluten.sql.columnar.writeToDataSourceV2 | 🔄 Dynamic | true | Enable or disable columnar v2 command write to data source v2. | -| spark.gluten.sql.columnarSampleEnabled | 🔄 Dynamic | false | Disable or enable columnar sample. | -| spark.gluten.sql.columnarToRowMemoryThreshold | 🔄 Dynamic | 64MB | -| spark.gluten.sql.countDistinctWithoutExpand | 🔄 Dynamic | false | Convert Count Distinct to a UDAF called count_distinct to prevent SparkPlanner converting it to Expand+Count. WARNING: When enabled, count distinct queries will fail to fallback!!! | -| spark.gluten.sql.extendedColumnPruning.enabled | 🔄 Dynamic | true | Do extended nested column pruning for cases ignored by vanilla Spark. | -| spark.gluten.sql.fallbackRegexpExpressions | 🔄 Dynamic | false | If true, fall back all regexp expressions. There are a few incompatible cases between RE2 (used by native engine) and java.util.regex (used by Spark). User should enable this property if their incompatibility is intolerable. | -| spark.gluten.sql.fallbackUnexpectedMetadataParquet | 🔄 Dynamic | false | If enabled, Gluten will not offload scan when unexpected metadata is detected. | -| spark.gluten.sql.fallbackUnexpectedMetadataParquet.limit | 🔄 Dynamic | 10 | If supplied, metadata of `limit` number of Parquet files will be checked to determine whether to fall back to java scan. | -| spark.gluten.sql.fallbackUnexpectedMetadataParquet.samplePercentage | 🔄 Dynamic | 0.1 | The percentage of root paths to sample for metadata validation when the number of root paths is large. Value range is (0, 1.0]. 1.0 means check all paths (no sampling). A smaller value reduces validation cost for tables with many partitions. | -| spark.gluten.sql.injectNativePlanStringToExplain | 🔄 Dynamic | false | When true, Gluten will inject native plan tree to Spark's explain output. | -| spark.gluten.sql.mergeTwoPhasesAggregate.enabled | 🔄 Dynamic | true | Whether to merge two phases aggregate if there are no other operators between them. | -| spark.gluten.sql.native.bloomFilter | 🔄 Dynamic | true | -| spark.gluten.sql.native.hive.writer.enabled | 🔄 Dynamic | true | This is config to specify whether to enable the native columnar writer for HiveFileFormat. Currently only supports HiveFileFormat with Parquet as the output file type. | -| spark.gluten.sql.native.hyperLogLog.Aggregate | 🔄 Dynamic | true | -| spark.gluten.sql.native.parquet.write.blockRows | 🔄 Dynamic | 100000000 | -| spark.gluten.sql.native.union | 🔄 Dynamic | false | Enable or disable native union where computation is completely offloaded to backend. | -| spark.gluten.sql.native.writeColumnMetadataExclusionList | 🔄 Dynamic | comment | Native write files does not support column metadata. Metadata in list would be removed to support native write files. Multiple values separated by commas. | -| spark.gluten.sql.native.writer.enabled | 🔄 Dynamic | <undefined> | This is config to specify whether to enable the native columnar parquet/orc writer | -| spark.gluten.sql.orc.charType.scan.fallback.enabled | 🔄 Dynamic | true | Force fallback for orc char type scan. | -| spark.gluten.sql.pushAggregateThroughJoin.enabled | 🔄 Dynamic | false | Enables the push-aggregate-through-join optimization in Gluten. When enabled, aggregate operators may be pushed below joins during logical optimization and corresponding physical plans may be rewritten to execute the aggregation earlier. | -| spark.gluten.sql.pushAggregateThroughJoin.maxDepth | 🔄 Dynamic | 2147483647 | Maximum join traversal depth when applying the push-aggregate-through-join optimization. A value of 1 allows pushing an aggregate through a single join; larger values allow the rule to traverse and push through multiple consecutive joins. | -| spark.gluten.sql.removeNativeWriteFilesSortAndProject | 🔄 Dynamic | true | When true, Gluten will remove the vanilla Spark V1Writes added sort and project for velox backend. | -| spark.gluten.sql.rewrite.dateTimestampComparison | 🔄 Dynamic | true | Rewrite the comparision between date and timestamp to timestamp comparison.For example `from_unixtime(ts) > date` will be rewritten to `ts > to_unixtime(date)` | -| spark.gluten.sql.scan.detailedMetrics.enabled | 🔄 Dynamic | true | When true (default), Velox backend scan operators register all detailed SQL metrics. When false, only essential scan metrics are registered to reduce driver memory usage. Also enabled automatically when spark.gluten.sql.debug is true. Does not affect the ClickHouse backend. | -| spark.gluten.sql.scan.fileSchemeValidation.enabled | 🔄 Dynamic | true | When true, enable file path scheme validation for scan. Validation will fail if file scheme is not supported by registered file systems, which will cause scan operator fall back. | -| spark.gluten.sql.supported.flattenNestedFunctions | 🔄 Dynamic | and,or | Flatten nested functions as one for optimization. | -| spark.gluten.sql.text.input.empty.as.default | 🔄 Dynamic | false | treat empty fields in CSV input as default values. | -| spark.gluten.sql.text.input.max.block.size | 🔄 Dynamic | 8KB | the max block size for text input rows | -| spark.gluten.sql.validation.printStackOnFailure | 🔄 Dynamic | false | -| spark.gluten.storage.hdfsViewfs.enabled | ⚓ Static | false | If enabled, gluten will convert the viewfs path to hdfs path in scala side | -| spark.gluten.supported.hive.udfs | 🔄 Dynamic || Supported hive udf names. | -| spark.gluten.supported.python.udfs | 🔄 Dynamic || Supported python udf names. | -| spark.gluten.supported.scala.udfs | 🔄 Dynamic || Supported scala udf names. | -| spark.gluten.ui.enabled | ⚓ Static | true | Whether to enable the gluten web UI, If true, attach the gluten UI page to the Spark web UI. | +| Key | Modifiability | Default | Description | +|-------------------------------------------------------------------------|---------------|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| spark.gluten.costModel | 🔄 Dynamic | legacy | The class name of user-defined cost model that will be used by Gluten's transition planner. If not specified, a legacy built-in cost model will be used. The legacy cost model exhaustively offloads computations, and helps transition planner choose columnar-to-columnar transition over others. | +| spark.gluten.enabled | 🔄 Dynamic | true | Whether to enable gluten. Default value is true. Just an experimental property. Recommend to enable/disable Gluten through the setting for spark.plugins. | +| spark.gluten.execution.resource.expired.time | 🔄 Dynamic | 86400 | Expired time of execution with resource relation has cached. | +| spark.gluten.expression.blacklist | 🔄 Dynamic | <undefined> | A black list of expression to skip transform, multiple values separated by commas. | +| spark.gluten.loadLibFromJar | 🔄 Dynamic | false | Whether to load shared libraries from jars. | +| spark.gluten.loadLibOS | 🔄 Dynamic | <undefined> | The shared library loader's OS name. | +| spark.gluten.loadLibOSVersion | 🔄 Dynamic | <undefined> | The shared library loader's OS version. | +| spark.gluten.memory.isolation | 🔄 Dynamic | false | Enable isolated memory mode. If true, Gluten controls the maximum off-heap memory can be used by each task to X, X = executor memory / max task slots. It's recommended to set true if Gluten serves concurrent queries within a single session, since not all memory Gluten allocated is guaranteed to be spillable. In the case, the feature should be enabled to avoid OOM. | +| spark.gluten.memory.overAcquiredMemoryRatio | 🔄 Dynamic | 0.3 | If larger than 0, Velox backend will try over-acquire this ratio of the total allocated memory as backup to avoid OOM. | +| spark.gluten.memory.reservationBlockSize | 🔄 Dynamic | 8MB | Block size of native reservation listener reserve memory from Spark. | +| spark.gluten.numTaskSlotsPerExecutor | 🔄 Dynamic | -1 | Must provide default value since non-execution operations (e.g. org.apache.spark.sql.Dataset#summary) doesn't propagate configurations using org.apache.spark.sql.execution.SQLExecution#withSQLConfPropagated | +| spark.gluten.shuffleWriter.bufferSize | 🔄 Dynamic | <undefined> | +| spark.gluten.soft-affinity.duplicateReading.maxCacheItems | 🔄 Dynamic | 10000 | Enable Soft Affinity duplicate reading detection | +| spark.gluten.soft-affinity.duplicateReadingDetect.enabled | 🔄 Dynamic | false | If true, Enable Soft Affinity duplicate reading detection | +| spark.gluten.soft-affinity.enabled | 🔄 Dynamic | false | Whether to enable Soft Affinity scheduling. | +| spark.gluten.soft-affinity.min.target-hosts | 🔄 Dynamic | 1 | For on HDFS, if there are already target hosts, and then prefer to use the original target hosts to schedule | +| spark.gluten.soft-affinity.replications.num | 🔄 Dynamic | 2 | Calculate the number of the replications for scheduling to the target executors per file | +| spark.gluten.sql.adaptive.costEvaluator.enabled | ⚓ Static | true | If true, use org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator as custom cost evaluator class, else follow the configuration spark.sql.adaptive.customCostEvaluatorClass. | +| spark.gluten.sql.ansiFallback.enabled | 🔄 Dynamic | true | When true (default), Gluten will fall back to Spark when ANSI mode is enabled. When false, Gluten will attempt to execute in ANSI mode. | +| spark.gluten.sql.cacheWholeStageTransformerContext | 🔄 Dynamic | false | When true, `WholeStageTransformer` will cache the `WholeStageTransformerContext` when executing. It is used to get substrait plan node and native plan string. | +| spark.gluten.sql.collapseGetJsonObject.enabled | 🔄 Dynamic | false | Collapse nested get_json_object functions as one for optimization. | +| spark.gluten.sql.columnar.appendData | 🔄 Dynamic | true | Enable or disable columnar v2 command append data. | +| spark.gluten.sql.columnar.arrowUdf | 🔄 Dynamic | true | Enable or disable columnar arrow udf. | +| spark.gluten.sql.columnar.batchscan | 🔄 Dynamic | true | Enable or disable columnar batchscan. | +| spark.gluten.sql.columnar.batchscan.maxInputPartitions | 🔄 Dynamic | 2147483647 | Maximum number of Spark task partitions for supported DataSource V2 batch scans. | +| spark.gluten.sql.columnar.broadcastExchange | 🔄 Dynamic | true | Enable or disable columnar broadcastExchange. | +| spark.gluten.sql.columnar.broadcastJoin | 🔄 Dynamic | true | Enable or disable columnar broadcastJoin. | +| spark.gluten.sql.columnar.broadcastNestedLoopJoin.enabled | 🔄 Dynamic | true | Enable or disable columnar broadcastNestedLoopJoin. | +| spark.gluten.sql.columnar.cartesianProduct.enabled | 🔄 Dynamic | true | Enable or disable columnar cartesianProduct. | +| spark.gluten.sql.columnar.cast.avg | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.coalesce | 🔄 Dynamic | true | Enable or disable columnar coalesce. | +| spark.gluten.sql.columnar.collectLimit | 🔄 Dynamic | true | Enable or disable columnar collectLimit. | +| spark.gluten.sql.columnar.collectTail | 🔄 Dynamic | true | Enable or disable columnar collectTail. | +| spark.gluten.sql.columnar.delta.deletionVector.deferPayloadRead.enabled | 🔄 Dynamic | true | When true, defer loading on-disk Delta deletion vector payloads until the native split is serialized on an executor. This removes remote deletion vector I/O from driver planning. Inline deletion vectors remain eagerly decoded because their bytes are already present in Delta metadata. | +| spark.gluten.sql.columnar.delta.deletionVector.nativeRangeRead.enabled | 🔄 Dynamic | false | When true, pass each on-disk Delta deletion vector's absolute path, offset, and size to Velox instead of materializing its payload in the JVM. Velox loads the range through its file-handle and buffered-input path during split preparation. This takes precedence over deferPayloadRead.enabled. Inline deletion vectors remain JVM-decoded. | +| spark.gluten.sql.columnar.enableNestedColumnPruningInHiveTableScan | 🔄 Dynamic | true | Enable or disable nested column pruning in hivetablescan. | +| spark.gluten.sql.columnar.enableVanillaVectorizedReaders | ⚓ Static | true | Enable or disable vanilla vectorized scan. | +| spark.gluten.sql.columnar.executor.libpath | 🔄 Dynamic || The gluten executor library path. | +| spark.gluten.sql.columnar.expand | 🔄 Dynamic | true | Enable or disable columnar expand. | +| spark.gluten.sql.columnar.fallback.expressions.threshold | 🔄 Dynamic | 50 | Fall back filter/project if number of nested expressions reaches this threshold, considering Spark codegen can bring better performance for such case. | +| spark.gluten.sql.columnar.fallback.ignoreRowToColumnar | 🔄 Dynamic | true | When true, the fallback policy ignores the RowToColumnar when counting fallback number. | +| spark.gluten.sql.columnar.fallback.preferColumnar | 🔄 Dynamic | true | When true, the fallback policy prefers to use Gluten plan rather than vanilla Spark plan if the both of them contains ColumnarToRow and the vanilla Spark plan ColumnarToRow number is not smaller than Gluten plan. | +| spark.gluten.sql.columnar.filescan | 🔄 Dynamic | true | Enable or disable columnar filescan. | +| spark.gluten.sql.columnar.filter | 🔄 Dynamic | true | Enable or disable columnar filter. | +| spark.gluten.sql.columnar.force.hashagg | 🔄 Dynamic | true | Whether to force to use gluten's hash agg for replacing vanilla spark's sort agg. | +| spark.gluten.sql.columnar.forceShuffledHashJoin | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.generate | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.hashagg | 🔄 Dynamic | true | Enable or disable columnar hashagg. | +| spark.gluten.sql.columnar.hivetablescan | 🔄 Dynamic | true | Enable or disable columnar hivetablescan. | +| spark.gluten.sql.columnar.libname | 🔄 Dynamic | gluten | The gluten library name. | +| spark.gluten.sql.columnar.libpath | 🔄 Dynamic || The gluten library path. | +| spark.gluten.sql.columnar.limit | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.localTableScan | 🔄 Dynamic | false | Enable or disable native columnar execution of LocalTableScanExec. When true, Gluten attempts to replace LocalTableScanExec (a driver-side local collection) with a backend transformer that converts the rows into columnar batches natively. | +| spark.gluten.sql.columnar.maxBatchSize | 🔄 Dynamic | 4096 | +| spark.gluten.sql.columnar.overwriteByExpression | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite by expression. | +| spark.gluten.sql.columnar.overwritePartitionsDynamic | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite partitions dynamic. | +| spark.gluten.sql.columnar.parquet.write.blockSize | 🔄 Dynamic | 128MB | +| spark.gluten.sql.columnar.partial.generate | 🔄 Dynamic | true | Evaluates the non-offload-able HiveUDTF using vanilla Spark generator | +| spark.gluten.sql.columnar.partial.project | 🔄 Dynamic | true | Break up one project node into 2 phases when some of the expressions are non offload-able. Phase one is a regular offloaded project transformer that evaluates the offload-able expressions in native, phase two preserves the output from phase one and evaluates the remaining non-offload-able expressions using vanilla Spark projections | +| spark.gluten.sql.columnar.physicalJoinOptimizationLevel | 🔄 Dynamic | 12 | Fallback to row operators if there are several continuous joins. | +| spark.gluten.sql.columnar.physicalJoinOptimizationOutputSize | 🔄 Dynamic | 52 | Fallback to row operators if there are several continuous joins and matched output size. | +| spark.gluten.sql.columnar.physicalJoinOptimizeEnable | 🔄 Dynamic | false | Enable or disable columnar physicalJoinOptimize. | +| spark.gluten.sql.columnar.preferStreamingAggregate | 🔄 Dynamic | true | Velox backend supports `StreamingAggregate`. `StreamingAggregate` uses the less memory as it does not need to hold all groups in memory, so it could avoid spill. When true and the child output ordering satisfies the grouping key then Gluten will choose `StreamingAggregate` as the native operator. | +| spark.gluten.sql.columnar.project | 🔄 Dynamic | true | Enable or disable columnar project. | +| spark.gluten.sql.columnar.project.collapse | 🔄 Dynamic | true | Combines two columnar project operators into one and perform alias substitution | +| spark.gluten.sql.columnar.query.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether query will fall back by counting the number of ColumnarToRow & vanilla leaf node. | +| spark.gluten.sql.columnar.range | 🔄 Dynamic | true | Enable or disable columnar range. | +| spark.gluten.sql.columnar.replaceData | 🔄 Dynamic | true | Enable or disable columnar v2 command replace data. | +| spark.gluten.sql.columnar.scanOnly | 🔄 Dynamic | false | When enabled, only scan and the filter after scan will be offloaded to native. | +| spark.gluten.sql.columnar.shuffle | 🔄 Dynamic | true | Enable or disable columnar shuffle. | +| spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled | ⚓ Static | true | If enabled, fall back to ColumnarShuffleManager when celeborn service is unavailable.Otherwise, throw an exception. | +| spark.gluten.sql.columnar.shuffle.celeborn.useRssSort | 🔄 Dynamic | true | If true, use RSS sort implementation for Celeborn sort-based shuffle.If false, use Gluten's row-based sort implementation. Only valid when `spark.celeborn.client.spark.shuffle.writer` is set to `sort`. | +| spark.gluten.sql.columnar.shuffle.codec | 🔄 Dynamic | <undefined> | By default, the supported codecs are lz4 and zstd. When spark.gluten.sql.columnar.shuffle.codecBackend=qat,the supported codecs are gzip and zstd. | +| spark.gluten.sql.columnar.shuffle.codecBackend | 🔄 Dynamic | <undefined> | +| spark.gluten.sql.columnar.shuffle.compression.threshold | 🔄 Dynamic | 100 | If number of rows in a batch falls below this threshold, will copy all buffers into one buffer to compress. | +| spark.gluten.sql.columnar.shuffle.dictionary.enabled | 🔄 Dynamic | false | Enable dictionary in hash-based shuffle. | +| spark.gluten.sql.columnar.shuffle.merge.threshold | 🔄 Dynamic | 0.25 | +| spark.gluten.sql.columnar.shuffle.partitionBufferEvictThreshold | 🔄 Dynamic | -1 | For Velox hash shuffle writer, evict partition buffers larger than this threshold after splitting an input batch. Use non-positive value to disable this feature. | +| spark.gluten.sql.columnar.shuffle.readerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for shuffle reader reading input stream from local or remote. | +| spark.gluten.sql.columnar.shuffle.realloc.threshold | 🔄 Dynamic | 0.25 | +| spark.gluten.sql.columnar.shuffle.sort.columns.threshold | 🔄 Dynamic | 100000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of columns is greater than this threshold. | +| spark.gluten.sql.columnar.shuffle.sort.deserializerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for sort-based shuffle reader deserializing raw input to columnar batch. | +| spark.gluten.sql.columnar.shuffle.sort.partitions.threshold | 🔄 Dynamic | 4000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of partitions is greater than this threshold. | +| spark.gluten.sql.columnar.shuffle.typeAwareCompress.enabled | 🔄 Dynamic | false | Enable type-aware compression (e.g. FFor for 64-bit integers) in shuffle. Not compatible with dictionary encoding; if both are enabled, type-aware compression is automatically disabled. | +| spark.gluten.sql.columnar.shuffledHashJoin | 🔄 Dynamic | true | Enable or disable columnar shuffledHashJoin. | +| spark.gluten.sql.columnar.shuffledHashJoin.optimizeBuildSide | 🔄 Dynamic | true | Whether to allow Gluten to choose an optimal build side for shuffled hash join. | +| spark.gluten.sql.columnar.smallFileThreshold | 🔄 Dynamic | 0.5 | The total size threshold of small files in table scan.To avoid small files being placed into the same partition, Gluten will try to distribute small files into different partitions when the total size of small files is below this threshold. | +| spark.gluten.sql.columnar.sort | 🔄 Dynamic | true | Enable or disable columnar sort. | +| spark.gluten.sql.columnar.sortMergeJoin | 🔄 Dynamic | true | Enable or disable columnar sortMergeJoin. This should be set with preferSortMergeJoin=false. | +| spark.gluten.sql.columnar.tableCache | ⚓ Static | true | Enable or disable columnar table cache. | +| spark.gluten.sql.columnar.tableCache.partitionStats.enabled | 🔄 Dynamic | false | When true, the Velox columnar cache serializer computes per-partition min/max/null/row-count stats and embeds them in the cached payload so that the Spark optimizer can prune whole partitions on equality / range predicates. When false (default), the serializer still writes the V3 per-column payload with empty stats so projected cache reads can lazily materialize only requested columns, while partition pruning is disabled. | +| spark.gluten.sql.columnar.takeOrderedAndProject | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.union | 🔄 Dynamic | true | Enable or disable columnar union. | +| spark.gluten.sql.columnar.wholeStage.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether whole stage will fall back in AQE supported case by counting the number of ColumnarToRow & vanilla leaf node. | +| spark.gluten.sql.columnar.window | 🔄 Dynamic | true | Enable or disable columnar window. | +| spark.gluten.sql.columnar.window.group.limit | 🔄 Dynamic | true | Enable or disable columnar window group limit. | +| spark.gluten.sql.columnar.writeToDataSourceV2 | 🔄 Dynamic | true | Enable or disable columnar v2 command write to data source v2. | +| spark.gluten.sql.columnarSampleEnabled | 🔄 Dynamic | false | Disable or enable columnar sample. | +| spark.gluten.sql.columnarToRowMemoryThreshold | 🔄 Dynamic | 64MB | +| spark.gluten.sql.countDistinctWithoutExpand | 🔄 Dynamic | false | Convert Count Distinct to a UDAF called count_distinct to prevent SparkPlanner converting it to Expand+Count. WARNING: When enabled, count distinct queries will fail to fallback!!! | +| spark.gluten.sql.extendedColumnPruning.enabled | 🔄 Dynamic | true | Do extended nested column pruning for cases ignored by vanilla Spark. | +| spark.gluten.sql.fallbackRegexpExpressions | 🔄 Dynamic | false | If true, fall back all regexp expressions. There are a few incompatible cases between RE2 (used by native engine) and java.util.regex (used by Spark). User should enable this property if their incompatibility is intolerable. | +| spark.gluten.sql.fallbackUnexpectedMetadataParquet | 🔄 Dynamic | false | If enabled, Gluten will not offload scan when unexpected metadata is detected. | +| spark.gluten.sql.fallbackUnexpectedMetadataParquet.limit | 🔄 Dynamic | 10 | If supplied, metadata of `limit` number of Parquet files will be checked to determine whether to fall back to java scan. | +| spark.gluten.sql.fallbackUnexpectedMetadataParquet.samplePercentage | 🔄 Dynamic | 0.1 | The percentage of root paths to sample for metadata validation when the number of root paths is large. Value range is (0, 1.0]. 1.0 means check all paths (no sampling). A smaller value reduces validation cost for tables with many partitions. | +| spark.gluten.sql.injectNativePlanStringToExplain | 🔄 Dynamic | false | When true, Gluten will inject native plan tree to Spark's explain output. | +| spark.gluten.sql.mergeTwoPhasesAggregate.enabled | 🔄 Dynamic | true | Whether to merge two phases aggregate if there are no other operators between them. | +| spark.gluten.sql.native.bloomFilter | 🔄 Dynamic | true | +| spark.gluten.sql.native.hive.writer.enabled | 🔄 Dynamic | true | This is config to specify whether to enable the native columnar writer for HiveFileFormat. Currently only supports HiveFileFormat with Parquet as the output file type. | +| spark.gluten.sql.native.hyperLogLog.Aggregate | 🔄 Dynamic | true | +| spark.gluten.sql.native.parquet.write.blockRows | 🔄 Dynamic | 100000000 | +| spark.gluten.sql.native.union | 🔄 Dynamic | false | Enable or disable native union where computation is completely offloaded to backend. | +| spark.gluten.sql.native.writeColumnMetadataExclusionList | 🔄 Dynamic | comment | Native write files does not support column metadata. Metadata in list would be removed to support native write files. Multiple values separated by commas. | +| spark.gluten.sql.native.writer.enabled | 🔄 Dynamic | <undefined> | This is config to specify whether to enable the native columnar parquet/orc writer | +| spark.gluten.sql.orc.charType.scan.fallback.enabled | 🔄 Dynamic | true | Force fallback for orc char type scan. | +| spark.gluten.sql.pushAggregateThroughJoin.enabled | 🔄 Dynamic | false | Enables the push-aggregate-through-join optimization in Gluten. When enabled, aggregate operators may be pushed below joins during logical optimization and corresponding physical plans may be rewritten to execute the aggregation earlier. | +| spark.gluten.sql.pushAggregateThroughJoin.maxDepth | 🔄 Dynamic | 2147483647 | Maximum join traversal depth when applying the push-aggregate-through-join optimization. A value of 1 allows pushing an aggregate through a single join; larger values allow the rule to traverse and push through multiple consecutive joins. | +| spark.gluten.sql.removeNativeWriteFilesSortAndProject | 🔄 Dynamic | true | When true, Gluten will remove the vanilla Spark V1Writes added sort and project for velox backend. | +| spark.gluten.sql.rewrite.dateTimestampComparison | 🔄 Dynamic | true | Rewrite the comparision between date and timestamp to timestamp comparison.For example `from_unixtime(ts) > date` will be rewritten to `ts > to_unixtime(date)` | +| spark.gluten.sql.scan.detailedMetrics.enabled | 🔄 Dynamic | true | When true (default), Velox backend scan operators register all detailed SQL metrics. When false, only essential scan metrics are registered to reduce driver memory usage. Also enabled automatically when spark.gluten.sql.debug is true. Does not affect the ClickHouse backend. | +| spark.gluten.sql.scan.fileSchemeValidation.enabled | 🔄 Dynamic | true | When true, enable file path scheme validation for scan. Validation will fail if file scheme is not supported by registered file systems, which will cause scan operator fall back. | +| spark.gluten.sql.supported.flattenNestedFunctions | 🔄 Dynamic | and,or | Flatten nested functions as one for optimization. | +| spark.gluten.sql.text.input.empty.as.default | 🔄 Dynamic | false | treat empty fields in CSV input as default values. | +| spark.gluten.sql.text.input.max.block.size | 🔄 Dynamic | 8KB | the max block size for text input rows | +| spark.gluten.sql.validation.printStackOnFailure | 🔄 Dynamic | false | +| spark.gluten.storage.hdfsViewfs.enabled | ⚓ Static | false | If enabled, gluten will convert the viewfs path to hdfs path in scala side | +| spark.gluten.supported.hive.udfs | 🔄 Dynamic || Supported hive udf names. | +| spark.gluten.supported.python.udfs | 🔄 Dynamic || Supported python udf names. | +| spark.gluten.supported.scala.udfs | 🔄 Dynamic || Supported scala udf names. | +| spark.gluten.ui.enabled | ⚓ Static | true | Whether to enable the gluten web UI, If true, attach the gluten UI page to the Spark web UI. | ## Gluten *experimental* configurations 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 af8e8df7da8..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 @@ -20,10 +20,22 @@ import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions import org.apache.spark.sql.execution.datasources.PartitionedFile +import org.apache.hadoop.fs.Path + import java.util.{Map => JMap} /** Reading deletion vectors natively requires Delta 3.3+, so there is nothing to materialize. */ object DeltaDeletionVectorScanInfo { - def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) + 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 af8e8df7da8..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 @@ -20,10 +20,22 @@ import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions import org.apache.spark.sql.execution.datasources.PartitionedFile +import org.apache.hadoop.fs.Path + import java.util.{Map => JMap} /** Reading deletion vectors natively requires Delta 3.3+, so there is nothing to materialize. */ object DeltaDeletionVectorScanInfo { - def normalize(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) + 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 812f9b9ec36..e0712540924 100644 --- a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -16,9 +16,10 @@ */ package org.apache.gluten.delta +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.rel.DeltaLocalFilesNode -import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions +import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayload, DeltaFileReadOptions, NativeDeletionVectorDescriptor, SerializedDeletionVectorPayload} import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat @@ -26,6 +27,7 @@ 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 @@ -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,24 +70,40 @@ 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. * - * 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. + * `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(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) + 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() - // 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 serializableHadoopConf = + if ( + GlutenConfig.get.deferDeltaDeletionVectorPayloadRead && + !GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + Some(new SerializableConfiguration(hadoopConf)) + } else { + None + } val scanInfos = partitionFiles.map { - file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath) + file => extract(file, hadoopConf, serializableHadoopConf, tablePath, readMetrics) } if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) { Some( @@ -96,30 +118,53 @@ 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 = { + file: PartitionedFile, + tablePath: Path): PartitionFileScanInfo = { val hadoopConf = spark.sessionState.newHadoopConf() - val tablePath = resolveTablePath(hadoopConf, partitionColumnCount, file) - extract(partitionColumnCount, file, hadoopConf, tablePath) + val serializableHadoopConf = + if ( + GlutenConfig.get.deferDeltaDeletionVectorPayloadRead && + !GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + Some(new SerializableConfiguration(hadoopConf)) + } else { + None + } + extract(file, hadoopConf, serializableHadoopConf, tablePath, None) } private def extract( - partitionColumnCount: Int, file: PartitionedFile, hadoopConf: Configuration, - tablePath: Path): PartitionFileScanInfo = { + serializableHadoopConf: Option[SerializableConfiguration], + tablePath: Path, + readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo = { val metadata = otherMetadataColumns(file) val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) - val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath) + val dvInfo = extractDeletionVectorInfo( + metadata, + hadoopConf, + serializableHadoopConf, + tablePath, + readMetrics) PartitionFileScanInfo(normalizedMetadata, dvInfo) } private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo): DeltaFileReadOptions = { - new DeltaFileReadOptions( - toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), - dvInfo.hasDeletionVector, - dvInfo.cardinality, - dvInfo.serializedDeletionVector) + dvInfo.deletionVectorPayload match { + case descriptor: NativeDeletionVectorDescriptor => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + descriptor) + case payload => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + payload) + } } private def toSubstraitRowIndexFilterType( @@ -143,21 +188,32 @@ object DeltaDeletionVectorScanInfo { private def extractDeletionVectorInfo( metadata: Map[String, Object], hadoopConf: Configuration, - tablePath: Path): DeletionVectorInfo = { + serializableHadoopConf: Option[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") @@ -193,6 +249,48 @@ object DeltaDeletionVectorScanInfo { } } + /** Selects a deferred source for on-disk DVs and eager bytes for inline or rollback mode. */ + private def deletionVectorPayload( + hadoopConf: Configuration, + serializableHadoopConf: Option[SerializableConfiguration], + tablePath: Path, + descriptor: DeletionVectorDescriptor, + readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorPayload = { + if (tablePath == null) { + throw new IllegalStateException( + "Unable to resolve Delta table path while preparing deletion vector payload") + } + if ( + descriptor.storageType != "i" && + GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + val dvPath = descriptor.absolutePath(tablePath) + new NativeDeletionVectorDescriptor( + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes.toLong) + } else if (descriptor.storageType != "i" && serializableHadoopConf.isDefined) { + val dvPath = descriptor.absolutePath(tablePath) + new OnDiskDeletionVectorPayload( + serializableHadoopConf.get, + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes, + readMetrics) + } else { + new SerializedDeletionVectorPayload(serializePayload(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 @@ -235,73 +333,71 @@ object DeltaDeletionVectorScanInfo { tablePath: Path, descriptor: DeletionVectorDescriptor): Array[Byte] = { val dvPath = descriptor.absolutePath(tablePath) + readRawDvBytes( + hadoopConf, + dvPath, + requiredOffset(descriptor), + descriptor.sizeInBytes) + } + + private def readRawDvBytes( + hadoopConf: Configuration, + dvPath: Path, + offset: Long, + sizeInBytes: Int): Array[Byte] = { val fs = dvPath.getFileSystem(hadoopConf) - val stream = new DataInputStream(fs.open(dvPath)) + val fileStream = fs.open(dvPath) try { - val offset = descriptor.offset.getOrElse(0) - if (offset > 0) { - stream.skipBytes(offset) - } - DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes) + fileStream.seek(offset) + DeletionVectorStore.readRangeFromStream(new DataInputStream(fileStream), sizeInBytes) } finally { - stream.close() + fileStream.close() } } - private def resolveTablePath( - hadoopConf: org.apache.hadoop.conf.Configuration, - partitionColumnCount: Int, - file: PartitionedFile): Path = { - val fileParent = new Path(unescapePathName(file.filePath.toString)).getParent - var tablePath = fileParent - for (_ <- 0 until partitionColumnCount) { - tablePath = tablePath.getParent - } - if (tablePath != null && isDeltaTablePath(hadoopConf, tablePath)) { - return tablePath - } - - var candidate = fileParent - while (candidate != null && !isDeltaTablePath(hadoopConf, candidate)) { - candidate = candidate.getParent - } - if (candidate != null) candidate else tablePath - } + /** + * 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") - private def isDeltaTablePath( - hadoopConf: org.apache.hadoop.conf.Configuration, - tablePath: Path): Boolean = { - val deltaLogPath = new Path(tablePath, "_delta_log") - try { - deltaLogPath.getFileSystem(hadoopConf).exists(deltaLogPath) - } catch { - case NonFatal(_) => false - } - } + @transient @volatile private var cachedPayload: Array[Byte] = _ - private def unescapePathName(path: String): String = { - if (path == null || path.indexOf('%') < 0) { - path - } else { - val builder = new StringBuilder(path.length) - var index = 0 - while (index < path.length) { - if (path.charAt(index) == '%' && index + 2 < path.length) { - val high = Character.digit(path.charAt(index + 1), 16) - val low = Character.digit(path.charAt(index + 2), 16) - if (high >= 0 && low >= 0) { - builder.append(((high << 4) | low).toChar) - index += 3 - } else { - builder.append(path.charAt(index)) - index += 1 + override def materialize(): Array[Byte] = { + var payload = cachedPayload + if (payload == null) { + this.synchronized { + payload = cachedPayload + if (payload == null) { + val startedAt = System.nanoTime() + 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)) + } } - } else { - builder.append(path.charAt(index)) - index += 1 } } - builder.toString() + 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 22ffb3c89a8..9f741944067 100644 --- a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala +++ b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala @@ -16,9 +16,10 @@ */ package org.apache.gluten.delta +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.rel.DeltaLocalFilesNode -import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions +import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayload, DeltaFileReadOptions, NativeDeletionVectorDescriptor, SerializedDeletionVectorPayload} import org.apache.spark.sql.SparkSession import org.apache.spark.sql.delta.DeltaParquetFileFormat @@ -26,6 +27,7 @@ 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 @@ -49,7 +51,11 @@ object DeltaDeletionVectorScanInfo { hasDeletionVector: Boolean, rowIndexFilterType: RowIndexFilterType, cardinality: Long, - serializedDeletionVector: Array[Byte]) + deletionVectorPayload: DeletionVectorPayload) { + def serializedDeletionVector: Array[Byte] = deletionVectorPayload.materialize() + + def isPayloadMaterialized: Boolean = deletionVectorPayload.isMaterialized() + } final case class PartitionFileScanInfo( normalizedOtherMetadataColumns: Map[String, Object], @@ -65,21 +71,40 @@ 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. * - * 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. + * `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(partitionColumnCount: Int, partitionFiles: Seq[PartitionedFile]) + 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 cachedTablePath = resolveTablePath(hadoopConf, partitionColumnCount, partitionFiles.head) + val serializableHadoopConf = + if ( + GlutenConfig.get.deferDeltaDeletionVectorPayloadRead && + !GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + Some(new SerializableConfiguration(hadoopConf)) + } else { + None + } val scanInfos = partitionFiles.map { - file => extract(partitionColumnCount, file, hadoopConf, cachedTablePath) + file => extract(file, hadoopConf, serializableHadoopConf, tablePath, readMetrics) } if (scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) { Some( @@ -94,30 +119,53 @@ 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 = { + file: PartitionedFile, + tablePath: Path): PartitionFileScanInfo = { val hadoopConf = spark.sessionState.newHadoopConf() - val tablePath = resolveTablePath(hadoopConf, partitionColumnCount, file) - extract(partitionColumnCount, file, hadoopConf, tablePath) + val serializableHadoopConf = + if ( + GlutenConfig.get.deferDeltaDeletionVectorPayloadRead && + !GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + Some(new SerializableConfiguration(hadoopConf)) + } else { + None + } + extract(file, hadoopConf, serializableHadoopConf, tablePath, None) } private def extract( - partitionColumnCount: Int, file: PartitionedFile, hadoopConf: Configuration, - tablePath: Path): PartitionFileScanInfo = { + serializableHadoopConf: Option[SerializableConfiguration], + tablePath: Path, + readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo = { val metadata = otherMetadataColumns(file) val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded, RowIndexFilterTypeKey) - val dvInfo = extractDeletionVectorInfo(metadata, hadoopConf, tablePath) + val dvInfo = extractDeletionVectorInfo( + metadata, + hadoopConf, + serializableHadoopConf, + tablePath, + readMetrics) PartitionFileScanInfo(normalizedMetadata, dvInfo) } private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo): DeltaFileReadOptions = { - new DeltaFileReadOptions( - toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), - dvInfo.hasDeletionVector, - dvInfo.cardinality, - dvInfo.serializedDeletionVector) + dvInfo.deletionVectorPayload match { + case descriptor: NativeDeletionVectorDescriptor => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + descriptor) + case payload => + new DeltaFileReadOptions( + toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType), + dvInfo.hasDeletionVector, + dvInfo.cardinality, + payload) + } } private def toSubstraitRowIndexFilterType( @@ -141,21 +189,32 @@ object DeltaDeletionVectorScanInfo { private def extractDeletionVectorInfo( metadata: Map[String, Object], hadoopConf: Configuration, - tablePath: Path): DeletionVectorInfo = { + serializableHadoopConf: Option[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") @@ -213,6 +272,48 @@ object DeltaDeletionVectorScanInfo { } } + /** Selects a deferred source for on-disk DVs and eager bytes for inline or rollback mode. */ + private def deletionVectorPayload( + hadoopConf: Configuration, + serializableHadoopConf: Option[SerializableConfiguration], + tablePath: Path, + descriptor: DeletionVectorDescriptor, + readMetrics: Option[DeletionVectorReadMetrics]): DeletionVectorPayload = { + if (tablePath == null) { + throw new IllegalStateException( + "Unable to resolve Delta table path while preparing deletion vector payload") + } + if ( + descriptor.storageType != "i" && + GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead + ) { + val dvPath = descriptor.absolutePath(tablePath) + new NativeDeletionVectorDescriptor( + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes.toLong) + } else if (descriptor.storageType != "i" && serializableHadoopConf.isDefined) { + val dvPath = descriptor.absolutePath(tablePath) + new OnDiskDeletionVectorPayload( + serializableHadoopConf.get, + dvPath.toString, + requiredOffset(descriptor), + descriptor.sizeInBytes, + readMetrics) + } else { + new SerializedDeletionVectorPayload(serializePayload(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") + } + } + private def serializePayload( hadoopConf: Configuration, tablePath: Path, @@ -239,73 +340,71 @@ object DeltaDeletionVectorScanInfo { tablePath: Path, descriptor: DeletionVectorDescriptor): Array[Byte] = { val dvPath = descriptor.absolutePath(tablePath) + readRawDvBytes( + hadoopConf, + dvPath, + requiredOffset(descriptor), + descriptor.sizeInBytes) + } + + private def readRawDvBytes( + hadoopConf: Configuration, + dvPath: Path, + offset: Long, + sizeInBytes: Int): Array[Byte] = { val fs = dvPath.getFileSystem(hadoopConf) - val stream = new DataInputStream(fs.open(dvPath)) + val fileStream = fs.open(dvPath) try { - val offset = descriptor.offset.getOrElse(0) - if (offset > 0) { - stream.skipBytes(offset) - } - DeletionVectorStore.readRangeFromStream(stream, descriptor.sizeInBytes) + fileStream.seek(offset) + DeletionVectorStore.readRangeFromStream(new DataInputStream(fileStream), sizeInBytes) } finally { - stream.close() + fileStream.close() } } - private def resolveTablePath( - hadoopConf: org.apache.hadoop.conf.Configuration, - partitionColumnCount: Int, - file: PartitionedFile): Path = { - val fileParent = new Path(unescapePathName(file.filePath.toString)).getParent - var tablePath = fileParent - for (_ <- 0 until partitionColumnCount) { - tablePath = tablePath.getParent - } - if (tablePath != null && isDeltaTablePath(hadoopConf, tablePath)) { - return tablePath - } + /** + * 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") - var candidate = fileParent - while (candidate != null && !isDeltaTablePath(hadoopConf, candidate)) { - candidate = candidate.getParent - } - if (candidate != null) candidate else tablePath - } + @transient @volatile private var cachedPayload: Array[Byte] = _ - private def isDeltaTablePath( - hadoopConf: org.apache.hadoop.conf.Configuration, - tablePath: Path): Boolean = { - val deltaLogPath = new Path(tablePath, "_delta_log") - try { - deltaLogPath.getFileSystem(hadoopConf).exists(deltaLogPath) - } catch { - case NonFatal(_) => false - } - } - - private def unescapePathName(path: String): String = { - if (path == null || path.indexOf('%') < 0) { - path - } else { - val builder = new StringBuilder(path.length) - var index = 0 - while (index < path.length) { - if (path.charAt(index) == '%' && index + 2 < path.length) { - val high = Character.digit(path.charAt(index + 1), 16) - val low = Character.digit(path.charAt(index + 2), 16) - if (high >= 0 && low >= 0) { - builder.append(((high << 4) | low).toChar) - index += 3 - } else { - builder.append(path.charAt(index)) - index += 1 + override def materialize(): Array[Byte] = { + var payload = cachedPayload + if (payload == null) { + this.synchronized { + payload = cachedPayload + if (payload == null) { + val startedAt = System.nanoTime() + 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)) + } } - } else { - builder.append(path.charAt(index)) - index += 1 } } - builder.toString() + payload } + + override def isMaterialized(): Boolean = cachedPayload != null } + } 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..153a36fc6bc --- /dev/null +++ b/gluten-delta/src/main/scala/org/apache/gluten/delta/DeletionVectorReadMetrics.scala @@ -0,0 +1,25 @@ +/* + * 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.sql.execution.metric.SQLMetric + +/** Metrics updated while an executor materializes an on-disk deletion-vector payload. */ +final case class DeletionVectorReadMetrics( + readTimeNanos: SQLMetric, + readBytes: SQLMetric, + readAttempts: SQLMetric) 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 c2cb6604db8..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 @@ -27,9 +27,10 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, import org.apache.spark.sql.catalyst.plans.QueryPlan import org.apache.spark.sql.connector.read.streaming.SparkDataStream import org.apache.spark.sql.delta.{DeltaParquetFileFormat, NoMapping} -import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeRemoveFileIndex} +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 @@ -120,20 +142,40 @@ case class DeltaScanTransformer( override def getSplitInfosFromPartitions( partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = { val splitInfos = super.getSplitInfosFromPartitions(partitions) - val partitionColumnCount = getPartitionSchema.fields.length - splitInfos.zip(partitions).map { - case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) => - DeltaDeletionVectorScanInfo - .normalize(partitionColumnCount, filePartition.files.toSeq) - .map { - case (otherMetadataColumns, deltaReadOptions) => - DeltaLocalFilesBuilder.makeDeltaLocalFiles( - localFiles, - otherMetadataColumns.asJava, - deltaReadOptions.asJava): SplitInfo - } - .getOrElse(localFiles) - case (splitInfo, _) => splitInfo + // Deletion vectors only exist on Delta tables read through a TahoeFileIndex (which also covers + // PreparedDeltaFileIndex). Its `path` is the authoritative table root and is used to resolve + // per-file DV locations. Any other location cannot carry Delta DV metadata, so the generic + // split representation is returned unchanged. + relation.location match { + case tahoe: TahoeFileIndex => + val tableRootPath = tahoe.path + splitInfos.zip(partitions).map { + case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) => + 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, + deltaReadOptions.asJava): SplitInfo + } + .getOrElse(localFiles) + case (splitInfo, _) => splitInfo + } + case _ => + splitInfos } } 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 138d318cffd..284f9c8ecd6 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 @@ -20,9 +20,12 @@ import org.apache.gluten.extension.DeltaPostTransformRules import org.apache.spark.SparkConf import org.apache.spark.sql.{DataFrame, Row} +import org.apache.spark.sql.delta.DeltaLog import org.apache.spark.sql.types._ import org.apache.spark.util.SparkVersionUtil +import org.apache.hadoop.fs.Path + import scala.collection.JavaConverters._ abstract class DeltaSuite extends WholeStageTransformerSuite { @@ -628,6 +631,41 @@ abstract class DeltaSuite extends WholeStageTransformerSuite { } } + testWithMinSparkVersion("deletion vector on partitioned table", "3.4") { + withTempPath { + p => + import testImplicits._ + val path = p.getCanonicalPath + // Partitioned so data files live under partition subdirs (region=.../...). The DV path is + // resolved from the table root (TahoeFileIndex.path) regardless of partition nesting; this + // guards the removal of the old partition-count-based table-path walk-up. + val data = + Seq((1, "a"), (2, "a"), (3, "b"), (4, "b"), (5, "a"), (6, "b")).toDF("id", "region") + data.write.format("delta").partitionBy("region").save(path) + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)") + spark.sql(s"DELETE FROM delta.`$path` WHERE id IN (2, 3, 6)") + val deletionVectors = DeltaLog + .forTable(spark, new Path(path)) + .update() + .allFiles + .collect() + .flatMap(file => Option(file.deletionVector)) + assert(deletionVectors.nonEmpty, "DELETE should produce deletion vectors") + assert( + deletionVectors.exists(_.storageType == "u"), + "DELETE should produce a table-root-relative UUID deletion vector") + val df = spark.read.format("delta").load(path) + if (SparkVersionUtil.gteSpark35) { + assert( + df.queryExecution.executedPlan + .collect { case _: DeltaScanTransformer => true } + .nonEmpty) + } + checkAnswer(df, Seq((1, "a"), (4, "b"), (5, "a")).toDF("id", "region")) + } + } + test("delta: push down input_file_name expression") { withTable("source_table") { withTable("target_table") { 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..202f1ffbadc 100644 --- a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java +++ b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java @@ -51,10 +51,18 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui .setHasDeletionVector(options.hasDeletionVector()); if (options.hasDeletionVector()) { - deltaBuilder - .setDeletionVectorCardinality(options.deletionVectorCardinality()) - .setSerializedDeletionVector( - UnsafeByteOperations.unsafeWrap(options.serializedDeletionVector())); + deltaBuilder.setDeletionVectorCardinality(options.deletionVectorCardinality()); + if (options.hasNativeDeletionVectorDescriptor()) { + NativeDeletionVectorDescriptor descriptor = options.nativeDeletionVectorDescriptor(); + deltaBuilder.setDeletionVectorDescriptor( + ReadRel.LocalFiles.FileOrFiles.DeltaReadOptions.DeletionVectorDescriptor.newBuilder() + .setAbsolutePath(descriptor.absolutePath()) + .setOffset(descriptor.offset()) + .setPayloadSize(descriptor.payloadSize())); + } else { + deltaBuilder.setSerializedDeletionVector( + UnsafeByteOperations.unsafeWrap(options.serializedDeletionVector())); + } } fileBuilder.setDelta(deltaBuilder.build()); @@ -79,24 +87,147 @@ public enum RowIndexFilterType { IF_NOT_CONTAINED } + /** + * Serializable source for a deletion-vector payload. + * + *

The source travels inside a Spark input partition. Implementations may therefore defer + * remote I/O until {@link #materialize()} is called while the split is converted to protobuf on + * an executor. The returned byte array must not be modified: protobuf wraps it without copying. + */ + public interface DeletionVectorPayload extends Serializable { + byte[] materialize(); + + /** Returns whether the payload bytes are already resident in this object. */ + boolean isMaterialized(); + } + + /** A payload source for inline DVs and the eager compatibility path. */ + 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; + } + + @Override + public byte[] materialize() { + return payload; + } + + @Override + public boolean isMaterialized() { + return true; + } + } + + /** Immutable executor-native source for an on-disk deletion vector. */ + public static final class NativeDeletionVectorDescriptor implements DeletionVectorPayload { + private static final long serialVersionUID = 1L; + + private final String absolutePath; + private final long offset; + private final long payloadSize; + + public NativeDeletionVectorDescriptor(String absolutePath, long offset, long payloadSize) { + if (absolutePath == null || absolutePath.isEmpty()) { + throw new IllegalArgumentException("absolutePath must not be empty"); + } + if (offset < 0) { + throw new IllegalArgumentException("offset must be non-negative"); + } + if (payloadSize <= 0) { + throw new IllegalArgumentException("payloadSize must be positive"); + } + this.absolutePath = absolutePath; + this.offset = offset; + this.payloadSize = payloadSize; + } + + public String absolutePath() { + return absolutePath; + } + + public long offset() { + return offset; + } + + public long payloadSize() { + return payloadSize; + } + + @Override + public byte[] materialize() { + throw new IllegalStateException( + "Native deletion vector descriptors do not contain JVM payload bytes"); + } + + @Override + public boolean isMaterialized() { + return false; + } + } + public static class DeltaFileReadOptions implements Serializable { private static final long serialVersionUID = 1L; private final RowIndexFilterType rowIndexFilterType; private final boolean hasDeletionVector; private final long deletionVectorCardinality; - private final byte[] serializedDeletionVector; + private final DeletionVectorPayload deletionVectorPayload; + private final NativeDeletionVectorDescriptor nativeDeletionVectorDescriptor; public DeltaFileReadOptions( RowIndexFilterType rowIndexFilterType, boolean hasDeletionVector, long deletionVectorCardinality, byte[] serializedDeletionVector) { + this( + rowIndexFilterType, + hasDeletionVector, + deletionVectorCardinality, + new SerializedDeletionVectorPayload(serializedDeletionVector)); + } + + public DeltaFileReadOptions( + RowIndexFilterType rowIndexFilterType, + boolean hasDeletionVector, + long deletionVectorCardinality, + DeletionVectorPayload deletionVectorPayload) { + if (rowIndexFilterType == null) { + throw new IllegalArgumentException("rowIndexFilterType must not be null"); + } + if (deletionVectorPayload == null) { + throw new IllegalArgumentException("deletionVectorPayload must not be null"); + } this.rowIndexFilterType = rowIndexFilterType; this.hasDeletionVector = hasDeletionVector; this.deletionVectorCardinality = deletionVectorCardinality; - this.serializedDeletionVector = - serializedDeletionVector == null ? new byte[0] : serializedDeletionVector; + this.deletionVectorPayload = deletionVectorPayload; + this.nativeDeletionVectorDescriptor = null; + } + + public DeltaFileReadOptions( + RowIndexFilterType rowIndexFilterType, + boolean hasDeletionVector, + long deletionVectorCardinality, + NativeDeletionVectorDescriptor nativeDeletionVectorDescriptor) { + if (rowIndexFilterType == null) { + throw new IllegalArgumentException("rowIndexFilterType must not be null"); + } + if (!hasDeletionVector) { + throw new IllegalArgumentException( + "A native deletion vector descriptor requires hasDeletionVector=true"); + } + if (nativeDeletionVectorDescriptor == null) { + throw new IllegalArgumentException("nativeDeletionVectorDescriptor must not be null"); + } + this.rowIndexFilterType = rowIndexFilterType; + this.hasDeletionVector = hasDeletionVector; + this.deletionVectorCardinality = deletionVectorCardinality; + this.deletionVectorPayload = null; + this.nativeDeletionVectorDescriptor = nativeDeletionVectorDescriptor; } public RowIndexFilterType rowIndexFilterType() { @@ -112,7 +243,23 @@ public long deletionVectorCardinality() { } public byte[] serializedDeletionVector() { - return serializedDeletionVector; + if (nativeDeletionVectorDescriptor != null) { + throw new IllegalStateException( + "Native deletion vector descriptors do not contain JVM payload bytes"); + } + return deletionVectorPayload.materialize(); + } + + public boolean isDeletionVectorPayloadMaterialized() { + return deletionVectorPayload != null && deletionVectorPayload.isMaterialized(); + } + + public boolean hasNativeDeletionVectorDescriptor() { + return nativeDeletionVectorDescriptor != null; + } + + public NativeDeletionVectorDescriptor nativeDeletionVectorDescriptor() { + return nativeDeletionVectorDescriptor; } } } diff --git a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto index e2396912085..34166904b6b 100644 --- a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto +++ b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto @@ -207,6 +207,15 @@ message ReadRel { bool has_deletion_vector = 2; uint64 deletion_vector_cardinality = 3; bytes serialized_deletion_vector = 4; + message DeletionVectorDescriptor { + // Authoritative absolute URI resolved by Delta on the JVM. + string absolute_path = 1; + // Offset of the 4-byte stored-payload length prefix. + uint64 offset = 2; + // Bitmap payload size, excluding the length prefix and CRC32. + uint64 payload_size = 3; + } + DeletionVectorDescriptor deletion_vector_descriptor = 5; } // File reading options diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 21a60b57bf3..0726d89a92c 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala @@ -74,6 +74,12 @@ class GlutenConfig(conf: SQLConf) extends GlutenCoreConfig(conf) { def enableColumnarFileScan: Boolean = getConf(COLUMNAR_FILESCAN_ENABLED) + def deferDeltaDeletionVectorPayloadRead: Boolean = + getConf(DELTA_DELETION_VECTOR_DEFER_PAYLOAD_READ_ENABLED) + + def enableNativeDeltaDeletionVectorPayloadRead: Boolean = + getConf(DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED) + def enableColumnarHiveTableScan: Boolean = getConf(COLUMNAR_HIVETABLESCAN_ENABLED) def enableColumnarHiveTableScanNestedColumnPruning: Boolean = @@ -880,6 +886,26 @@ object GlutenConfig extends ConfigRegistry { .booleanConf .createWithDefault(true) + val DELTA_DELETION_VECTOR_DEFER_PAYLOAD_READ_ENABLED = + buildConf("spark.gluten.sql.columnar.delta.deletionVector.deferPayloadRead.enabled") + .doc( + "When true, defer loading on-disk Delta deletion vector payloads until the native split " + + "is serialized on an executor. This removes remote deletion vector I/O from driver " + + "planning. Inline deletion vectors remain eagerly decoded because their bytes are " + + "already present in Delta metadata.") + .booleanConf + .createWithDefault(true) + + val DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED = + buildConf("spark.gluten.sql.columnar.delta.deletionVector.nativeRangeRead.enabled") + .doc( + "When true, pass each on-disk Delta deletion vector's absolute path, offset, and size " + + "to Velox instead of materializing its payload in the JVM. Velox loads the range " + + "through its file-handle and buffered-input path during split preparation. This takes " + + "precedence over deferPayloadRead.enabled. Inline deletion vectors remain JVM-decoded.") + .booleanConf + .createWithDefault(false) + val COLUMNAR_HIVETABLESCAN_ENABLED = buildConf("spark.gluten.sql.columnar.hivetablescan") .doc("Enable or disable columnar hivetablescan.") diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala index 83023f027ac..be7835695f3 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,11 +116,14 @@ abstract class FileSourceScanExecTransformerBase( disableBucketedScan) with DatasourceScanTransformer { + /** Format-specific metrics that should be displayed with the native file scan. */ + protected def additionalScanMetrics: Map[String, SQLMetric] = Map.empty + // Executor-side metrics only (excludes driverMetricsAlias). @transient private lazy val executorSideScanMetrics: 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] = diff --git a/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala b/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala index 085b39973e8..7253539f211 100644 --- a/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala +++ b/gluten-ut/test/src/test/scala/org/apache/gluten/config/GlutenRuntimeConfigSuite.scala @@ -48,6 +48,36 @@ class GlutenRuntimeConfigSuite extends GlutenQueryTest with SharedSparkSession { } } + test("Delta deletion vector payload reads are deferred by default and configurable") { + val conf = SparkSession.active.conf + val key = GlutenConfig.DELTA_DELETION_VECTOR_DEFER_PAYLOAD_READ_ENABLED.key + val original = conf.get(key) + try { + assert(GlutenConfig.get.deferDeltaDeletionVectorPayloadRead) + conf.set(key, false) + assert(!GlutenConfig.get.deferDeltaDeletionVectorPayloadRead) + conf.set(key, true) + assert(GlutenConfig.get.deferDeltaDeletionVectorPayloadRead) + } finally { + conf.set(key, original) + } + } + + test("native Delta deletion vector payload reads are opt-in and configurable") { + val conf = SparkSession.active.conf + val key = GlutenConfig.DELTA_DELETION_VECTOR_NATIVE_PAYLOAD_READ_ENABLED.key + val original = conf.get(key) + try { + assert(!GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead) + conf.set(key, true) + assert(GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead) + conf.set(key, false) + assert(!GlutenConfig.get.enableNativeDeltaDeletionVectorPayloadRead) + } finally { + conf.set(key, original) + } + } + test("Memory manager capacity ratio config validation") { assert(GlutenConfig.MEMORY_MANAGER_CAPACITY_RATIO.defaultValue.get == 0.75)