From d0f06b32057454e19a2f9af73b2b5be8728ccec7 Mon Sep 17 00:00:00 2001 From: wecharyu Date: Wed, 12 Aug 2026 22:28:49 +0800 Subject: [PATCH 1/2] [VL] Add new config spark.gluten.sql.columnar.backend.velox.spillNumMaxMergeFiles --- .../apache/gluten/config/VeloxConfig.scala | 9 ++++++++ cpp/velox/compute/WholeStageResultIterator.cc | 23 +++++++++++-------- cpp/velox/config/VeloxConfig.h | 1 + docs/get-started/Velox.md | 1 + docs/velox-configuration.md | 1 + 5 files changed, 25 insertions(+), 10 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala index fade4402cfc..9cdfa4c5c5f 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala @@ -280,6 +280,15 @@ object VeloxConfig extends ConfigRegistry { .bytesConf(ByteUnit.BYTE) .createWithDefaultString("1GB") + val COLUMNAR_VELOX_SPILL_NUM_MAX_MERGE_FILES = + buildConf("spark.gluten.sql.columnar.backend.velox.spillNumMaxMergeFiles") + .doc( + "The max number of files to merge at a time when merging sorted files " + + "into a single ordered stream. 0 means unlimited.") + .intConf + .checkValue(_ >= 0, "must be non-negative") + .createWithDefault(0) + val COLUMNAR_VELOX_SPILL_FILE_SYSTEM = buildConf("spark.gluten.sql.columnar.backend.velox.spillFileSystem") .doc( diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 0cb840c7b57..c38322e3081 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -491,16 +491,17 @@ void WholeStageResultIterator::collectMetrics() { "count", customMetric.second.count)("min", customMetric.second.min)("max", customMetric.second.max); } - operatorStats.push_back(folly::dynamic::object("inputRows", opStats->inputRows)( - "inputVectors", opStats->inputVectors)("inputBytes", opStats->inputBytes)( - "rawInputRows", opStats->rawInputRows)("rawInputBytes", opStats->rawInputBytes)( - "outputRows", opStats->outputRows)("outputVectors", opStats->outputVectors)( - "outputBytes", opStats->outputBytes)("cpuCount", opStats->cpuWallTiming.count)( - "wallNanos", opStats->cpuWallTiming.wallNanos)("peakMemoryBytes", opStats->peakMemoryBytes)( - "numMemoryAllocations", opStats->numMemoryAllocations)("spilledInputBytes", opStats->spilledInputBytes)( - "spilledBytes", opStats->spilledBytes)("spilledRows", opStats->spilledRows)( - "spilledPartitions", opStats->spilledPartitions)("spilledFiles", opStats->spilledFiles)( - "physicalWrittenBytes", opStats->physicalWrittenBytes)("customStats", customStats)); + operatorStats.push_back( + folly::dynamic::object("inputRows", opStats->inputRows)("inputVectors", opStats->inputVectors)( + "inputBytes", opStats->inputBytes)("rawInputRows", opStats->rawInputRows)( + "rawInputBytes", opStats->rawInputBytes)("outputRows", opStats->outputRows)( + "outputVectors", opStats->outputVectors)("outputBytes", opStats->outputBytes)( + "cpuCount", opStats->cpuWallTiming.count)("wallNanos", opStats->cpuWallTiming.wallNanos)( + "peakMemoryBytes", opStats->peakMemoryBytes)("numMemoryAllocations", opStats->numMemoryAllocations)( + "spilledInputBytes", opStats->spilledInputBytes)("spilledBytes", opStats->spilledBytes)( + "spilledRows", opStats->spilledRows)("spilledPartitions", opStats->spilledPartitions)( + "spilledFiles", opStats->spilledFiles)("physicalWrittenBytes", opStats->physicalWrittenBytes)( + "customStats", customStats)); } statsNum += static_cast(operatorStats.size()); @@ -580,6 +581,8 @@ std::unordered_map WholeStageResultIterator::getQueryC configs[velox::core::QueryConfig::kMaxSpillLevel] = std::to_string(veloxCfg_->get(kMaxSpillLevel, 4)); configs[velox::core::QueryConfig::kMaxSpillFileSize] = std::to_string(veloxCfg_->get(kMaxSpillFileSize, 1L * 1024 * 1024 * 1024)); + configs[velox::core::QueryConfig::kSpillNumMaxMergeFiles] = + std::to_string(veloxCfg_->get(kSpillNumMaxMergeFiles, 0)); configs[velox::core::QueryConfig::kMaxSpillRunRows] = std::to_string(veloxCfg_->get(kMaxSpillRunRows, 3L * 1024 * 1024)); configs[velox::core::QueryConfig::kMaxSpillBytes] = diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index b6dd5f9fa00..3e230668b75 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/cpp/velox/config/VeloxConfig.h @@ -35,6 +35,7 @@ const std::string kWindowSpillEnabled = "spark.gluten.sql.columnar.backend.velox // https://github.com/facebookincubator/velox/blob/95f3e80e77d046c12fbc79dc529366be402e9c2b/velox/docs/configs.rst#spilling const std::string kMaxSpillLevel = "spark.gluten.sql.columnar.backend.velox.maxSpillLevel"; const std::string kMaxSpillFileSize = "spark.gluten.sql.columnar.backend.velox.maxSpillFileSize"; +const std::string kSpillNumMaxMergeFiles = "spark.gluten.sql.columnar.backend.velox.spillNumMaxMergeFiles"; const std::string kSpillStartPartitionBit = "spark.gluten.sql.columnar.backend.velox.spillStartPartitionBit"; const std::string kSpillPartitionBits = "spark.gluten.sql.columnar.backend.velox.spillPartitionBits"; const std::string kMaxSpillRunRows = "spark.gluten.sql.columnar.backend.velox.MaxSpillRunRows"; diff --git a/docs/get-started/Velox.md b/docs/get-started/Velox.md index 9eead57d0d3..03aa02ef007 100644 --- a/docs/get-started/Velox.md +++ b/docs/get-started/Velox.md @@ -474,6 +474,7 @@ Using the following configuration options to customize spilling: | spark.gluten.sql.columnar.backend.velox.orderBySpillEnabled | true | Whether spill is enabled on sorts | | spark.gluten.sql.columnar.backend.velox.maxSpillLevel | 4 | The max allowed spilling level with zero being the initial spilling level | | spark.gluten.sql.columnar.backend.velox.maxSpillFileSize | 1GB | The max allowed spill file size. If it is zero, then there is no limit | +| spark.gluten.sql.columnar.backend.velox.spillNumMaxMergeFiles | 0 | The max number of files to merge at a time when merging sorted files into a single ordered stream. 0 means unlimited. | | spark.gluten.sql.columnar.backend.velox.spillStartPartitionBit | 48 | The start partition bit which is used with 'spillPartitionBits' together to calculate the spilling partition number | | spark.gluten.sql.columnar.backend.velox.spillPartitionBits | 3 | The number of bits used to calculate the spilling partition number. The number of spilling partitions will be power of two | | spark.gluten.sql.columnar.backend.velox.spillableReservationGrowthPct | 25 | The spillable memory reservation growth percentage of the previous memory reservation size | diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 8f80bb2d384..7f78397556c 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -75,6 +75,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput | 🔄 Dynamic | false | If true, combine small columnar batches together right after shuffle read. The default minimum output batch size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize | | spark.gluten.sql.columnar.backend.velox.showTaskMetricsWhenFinished | 🔄 Dynamic | false | Show velox full task metrics when finished. | | spark.gluten.sql.columnar.backend.velox.spillFileSystem | 🔄 Dynamic | local | The filesystem used to store spill data. local: The local file system. heap-over-local: Write file to JVM heap if having extra heap space. Otherwise write to local file system. | +| spark.gluten.sql.columnar.backend.velox.spillNumMaxMergeFiles | 🔄 Dynamic | 0 | The max number of files to merge at a time when merging sorted files into a single ordered stream. 0 means unlimited. | | spark.gluten.sql.columnar.backend.velox.spillStrategy | 🔄 Dynamic | auto | none: Disable spill on Velox backend; auto: Let Spark memory manager manage Velox's spilling | | spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads | ⚓ Static | 4 | The number of IO threads for SSD cache read/write operations | | spark.gluten.sql.columnar.backend.velox.ssdCachePath | ⚓ Static | /tmp | The folder to store the cache files, better on SSD | From 3d58bdbdb941d827b6fd834491c1b49481834e6c Mon Sep 17 00:00:00 2001 From: wecharyu Date: Wed, 12 Aug 2026 23:27:53 +0800 Subject: [PATCH 2/2] fix format --- cpp/velox/compute/WholeStageResultIterator.cc | 21 +++++++++---------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index c38322e3081..d29ac7c8bbb 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -491,17 +491,16 @@ void WholeStageResultIterator::collectMetrics() { "count", customMetric.second.count)("min", customMetric.second.min)("max", customMetric.second.max); } - operatorStats.push_back( - folly::dynamic::object("inputRows", opStats->inputRows)("inputVectors", opStats->inputVectors)( - "inputBytes", opStats->inputBytes)("rawInputRows", opStats->rawInputRows)( - "rawInputBytes", opStats->rawInputBytes)("outputRows", opStats->outputRows)( - "outputVectors", opStats->outputVectors)("outputBytes", opStats->outputBytes)( - "cpuCount", opStats->cpuWallTiming.count)("wallNanos", opStats->cpuWallTiming.wallNanos)( - "peakMemoryBytes", opStats->peakMemoryBytes)("numMemoryAllocations", opStats->numMemoryAllocations)( - "spilledInputBytes", opStats->spilledInputBytes)("spilledBytes", opStats->spilledBytes)( - "spilledRows", opStats->spilledRows)("spilledPartitions", opStats->spilledPartitions)( - "spilledFiles", opStats->spilledFiles)("physicalWrittenBytes", opStats->physicalWrittenBytes)( - "customStats", customStats)); + operatorStats.push_back(folly::dynamic::object("inputRows", opStats->inputRows)( + "inputVectors", opStats->inputVectors)("inputBytes", opStats->inputBytes)( + "rawInputRows", opStats->rawInputRows)("rawInputBytes", opStats->rawInputBytes)( + "outputRows", opStats->outputRows)("outputVectors", opStats->outputVectors)( + "outputBytes", opStats->outputBytes)("cpuCount", opStats->cpuWallTiming.count)( + "wallNanos", opStats->cpuWallTiming.wallNanos)("peakMemoryBytes", opStats->peakMemoryBytes)( + "numMemoryAllocations", opStats->numMemoryAllocations)("spilledInputBytes", opStats->spilledInputBytes)( + "spilledBytes", opStats->spilledBytes)("spilledRows", opStats->spilledRows)( + "spilledPartitions", opStats->spilledPartitions)("spilledFiles", opStats->spilledFiles)( + "physicalWrittenBytes", opStats->physicalWrittenBytes)("customStats", customStats)); } statsNum += static_cast(operatorStats.size());