iemejia commented on code in PR #12400: URL: https://github.com/apache/gluten/pull/12400#discussion_r3616783046
########## backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxFileHandleCacheSuite.scala: ########## @@ -0,0 +1,321 @@ +/* + * 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") + } + + 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 (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) Review Comment: Fixed in `66f50229d`. Reduced `ttlMs` 2000->500 and the total wait 3000ms->1000ms, which cuts the suite time noticeably (24s->17s locally). I didn't switch to `eventually`/polling because the test has no JVM-visible eviction signal to poll on -- it just re-scans and asserts correctness, and that assertion holds whether or not eviction has happened yet, so a short fixed wait past the TTL is sufficient and not flaky. ########## 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. | Review Comment: `docs/velox-configuration.md` is auto-generated, and `600000ms` is the rendering of the `timeConf(MILLISECONDS)` default -- consistent with how every duration config is rendered in that table. Hand-editing it to `600000` would be overwritten on the next regeneration (and would only change if we switched to `longConf`, which we decided against to keep the `timeConf` convention). The description already states the value is in milliseconds, so I've left the generated output as-is. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
