This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new e5dce2956616 [SPARK-57531][SQL] Reuse OrcTail in non-vectorized ORC
reader path
e5dce2956616 is described below
commit e5dce2956616bf112c7b6add9b99b7e9b4bd4fda
Author: cxzl25 <[email protected]>
AuthorDate: Fri Jun 26 11:47:53 2026 -0700
[SPARK-57531][SQL] Reuse OrcTail in non-vectorized ORC reader path
### What changes were proposed in this pull request?
This PR mirrors what `buildColumnarReader` already does since SPARK-44556:
capture the `readerOptions` returned by `createORCReader`, then pass
`readerOptions.getOrcTail` when constructing the per-split record reader so the
footer is not re-read.
### Why are the changes needed?
`OrcPartitionReaderFactory.buildReader` (the non-vectorized / row-based
read path) previously called `OrcInputFormat.createRecordReader(fileSplit,
taskAttemptContext)`, which internally calls `OrcFile.createReader` without an
`OrcTail` and therefore re-parses the file footer from storage on every split.
Without OrcTail reuse the non-vectorized path pays this cost a second time
when opening the data reader for each split, while the vectorized path has been
avoiding it since SPARK-44556.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
GHA
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #56591 from cxzl25/SPARK-57531.
Authored-by: cxzl25 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 01a895d89894b420b8315beba42056f5462f43b6)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../execution/datasources/orc/OrcFileFormat.scala | 4 ++--
.../sql/execution/datasources/orc/OrcUtils.scala | 21 +++++++++++++++++++++
.../v2/orc/OrcPartitionReaderFactory.scala | 11 +++++------
3 files changed, 28 insertions(+), 8 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
index 633ac8107f32..bf0729bf4a0a 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
@@ -225,8 +225,8 @@ class OrcFileFormat
iter.asInstanceOf[Iterator[InternalRow]]
} else {
- val orcRecordReader = new OrcInputFormat[OrcStruct]
- .createRecordReader(fileSplit, taskAttemptContext)
+ val orcRecordReader = OrcUtils.createOrcMapreduceRecordReader(
+ filePath, taskConf, fileSplit, readerOptions)
val iter = new RecordReaderIterator[OrcStruct](orcRecordReader)
Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
iter.close()))
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 2c1e8beeb2b1..c1ed3864b63a 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
@@ -29,7 +29,10 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileStatus, Path}
import org.apache.hadoop.hive.serde2.io.DateWritable
import org.apache.hadoop.io.{BooleanWritable, ByteWritable, DoubleWritable,
FloatWritable, IntWritable, LongWritable, ShortWritable, WritableComparable}
+import org.apache.hadoop.mapreduce.lib.input.FileSplit
import org.apache.orc.{BooleanColumnStatistics, ColumnStatistics,
DateColumnStatistics, DoubleColumnStatistics, IntegerColumnStatistics, OrcConf,
OrcFile, Reader, TypeDescription, Writer}
+import org.apache.orc.mapred.{OrcInputFormat => OrcMapredInputFormat,
OrcStruct}
+import org.apache.orc.mapreduce.OrcMapreduceRecordReader
import org.apache.spark.{SPARK_VERSION_SHORT, SparkException}
import org.apache.spark.deploy.SparkHadoopUtil
@@ -574,4 +577,22 @@ object OrcUtils extends Logging {
resultRow
}
}
+
+ def createOrcMapreduceRecordReader(
+ filePath: Path,
+ conf: Configuration,
+ fileSplit: FileSplit,
+ readerOptions: OrcFile.ReaderOptions):
OrcMapreduceRecordReader[OrcStruct] = {
+ val fs = filePath.getFileSystem(conf)
+ val orcReader = OrcFile.createReader(
+ filePath,
+ OrcFile.readerOptions(conf)
+ .maxLength(OrcConf.MAX_FILE_LENGTH.getLong(conf))
+ .filesystem(fs)
+ .orcTail(readerOptions.getOrcTail))
+ val options = OrcMapredInputFormat
+ .buildOptions(conf, orcReader, fileSplit.getStart, fileSplit.getLength)
+ .useSelected(true)
+ new OrcMapreduceRecordReader[OrcStruct](orcReader, options)
+ }
}
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 8543fa9ca1d5..6b75b1473bd2 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
@@ -22,7 +22,6 @@ import org.apache.hadoop.mapreduce.{JobID, TaskAttemptID,
TaskID, TaskType}
import org.apache.hadoop.mapreduce.lib.input.FileSplit
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.orc.{OrcConf, OrcFile, Reader, TypeDescription}
-import org.apache.orc.mapred.OrcStruct
import org.apache.orc.mapreduce.OrcInputFormat
import org.apache.spark.broadcast.Broadcast
@@ -89,7 +88,8 @@ case class OrcPartitionReaderFactory(
}
val filePath = file.toPath
- val orcSchema = Utils.tryWithResource(createORCReader(filePath,
conf)._1)(_.getSchema)
+ val (orcSchemaReader, readerOptions) = createORCReader(filePath, conf)
+ val orcSchema = Utils.tryWithResource(orcSchemaReader)(_.getSchema)
val resultedColPruneInfo = OrcUtils.requestedColumnIds(
isCaseSensitive, dataSchema, readDataSchema, orcSchema, conf)
@@ -104,11 +104,10 @@ case class OrcPartitionReaderFactory(
val taskConf = new Configuration(conf)
val fileSplit = new FileSplit(filePath, file.start, file.length,
Array.empty)
- val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP,
0), 0)
- val taskAttemptContext = new TaskAttemptContextImpl(taskConf, attemptId)
- val orcRecordReader = new OrcInputFormat[OrcStruct]
- .createRecordReader(fileSplit, taskAttemptContext)
+ val orcRecordReader = OrcUtils.createOrcMapreduceRecordReader(
+ filePath, taskConf, fileSplit, readerOptions)
+
val deserializer = new OrcDeserializer(readDataSchema, requestedColIds)
val fileReader = new PartitionReader[InternalRow] {
override def next(): Boolean = orcRecordReader.nextKeyValue()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
