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 2a3f34483e..d8d80073b8 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,10 @@ 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) + .checkValue(_ > 0, "must be a positive number") + .createWithDefault(4) val COLUMNAR_VELOX_SSD_ODIRECT_ENABLED = buildStaticConf("spark.gluten.sql.columnar.backend.velox.ssdODirect") @@ -534,10 +535,35 @@ 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). 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).") + .intConf + .checkValue(_ > 0, "must be a positive number") + .createWithDefault(10000) + + 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.") + .timeConf(TimeUnit.MILLISECONDS) + .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based eviction)") + .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 0000000000..4ece34c7aa --- /dev/null +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala @@ -0,0 +1,342 @@ +/* + * 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 + +import java.io.FileNotFoundException +import java.nio.file.NoSuchFileException + +/** + * 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" + + // 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 + + /** 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") + .set(VeloxConfig.COLUMNAR_VELOX_FILE_HANDLE_EXPIRATION_DURATION_MS.key, ttlMs.toString) + .set(VeloxConfig.COLUMNAR_VELOX_NUM_CACHE_FILE_HANDLES.key, "10000") + } + + 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", + "3.5", + "3.5") { + // 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 + .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) + + // Verify scans go through Gluten/Velox + checkGlutenPlan[BasicScanExecTransformer](spark.read.parquet(path)) + + // Scan the same files multiple times - results must be consistent + 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 >= 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) + + // Scan again - results must remain consistent + 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 + + // 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") + + // 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 - results must remain consistent + 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 does not silently return wrong data", + "3.5", + "3.5") { + // If a file is deleted between scans, the next scan should either: + // - 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 + .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) + + // 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) + val deletedFile = parquetFiles.head + val deletedRows = spark.read.parquet(deletedFile.getCanonicalPath).count() + 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. + // 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 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")) => + // Fallback: message-based matching for FS implementations that use + // custom exception types (e.g., Hadoop, Velox native errors). + } + } + } + + testWithSpecifiedSparkVersion( + "scans remain correct after TTL expiration window", + "3.5", + "3.5") { + // 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 + .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) + + // 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 + Thread.sleep(ttlWaitMs) + + // 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) + assert(sum2 == sum1, s"Sum mismatch after TTL expiration: expected $sum1, got $sum2") + } + } + + 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 + + // 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) + + // 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.count() == 5000) + + // 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 0000000000..bd0fdb7844 --- /dev/null +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala @@ -0,0 +1,127 @@ +/* + * 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 ba9bcda5a7..36d029e8ff 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 = 10000; + +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 7ee2deae8a..240033badb 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/docs/get-started/VeloxLocalCache.md b/docs/get-started/VeloxLocalCache.md index 1c7c40ced0..432bfc38d2 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 965b03802e..0231aae474 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 | 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. | @@ -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 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 | @@ -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 | 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 0000000000..aabb861425 --- /dev/null +++ b/ep/build-velox/src/file-handle-cache-ttl.patch @@ -0,0 +1,14 @@ +diff --git a/velox/connectors/hive/HiveConnector.cpp b/velox/connectors/hive/HiveConnector.cpp +index abc13acea..861f95c7f 100644 +--- a/velox/connectors/hive/HiveConnector.cpp ++++ b/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 6443710a93..ce81dc6aa2 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -156,9 +156,29 @@ 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 :/ + # 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 + 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 + popd > /dev/null + echo "ERROR: file-handle-cache-ttl.patch failed to apply and is not present upstream" >&2 + exit 1 + fi + fi + + popd > /dev/null } 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 93fb3888e9..77ea6a1054 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,9 @@ 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.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"),