From eca89da73f2269a6bba5543c9c065a0db26dc355 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Mon, 29 Jun 2026 15:18:25 +0200 Subject: [PATCH 01/25] [VL] Enable file handle cache by default with TTL-based eviction Enable fileHandleCacheEnabled by default (was false) and increase ssdCacheIOThreads from 1 to 4. Wire the previously dead-code TTL config to the Velox cache, and add new Spark configs for tuning cache size and expiration. Add a test suite and benchmark to validate correctness and measure performance. Changes: 1. Default config changes: - fileHandleCacheEnabled: false -> true - ssdCacheIOThreads: 1 -> 4 2. Fix Velox TTL wiring (file-handle-cache-ttl.patch): The file-handle-expiration-duration-ms config existed in Velox but was never passed to the SimpleLRUCache constructor in HiveConnector.cpp. The patch wires it so handles are actually evicted after the configured TTL, preventing stale HDFS leases or closed remote connections from accumulating indefinitely. 3. New Spark configs exposed: - spark.gluten.sql.columnar.backend.velox.numCacheFileHandles (default: 20000) - max entries in the LRU cache - spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs (default: 600000 / 10 min) - TTL per handle; idle handles are evicted 4. Test suite (VeloxFileHandleCacheSuite, 6 tests): - Basic scan correctness with cache enabled - Repeated scans produce consistent results (cache hit path) - Many small files (200) do not cause resource errors - Filtered scan correctness with predicate pushdown - Graceful behavior when files are deleted between scans - Column pruning with different projections on cached handles 5. Benchmark (FileHandleCacheBenchmark): Measures repeated scans of 200 small Parquet files. Run twice with different --conf to compare enabled vs disabled (static config). Rationale: Data lake files (Parquet, Delta, Iceberg) are immutable once written, making file handle caching safe for production workloads. Caching avoids repeated open/close per file, which is costly on remote filesystems (S3, HDFS, ABFS) where handle creation involves network round-trips. Benchmark results (200 Parquet files, 10 repeated scans, local FS): Cache OFF Cache ON Improvement Full scan 1586 ms 1475 ms 7.0% Filtered scan 1915 ms 1757 ms 8.3% Column pruning 1484 ms 1378 ms 7.1% The measured per-file-open saving is ~55us on local FS (111ms across 2000 file opens). On object stores such as S3, each file open involves HTTP HEAD + GET with typical first-byte latency of 20-100ms, making the per-file-open cost ~500-2000x higher than local FS. For the same workload (200 files, 10 repeated scans), this translates to 36-180s of avoidable overhead on cache hits, yielding an estimated 40-70% improvement on remote storage for repeated scans of many small files. Users who work with mutable files can set spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled=false. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../apache/gluten/config/VeloxConfig.scala | 29 ++- .../execution/VeloxFileHandleCacheSuite.scala | 244 ++++++++++++++++++ .../benchmark/FileHandleCacheBenchmark.scala | 125 +++++++++ cpp/velox/config/VeloxConfig.h | 13 +- cpp/velox/utils/ConfigExtractor.cc | 4 + .../src/file-handle-cache-ttl.patch | 14 + ep/build-velox/src/get-velox.sh | 10 + .../apache/gluten/config/GlutenConfig.scala | 2 +- 8 files changed, 433 insertions(+), 8 deletions(-) create mode 100644 backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala create mode 100644 backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala create mode 100644 ep/build-velox/src/file-handle-cache-ttl.patch 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 2a3f34483ec..24de2fd086d 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 @@ -180,9 +180,9 @@ object VeloxConfig extends ConfigRegistry { val COLUMNAR_VELOX_SSD_CACHE_IO_THREADS = buildStaticConf("spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads") - .doc("The IO threads for cache promoting") + .doc("The number of IO threads for SSD cache read/write operations") .intConf - .createWithDefault(1) + .createWithDefault(4) val COLUMNAR_VELOX_SSD_ODIRECT_ENABLED = buildStaticConf("spark.gluten.sql.columnar.backend.velox.ssdODirect") @@ -534,10 +534,29 @@ object VeloxConfig extends ConfigRegistry { val COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED = buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled") .doc( - "Disables caching if false. File handle cache should be disabled " + - "if files are mutable, i.e. file content may change while file path stays the same.") + "Enables caching of file handles to avoid repeated open/close overhead on remote " + + "filesystems. Should be disabled if files are mutable, i.e. file content may " + + "change while file path stays the same.") .booleanConf - .createWithDefault(false) + .createWithDefault(true) + + val COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES = + buildStaticConf("spark.gluten.sql.columnar.backend.velox.numCacheFileHandles") + .doc( + "Maximum number of entries in the file handle cache. Each entry holds an open " + + "file descriptor (local FS) or connection state (remote FS).") + .intConf + .createWithDefault(20000) + + val COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS = + buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs") + .doc( + "Expiration time in milliseconds for cached file handles. Handles not accessed " + + "within this duration are evicted from the cache. This prevents stale handles " + + "from accumulating (e.g., expired HDFS leases, closed remote connections). " + + "A value of 0 disables TTL-based eviction.") + .longConf + .createWithDefault(600000L) // 10 minutes val DIRECTORY_SIZE_GUESS = buildStaticConf("spark.gluten.sql.columnar.backend.velox.directorySizeGuess") diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala new file mode 100644 index 00000000000..d29d88d0ff5 --- /dev/null +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -0,0 +1,244 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.spark.sql.execution + +import org.apache.gluten.config.VeloxConfig +import org.apache.gluten.execution.{BasicScanExecTransformer, VeloxWholeStageTransformerSuite} + +import org.apache.spark.SparkConf + +/** + * Test suite for Velox file handle cache behavior. + * + * Tests correctness, config propagation, and edge cases for the file handle cache which caches open + * file handles (descriptors) to avoid repeated open/close overhead. + */ +class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { + override protected val resourcePath: String = "/parquet-for-read" + override protected val fileFormat: String = "parquet" + + override protected def sparkConf: SparkConf = { + super.sparkConf + .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED.key, "true") + .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, "600000") + .set(VeloxConfig.COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES.key, "20000") + } + + testWithSpecifiedSparkVersion( + "basic scan correctness with file handle cache enabled", + "3.5", + "3.5") { + // Verify that enabling file handle cache produces correct scan results + withTempPath { + dir => + spark + .range(10000) + .selectExpr("id", "cast(id % 7 as int) as category", "id * 1.5 as value") + .repartition(10) + .write + .parquet(dir.getCanonicalPath) + + val df = spark.read.parquet(dir.getCanonicalPath) + df.createOrReplaceTempView("t") + + runQueryAndCompare("SELECT count(*) FROM t") { + checkGlutenPlan[BasicScanExecTransformer] + } + runQueryAndCompare("SELECT sum(value) FROM t WHERE category = 3") { + checkGlutenPlan[BasicScanExecTransformer] + } + runQueryAndCompare("SELECT category, count(*) FROM t GROUP BY category") { + checkGlutenPlan[BasicScanExecTransformer] + } + } + } + + testWithSpecifiedSparkVersion( + "repeated scans produce consistent results (cache hit path)", + "3.5", + "3.5") { + // When file handles are cached, repeated scans of the same files must produce + // identical results. This exercises the cache hit path. + withTempPath { + dir => + spark + .range(5000) + .selectExpr("id", "cast(id as string) as name") + .repartition(50) // 50 files to exercise many cache entries + .write + .parquet(dir.getCanonicalPath) + + val path = dir.getCanonicalPath + val expected = spark.read.parquet(path).count() + assert(expected == 5000) + + // Scan the same files multiple times - each should hit the cache + for (i <- 1 to 5) { + val count = spark.read.parquet(path).count() + assert( + count == expected, + s"Iteration $i: expected $expected rows but got $count") + } + + // Verify aggregation consistency across repeated scans + val firstSum = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) + for (i <- 1 to 3) { + val sum = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) + assert( + sum == firstSum, + s"Iteration $i: sum mismatch, expected $firstSum but got $sum") + } + } + } + + testWithSpecifiedSparkVersion( + "many small files do not cause errors with file handle cache", + "3.5", + "3.5") { + // Verify that scanning many small files with caching enabled does not cause + // file descriptor exhaustion or other resource-related errors. + withTempPath { + dir => + // Create 200 small parquet files + spark + .range(20000) + .selectExpr("id", "uuid() as payload") + .repartition(200) + .write + .parquet(dir.getCanonicalPath) + + val fileCount = dir.listFiles().count(_.getName.endsWith(".parquet")) + assert(fileCount >= 100, s"Expected at least 100 files, got $fileCount") + + // Scan all files - should work without resource errors + val count = spark.read.parquet(dir.getCanonicalPath).count() + assert(count == 20000) + + // Scan again (cache hit path) - should also work + val count2 = spark.read.parquet(dir.getCanonicalPath).count() + assert(count2 == 20000) + } + } + + testWithSpecifiedSparkVersion( + "filtered scan correctness with file handle cache", + "3.5", + "3.5") { + // Verify that predicate pushdown works correctly with cached file handles. + // This exercises the row group skipping path through cached handles. + withTempPath { + dir => + spark + .range(100000) + .selectExpr( + "id", + "cast(id % 10 as int) as partition_key", + "cast(id * 0.01 as double) as metric") + .repartition(20) + .write + .parquet(dir.getCanonicalPath) + + val path = dir.getCanonicalPath + + // Filter that matches ~10% of rows + val filtered = spark.read.parquet(path).where("partition_key = 5").count() + assert(filtered == 10000, s"Expected 10000 filtered rows, got $filtered") + + // Range filter + val rangeFiltered = spark.read.parquet(path).where("id >= 50000").count() + assert(rangeFiltered == 50000, s"Expected 50000 range-filtered rows, got $rangeFiltered") + + // Re-run same filters (cache hit path) + val filtered2 = spark.read.parquet(path).where("partition_key = 5").count() + assert(filtered2 == filtered, "Filtered count mismatch on repeated scan") + } + } + + testWithSpecifiedSparkVersion( + "scan after file deletion produces appropriate error or empty result", + "3.5", + "3.5") { + // If a file is deleted between scans, the next scan should either: + // - Succeed (if the cached FD still works on Linux with unlinked inodes) + // - Produce an error (not silently return wrong data) + withTempPath { + dir => + spark + .range(1000) + .selectExpr("id") + .repartition(5) + .write + .parquet(dir.getCanonicalPath) + + val path = dir.getCanonicalPath + // First scan populates the cache + val count1 = spark.read.parquet(path).count() + assert(count1 == 1000) + + // Delete one parquet file + val parquetFiles = dir.listFiles().filter(_.getName.endsWith(".parquet")) + assert(parquetFiles.nonEmpty) + val deletedFile = parquetFiles.head + val deletedRows = spark.read.parquet(deletedFile.getCanonicalPath).count() + deletedFile.delete() + + // On Linux, the cached FD to the deleted file may still work (unlinked inode). + // Either way, the remaining files should be readable. + // We don't assert on exact count because the deleted file's FD might still be valid. + val count2 = spark.read.parquet(path).count() + // The count should be either (count1 - deletedRows) or count1 + // depending on whether the OS kept the inode accessible + assert( + count2 == count1 || count2 == count1 - deletedRows, + s"Unexpected count after deletion: $count2 (original: $count1, deleted: $deletedRows)") + } + } + + testWithSpecifiedSparkVersion( + "column pruning with cached file handles", + "3.5", + "3.5") { + // Verify that column pruning works correctly when file handles are cached. + // The cache key includes the file path but not the projected columns, so + // different projections on the same file must still work correctly. + withTempPath { + dir => + spark + .range(5000) + .selectExpr("id", "id * 2 as doubled", "id * 3 as tripled", "uuid() as text") + .repartition(10) + .write + .parquet(dir.getCanonicalPath) + + val path = dir.getCanonicalPath + + // Read all columns + val allCols = spark.read.parquet(path).select("id", "doubled", "tripled", "text").count() + assert(allCols == 5000) + + // Read subset of columns (same file handles, different projection) + val subset1 = spark.read.parquet(path).select("id").collect() + assert(subset1.length == 5000) + assert(subset1.head.schema.fieldNames.sameElements(Array("id"))) + + // Different subset + val subset2 = spark.read.parquet(path).selectExpr("sum(doubled)").collect() + val expectedSum = (0L until 5000L).map(_ * 2).sum + assert(subset2(0).getLong(0) == expectedSum) + } + } +} diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala new file mode 100644 index 00000000000..36166b46e99 --- /dev/null +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala @@ -0,0 +1,125 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.spark.sql.execution.benchmark + +import org.apache.gluten.config.VeloxConfig + +import org.apache.spark.benchmark.Benchmark + +/** + * Benchmark to measure the effect of Velox file handle caching on repeated scans of many small + * Parquet files. File handle caching avoids repeated open/close overhead per file, which is + * especially beneficial for remote filesystems (S3, HDFS, ABFS) where handle creation involves + * network round-trips (DNS, TCP, auth). + * + * Even on local filesystems the overhead is measurable when scanning hundreds of small files + * multiple times (e.g., repeated queries on a heavily-partitioned table). + * + * NOTE: `fileHandleCacheEnabled` is a static config (read at backend init). To compare on vs off, + * run this benchmark twice with different Spark configurations: + * {{{ + * # With file handle cache enabled (default): + * bin/spark-submit --class \ + * --conf spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled=true \ + * --jars + * + * # With file handle cache disabled: + * bin/spark-submit --class \ + * --conf spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled=false \ + * --jars + * }}} + * + * Expected result: with caching enabled, repeated scans should show lower wall-clock time due to + * avoiding per-file open() syscalls (or remote filesystem connection establishment) on subsequent + * scans of the same files. + */ +object FileHandleCacheBenchmark extends SqlBasedBenchmark { + + // Number of small files to generate (simulates a heavily-partitioned table) + private val numFiles = 200 + // Rows per file (small to emphasize per-file overhead over per-row compute) + private val rowsPerFile = 5000 + // Number of repeated scans per benchmark iteration + private val scanIterations = 10 + + override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { + val totalRows = numFiles.toLong * rowsPerFile + + val cacheEnabled = spark.conf.get( + VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED.key, + "true") + val label = s"fileHandleCache=$cacheEnabled" + + withTempPath { + dir => + val path = dir.getCanonicalPath + + // Generate many small Parquet files by repartitioning + spark + .range(totalRows) + .selectExpr( + "id", + "cast(id % 100 as int) as category", + "cast(id * 1.5 as double) as value", + "uuid() as payload" + ) + .repartition(numFiles) + .write + .parquet(path) + + val fileCount = dir.listFiles().count(_.getName.endsWith(".parquet")) + // scalastyle:off println + println(s"Generated $fileCount parquet files with ~$rowsPerFile rows each") + println(s"Config: $label") + // scalastyle:on println + + val benchmark = new Benchmark( + s"Repeated scan of $fileCount small files ($label)", + totalRows * scanIterations, + output = output) + + // Warm up: first scan populates any one-time init (Velox backend, JIT, etc.) + spark.read.parquet(path).selectExpr("sum(value)").collect() + + benchmark.addCase(s"full scan ($scanIterations iterations)", 5) { + _ => + for (_ <- 1 to scanIterations) { + spark.read.parquet(path).selectExpr("sum(value)", "count(*)").collect() + } + } + + benchmark.addCase(s"filtered scan ($scanIterations iterations)", 5) { + _ => + for (_ <- 1 to scanIterations) { + spark.read.parquet(path) + .where("category < 10") + .selectExpr("sum(value)", "count(*)") + .collect() + } + } + + benchmark.addCase(s"column pruning scan ($scanIterations iterations)", 5) { + _ => + for (_ <- 1 to scanIterations) { + spark.read.parquet(path).selectExpr("sum(id)").collect() + } + } + + benchmark.run() + } + } +} diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index ba9bcda5a7e..03f1d22bd0a 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/cpp/velox/config/VeloxConfig.h @@ -139,7 +139,7 @@ const std::string kVeloxSsdCachePathDefault = "/tmp/"; const std::string kVeloxSsdCacheShards = "spark.gluten.sql.columnar.backend.velox.ssdCacheShards"; const uint32_t kVeloxSsdCacheShardsDefault = 1; const std::string kVeloxSsdCacheIOThreads = "spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads"; -const uint32_t kVeloxSsdCacheIOThreadsDefault = 1; +const uint32_t kVeloxSsdCacheIOThreadsDefault = 4; const std::string kVeloxSsdODirectEnabled = "spark.gluten.sql.columnar.backend.velox.ssdODirect"; const std::string kVeloxSsdCheckpointIntervalBytes = "spark.gluten.sql.columnar.backend.velox.ssdCheckpointIntervalBytes"; @@ -162,7 +162,16 @@ const std::string kVeloxUdfLibraryPaths = "spark.gluten.sql.columnar.backend.vel const std::string kVeloxShuffleReaderPrintFlag = "spark.gluten.velox.shuffleReaderPrintFlag"; const std::string kVeloxFileHandleCacheEnabled = "spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled"; -const bool kVeloxFileHandleCacheEnabledDefault = false; +const bool kVeloxFileHandleCacheEnabledDefault = true; + +const std::string kVeloxNumCacheFileHandles = "spark.gluten.sql.columnar.backend.velox.numCacheFileHandles"; +const int32_t kVeloxNumCacheFileHandlesDefault = 20000; + +const std::string kVeloxFileHandleExpirationDurationMs = + "spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs"; +// 10 minutes default TTL — ensures stale handles (e.g., expired HDFS leases, +// closed remote connections) are evicted from the cache. +const int64_t kVeloxFileHandleExpirationDurationMsDefault = 600000; /* configs for file read in velox*/ const std::string kDirectorySizeGuess = "spark.gluten.sql.columnar.backend.velox.directorySizeGuess"; diff --git a/cpp/velox/utils/ConfigExtractor.cc b/cpp/velox/utils/ConfigExtractor.cc index 7ee2deae8a8..240033badb0 100644 --- a/cpp/velox/utils/ConfigExtractor.cc +++ b/cpp/velox/utils/ConfigExtractor.cc @@ -322,6 +322,10 @@ std::shared_ptr createHiveConnectorConfig( hiveConfMap[facebook::velox::connector::hive::HiveConfig::kEnableFileHandleCache] = conf->get(kVeloxFileHandleCacheEnabled, kVeloxFileHandleCacheEnabledDefault) ? "true" : "false"; + hiveConfMap[facebook::velox::connector::hive::HiveConfig::kNumCacheFileHandles] = + std::to_string(conf->get(kVeloxNumCacheFileHandles, kVeloxNumCacheFileHandlesDefault)); + hiveConfMap[facebook::velox::connector::hive::HiveConfig::kFileHandleExpirationDurationMs] = std::to_string( + conf->get(kVeloxFileHandleExpirationDurationMs, kVeloxFileHandleExpirationDurationMsDefault)); hiveConfMap[facebook::velox::connector::hive::HiveConfig::kMaxCoalescedBytes] = conf->get(kMaxCoalescedBytes, "67108864"); // 64M hiveConfMap[facebook::velox::connector::hive::HiveConfig::kMaxCoalescedDistance] = diff --git a/ep/build-velox/src/file-handle-cache-ttl.patch b/ep/build-velox/src/file-handle-cache-ttl.patch new file mode 100644 index 00000000000..257bef55db1 --- /dev/null +++ b/ep/build-velox/src/file-handle-cache-ttl.patch @@ -0,0 +1,14 @@ +diff --git i/velox/connectors/hive/HiveConnector.cpp w/velox/connectors/hive/HiveConnector.cpp +index abc13acea..861f95c7f 100644 +--- i/velox/connectors/hive/HiveConnector.cpp ++++ w/velox/connectors/hive/HiveConnector.cpp +@@ -40,7 +40,8 @@ HiveConnector::HiveConnector( + fileHandleFactory_( + hiveConfig_->isFileHandleCacheEnabled() + ? std::make_unique>( +- hiveConfig_->numCacheFileHandles()) ++ hiveConfig_->numCacheFileHandles(), ++ hiveConfig_->fileHandleExpirationDurationMs()) + : nullptr, + std::make_unique(hiveConfig_->config())), + ioExecutor_(ioExecutor) { diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index 6443710a936..4f4c3ac711f 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -159,6 +159,16 @@ function apply_compilation_fixes { $SUDO_CMD cp ${CURRENT_DIR}/modify_arrow.patch ${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/ git add ${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/modify_arrow.patch # to avoid the file from being deleted by git clean -dffx :/ + + # Wire file handle cache TTL config to SimpleLRUCache constructor. + if [ -f "${CURRENT_DIR}/file-handle-cache-ttl.patch" ]; then + pushd $VELOX_HOME + git apply --check ${CURRENT_DIR}/file-handle-cache-ttl.patch 2>/dev/null && \ + git apply ${CURRENT_DIR}/file-handle-cache-ttl.patch && \ + echo "Applied file-handle-cache-ttl.patch" || \ + echo "file-handle-cache-ttl.patch already applied or not applicable, skipping" + popd + fi } function setup_linux { 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 93fb3888e99..cdd1a3d9e9e 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 @@ -653,7 +653,7 @@ object GlutenConfig extends ConfigRegistry { ("spark.hadoop.dfs.client.log.severity", "INFO"), ("spark.sql.orc.compression.codec", "snappy"), ("spark.sql.decimalOperations.allowPrecisionLoss", "true"), - ("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled", "false"), + ("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled", "true"), ("spark.gluten.velox.awsSdkLogLevel", "FATAL"), ("spark.gluten.velox.s3UseProxyFromEnv", "false"), ("spark.gluten.velox.s3PayloadSigningPolicy", "Never"), From 78bfa6664ec22184755fc6ad2a1db88f947f7717 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 30 Jun 2026 13:41:05 +0200 Subject: [PATCH 02/25] Address review comments: fix schema assertion, deletion test, patch guard, cache default - Fix subset1.head.schema: assert schema on DataFrame before collect() - Assert file deletion succeeded; wrap second scan in try-catch - get-velox.sh: fail fast if TTL patch doesn't apply and isn't upstream - Reduce numCacheFileHandles default from 20000 to 10000 - Expand doc to clarify FD vs HTTP connection distinction --- .../apache/gluten/config/VeloxConfig.scala | 7 +++-- .../execution/VeloxFileHandleCacheSuite.scala | 30 +++++++++++-------- cpp/velox/config/VeloxConfig.h | 2 +- ep/build-velox/src/get-velox.sh | 13 +++++--- 4 files changed, 33 insertions(+), 19 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 24de2fd086d..1d875f97535 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 @@ -544,9 +544,12 @@ object VeloxConfig extends ConfigRegistry { buildStaticConf("spark.gluten.sql.columnar.backend.velox.numCacheFileHandles") .doc( "Maximum number of entries in the file handle cache. Each entry holds an open " + - "file descriptor (local FS) or connection state (remote FS).") + "file descriptor (local FS) or connection state (remote FS). Note that on " + + "local filesystems, high values may approach the OS file descriptor limit " + + "(ulimit -n). On remote object stores (S3, ABFS, GCS) entries are HTTP " + + "connections, not OS file descriptors.") .intConf - .createWithDefault(20000) + .createWithDefault(10000) val COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS = buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs") diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index d29d88d0ff5..b351903c51c 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -35,7 +35,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { super.sparkConf .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED.key, "true") .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, "600000") - .set(VeloxConfig.COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES.key, "20000") + .set(VeloxConfig.COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES.key, "10000") } testWithSpecifiedSparkVersion( @@ -194,17 +194,23 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { assert(parquetFiles.nonEmpty) val deletedFile = parquetFiles.head val deletedRows = spark.read.parquet(deletedFile.getCanonicalPath).count() - deletedFile.delete() + assert(deletedFile.delete(), s"Failed to delete ${deletedFile.getCanonicalPath}") // On Linux, the cached FD to the deleted file may still work (unlinked inode). // Either way, the remaining files should be readable. - // We don't assert on exact count because the deleted file's FD might still be valid. - val count2 = spark.read.parquet(path).count() - // The count should be either (count1 - deletedRows) or count1 - // depending on whether the OS kept the inode accessible - assert( - count2 == count1 || count2 == count1 - deletedRows, - s"Unexpected count after deletion: $count2 (original: $count1, deleted: $deletedRows)") + // The scan may also throw if the FS detects the missing file. + try { + val count2 = spark.read.parquet(path).count() + // The count should be either (count1 - deletedRows) or count1 + // depending on whether the OS kept the inode accessible + assert( + count2 == count1 || count2 == count1 - deletedRows, + s"Unexpected count after deletion: $count2 (original: $count1, deleted: $deletedRows)") + } catch { + case _: Exception => + // Acceptable: the scan failed because the deleted file is no longer accessible. + // The important thing is that it does not silently return wrong data. + } } } @@ -231,9 +237,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { assert(allCols == 5000) // Read subset of columns (same file handles, different projection) - val subset1 = spark.read.parquet(path).select("id").collect() - assert(subset1.length == 5000) - assert(subset1.head.schema.fieldNames.sameElements(Array("id"))) + val subset1Df = spark.read.parquet(path).select("id") + assert(subset1Df.schema.fieldNames.sameElements(Array("id"))) + assert(subset1Df.collect().length == 5000) // Different subset val subset2 = spark.read.parquet(path).selectExpr("sum(doubled)").collect() diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index 03f1d22bd0a..36d029e8ff4 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/cpp/velox/config/VeloxConfig.h @@ -165,7 +165,7 @@ const std::string kVeloxFileHandleCacheEnabled = "spark.gluten.sql.columnar.back const bool kVeloxFileHandleCacheEnabledDefault = true; const std::string kVeloxNumCacheFileHandles = "spark.gluten.sql.columnar.backend.velox.numCacheFileHandles"; -const int32_t kVeloxNumCacheFileHandlesDefault = 20000; +const int32_t kVeloxNumCacheFileHandlesDefault = 10000; const std::string kVeloxFileHandleExpirationDurationMs = "spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs"; diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index 4f4c3ac711f..33390e31ae2 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -163,10 +163,15 @@ function apply_compilation_fixes { # Wire file handle cache TTL config to SimpleLRUCache constructor. if [ -f "${CURRENT_DIR}/file-handle-cache-ttl.patch" ]; then pushd $VELOX_HOME - git apply --check ${CURRENT_DIR}/file-handle-cache-ttl.patch 2>/dev/null && \ - git apply ${CURRENT_DIR}/file-handle-cache-ttl.patch && \ - echo "Applied file-handle-cache-ttl.patch" || \ - echo "file-handle-cache-ttl.patch already applied or not applicable, skipping" + if git apply --check ${CURRENT_DIR}/file-handle-cache-ttl.patch 2>/dev/null; then + git apply ${CURRENT_DIR}/file-handle-cache-ttl.patch + echo "Applied file-handle-cache-ttl.patch" + elif git apply --check -R ${CURRENT_DIR}/file-handle-cache-ttl.patch 2>/dev/null; then + echo "file-handle-cache-ttl.patch already applied upstream, skipping" + else + echo "ERROR: file-handle-cache-ttl.patch failed to apply and is not present upstream" >&2 + exit 1 + fi popd fi } From e1205d5924d0e82dd6e4c0dedb24707345826706 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 30 Jun 2026 13:52:36 +0200 Subject: [PATCH 03/25] Tighten fileCount assertion to >= 200 to match repartition(200) --- .../apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index b351903c51c..e510442e54f 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -122,7 +122,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { .parquet(dir.getCanonicalPath) val fileCount = dir.listFiles().count(_.getName.endsWith(".parquet")) - assert(fileCount >= 100, s"Expected at least 100 files, got $fileCount") + assert(fileCount >= 200, s"Expected at least 200 files, got $fileCount") // Scan all files - should work without resource errors val count = spark.read.parquet(dir.getCanonicalPath).count() From 389de4c584cbf4901d22f7b77f4dbbb8886854eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 30 Jun 2026 13:56:16 +0200 Subject: [PATCH 04/25] Add TTL eviction test: verify scans succeed after cached handles expire Set suite-level TTL to 2s, add test that scans files, waits 3s for expiration, then verifies scans still return correct results after handles are evicted and re-opened. --- .../execution/VeloxFileHandleCacheSuite.scala | 35 ++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index e510442e54f..c6956054705 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -34,7 +34,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { override protected def sparkConf: SparkConf = { super.sparkConf .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED.key, "true") - .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, "600000") + .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, "2000") .set(VeloxConfig.COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES.key, "10000") } @@ -214,6 +214,39 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { } } + testWithSpecifiedSparkVersion( + "TTL-based eviction: scans succeed after cached handles expire", + "3.5", + "3.5") { + // Verify that after the TTL expires (2s, set in sparkConf), cached handles + // are evicted and subsequent scans re-open files correctly. + withTempPath { + dir => + spark + .range(5000) + .selectExpr("id", "id * 2 as doubled") + .repartition(20) + .write + .parquet(dir.getCanonicalPath) + + val path = dir.getCanonicalPath + + // First scan populates the cache + val count1 = spark.read.parquet(path).count() + assert(count1 == 5000) + val sum1 = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) + + // Wait for TTL to expire (configured to 2s in sparkConf) + Thread.sleep(3000) + + // Scan after expiration: handles should be evicted and re-opened + val count2 = spark.read.parquet(path).count() + assert(count2 == 5000, s"Count mismatch after TTL expiration: expected 5000, got $count2") + val sum2 = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) + assert(sum2 == sum1, s"Sum mismatch after TTL expiration: expected $sum1, got $sum2") + } + } + testWithSpecifiedSparkVersion( "column pruning with cached file handles", "3.5", From 7bcad1aaceed2f746808b2fac7cbe5da9e76a893 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 30 Jun 2026 14:03:58 +0200 Subject: [PATCH 05/25] Narrow catch to file-not-found errors; use count() instead of collect().length --- .../spark/sql/execution/VeloxFileHandleCacheSuite.scala | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index c6956054705..d5c54a5febf 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -207,7 +207,12 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { count2 == count1 || count2 == count1 - deletedRows, s"Unexpected count after deletion: $count2 (original: $count1, deleted: $deletedRows)") } catch { - case _: Exception => + case e: Exception + if e.getMessage != null && + (e.getMessage.contains("FileNotFoundException") || + e.getMessage.contains("No such file") || + e.getMessage.contains("Path does not exist") || + e.getMessage.contains("does not exist")) => // Acceptable: the scan failed because the deleted file is no longer accessible. // The important thing is that it does not silently return wrong data. } @@ -272,7 +277,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { // Read subset of columns (same file handles, different projection) val subset1Df = spark.read.parquet(path).select("id") assert(subset1Df.schema.fieldNames.sameElements(Array("id"))) - assert(subset1Df.collect().length == 5000) + assert(subset1Df.count() == 5000) // Different subset val subset2 = spark.read.parquet(path).selectExpr("sum(doubled)").collect() From 1c8325ba653c8f264939824f4db4da4bacbf1549 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 30 Jun 2026 16:13:55 +0200 Subject: [PATCH 06/25] Regenerate configuration docs for new file handle cache configs --- docs/velox-configuration.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 965b03802e5..270387cdab4 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -30,7 +30,8 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.directorySizeGuess | ⚓ Static | 32KB | Deprecated, rename to spark.gluten.sql.columnar.backend.velox.footerEstimatedSize | | spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild | 🔄 Dynamic | false | Enable driver-side broadcast hash table build. When enabled, the hash table is built and serialized on the driver, then broadcast to executors. When disabled, each executor builds its own hash table from the broadcast data. | | spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation | 🔄 Dynamic | false | Enable validation fallback for TimestampNTZ type. When true, any plan containing TimestampNTZ will fall back to Spark execution. When false, allows native execution for TimestampNTZ scan. | -| spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | false | Disables caching if false. File handle cache should be disabled if files are mutable, i.e. file content may change while file path stays the same. | +| spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | true | Enables caching of file handles to avoid repeated open/close overhead on remote filesystems. Should be disabled if files are mutable, i.e. file content may change while file path stays the same. | +| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000 | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | | spark.gluten.sql.columnar.backend.velox.filePreloadThreshold | ⚓ Static | 1MB | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold | | spark.gluten.sql.columnar.backend.velox.floatingPointMode | 🔄 Dynamic | loose | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled. | | spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation | 🔄 Dynamic | true | Enable flushable aggregation. If true, Gluten will try converting regular aggregation into Velox's flushable aggregation when applicable. A flushable aggregation could emit intermediate result at anytime when memory is full / data reduction ratio is low. | @@ -58,6 +59,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.memInitCapacity | 🔄 Dynamic | 8MB | The initial memory capacity to reserve for a newly created Velox query memory pool. | | spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks | 🔄 Dynamic | true | Whether to allow memory capacity transfer between memory pools from different tasks. | | spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. | +| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries are HTTP connections, not OS file descriptors. | | spark.gluten.sql.columnar.backend.velox.orc.scan.enabled | 🔄 Dynamic | true | Enable velox orc scan. If disabled, vanilla spark orc scan will be used. | | spark.gluten.sql.columnar.backend.velox.orcUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for ORC files. | | spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes | 🔄 Dynamic | 2MB | The maximum size in bytes for a Parquet dictionary page | @@ -75,7 +77,7 @@ nav_order: 16 | 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.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 | 1 | The IO threads for cache promoting | +| 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 | | spark.gluten.sql.columnar.backend.velox.ssdCacheShards | ⚓ Static | 1 | The cache shards | | spark.gluten.sql.columnar.backend.velox.ssdCacheSize | ⚓ Static | 1GB | The SSD cache size, will do memory caching only if this value = 0 | From 06ba77b959d372eed8a29ec93b1a11fb4943ab6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 30 Jun 2026 17:16:28 +0200 Subject: [PATCH 07/25] Add value validation for numCacheFileHandles and fileHandleExpirationDurationMs --- .../src/main/scala/org/apache/gluten/config/VeloxConfig.scala | 2 ++ 1 file changed, 2 insertions(+) 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 1d875f97535..7ec26366290 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 @@ -549,6 +549,7 @@ object VeloxConfig extends ConfigRegistry { "(ulimit -n). On remote object stores (S3, ABFS, GCS) entries are HTTP " + "connections, not OS file descriptors.") .intConf + .checkValue(_ > 0, "must be a positive number") .createWithDefault(10000) val COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS = @@ -559,6 +560,7 @@ object VeloxConfig extends ConfigRegistry { "from accumulating (e.g., expired HDFS leases, closed remote connections). " + "A value of 0 disables TTL-based eviction.") .longConf + .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based eviction)") .createWithDefault(600000L) // 10 minutes val DIRECTORY_SIZE_GUESS = From e92bc7708b00ca83e147e852ebed572c4770ca97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Wed, 8 Jul 2026 21:33:45 +0200 Subject: [PATCH 08/25] Clarify that remote FS cache entries still count toward OS limits Reword the numCacheFileHandles doc in both VeloxConfig.scala and velox-configuration.md to make clear that remote object store entries (S3, ABFS, GCS) represent network connections/sockets that can still count toward OS resource limits (ulimit -n), rather than implying they are exempt from FD limits. Addresses review feedback on PR #12400. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../main/scala/org/apache/gluten/config/VeloxConfig.scala | 5 +++-- docs/velox-configuration.md | 2 +- 2 files changed, 4 insertions(+), 3 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 7ec26366290..171dae27ed1 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 @@ -546,8 +546,9 @@ object VeloxConfig extends ConfigRegistry { "Maximum number of entries in the file handle cache. Each entry holds an open " + "file descriptor (local FS) or connection state (remote FS). Note that on " + "local filesystems, high values may approach the OS file descriptor limit " + - "(ulimit -n). On remote object stores (S3, ABFS, GCS) entries are HTTP " + - "connections, not OS file descriptors.") + "(ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent " + + "network connections/sockets rather than per-file OS file descriptors, but " + + "they can still count toward OS resource limits (ulimit -n).") .intConf .checkValue(_ > 0, "must be a positive number") .createWithDefault(10000) diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 270387cdab4..b100e35726f 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -59,7 +59,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.memInitCapacity | 🔄 Dynamic | 8MB | The initial memory capacity to reserve for a newly created Velox query memory pool. | | spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks | 🔄 Dynamic | true | Whether to allow memory capacity transfer between memory pools from different tasks. | | spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. | -| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries are HTTP connections, not OS file descriptors. | +| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent network connections/sockets rather than per-file OS file descriptors, but they can still count toward OS resource limits (ulimit -n). | | spark.gluten.sql.columnar.backend.velox.orc.scan.enabled | 🔄 Dynamic | true | Enable velox orc scan. If disabled, vanilla spark orc scan will be used. | | spark.gluten.sql.columnar.backend.velox.orcUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for ORC files. | | spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes | 🔄 Dynamic | 2MB | The maximum size in bytes for a Parquet dictionary page | From 6713dafff6c584236b3e22e6202a07e46ed2764e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Wed, 8 Jul 2026 21:59:50 +0200 Subject: [PATCH 09/25] Add checkGlutenPlan assertions to all file handle cache tests Each test now verifies that parquet scans execute through Gluten/Velox native operators rather than falling back to vanilla Spark. This ensures the tests actually exercise the file-handle cache behavior being validated. Addresses Copilot review feedback on PR #12400. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../execution/VeloxFileHandleCacheSuite.scala | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index d5c54a5febf..d776128bd3a 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -86,6 +86,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val expected = spark.read.parquet(path).count() assert(expected == 5000) + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(path)) + // Scan the same files multiple times - each should hit the cache for (i <- 1 to 5) { val count = spark.read.parquet(path).count() @@ -124,6 +127,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val fileCount = dir.listFiles().count(_.getName.endsWith(".parquet")) assert(fileCount >= 200, s"Expected at least 200 files, got $fileCount") + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(dir.getCanonicalPath)) + // Scan all files - should work without resource errors val count = spark.read.parquet(dir.getCanonicalPath).count() assert(count == 20000) @@ -154,6 +160,10 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val path = dir.getCanonicalPath + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer]( + spark.read.parquet(path).where("partition_key = 5")) + // Filter that matches ~10% of rows val filtered = spark.read.parquet(path).where("partition_key = 5").count() assert(filtered == 10000, s"Expected 10000 filtered rows, got $filtered") @@ -189,6 +199,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val count1 = spark.read.parquet(path).count() assert(count1 == 1000) + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(path)) + // Delete one parquet file val parquetFiles = dir.listFiles().filter(_.getName.endsWith(".parquet")) assert(parquetFiles.nonEmpty) @@ -239,6 +252,10 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { // First scan populates the cache val count1 = spark.read.parquet(path).count() assert(count1 == 5000) + + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(path)) + val sum1 = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) // Wait for TTL to expire (configured to 2s in sparkConf) @@ -270,6 +287,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val path = dir.getCanonicalPath + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(path)) + // Read all columns val allCols = spark.read.parquet(path).select("id", "doubled", "tripled", "text").count() assert(allCols == 5000) From 009907072f221f6c09aaec99615179e81f96eea5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 01:52:32 +0200 Subject: [PATCH 10/25] Fix config doc padding and address review comments - Fix velox-configuration.md line 59 trailing space padding to match flexmark-generated column width (was manually edited without regenerating, causing AllVeloxConfiguration test to fail in CI) - Quote variables and silence pushd/popd in get-velox.sh patch block; ensure popd runs before exit on error path - Update VeloxLocalCache.md ssdCacheIOThreads to reflect new default (4) and updated description - Reword TTL test comments to avoid overclaiming eviction verification; the test asserts scan correctness, not eviction observability Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../sql/execution/VeloxFileHandleCacheSuite.scala | 8 +++++--- docs/get-started/VeloxLocalCache.md | 2 +- docs/velox-configuration.md | 2 +- ep/build-velox/src/get-velox.sh | 11 ++++++----- 4 files changed, 13 insertions(+), 10 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index d776128bd3a..bd14dae9822 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -236,8 +236,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { "TTL-based eviction: scans succeed after cached handles expire", "3.5", "3.5") { - // Verify that after the TTL expires (2s, set in sparkConf), cached handles - // are evicted and subsequent scans re-open files correctly. + // Verify that scans still produce correct results after the configured TTL + // (2s, set in sparkConf) has elapsed. This exercises the path where cached + // handles may have been evicted and must be re-opened transparently. withTempPath { dir => spark @@ -261,7 +262,8 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { // Wait for TTL to expire (configured to 2s in sparkConf) Thread.sleep(3000) - // Scan after expiration: handles should be evicted and re-opened + // Scan after TTL expiration: verify results remain correct + // (handles may have been evicted and transparently re-opened) val count2 = spark.read.parquet(path).count() assert(count2 == 5000, s"Count mismatch after TTL expiration: expected 5000, got $count2") val sum2 = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) diff --git a/docs/get-started/VeloxLocalCache.md b/docs/get-started/VeloxLocalCache.md index 1c7c40ced02..e67e8f01f2d 100644 --- a/docs/get-started/VeloxLocalCache.md +++ b/docs/get-started/VeloxLocalCache.md @@ -13,7 +13,7 @@ spark.gluten.sql.columnar.backend.velox.memCacheSize // the total size of i spark.gluten.sql.columnar.backend.velox.ssdCachePath // the folder to store the cache files, default is "/tmp". spark.gluten.sql.columnar.backend.velox.ssdCacheSize // the total size of the SSD cache, default is 128MB. Velox will do in-mem cache only if this value is 0. spark.gluten.sql.columnar.backend.velox.ssdCacheShards // the shards of the SSD cache, default is 1. -spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads // the IO threads for cache promoting, default is 1. Velox will try to do "read-ahead" if this value is bigger than 1 +spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads // the number of IO threads for SSD cache read/write operations, default is 4. Velox will try to do "read-ahead" if this value is bigger than 1 spark.gluten.sql.columnar.backend.velox.ssdODirect // enable or disable O_DIRECT on cache write, default false. ``` diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index b100e35726f..fd7d2208944 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -59,7 +59,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.memInitCapacity | 🔄 Dynamic | 8MB | The initial memory capacity to reserve for a newly created Velox query memory pool. | | spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks | 🔄 Dynamic | true | Whether to allow memory capacity transfer between memory pools from different tasks. | | spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. | -| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent network connections/sockets rather than per-file OS file descriptors, but they can still count toward OS resource limits (ulimit -n). | +| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent network connections/sockets rather than per-file OS file descriptors, but they can still count toward OS resource limits (ulimit -n). | | spark.gluten.sql.columnar.backend.velox.orc.scan.enabled | 🔄 Dynamic | true | Enable velox orc scan. If disabled, vanilla spark orc scan will be used. | | spark.gluten.sql.columnar.backend.velox.orcUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for ORC files. | | spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes | 🔄 Dynamic | 2MB | The maximum size in bytes for a Parquet dictionary page | diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index 33390e31ae2..97eed3f1d43 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -162,17 +162,18 @@ function apply_compilation_fixes { # Wire file handle cache TTL config to SimpleLRUCache constructor. if [ -f "${CURRENT_DIR}/file-handle-cache-ttl.patch" ]; then - pushd $VELOX_HOME - if git apply --check ${CURRENT_DIR}/file-handle-cache-ttl.patch 2>/dev/null; then - git apply ${CURRENT_DIR}/file-handle-cache-ttl.patch + pushd "${VELOX_HOME}" > /dev/null + if git apply --check "${CURRENT_DIR}/file-handle-cache-ttl.patch" 2>/dev/null; then + git apply "${CURRENT_DIR}/file-handle-cache-ttl.patch" echo "Applied file-handle-cache-ttl.patch" - elif git apply --check -R ${CURRENT_DIR}/file-handle-cache-ttl.patch 2>/dev/null; then + elif git apply --check -R "${CURRENT_DIR}/file-handle-cache-ttl.patch" 2>/dev/null; then echo "file-handle-cache-ttl.patch already applied upstream, skipping" else + popd > /dev/null echo "ERROR: file-handle-cache-ttl.patch failed to apply and is not present upstream" >&2 exit 1 fi - popd + popd > /dev/null fi } From 6d12d2503a77a28d3f1f75df3dd9937dfcb8e146 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 07:03:11 +0200 Subject: [PATCH 11/25] Remove trailing whitespace; reword cache-hit comments - Remove trailing space from VeloxLocalCache.md line 16 - Reword repeated-scan test comments to focus on result consistency rather than assuming cache hits (TTL eviction may occur between iterations) Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../spark/sql/execution/VeloxFileHandleCacheSuite.scala | 6 +++--- docs/get-started/VeloxLocalCache.md | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index bd14dae9822..45ad7463c12 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -71,8 +71,8 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { "repeated scans produce consistent results (cache hit path)", "3.5", "3.5") { - // When file handles are cached, repeated scans of the same files must produce - // identical results. This exercises the cache hit path. + // Repeated scans of the same files must produce identical results regardless + // of whether handles are served from cache or re-opened after TTL eviction. withTempPath { dir => spark @@ -89,7 +89,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { // Verify scans go through Gluten/Velox checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(path)) - // Scan the same files multiple times - each should hit the cache + // Scan the same files multiple times - results must be consistent for (i <- 1 to 5) { val count = spark.read.parquet(path).count() assert( diff --git a/docs/get-started/VeloxLocalCache.md b/docs/get-started/VeloxLocalCache.md index e67e8f01f2d..432bfc38d2b 100644 --- a/docs/get-started/VeloxLocalCache.md +++ b/docs/get-started/VeloxLocalCache.md @@ -13,7 +13,7 @@ spark.gluten.sql.columnar.backend.velox.memCacheSize // the total size of i spark.gluten.sql.columnar.backend.velox.ssdCachePath // the folder to store the cache files, default is "/tmp". spark.gluten.sql.columnar.backend.velox.ssdCacheSize // the total size of the SSD cache, default is 128MB. Velox will do in-mem cache only if this value is 0. spark.gluten.sql.columnar.backend.velox.ssdCacheShards // the shards of the SSD cache, default is 1. -spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads // the number of IO threads for SSD cache read/write operations, default is 4. Velox will try to do "read-ahead" if this value is bigger than 1 +spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads // the number of IO threads for SSD cache read/write operations, default is 4. Velox will try to do "read-ahead" if this value is bigger than 1 spark.gluten.sql.columnar.backend.velox.ssdODirect // enable or disable O_DIRECT on cache write, default false. ``` From db3bcfff6b78e183b43a2e10acc33f7b0b7b8641 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 07:14:44 +0200 Subject: [PATCH 12/25] Fix spark-submit example in FileHandleCacheBenchmark scaladoc Use comma-separated --jars and add application jar as positional arg, matching spark-submit syntax. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../sql/execution/benchmark/FileHandleCacheBenchmark.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala index 36166b46e99..bd0fdb7844a 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala @@ -35,12 +35,14 @@ import org.apache.spark.benchmark.Benchmark * # With file handle cache enabled (default): * bin/spark-submit --class \ * --conf spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled=true \ - * --jars + * --jars , \ + * * * # With file handle cache disabled: * bin/spark-submit --class \ * --conf spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled=false \ - * --jars + * --jars , \ + * * }}} * * Expected result: with caching enabled, repeated scans should show lower wall-clock time due to From 2adbf271d97cdc46af836521e09d464c7a1be1c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 07:35:03 +0200 Subject: [PATCH 13/25] Add file handle cache configs to native conf defaults list Include numCacheFileHandles and fileHandleExpirationDurationMs in getNativeBackendConf defaults so the native conf map is consistent regardless of whether users explicitly set these to their default values. This prevents unnecessary backend/connector reuse misses. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../src/main/scala/org/apache/gluten/config/GlutenConfig.scala | 2 ++ 1 file changed, 2 insertions(+) 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 cdd1a3d9e9e..77ea6a10546 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 @@ -654,6 +654,8 @@ object GlutenConfig extends ConfigRegistry { ("spark.sql.orc.compression.codec", "snappy"), ("spark.sql.decimalOperations.allowPrecisionLoss", "true"), ("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled", "true"), + ("spark.gluten.sql.columnar.backend.velox.numCacheFileHandles", "10000"), + ("spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs", "600000"), ("spark.gluten.velox.awsSdkLogLevel", "FATAL"), ("spark.gluten.velox.s3UseProxyFromEnv", "false"), ("spark.gluten.velox.s3PayloadSigningPolicy", "Never"), From 7f4757cfd515cdeb428a6e026c74ee53ad334a36 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 08:15:27 +0200 Subject: [PATCH 14/25] Use standard a/b prefixes in file-handle-cache-ttl.patch Switch from i/w (mnemonic) to a/b prefixes for consistency with other patches under ep/build-velox/src/. Assisted-by: GitHub Copilot:claude-opus-4.6 --- ep/build-velox/src/file-handle-cache-ttl.patch | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/ep/build-velox/src/file-handle-cache-ttl.patch b/ep/build-velox/src/file-handle-cache-ttl.patch index 257bef55db1..aabb8614251 100644 --- a/ep/build-velox/src/file-handle-cache-ttl.patch +++ b/ep/build-velox/src/file-handle-cache-ttl.patch @@ -1,7 +1,7 @@ -diff --git i/velox/connectors/hive/HiveConnector.cpp w/velox/connectors/hive/HiveConnector.cpp +diff --git a/velox/connectors/hive/HiveConnector.cpp b/velox/connectors/hive/HiveConnector.cpp index abc13acea..861f95c7f 100644 ---- i/velox/connectors/hive/HiveConnector.cpp -+++ w/velox/connectors/hive/HiveConnector.cpp +--- a/velox/connectors/hive/HiveConnector.cpp ++++ b/velox/connectors/hive/HiveConnector.cpp @@ -40,7 +40,8 @@ HiveConnector::HiveConnector( fileHandleFactory_( hiveConfig_->isFileHandleCacheEnabled() From 1856ae659c584050194c4833dbb9a98361853ed2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 08:32:26 +0200 Subject: [PATCH 15/25] Rename repeated-scan test to drop 'cache hit path' claim The test asserts result consistency, not cache hits. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index 45ad7463c12..364cfed20fc 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -68,7 +68,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { } testWithSpecifiedSparkVersion( - "repeated scans produce consistent results (cache hit path)", + "repeated scans produce consistent results", "3.5", "3.5") { // Repeated scans of the same files must produce identical results regardless From bff2a8c93527a92fb4a7f75bc68aa447b7531775 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 08:44:44 +0200 Subject: [PATCH 16/25] Use timeConf for fileHandleExpirationDurationMs Change from .longConf to .timeConf(TimeUnit.MILLISECONDS) for consistency with reclaimMaxWaitMs and asyncTimeoutOnTaskStopping. Update docs default display from '600000' to '600000ms'. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../src/main/scala/org/apache/gluten/config/VeloxConfig.scala | 2 +- docs/velox-configuration.md | 2 +- 2 files changed, 2 insertions(+), 2 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 171dae27ed1..eaa411c1173 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 @@ -560,7 +560,7 @@ object VeloxConfig extends ConfigRegistry { "within this duration are evicted from the cache. This prevents stale handles " + "from accumulating (e.g., expired HDFS leases, closed remote connections). " + "A value of 0 disables TTL-based eviction.") - .longConf + .timeConf(TimeUnit.MILLISECONDS) .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based eviction)") .createWithDefault(600000L) // 10 minutes diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index fd7d2208944..0231aae4742 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -31,7 +31,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild | 🔄 Dynamic | false | Enable driver-side broadcast hash table build. When enabled, the hash table is built and serialized on the driver, then broadcast to executors. When disabled, each executor builds its own hash table from the broadcast data. | | spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation | 🔄 Dynamic | false | Enable validation fallback for TimestampNTZ type. When true, any plan containing TimestampNTZ will fall back to Spark execution. When false, allows native execution for TimestampNTZ scan. | | spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | true | Enables caching of file handles to avoid repeated open/close overhead on remote filesystems. Should be disabled if files are mutable, i.e. file content may change while file path stays the same. | -| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000 | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | +| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000ms | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | | spark.gluten.sql.columnar.backend.velox.filePreloadThreshold | ⚓ Static | 1MB | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold | | spark.gluten.sql.columnar.backend.velox.floatingPointMode | 🔄 Dynamic | loose | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled. | | spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation | 🔄 Dynamic | true | Enable flushable aggregation. If true, Gluten will try converting regular aggregation into Velox's flushable aggregation when applicable. A flushable aggregation could emit intermediate result at anytime when memory is full / data reduction ratio is low. | From 6abaf05a87cf42b32c7d62c3edcc5575746fbd44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 08:49:12 +0200 Subject: [PATCH 17/25] Extract TTL constant; remove remaining 'cache hit path' comments - Define ttlMs and ttlWaitMs constants to avoid duplicated magic numbers between sparkConf and Thread.sleep - Reword remaining 'cache hit path' comments to 'results must remain consistent' Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../sql/execution/VeloxFileHandleCacheSuite.scala | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index 364cfed20fc..e2beee8f27c 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -31,10 +31,14 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { override protected val resourcePath: String = "/parquet-for-read" override protected val fileFormat: String = "parquet" + // TTL for file handle cache eviction (used in sparkConf and sleep calculations) + private val ttlMs = 2000 + private val ttlWaitMs = ttlMs + 1000 // TTL + buffer for eviction to take effect + override protected def sparkConf: SparkConf = { super.sparkConf .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED.key, "true") - .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, "2000") + .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, ttlMs.toString) .set(VeloxConfig.COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES.key, "10000") } @@ -134,7 +138,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val count = spark.read.parquet(dir.getCanonicalPath).count() assert(count == 20000) - // Scan again (cache hit path) - should also work + // Scan again - results must remain consistent val count2 = spark.read.parquet(dir.getCanonicalPath).count() assert(count2 == 20000) } @@ -172,7 +176,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val rangeFiltered = spark.read.parquet(path).where("id >= 50000").count() assert(rangeFiltered == 50000, s"Expected 50000 range-filtered rows, got $rangeFiltered") - // Re-run same filters (cache hit path) + // Re-run same filters - results must remain consistent val filtered2 = spark.read.parquet(path).where("partition_key = 5").count() assert(filtered2 == filtered, "Filtered count mismatch on repeated scan") } @@ -259,8 +263,8 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { val sum1 = spark.read.parquet(path).selectExpr("sum(id)").collect()(0).getLong(0) - // Wait for TTL to expire (configured to 2s in sparkConf) - Thread.sleep(3000) + // Wait for TTL to expire + Thread.sleep(ttlWaitMs) // Scan after TTL expiration: verify results remain correct // (handles may have been evicted and transparently re-opened) From 3b2944d46dd0eb8d115b2a4287d804043ca3b10e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 08:52:56 +0200 Subject: [PATCH 18/25] Add checkValue validation for ssdCacheIOThreads Reject 0 or negative values early since the config is used to construct folly::IOThreadPoolExecutor. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../src/main/scala/org/apache/gluten/config/VeloxConfig.scala | 1 + 1 file changed, 1 insertion(+) 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 eaa411c1173..d8d80073b8b 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 @@ -182,6 +182,7 @@ object VeloxConfig extends ConfigRegistry { buildStaticConf("spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads") .doc("The number of IO threads for SSD cache read/write operations") .intConf + .checkValue(_ > 0, "must be a positive number") .createWithDefault(4) val COLUMNAR_VELOX_SSD_ODIRECT_ENABLED = From 103197e9fa521598219e24a9adb28a4f2c9e2f66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 9 Jul 2026 09:12:40 +0200 Subject: [PATCH 19/25] Revert timeConf back to longConf for fileHandleExpirationDurationMs The native layer reads this config via conf->get() which only parses plain integers. Using timeConf would allow users to set values like '10min' that pass Spark-side validation but fail native init. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../src/main/scala/org/apache/gluten/config/VeloxConfig.scala | 2 +- docs/velox-configuration.md | 2 +- 2 files changed, 2 insertions(+), 2 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 d8d80073b8b..2968b4cc320 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 @@ -561,7 +561,7 @@ object VeloxConfig extends ConfigRegistry { "within this duration are evicted from the cache. This prevents stale handles " + "from accumulating (e.g., expired HDFS leases, closed remote connections). " + "A value of 0 disables TTL-based eviction.") - .timeConf(TimeUnit.MILLISECONDS) + .longConf .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based eviction)") .createWithDefault(600000L) // 10 minutes diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 0231aae4742..fd7d2208944 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -31,7 +31,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild | 🔄 Dynamic | false | Enable driver-side broadcast hash table build. When enabled, the hash table is built and serialized on the driver, then broadcast to executors. When disabled, each executor builds its own hash table from the broadcast data. | | spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation | 🔄 Dynamic | false | Enable validation fallback for TimestampNTZ type. When true, any plan containing TimestampNTZ will fall back to Spark execution. When false, allows native execution for TimestampNTZ scan. | | spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | true | Enables caching of file handles to avoid repeated open/close overhead on remote filesystems. Should be disabled if files are mutable, i.e. file content may change while file path stays the same. | -| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000ms | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | +| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000 | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | | spark.gluten.sql.columnar.backend.velox.filePreloadThreshold | ⚓ Static | 1MB | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold | | spark.gluten.sql.columnar.backend.velox.floatingPointMode | 🔄 Dynamic | loose | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled. | | spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation | 🔄 Dynamic | true | Enable flushable aggregation. If true, Gluten will try converting regular aggregation into Velox's flushable aggregation when applicable. A flushable aggregation could emit intermediate result at anytime when memory is full / data reduction ratio is low. | From 7cd516a40e0767691baf68a5feea5fc875575a10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 10 Jul 2026 10:04:57 +0200 Subject: [PATCH 20/25] Use timeConf for fileHandleExpirationDurationMs Change from .longConf to .timeConf(TimeUnit.MILLISECONDS) as suggested by reviewer. GlutenConfigUtil.parseConfig converts time strings (e.g. "10m") to plain millisecond values before passing to native, so conf->get() on the C++ side always receives a numeric string. This is consistent with reclaimMaxWaitMs and asyncTimeoutOnTaskStopping. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../src/main/scala/org/apache/gluten/config/VeloxConfig.scala | 2 +- docs/velox-configuration.md | 2 +- 2 files changed, 2 insertions(+), 2 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 2968b4cc320..d8d80073b8b 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 @@ -561,7 +561,7 @@ object VeloxConfig extends ConfigRegistry { "within this duration are evicted from the cache. This prevents stale handles " + "from accumulating (e.g., expired HDFS leases, closed remote connections). " + "A value of 0 disables TTL-based eviction.") - .longConf + .timeConf(TimeUnit.MILLISECONDS) .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based eviction)") .createWithDefault(600000L) // 10 minutes diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index fd7d2208944..0231aae4742 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -31,7 +31,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild | 🔄 Dynamic | false | Enable driver-side broadcast hash table build. When enabled, the hash table is built and serialized on the driver, then broadcast to executors. When disabled, each executor builds its own hash table from the broadcast data. | | spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation | 🔄 Dynamic | false | Enable validation fallback for TimestampNTZ type. When true, any plan containing TimestampNTZ will fall back to Spark execution. When false, allows native execution for TimestampNTZ scan. | | spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | true | Enables caching of file handles to avoid repeated open/close overhead on remote filesystems. Should be disabled if files are mutable, i.e. file content may change while file path stays the same. | -| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000 | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | +| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 600000ms | Expiration time in milliseconds for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). A value of 0 disables TTL-based eviction. | | spark.gluten.sql.columnar.backend.velox.filePreloadThreshold | ⚓ Static | 1MB | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold | | spark.gluten.sql.columnar.backend.velox.floatingPointMode | 🔄 Dynamic | loose | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled. | | spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation | 🔄 Dynamic | true | Enable flushable aggregation. If true, Gluten will try converting regular aggregation into Velox's flushable aggregation when applicable. A flushable aggregation could emit intermediate result at anytime when memory is full / data reduction ratio is low. | From 1152a16429fbe8c64c5dbd734a6243e928f39509 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 10 Jul 2026 14:15:42 +0200 Subject: [PATCH 21/25] Quote path expansions in apply_compilation_fixes for safety Quote ${CURRENT_DIR} and ${VELOX_HOME} in cp and git-add commands to prevent word-splitting if paths contain spaces, consistent with the file-handle-cache-ttl patch block below. Assisted-by: GitHub Copilot:claude-opus-4.6 --- ep/build-velox/src/get-velox.sh | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index 97eed3f1d43..b94312dca4a 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -156,9 +156,9 @@ function apply_compilation_fixes { if [ "$OS" == "Linux" ]; then SUDO_CMD="sudo" fi - $SUDO_CMD cp ${CURRENT_DIR}/modify_arrow.patch ${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/ + $SUDO_CMD cp "${CURRENT_DIR}/modify_arrow.patch" "${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/" - git add ${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/modify_arrow.patch # to avoid the file from being deleted by git clean -dffx :/ + git add "${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/modify_arrow.patch" # to avoid the file from being deleted by git clean -dffx :/ # Wire file handle cache TTL config to SimpleLRUCache constructor. if [ -f "${CURRENT_DIR}/file-handle-cache-ttl.patch" ]; then From 2f5abbf69123116ad8ca72baf37d48a9c1afbb17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 10 Jul 2026 15:40:25 +0200 Subject: [PATCH 22/25] Make apply_compilation_fixes self-contained; clarify TTL test scope Move pushd into $VELOX_HOME to the top of apply_compilation_fixes so git add targets the correct working tree regardless of the caller's current directory. The git add path is now repo-relative. Update the TTL eviction test comment to clarify that this is a correctness guard (scans produce correct results after TTL expiration), not an eviction-observability test (Velox exposes no JVM-visible eviction counter). Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../sql/execution/VeloxFileHandleCacheSuite.scala | 11 ++++++++--- ep/build-velox/src/get-velox.sh | 10 +++++++--- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index e2beee8f27c..da2a9fa8555 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -240,9 +240,14 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { "TTL-based eviction: scans succeed after cached handles expire", "3.5", "3.5") { - // Verify that scans still produce correct results after the configured TTL - // (2s, set in sparkConf) has elapsed. This exercises the path where cached - // handles may have been evicted and must be re-opened transparently. + // Correctness guard: verify that scans produce correct results after the + // configured TTL (2s, set in sparkConf) has elapsed and cached handles may + // have been evicted. This does NOT directly assert that eviction occurred + // (Velox exposes no JVM-visible eviction counter), but it exercises the + // re-open path: if a handle was evicted, the scan must transparently + // re-open the file and return the same data. Combined with the "scan after + // file deletion" test -- which proves cached handles keep the inode alive -- + // this gives reasonable coverage that the TTL wiring works end-to-end. withTempPath { dir => spark diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index b94312dca4a..ce81dc6aa23 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -158,11 +158,14 @@ function apply_compilation_fixes { fi $SUDO_CMD cp "${CURRENT_DIR}/modify_arrow.patch" "${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/" - git add "${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/modify_arrow.patch" # to avoid the file from being deleted by git clean -dffx :/ + # Run git commands inside the Velox repo so git add targets the correct + # working tree regardless of the caller's current directory. + pushd "${VELOX_HOME}" > /dev/null + + git add "CMake/resolve_dependency_modules/arrow/modify_arrow.patch" # to avoid the file from being deleted by git clean -dffx :/ # Wire file handle cache TTL config to SimpleLRUCache constructor. if [ -f "${CURRENT_DIR}/file-handle-cache-ttl.patch" ]; then - pushd "${VELOX_HOME}" > /dev/null if git apply --check "${CURRENT_DIR}/file-handle-cache-ttl.patch" 2>/dev/null; then git apply "${CURRENT_DIR}/file-handle-cache-ttl.patch" echo "Applied file-handle-cache-ttl.patch" @@ -173,8 +176,9 @@ function apply_compilation_fixes { echo "ERROR: file-handle-cache-ttl.patch failed to apply and is not present upstream" >&2 exit 1 fi - popd > /dev/null fi + + popd > /dev/null } function setup_linux { From 1436dbc12f056aac6681156887a7f68b9a57cd0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 10 Jul 2026 16:44:06 +0200 Subject: [PATCH 23/25] Harden deletion test: check cause chain before message matching Rename test to 'scan after file deletion does not silently return wrong data' to match what the assertions actually validate. Improve the catch block to check for FileNotFoundException and NoSuchFileException directly, then walk the exception cause chain for wrapped exceptions (e.g., SparkException), before falling back to message-based matching for FS implementations with custom exception types. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../execution/VeloxFileHandleCacheSuite.scala | 32 ++++++++++++++++--- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index da2a9fa8555..00f1f8f79f9 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -21,6 +21,9 @@ import org.apache.gluten.execution.{BasicScanExecTransformer, VeloxWholeStageTra import org.apache.spark.SparkConf +import java.io.FileNotFoundException +import java.nio.file.NoSuchFileException + /** * Test suite for Velox file handle cache behavior. * @@ -35,6 +38,16 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { private val ttlMs = 2000 private val ttlWaitMs = ttlMs + 1000 // TTL + buffer for eviction to take effect + /** Walks the exception cause chain looking for an instance of the given type. */ + private def hasCauseOfType(e: Throwable, cls: Class[_ <: Throwable]): Boolean = { + var cause = e.getCause + while (cause != null) { + if (cls.isInstance(cause)) return true + cause = cause.getCause + } + false + } + override protected def sparkConf: SparkConf = { super.sparkConf .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED.key, "true") @@ -183,12 +196,14 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { } testWithSpecifiedSparkVersion( - "scan after file deletion produces appropriate error or empty result", + "scan after file deletion does not silently return wrong data", "3.5", "3.5") { // If a file is deleted between scans, the next scan should either: - // - Succeed (if the cached FD still works on Linux with unlinked inodes) - // - Produce an error (not silently return wrong data) + // - Succeed with the original count (cached FD keeps inode alive on Linux) + // - Succeed with a reduced count (deleted file not accessible) + // - Throw a file-not-found error + // The key invariant: it must NOT silently return incorrect data. withTempPath { dir => spark @@ -224,14 +239,21 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { count2 == count1 || count2 == count1 - deletedRows, s"Unexpected count after deletion: $count2 (original: $count1, deleted: $deletedRows)") } catch { + case e: FileNotFoundException => + // Direct file-not-found exception. + case e: NoSuchFileException => + // NIO equivalent of FileNotFoundException. + case e: Exception if hasCauseOfType(e, classOf[FileNotFoundException]) || + hasCauseOfType(e, classOf[NoSuchFileException]) => + // Wrapped file-not-found in the cause chain (e.g., SparkException wrapping). case e: Exception if e.getMessage != null && (e.getMessage.contains("FileNotFoundException") || e.getMessage.contains("No such file") || e.getMessage.contains("Path does not exist") || e.getMessage.contains("does not exist")) => - // Acceptable: the scan failed because the deleted file is no longer accessible. - // The important thing is that it does not silently return wrong data. + // Fallback: message-based matching for FS implementations that use + // custom exception types (e.g., Hadoop, Velox native errors). } } } From c6a17a5d9ed34daf9982dcf6687c06fa06bc627d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 10 Jul 2026 19:08:39 +0200 Subject: [PATCH 24/25] Rename TTL test to reflect what it actually asserts Rename from 'TTL-based eviction: scans succeed after cached handles expire' to 'scans remain correct after TTL expiration window' since the test verifies correctness, not that eviction occurred (Velox exposes no JVM-visible eviction counter). Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index 00f1f8f79f9..e29a861fa17 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -259,7 +259,7 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { } testWithSpecifiedSparkVersion( - "TTL-based eviction: scans succeed after cached handles expire", + "scans remain correct after TTL expiration window", "3.5", "3.5") { // Correctness guard: verify that scans produce correct results after the From 99802e5612031868db1560b638f1ede36c506f8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 10 Jul 2026 21:00:40 +0200 Subject: [PATCH 25/25] Fix scalafmt: move case guard to its own line The CI spotless plugin (v2.27.2) requires the 'if' guard on a case pattern to be on a separate line, not inline with 'case'. Assisted-by: GitHub Copilot:claude-opus-4.6 --- .../spark/sql/execution/VeloxFileHandleCacheSuite.scala | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala index e29a861fa17..4ece34c7aa9 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -243,8 +243,9 @@ class VeloxFileHandleCacheSuite extends VeloxWholeStageTransformerSuite { // Direct file-not-found exception. case e: NoSuchFileException => // NIO equivalent of FileNotFoundException. - case e: Exception if hasCauseOfType(e, classOf[FileNotFoundException]) || - hasCauseOfType(e, classOf[NoSuchFileException]) => + case e: Exception + if hasCauseOfType(e, classOf[FileNotFoundException]) || + hasCauseOfType(e, classOf[NoSuchFileException]) => // Wrapped file-not-found in the cause chain (e.g., SparkException wrapping). case e: Exception if e.getMessage != null &&