(spark) branch branch-3.5 updated: [SPARK-57958][SQL][FOLLOWUP] Close ORC schema-inference Reader uninterruptibly to fix leaked file streams

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

HyukjinKwon pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-3.5 by this push:
     new 834bfe6adf5e [SPARK-57958][SQL][FOLLOWUP] Close ORC schema-inference Reader uninterruptibly to fix leaked file streams
834bfe6adf5e is described below

commit 834bfe6adf5eba83f59d6aa7b9d7f5c551abe7c3
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Tue Jul 7 16:40:31 2026 +0900

    [SPARK-57958][SQL][FOLLOWUP] Close ORC schema-inference Reader uninterruptibly to fix leaked file streams
    
    `OrcUtils.readSchema` previously read the ORC footer via `Utils.tryWithResource(OrcFile.createReader(...))`. This PR holds the `Reader` in an explicit variable and closes it in a `finally` block that **clears the thread's interrupt status around `Reader.close()`** (restoring it afterwards) and logs-and-swallows a non-fatal close failure, so the underlying file stream is always released.
    
    `readSchema` runs on the `readingOrcSchemas` `parmap` worker threads (and Spark task threads) during parallel / merged ORC schema inference (`OrcUtils.readOrcSchemasInParallel` → `SchemaMergeUtils.mergeSchemasInParallel`). When a sibling file fails — e.g. the `SPARK-11412 read and merge orc schemas in parallel` test deliberately mixes a corrupt (JSON) file with `ignoreCorruptFiles=false` and asserts the Spark job aborts — the enclosing job is cancelled and a worker that has already op [...]
    
    In test builds the leaked stream is caught by `SharedSparkSessionBase.afterEach` → `DebugFilesystem.assertNoOpenStreams`, which retries for 10s then aborts the **whole suite** with `There are 1 possibly leaked file streams`. Because the leak is registered asynchronously, it surfaces nondeterministically in a *later* ORC suite's `afterEach`, so `OrcSourceV1Suite` / `OrcSourceV2Suite` abort intermittently on `build_java21` for `branch-4.0`:
    
    - https://github.com/apache/spark/actions/runs/28766468775 — `OrcSourceV1Suite *** ABORTED ***`
    - https://github.com/apache/spark/actions/runs/28731799013 — `OrcSourceV2Suite *** ABORTED ***`
    
    In both runs the leaked-connection stack logged by `DebugFilesystem` points exactly at:
    
    ```
    DebugFilesystem.open → org.apache.orc.impl.ReaderImpl.extractFileTail
      → OrcFile.createReader → OrcUtils.readSchema(OrcUtils.scala:80)
      → OrcUtils.readOrcSchemasInParallel → ThreadUtils.parmap
    ```
    
    [SPARK-57958](https://issues.apache.org/jira/browse/SPARK-57958) fixed a *sibling* reader leak in the `OrcSuite` test helpers (`testBloomFilterCreation` / `testSelectiveDictionaryEncoding`); those helpers do **not** appear as leak sites in the failing runs. This PR addresses the remaining production leak site those runs actually hit.
    
    No.
    
    Existing `OrcSourceV1Suite` / `OrcSourceV2Suite`. The flake is a rare job-cancellation race that does not reproduce deterministically (verified: it did not surface across a standalone ORC repro, a 300× in-JVM loop of the exact merge-abort scenario, or 10× dedicated-JVM suite runs on GitHub Actions). This change makes the reader/stream release robust to interruption at the leak site the failing CI runs identified.
    
    This pull request and its description were written by Isaac.
    
    Closes #57062 from HyukjinKwon/DO-NOT-MERGE/orc-schema-stream-leak.
    
    Authored-by: Hyukjin Kwon <gu...@apache.org>
    Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
    (cherry picked from commit 11669825fc8382168cffde2e0ccd9e7a252f71ba)
    Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
 .../sql/execution/datasources/orc/OrcUtils.scala   | 35 +++++++++++++++++++---
 1 file changed, 31 insertions(+), 4 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
index 89a9ba6ec7aa..fe3721a9fccc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
@@ -22,6 +22,7 @@ import java.util.Locale
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
+import scala.util.control.NonFatal
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, Path}
@@ -43,7 +44,7 @@ import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, SchemaMergeUtils}
 import org.apache.spark.sql.execution.datasources.v2.V2ColumnUtils
 import org.apache.spark.sql.types._
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.ThreadUtils
 
 object OrcUtils extends Logging {
 
@@ -73,10 +74,20 @@ object OrcUtils extends Logging {
       : Option[TypeDescription] = {
     val fs = file.getFileSystem(conf)
     val readerOptions = OrcFile.readerOptions(conf).filesystem(fs)
+    // Follow-up to SPARK-57958: This runs on `readingOrcSchemas` parmap workers (Spark task
+    // threads) during parallel/merged schema inference. When a sibling file fails and
+    // the enclosing Spark job is aborted, the worker can be interrupted between opening
+    // the ORC `Reader` (which holds an `FSDataInputStream` opened in
+    // `ReaderImpl.extractFileTail`) and closing it. `Reader.close()` performs file-system
+    // I/O, so a pending interrupt can turn it into a no-op/`ClosedByInterruptException`
+    // and orphan the stream, which later surfaces as
+    // `DebugFilesystem.assertNoOpenStreams` "possibly leaked file streams" aborting an
+    // unrelated ORC suite. Hold the reader explicitly and close it uninterruptibly so the
+    // handle is always released regardless of the enclosing task's interrupt state.
+    var reader: Reader = null
     try {
-      val schema = Utils.tryWithResource(OrcFile.createReader(file, readerOptions)) { reader =>
-        reader.getSchema
-      }
+      reader = OrcFile.createReader(file, readerOptions)
+      val schema = reader.getSchema
       if (schema.getFieldNames.size == 0) {
         None
       } else {
@@ -90,6 +101,22 @@ object OrcUtils extends Logging {
         } else {
           throw QueryExecutionErrors.cannotReadFooterForFileError(file, e)
         }
+    } finally {
+      if (reader != null) {
+        // Close without being aborted by a pending interrupt from job cancellation,
+        // then restore the interrupt status for the caller.
+        val interrupted = Thread.interrupted()
+        try {
+          reader.close()
+        } catch {
+          case NonFatal(t) =>
+            logWarning(s"Failed to close the ORC reader for $file", t)
+        } finally {
+          if (interrupted) {
+            Thread.currentThread().interrupt()
+          }
+        }
+      }
     }
   }
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org