You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/22 03:51:57 UTC
(spark) branch branch-4.2 updated: [SPARK-60436][SQL] Fix ORC filter pushdown on timestamp literals with sub-millisecond digits
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new ff36f21abca5 [SPARK-60436][SQL] Fix ORC filter pushdown on timestamp literals with sub-millisecond digits
ff36f21abca5 is described below
commit ff36f21abca5f8b243072573835991c78a07c72c
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Wed Oct 21 20:21:10 2026 -0700
[SPARK-60436][SQL] Fix ORC filter pushdown on timestamp literals with sub-millisecond digits
### What changes were proposed in this pull request?
This PR aims to fix the native ORC reader (`spark.sql.orc.impl=native`) to push down the ORC `SearchArgument` without losing the sub-millisecond digits of timestamp literals.
`OrcFileFormat` (DSv1) and `OrcPartitionReaderFactory` (DSv2) now pass the `SearchArgument` to the ORC readers, which set it directly on the ORC `Reader.Options` after `OrcInputFormat.buildOptions`. Previously, it was passed through the Hadoop `Configuration` via `OrcInputFormat.setSearchArgument`.
- `OrcColumnarBatchReader.initialize` has a new overload with the `SearchArgument`.
- `OrcUtils.createOrcMapreduceRecordReader` has a new `SearchArgument` parameter.
- `OrcPartitionReaderFactory.createORCReader` no longer builds the `SearchArgument`, so it does not need to close the reader on a failure (SPARK-57529) any more. The `SearchArgument` is built while no ORC reader is open.
The `SearchArgument` itself is built in the same way as before, for each file in the task.
### Why are the changes needed?
`OrcInputFormat.setSearchArgument` writes the `SearchArgument` into `orc.kryo.sarg` with a plain `Kryo`, and `OrcInputFormat.buildOptions` reads it back. Kryo's default serializer of `java.util.Date` keeps only `getTime()`, so a `java.sql.Timestamp` literal loses its sub-millisecond digits, e.g. `2020-01-01 00:00:00.123456` becomes `2020-01-01 00:00:00.123`.
Since ORC 1.6.4 (ORC-611), ORC timestamp statistics have nanosecond precision, so ORC compares the truncated literal with the nanosecond minimums and maximums.
- `=`, `<=>`, `IN`, `<`, and `<=` skip a stripe or row group whose minimum is between the truncated literal and the literal, although it can hold matching rows.
- `!=` and `NOT IN` skip a stripe or row group whose values are all equal to the truncated literal, because ORC evaluates `=` to `YES` for it.
- `>` and `>=` stay correct, because the truncated literal only makes them less selective.
This affects `TIMESTAMP` columns since Apache Spark 3.2.0 ([v3.2.0](https://github.com/apache/spark/releases/tag/v3.2.0), 2021-10-06), the first release with ORC 1.6 (SPARK-33295). It also affects the unreleased nanosecond timestamp types, `TIMESTAMP_NTZ(p)` and `TIMESTAMP_LTZ(p)` (SPARK-57823), even when they are read with the precision they are written with. Both DSv1 and DSv2 are affected, with the vectorized ORC reader on and off.
For example, the following query returns 0 instead of 1. The result is correct with `spark.sql.orc.filterPushdown=false`.
```scala
spark.sql("SELECT TIMESTAMP '2020-01-01 00:00:00.123456' AS ts").write.orc(path)
spark.read.orc(path).where("ts = TIMESTAMP '2020-01-01 00:00:00.123456'").count()
```
The same issue is reproduced with Apache Spark 4.0.4, 4.1.3, 4.2.0, and 4.3.0-rc1.
The Hive ORC reader (`spark.sql.orc.impl=hive`) has the same Kryo issue, but Hive's reader gets the `SearchArgument` only from the Hadoop configuration. So SPARK-60396 and SPARK-60423 stopped pushing down the affected predicates there instead.
### Does this PR introduce _any_ user-facing change?
Yes. With the native ORC reader, predicates that compare a timestamp column with a literal with sub-millisecond digits return correct results. Previously, they could skip matching rows silently since Apache Spark 3.2.0.
### How was this patch tested?
Pass the CIs with the newly added test case in `OrcQuerySuite`, which runs in `OrcV1QuerySuite` and `OrcV2QuerySuite`. It checks `=`, `<=>`, `IN`, `<`, `<=`, `>`, `>=`, `!=`, and `NOT IN` on `TIMESTAMP`, `TIMESTAMP_NTZ(p)`, and `TIMESTAMP_LTZ(p)` for all precisions, with the vectorized ORC reader on and off. It also checks that a pushed down `=` keeps skipping the files without the value. Without the fix, it fails.
```
[info] - SPARK-60436: ORC predicate pushdown on timestamps with sub-millisecond digits *** FAILED *** (2 seconds, 144 milliseconds)
[info] Results do not match for query:
...
[info] 'Filter ('ts = cast(2020-01-01 00:00:00.123456789 as timestamp))
...
[info] !== Correct Answer - 1 == == Spark Answer - 0 ==
[info] !struct<ts:timestamp> struct<>
[info] ![2020-01-01 00:00:00.123456] (CheckAnswerHelper.scala:61)
```
The related test suites also pass.
```
$ build/sbt "sql/testOnly org.apache.spark.sql.execution.datasources.orc.* *OrcV1AggregatePushDownSuite *OrcV2AggregatePushDownSuite *OrcCodecSuite *OrcReadSchemaSuite *MergedOrcReadSchemaSuite *VectorizedOrcReadSchemaSuite *OrcZipArchiveReadSuite *OrcTarArchiveReadSuite *OrcSevenZArchiveReadSuite *FileBasedDataSourceSuite"
$ build/sbt -Phive "hive/testOnly org.apache.spark.sql.hive.orc.*"
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59682 from dongjoon-hyun/SPARK-60436.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 2b46c2b896789df7cf0d48a2fc442a346a87a5f5)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../datasources/orc/OrcColumnarBatchReader.java | 19 ++++++++++++
.../execution/datasources/orc/OrcFileFormat.scala | 16 +++++-----
.../sql/execution/datasources/orc/OrcUtils.scala | 24 +++++++++++++++
.../v2/orc/OrcPartitionReaderFactory.scala | 31 +++++++-------------
.../execution/datasources/orc/OrcQuerySuite.scala | 34 ++++++++++++++++++++++
5 files changed, 96 insertions(+), 28 deletions(-)
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnarBatchReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnarBatchReader.java
index ea33aa1e2325..d0cdff895b5f 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnarBatchReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnarBatchReader.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
@@ -130,6 +131,19 @@ public class OrcColumnarBatchReader extends RecordReader<Void, ColumnarBatch> {
InputSplit inputSplit,
TaskAttemptContext taskAttemptContext,
OrcTail orcTail) throws IOException {
+ initialize(inputSplit, taskAttemptContext, orcTail, null);
+ }
+
+ /**
+ * Initialize ORC file reader and batch record reader with the search argument to push down,
+ * or null if there is none.
+ * Please note that `initBatch` is needed to be called after this.
+ */
+ public void initialize(
+ InputSplit inputSplit,
+ TaskAttemptContext taskAttemptContext,
+ OrcTail orcTail,
+ SearchArgument sarg) throws IOException {
FileSplit fileSplit = (FileSplit)inputSplit;
Configuration conf = taskAttemptContext.getConfiguration();
Reader reader = OrcFile.createReader(
@@ -140,6 +154,11 @@ public class OrcColumnarBatchReader extends RecordReader<Void, ColumnarBatch> {
.orcTail(orcTail));
Reader.Options options =
OrcInputFormat.buildOptions(conf, reader, fileSplit.getStart(), fileSplit.getLength());
+ if (sarg != null) {
+ // Set it directly rather than with `OrcInputFormat.setSearchArgument`, whose Kryo round
+ // trip truncates timestamp values to milliseconds.
+ options.searchArgument(sarg, reader.getSchema().getFieldNames().toArray(new String[0]));
+ }
recordReader = reader.rows(options);
}
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..21299c61c581 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
@@ -28,7 +28,6 @@ import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.orc.{OrcUtils => _, _}
import org.apache.orc.OrcConf.COMPRESS
import org.apache.orc.mapred.OrcStruct
-import org.apache.orc.mapreduce._
import org.apache.spark.TaskContext
import org.apache.spark.memory.MemoryMode
@@ -185,11 +184,11 @@ class OrcFileFormat
Iterator.empty
} else {
// ORC predicate pushdown
- if (orcFilterPushDown && filters.nonEmpty) {
+ val searchArgument = if (orcFilterPushDown && filters.nonEmpty) {
val fileSchema = OrcUtils.toCatalystSchema(orcSchema)
- OrcFilters.createFilter(fileSchema, filters).foreach { f =>
- OrcInputFormat.setSearchArgument(conf, f, fileSchema.fieldNames)
- }
+ OrcFilters.createFilter(fileSchema, filters)
+ } else {
+ None
}
val (requestedColIds, canPruneCols) = resultedColPruneInfo.get
@@ -215,7 +214,8 @@ class OrcFileFormat
val requestedDataColIds = requestedColIds ++ Array.fill(partitionSchema.length)(-1)
val requestedPartitionColIds =
Array.fill(requiredSchema.length)(-1) ++ Range(0, partitionSchema.length)
- batchReader.initialize(fileSplit, taskAttemptContext, readerOptions.getOrcTail)
+ batchReader.initialize(
+ fileSplit, taskAttemptContext, readerOptions.getOrcTail, searchArgument.orNull)
batchReader.initBatch(
TypeDescription.fromString(resultSchemaString),
resultSchema.fields,
@@ -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, searchArgument)
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 5e47bde0e8fd..54c70923d23c 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
@@ -28,9 +28,13 @@ import scala.util.control.NonFatal
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument
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
@@ -567,4 +571,24 @@ object OrcUtils extends Logging {
resultRow
}
}
+
+ def createOrcMapreduceRecordReader(
+ filePath: Path,
+ conf: Configuration,
+ fileSplit: FileSplit,
+ searchArgument: Option[SearchArgument]): OrcMapreduceRecordReader[OrcStruct] = {
+ val orcReader = OrcFile.createReader(
+ filePath,
+ OrcFile.readerOptions(conf)
+ .maxLength(OrcConf.MAX_FILE_LENGTH.getLong(conf)))
+ val options = OrcMapredInputFormat
+ .buildOptions(conf, orcReader, fileSplit.getStart, fileSplit.getLength)
+ .useSelected(true)
+ // Set it directly rather than with `OrcInputFormat.setSearchArgument`, whose Kryo round trip
+ // truncates timestamp values to milliseconds.
+ searchArgument.foreach { sarg =>
+ options.searchArgument(sarg, orcReader.getSchema.getFieldNames.asScala.toArray)
+ }
+ 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..bc658ebf191c 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
@@ -18,12 +18,11 @@ package org.apache.spark.sql.execution.datasources.v2.orc
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument
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
import org.apache.spark.memory.MemoryMode
@@ -73,12 +72,12 @@ case class OrcPartitionReaderFactory(
s.dataType, sqlConf.orcVectorizedReaderNestedColumnEnabled))
}
- private def pushDownPredicates(orcSchema: TypeDescription, conf: Configuration): Unit = {
+ private def pushDownPredicates(orcSchema: TypeDescription): Option[SearchArgument] = {
if (orcFilterPushDown && filters.nonEmpty) {
val fileSchema = OrcUtils.toCatalystSchema(orcSchema)
- OrcFilters.createFilter(fileSchema, filters.toImmutableArraySeq).foreach { f =>
- OrcInputFormat.setSearchArgument(conf, f, fileSchema.fieldNames)
- }
+ OrcFilters.createFilter(fileSchema, filters.toImmutableArraySeq)
+ } else {
+ None
}
}
@@ -104,11 +103,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, pushDownPredicates(orcSchema))
+
val deserializer = new OrcDeserializer(readDataSchema, requestedColIds)
val fileReader = new PartitionReader[InternalRow] {
override def next(): Boolean = orcRecordReader.nextKeyValue()
@@ -150,7 +148,8 @@ case class OrcPartitionReaderFactory(
val taskAttemptContext = new TaskAttemptContextImpl(taskConf, attemptId)
val batchReader = new OrcColumnarBatchReader(capacity, memoryMode)
- batchReader.initialize(fileSplit, taskAttemptContext, readerOptions.getOrcTail)
+ batchReader.initialize(fileSplit, taskAttemptContext, readerOptions.getOrcTail,
+ pushDownPredicates(orcSchema).orNull)
val requestedPartitionColIds =
Array.fill(readDataSchema.length)(-1) ++ Range(0, partitionSchema.length)
@@ -171,15 +170,7 @@ case class OrcPartitionReaderFactory(
val fs = filePath.getFileSystem(conf)
val readerOptions = OrcFile.readerOptions(conf).filesystem(fs)
- val reader = OrcFile.createReader(filePath, readerOptions)
- try {
- pushDownPredicates(reader.getSchema, conf)
- (reader, readerOptions)
- } catch {
- case e: Throwable =>
- reader.close()
- throw e
- }
+ (OrcFile.createReader(filePath, readerOptions), readerOptions)
}
/**
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
index a42c004e3aaf..77676dbf572d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
@@ -917,6 +917,40 @@ abstract class OrcQuerySuite extends OrcQueryTest with SharedSparkSession {
}
}
+ test("SPARK-60436: ORC predicate pushdown on timestamps with sub-millisecond digits") {
+ // Each value is written to its own file, so a pushed down predicate that skips a file with a
+ // matching row changes the result. `00:00:00.123` is `value` truncated to milliseconds.
+ def ts(time: String): String = s"CAST('2020-01-01 $time' AS TIMESTAMP)"
+ val value = ts("00:00:00.123456789")
+ val input = sql(s"SELECT ${ts("00:00:00.123")} AS ts UNION ALL SELECT $value " +
+ s"UNION ALL SELECT ${ts("00:00:01")}")
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ input.coalesce(1).write.option("maxRecordsPerFile", 1).orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> vectorized.toString) {
+ val df = spark.read.schema(input.schema).orc(path)
+ Seq(
+ s"ts = $value",
+ s"ts <=> $value",
+ s"ts IN ($value, ${ts("00:00:01")})",
+ s"ts < $value",
+ s"ts <= $value",
+ s"ts > $value",
+ s"ts >= $value",
+ s"ts != $value",
+ s"ts NOT IN ($value, ${ts("00:00:01")})"
+ ).foreach { condition =>
+ checkAnswer(df.where(condition), input.where(condition))
+ }
+ // The pushed down value keeps its sub-millisecond digits and skips the other files.
+ checkAnswer(
+ stripSparkFilter(df.where(s"ts = $value")), input.where(s"ts = $value"))
+ }
+ }
+ }
+ }
+
// SPARK-39519: Ignore this case because it requires more than 4g heap memory to ensure test
// stability when use Java 11. Should test it manually when upgrading `hive-storage-api`
ignore("SPARK-39387: BytesColumnVector should not throw RuntimeException due to overflow") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org