This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 3c2b57a0fa25 [SPARK-57958][SQL][FOLLOWUP] Close ORC schema-inference
Reader uninterruptibly to fix leaked file streams
3c2b57a0fa25 is described below
commit 3c2b57a0fa250f081ddaae22ab9c03ae7b4df69c
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Tue Jul 7 16:40:31 2026 +0900
[SPARK-57958][SQL][FOLLOWUP] Close ORC schema-inference Reader
uninterruptibly to fix leaked file streams
### What changes were proposed in this pull request?
`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.
### Why are the changes needed?
`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.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
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 <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 11669825fc8382168cffde2e0ccd9e7a252f71ba)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../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 50c28c783b4c..3ca39f681b58 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.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileStatus, Path}
@@ -44,8 +45,8 @@ 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.ArrayImplicits._
+import org.apache.spark.util.ThreadUtils
object OrcUtils extends Logging {
@@ -76,10 +77,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 {
@@ -93,6 +104,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(log"Failed to close the ORC reader for ${MDC(PATH,
file)}", t)
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt()
+ }
+ }
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
