This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new e3132f8cdf7d [SPARK-57529][SQL] Fix possible ORC reader leak in
OrcPartitionReaderFactory
e3132f8cdf7d is described below
commit e3132f8cdf7d6106976d9bad580d68b88f5d2809
Author: cxzl25 <[email protected]>
AuthorDate: Thu Jun 18 08:32:39 2026 -0700
[SPARK-57529][SQL] Fix possible ORC reader leak in OrcPartitionReaderFactory
### What changes were proposed in this pull request?
1. `OrcPartitionReaderFactory.createORCReader`: wrap the body after
`OrcFile.createReader` in a try/catch; on any exception, close the reader
before re-throwing.
2. `OrcPartitionReaderFactory.buildColumnarReader`: remove the `lazy`
qualifier from `val (reader, readerOptions)`, which was meaningless because
`reader` is accessed on the very next line.
### Why are the changes needed?
Reader may not be closed properly when an exception is thrown.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added `OrcPartitionReaderFactorySuite` with a test that reproduces the leak
scenario.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #56589 from cxzl25/SPARK-57529.
Authored-by: cxzl25 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 37b2ac292958c7af101820ccc0b0d8ee9bc26963)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../v2/orc/OrcPartitionReaderFactory.scala | 14 ++--
.../orc/OrcPartitionReaderFactorySuite.scala | 82 ++++++++++++++++++++++
2 files changed, 91 insertions(+), 5 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala
index c44a5d30cafe..8543fa9ca1d5 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala
@@ -129,7 +129,7 @@ case class OrcPartitionReaderFactory(
return buildColumnarReaderWithAggregates(file, conf)
}
val filePath = file.toPath
- lazy val (reader, readerOptions) = createORCReader(filePath, conf)
+ val (reader, readerOptions) = createORCReader(filePath, conf)
val orcSchema = Utils.tryWithResource(reader)(_.getSchema)
val resultedColPruneInfo = OrcUtils.requestedColumnIds(
isCaseSensitive, dataSchema, readDataSchema, orcSchema, conf)
@@ -172,10 +172,14 @@ case class OrcPartitionReaderFactory(
val fs = filePath.getFileSystem(conf)
val readerOptions = OrcFile.readerOptions(conf).filesystem(fs)
val reader = OrcFile.createReader(filePath, readerOptions)
-
- pushDownPredicates(reader.getSchema, conf)
-
- (reader, readerOptions)
+ try {
+ pushDownPredicates(reader.getSchema, conf)
+ (reader, readerOptions)
+ } catch {
+ case e: Throwable =>
+ reader.close()
+ throw e
+ }
}
/**
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcPartitionReaderFactorySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcPartitionReaderFactorySuite.scala
new file mode 100644
index 000000000000..ae161bb73f10
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcPartitionReaderFactorySuite.scala
@@ -0,0 +1,82 @@
+/*
+ * 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.datasources.orc
+
+import org.apache.spark.DebugFilesystem
+import org.apache.spark.memory.MemoryMode
+import org.apache.spark.paths.SparkPath
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import
org.apache.spark.sql.execution.datasources.v2.orc.OrcPartitionReaderFactory
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.sources.EqualTo
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+import org.apache.spark.util.SerializableConfiguration
+
+class OrcPartitionReaderFactorySuite extends OrcTest with SharedSparkSession {
+
+ import testImplicits._
+
+ test("SPARK-57529: Fix possible ORC reader leak in
OrcPartitionReaderFactory") {
+ withTempPath { dir =>
+ val dataSchema = StructType(Array(StructField("value", StringType)))
+ spark.range(10)
+ .select($"id".cast(StringType).as("value"))
+ .write.orc(dir.getCanonicalPath)
+
+ val orcFile = dir.listFiles(_.getName.endsWith(".orc")).headOption
+ .getOrElse(fail("No ORC file written"))
+
+ withSQLConf(SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ val sqlConf = spark.sessionState.conf
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ // Route file I/O through DebugFilesystem so we can assert no streams
are leaked.
+ hadoopConf.set("fs.file.impl", classOf[DebugFilesystem].getName)
+ hadoopConf.set("fs.file.impl.disable.cache", "true")
+ val broadcastedConf =
+ spark.sparkContext.broadcast(new
SerializableConfiguration(hadoopConf))
+
+ val factory = OrcPartitionReaderFactory(
+ sqlConf = sqlConf,
+ broadcastedConf = broadcastedConf,
+ dataSchema = dataSchema,
+ readDataSchema = dataSchema,
+ partitionSchema = StructType(Seq.empty),
+ // Integer literal on a STRING column triggers
IllegalArgumentException in
+ // OrcFilters.createFilter -> buildLeafSearchArgument
+ filters = Array(EqualTo("value", 1)),
+ aggregation = None,
+ options = new OrcOptions(Map.empty[String, String], sqlConf),
+ memoryMode = MemoryMode.ON_HEAP)
+
+ val partFile = PartitionedFile(
+ partitionValues = InternalRow.empty,
+ filePath = SparkPath.fromPathString(orcFile.getAbsolutePath),
+ start = 0,
+ length = orcFile.length())
+
+ DebugFilesystem.clearOpenStreams()
+ intercept[IllegalArgumentException] {
+ factory.buildReader(partFile)
+ }
+ DebugFilesystem.assertNoOpenStreams()
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
