[VL] Enable file handle cache by default with TTL-based eviction#12400
[VL] Enable file handle cache by default with TTL-based eviction#12400iemejia wants to merge 29 commits into
Conversation
|
Run Gluten Clickhouse CI on x86 |
2808437 to
b794974
Compare
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR updates Velox backend defaults to enable file-handle caching by default, adds TTL-based eviction wiring in the Velox Hive connector (via an applied patch during Velox fetch), and exposes new Spark configs for tuning cache size and expiration. It also adds a dedicated test suite plus a benchmark to validate and measure the impact of the cache.
Changes:
- Enable
spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabledby default and increase SSD cache IO threads default from 1 to 4. - Propagate new cache tuning configs (
numCacheFileHandles,fileHandleExpirationDurationMs) into the Velox Hive connector configuration, and wire TTL into theSimpleLRUCacheconstructor via a build-time patch. - Add
VeloxFileHandleCacheSuiteandFileHandleCacheBenchmarkto validate correctness and measure performance.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala | Changes the default Spark-side config map to enable Velox file-handle caching by default. |
| ep/build-velox/src/get-velox.sh | Applies a new Velox patch (if present) to wire the file-handle TTL into the cache constructor. |
| ep/build-velox/src/file-handle-cache-ttl.patch | Patch that passes fileHandleExpirationDurationMs into Velox SimpleLRUCache for file handles. |
| cpp/velox/utils/ConfigExtractor.cc | Propagates numCacheFileHandles and fileHandleExpirationDurationMs into Velox Hive connector config. |
| cpp/velox/config/VeloxConfig.h | Adds new config keys/defaults and updates defaults for file-handle cache enablement and SSD cache IO threads. |
| backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala | Exposes new Spark configs and updates defaults/docs for SSD IO threads and file-handle cache. |
| backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala | Adds coverage for file-handle cache correctness and edge cases (but currently has issues that need fixing). |
| backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/FileHandleCacheBenchmark.scala | Adds a benchmark to compare repeated scans with file-handle cache enabled vs disabled. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // 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)") |
There was a problem hiding this comment.
Fixed. Wrapped the second scan in a try-catch — if the scan throws because the file is no longer accessible, that is acceptable behavior. The important invariant is that it must not silently return wrong data.
| // 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"))) |
There was a problem hiding this comment.
Fixed. Moved the schema assertion to the DataFrame before collect(): check subset1Df.schema.fieldNames first, then collect and assert row count.
| # 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 |
There was a problem hiding this comment.
Fixed. The script now distinguishes three cases: (1) patch applies cleanly — apply it, (2) reverse-apply check passes — patch is already present upstream, skip, (3) neither — fail the build with an error. This ensures the TTL wiring is never silently absent.
| // 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"))) |
There was a problem hiding this comment.
Fixed. Moved the schema assertion to the DataFrame before collect(): check subset1Df.schema.fieldNames first, then collect and assert row count.
| assert(parquetFiles.nonEmpty) | ||
| val deletedFile = parquetFiles.head | ||
| val deletedRows = spark.read.parquet(deletedFile.getCanonicalPath).count() | ||
| deletedFile.delete() |
There was a problem hiding this comment.
Fixed. Added assert(deletedFile.delete(), ...) to fail if deletion does not succeed.
| 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) |
There was a problem hiding this comment.
Good point. Reduced the default from 20000 to 10000. Also expanded the doc to clarify that on remote object stores (S3, ABFS, GCS) entries are HTTP connections, not OS file descriptors, so the FD concern primarily applies to local filesystems.
|
Run Gluten Clickhouse CI on x86 |
041b8ad to
aac388b
Compare
|
Run Gluten Clickhouse CI on x86 |
| hiveConfMap[facebook::velox::connector::hive::HiveConfig::kFileHandleExpirationDurationMs] = std::to_string( | ||
| conf->get<int64_t>(kVeloxFileHandleExpirationDurationMs, kVeloxFileHandleExpirationDurationMsDefault)); |
There was a problem hiding this comment.
This is the output of clang-format-15, which is the project's authoritative formatter. The line break is where clang-format places it given the column limit. Reformatting it differently would cause the format check to fail.
| val fileCount = dir.listFiles().count(_.getName.endsWith(".parquet")) | ||
| assert(fileCount >= 100, s"Expected at least 100 files, got $fileCount") |
There was a problem hiding this comment.
Fixed. Tightened the assertion from >= 100 to >= 200 to match the repartition(200) call.
| 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, "10000") | ||
| } |
There was a problem hiding this comment.
Fixed. Set the suite-level TTL to 2 seconds and added a dedicated test that scans files, waits 3 seconds for handle expiration, then verifies that subsequent scans still return correct results after handles are evicted and re-opened.
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
| .longConf | ||
| .createWithDefault(600000L) // 10 minutes |
There was a problem hiding this comment.
Fixed. Added .checkValue(_ >= 0, "must be a non-negative number (0 disables TTL-based eviction)") — rejects negative values while preserving the documented 0-to-disable behavior.
Reject 0 or negative values early since the config is used to construct folly::IOThreadPoolExecutor. Assisted-by: GitHub Copilot:claude-opus-4.6
The native layer reads this config via conf->get<int64_t>() 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
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<int64_t>() on the C++ side always receives a numeric string. This is consistent with reclaimMaxWaitMs and asyncTimeoutOnTaskStopping. Assisted-by: GitHub Copilot:claude-opus-4.6
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
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
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
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
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
The rebase bumped the pinned Velox branch to dft-2026_07_16, which already wires fileHandleExpirationDurationMs() into the SimpleLRUCache constructor in HiveConnector.cpp. The patch and its apply logic in get-velox.sh are now redundant, so remove them and revert get-velox.sh to origin/main.
99802e5 to
1d2d4f7
Compare
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (2)
docs/velox-configuration.md:34
- The default value is documented as
600000ms, but this config is passed to native code and parsed as an integer millisecond count (e.g., viaconf->get<int64_t>(...)). A value with a unit suffix likemswill fail to parse and can break backend initialization. Document the default as a plain number of milliseconds (no unit suffix), and consider explicitly noting that the value must be numeric ms.
| 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. |
backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala:564
- This config is defined with
timeConf(TimeUnit.MILLISECONDS), which encourages values like10m/600000ms, but the value is forwarded to native code and parsed as anint64(numeric milliseconds). If a user supplies a value with a unit suffix, native parsing will fail at init time. Prefer a numeric config type (longConf) and document it as a plain millisecond count.
.timeConf(TimeUnit.MILLISECONDS)
|
The failing All other Spark versions pass, including this PR's |
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (3)
backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala:255
- The message-based fallback in this catch block is overly broad: matching the substring "does not exist" can accidentally swallow unrelated failures (e.g., "Table does not exist"), causing the test to pass when it should fail. Narrow the message matching to file-not-found-specific phrases only.
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")) =>
backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala:566
- This config is defined as
timeConf(TimeUnit.MILLISECONDS), which lets users set values like10m, butGlutenConfig.getNativeBackendConfforwards the raw SparkConf string to native andConfigExtractor.ccreads it as anint64_t. If a user sets10m, Spark will accept it, but native parsing is likely to fail at runtime. Consider usinglongConfhere to enforce numeric milliseconds at SparkConf validation time (or alternatively teach the native side to parse Spark time strings).
"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
cpp/velox/utils/ConfigExtractor.cc:328
- Minor style/consistency: this assignment is formatted differently than the surrounding hiveConfMap entries (the
std::to_string(starts on the same line as=). Adjusting it to match the nearby formatting improves readability and makes clang-format output more predictable.
hiveConfMap[facebook::velox::connector::hive::HiveConfig::kFileHandleExpirationDurationMs] = std::to_string(
conf->get<int64_t>(kVeloxFileHandleExpirationDurationMs, kVeloxFileHandleExpirationDurationMsDefault));
| } | ||
| } | ||
|
|
||
| testWithSpecifiedSparkVersion( |
There was a problem hiding this comment.
This change should be version-agnostic, we can simply use test instead of testWithSpecifiedSparkVersion.
There was a problem hiding this comment.
Done in 0e3784a63. Replaced testWithSpecifiedSparkVersion(..., "3.5", "3.5") with plain test(...) for all tests in the suite — the file handle cache behavior is version-agnostic, so there's no reason to gate these to Spark 3.5. Verified locally against a full native Velox build: all 7 tests pass.
| * avoiding per-file open() syscalls (or remote filesystem connection establishment) on subsequent | ||
| * scans of the same files. | ||
| */ | ||
| object FileHandleCacheBenchmark extends SqlBasedBenchmark { |
There was a problem hiding this comment.
Can this benchmark file be deleted? I didn't see any mention of performance test results in the PR. If it's not necessary, we could consider removing it.
There was a problem hiding this comment.
Fair point on the missing numbers — I added them, and want to be upfront about what this benchmark does and doesn't show.
Local run (200 Parquet files × 5000 rows, 10 scans/iter, best of 5), cache ON vs OFF:
| Case | Cache OFF (best/avg ms) | Cache ON (best/avg ms) |
|---|---|---|
| full scan | 2625 / 3521 | 2549 / 2794 |
| filtered scan | 2734 / 2899 | 2710 / 2861 |
| column pruning | 2304 / 2434 | 2262 / 2312 |
The effect on local FS is within noise, which is expected — a local open() is cheap. I also traced the remote path in the Velox source to check whether the cache saves the open-time round-trip it's designed for, and on the Gluten path it mostly doesn't: Gluten passes each file's size into the split, so S3ReadFile/AbfsReadFile skip their HeadObject/GetProperties on open (they return early when size is known). The main exception is ABFS+OAuth, where the adapter rebuilds the token credential per open, so a cached handle avoids repeated AAD token fetches.
So this benchmark is best understood as a local-FS reproducibility harness, not proof of a remote win (it only exercises a local withTempPath). I'm inclined to keep it on that basis — it's not wired into CI and follows the existing SqlBasedBenchmark pattern — but since it doesn't demonstrate the remote case, I'm equally happy to remove it if you'd prefer. Your call as reviewer.
There was a problem hiding this comment.
Personally, I think it would be better to remove it. The benchmark does not seem particularly necessary at this point. Rather than benchmarking the file handle specifically, it would be more valuable to benchmark the Velox cache as a whole.
|
A little more comments, PTAL. Thanks. |
The file handle cache behavior is version-agnostic, so replace testWithSpecifiedSparkVersion(..., "3.5", "3.5") with plain test(...) in VeloxFileHandleCacheSuite so the tests run on every supported Spark version instead of only 3.5.
|
Run Gluten Clickhouse CI on x86 |
The TTL-expiration test only asserts scan correctness after the TTL window elapses (it passes whether or not eviction has occurred), so the wait does not need to be long. Reduce ttlMs 2000->500 and the total sleep 3000ms->1000ms, cutting CI time without introducing flakiness.
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (2)
cpp/velox/utils/ConfigExtractor.cc:328
- The new
numCacheFileHandlesandfileHandleExpirationDurationMsvalues are forwarded to Velox without any native-side validation. Because these settings can be provided as raw SparkConf strings (bypassing the ScalacheckValueguards), invalid values (e.g.,numCacheFileHandles<=0orfileHandleExpirationDurationMs<0) could reach Velox and trigger undefined behavior or hard-to-debug failures. Consider enforcing basic invariants here (fail fast with a clear message) before inserting intohiveConfMap.
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));
backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala:258
- This TTL test only sleeps and re-runs scans for correctness; it will still pass if the TTL value is ignored and no eviction ever happens (because the data is immutable and re-reading returns the same results). Since this PR is explicitly about wiring TTL-based eviction, it would be stronger to add an assertion that distinguishes "evicted" vs "not evicted" (e.g., delete/rename a previously scanned file, wait > TTL, then verify a scan using the same cached file listing fails or changes as expected).
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.
What changes are proposed in this pull request?
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.
Changes
Default config changes:
fileHandleCacheEnabled: false -> truessdCacheIOThreads: 1 -> 4TTL wiring for the file handle cache:
The
file-handle-expiration-duration-msconfig existed in Velox but was never passed to theSimpleLRUCacheconstructor inHiveConnector.cpp, so handles were never evicted by TTL. This wiring now lands upstream in Velox (as of the2026_07_16Velox bump), so the earlier Gluten-side patch has been dropped; handles are evicted after the configured TTL, preventing stale HDFS leases or closed remote connections from accumulating indefinitely.New Spark configs exposed:
spark.gluten.sql.columnar.backend.velox.numCacheFileHandles(default: 10000) - max entries in the LRU cachespark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs(default: 600000 / 10 min) - TTL per handle; idle handles are evictedTest suite (
VeloxFileHandleCacheSuite, 7 tests):Benchmark (
FileHandleCacheBenchmark):A local-FS reproducibility harness that measures repeated scans of 200 small Parquet files with the cache enabled vs disabled.
Rationale
Data lake files (Parquet, Delta, Iceberg) are immutable once written, which makes caching open file handles safe for these workloads; the TTL bounds staleness for overwrite-in-place cases (
INSERT OVERWRITE, compaction).A caveat on the size of the benefit: because Gluten passes each file's size into the Velox split (
VeloxIteratorApi->LocalFilesNode->FileProperties.fileSize), the S3 and ABFS read paths skip their open-time metadata request --S3ReadFile/AbfsReadFilereturn early when the size is known, avoidingHeadObject/GetProperties. So on the Gluten path the cache does not eliminate a per-open network round-trip in the common case. Its measurable effect is avoiding per-scanReadFile/client object construction and, for local files,open()/close()syscalls, on workloads that repeatedly scan the same set of many small files.The main case where the cache also avoids a real network round-trip is ABFS with OAuth auth, where the adapter currently rebuilds the token credential on every open, so a cached handle avoids repeated AAD token acquisition.
Users who work with mutable files, or who don't benefit, can disable it via
spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled=false.How was this patch tested?
VeloxFileHandleCacheSuite(7 tests) covering correctness, cache hits, many files, predicate pushdown, deleted files, TTL expiration, and column pruning -- all pass against a native Velox build.FileHandleCacheBenchmarkas a local-FS reproducibility harness. Local run (200 small Parquet files x 5000 rows, 10 scans/iteration, best of 5), cache ON vs OFF:The difference is within noise, consistent with the rationale above: on a local filesystem
open()is cheap, so caching the handle saves little. The benchmark only exercises a localwithTempPathand is not a remote-FS measurement.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude claude-opus-4.6