This is an automated email from the ASF dual-hosted git repository.
jackylee-ch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new ed81935c40 [VL] Enable file handle cache by default with TTL-based
eviction (#12400)
ed81935c40 is described below
commit ed81935c40db39b4d23f7d7f420113eccfbaa243
Author: IsmaΓ«l MejΓa <[email protected]>
AuthorDate: Thu Jul 23 15:17:13 2026 +0200
[VL] Enable file handle cache by default with TTL-based eviction (#12400)
---
.../org/apache/gluten/config/VeloxConfig.scala | 39 ++-
.../sql/execution/VeloxFileHandleCacheSuite.scala | 366 +++++++++++++++++++++
cpp/velox/config/VeloxConfig.h | 13 +-
cpp/velox/utils/ConfigExtractor.cc | 4 +
docs/get-started/VeloxLocalCache.md | 2 +-
docs/velox-configuration.md | 6 +-
.../org/apache/gluten/config/GlutenConfig.scala | 4 +-
7 files changed, 423 insertions(+), 11 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 2a3f34483e..0dc6f052e0 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,38 @@ 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 open file handles to avoid repeated open/close
overhead. " +
+ "Benefits both local filesystems (fewer open/close syscalls and file
descriptor " +
+ "churn) and remote filesystems/object stores (reused connection
state). Should be " +
+ "disabled if files are mutable, i.e. file content may change while
the 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 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). Accepts a Spark
duration string " +
+ "(e.g., \"10m\", \"600s\") or a plain number interpreted as
milliseconds. A value " +
+ "of 0 disables TTL-based eviction.")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based
eviction)")
+ .createWithDefaultString("10m")
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..c7685a4b93
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala
@@ -0,0 +1,366 @@
+/*
+ * 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).
+ // Kept small to minimize CI time; the TTL test only asserts scan
correctness after
+ // the window elapses (it passes whether or not eviction has occurred), so a
short
+ // wait is sufficient and does not introduce flakiness.
+ private val ttlMs = 500
+ private val ttlWaitMs = ttlMs + 500 // TTL + buffer for lazy eviction on
next access
+
+ /** 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")
+ }
+
+ test("basic scan correctness with file handle cache enabled") {
+ // 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]
+ }
+ }
+ }
+
+ test("repeated scans produce consistent results") {
+ // 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")
+ }
+ }
+ }
+
+ test("many small files do not cause errors with file handle cache") {
+ // 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)
+ }
+ }
+
+ test("filtered scan correctness with file handle cache") {
+ // 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")
+ }
+ }
+
+ test("scan after file deletion does not silently return wrong data") {
+ // 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).
+ }
+ }
+ }
+
+ test("scans remain correct after TTL expiration window") {
+ // Correctness guard: verify that scans produce correct results after the
+ // configured TTL (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")
+
+ // Best-effort eviction probe: delete a file, wait past the TTL, then
scan
+ // again. This drives the "cached handle expired -> re-open" path for
a file
+ // that no longer exists. We cannot assert that eviction definitively
+ // occurred (Velox exposes no JVM-visible eviction counter, and on
Linux a
+ // still-cached FD keeps the unlinked inode readable), so we assert
the only
+ // invariant that must always hold: the scan must NOT silently return
+ // corrupted data. It must return either the full count (FD kept the
inode
+ // alive), a reduced count (handle was evicted and the file is gone),
or
+ // throw a file-not-found error.
+ 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}")
+
+ Thread.sleep(ttlWaitMs)
+
+ try {
+ val count3 = spark.read.parquet(path).count()
+ assert(
+ count3 == count2 || count3 == count2 - deletedRows,
+ s"Unexpected count after TTL + deletion: $count3 " +
+ s"(pre-deletion: $count2, 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).
+ }
+ }
+ }
+
+ test("column pruning with cached file handles") {
+ // 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/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<facebook::velox::config::ConfigBase>
createHiveConnectorConfig(
hiveConfMap[facebook::velox::connector::hive::HiveConfig::kEnableFileHandleCache]
=
conf->get<bool>(kVeloxFileHandleCacheEnabled,
kVeloxFileHandleCacheEnabledDefault) ? "true" : "false";
+
hiveConfMap[facebook::velox::connector::hive::HiveConfig::kNumCacheFileHandles]
=
+ std::to_string(conf->get<int32_t>(kVeloxNumCacheFileHandles,
kVeloxNumCacheFileHandlesDefault));
+
hiveConfMap[facebook::velox::connector::hive::HiveConfig::kFileHandleExpirationDurationMs]
= std::to_string(
+ conf->get<int64_t>(kVeloxFileHandleExpirationDurationMs,
kVeloxFileHandleExpirationDurationMsDefault));
hiveConfMap[facebook::velox::connector::hive::HiveConfig::kMaxCoalescedBytes] =
conf->get<std::string>(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..4787b07554 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 open file handles
to avoid repeated open/close overhead. Benefits both local filesystems (fewer
open/close syscalls and file descriptor churn) and remote filesystems/object
stores (reused connection state). Should be disabled if files are mutable, i.e.
file content may change while the file path stays the same.
[...]
+| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs
| β Static | 10m | Expiration time 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). Accepts a Spark duration string (e.g., "10m",
"600s") or a plain number interpreted as milliseconds. 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 the [...]
| 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/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
index 98fe6a0657..e4fafa3599 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
@@ -649,7 +649,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"),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
