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 12:49:33 UTC
(spark) branch branch-4.x updated: [SPARK-60449][SQL] Fix ORC filter pushdown on `TIMESTAMP` values with sub-microsecond digits
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 124081b44c1e [SPARK-60449][SQL] Fix ORC filter pushdown on `TIMESTAMP` values with sub-microsecond digits
124081b44c1e is described below
commit 124081b44c1ebe70e6da3334b8bccf7942c305ca
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Oct 22 05:48:56 2026 -0700
[SPARK-60449][SQL] Fix ORC filter pushdown on `TIMESTAMP` values with sub-microsecond digits
### What changes were proposed in this pull request?
This PR aims to fix ORC filter pushdown on `TIMESTAMP` columns whose ORC files store sub-microsecond digits, like the files written by other engines.
Spark reads an ORC timestamp into `TimestampType` by truncating it to microseconds, so a literal `v` matches the stored values from `v` to `v + 999` nanoseconds. `OrcFilters` now builds the predicates on `TimestampType` with that range.
| Predicate | Before | After |
| --- | --- | --- |
| `ts = v` | `ts = v` | `ts BETWEEN v AND v + 999ns` |
| `ts <=> v` | `ts <=> v` | `NOT ts IS NULL AND ts BETWEEN v AND v + 999ns` |
| `ts <= v` | `ts <= v` | `ts <= v + 999ns` |
| `ts > v` | `NOT ts <= v` | `NOT ts <= v + 999ns` |
| `ts IN (v1, v2)` | `ts IN (v1, v2)` | `ts BETWEEN v1 AND v1 + 999ns OR ts BETWEEN v2 AND v2 + 999ns` |
`<`, `>=`, `IS NULL`, and `IS NOT NULL` are already exact, so they are unchanged.
This works because SPARK-60436 (#59682) stopped the native ORC reader from passing the `SearchArgument` through Kryo, which truncates the timestamp literals to milliseconds. Otherwise, `v + 999ns` would become `v` again.
### Why are the changes needed?
ORC stores timestamps with nanosecond precision, and its timestamp statistics have nanosecond precision since ORC 1.6.4 and 1.7.0 (ORC-611), and ORC C++ 1.7.0 (ORC-663). So ORC compares a pushed down literal with the nanosecond minimums and maximums, while Spark compares it with the values truncated to microseconds.
- `=`, `<=>`, `IN`, and `<=` skip a stripe or row group whose minimum is greater than `v` but at most `v + 999ns`, although such a minimum is read as `v`.
- `>` is pushed down as `NOT ts <= v`, which has the same issue when it is negated. So it is changed in the same way to stay exact.
For example, a file that stores `2020-05-25 10:11:13.000000789` is read as `2020-05-25 10:11:13`, but the following query returns 0 instead of 1. The result is correct with `spark.sql.orc.filterPushdown=false`.
```scala
spark.read.orc(path).where("ts = TIMESTAMP '2020-05-25 10:11:13'").count()
```
This affects files with sub-microsecond digits 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), which reads the nanosecond statistics. The example above returns wrong results with Apache Spark 4.0.4, 4.1.3, and 4.2.0, in DSv1 and DSv2, with the vectorized ORC reader on and off. Files written by Spark are not affected, because their `TIMESTAMP` values have microsecond precision. ORC `time [...]
The Hive ORC reader (`spark.sql.orc.impl=hive`) still gets the `SearchArgument` through Kryo, which truncates `v + 999ns` back to `v`. SPARK-60423 stops pushing down only the filters with sub-millisecond literals there, so this PR does not change the results of the Hive ORC reader, and it is out of the scope of this PR.
### Does this PR introduce _any_ user-facing change?
Yes. Predicates on `TIMESTAMP` columns of ORC files with sub-microsecond digits return correct results with filter pushdown. Previously, they could skip matching rows silently since Apache Spark 3.2.0.
Since ORC evaluates bloom filters only for `=`, `<=>`, and `IN`, these predicates on `TIMESTAMP` columns no longer use ORC bloom filters. The minimum and maximum statistics are used as before. In addition, an `IN` list with `k` values becomes `k` `BETWEEN` predicates, so ORC evaluates a long `IN` list more slowly, e.g., about 11 ms more per file with 200 row groups for 1,000 values.
### How was this patch tested?
Pass the CIs with the newly added test case in `OrcFilterSuite`, which runs in `OrcFilterSuite` (DSv2) and `OrcV1FilterSuite` (DSv1). It writes ORC files with sub-microsecond digits through the ORC writer API, including the last nanosecond of a microsecond and the next microsecond in their own files, and checks `=`, `<=>`, `IN`, `<`, `<=`, `>`, `>=`, `!=`, and `NOT IN` with the vectorized ORC reader and `spark.sql.datetime.java8API.enabled` on and off. It also checks that pushed down [...]
```
[info] OrcFilterSuite:
[info] - SPARK-60449: filter pushdown - timestamp with sub-microsecond digits *** FAILED *** (2 seconds, 103 milliseconds)
[info] 0 did not equal 1 ts = TIMESTAMP'2020-05-25 10:11:13' (OrcFilterSuite.scala:400)
...
[info] OrcV1FilterSuite:
[info] - SPARK-60449: filter pushdown - timestamp with sub-microsecond digits *** FAILED *** (377 milliseconds)
[info] 0 did not equal 1 ts = TIMESTAMP'2020-05-25 10:11:13' (OrcFilterSuite.scala:400)
```
`filter pushdown - timestamp` is updated for the new `BETWEEN` predicates. 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 #59699 from dongjoon-hyun/SPARK-60449.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 5300d76dcb6e186f94d500ea3edfd9f35335ec9d)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../sql/execution/datasources/orc/OrcFilters.scala | 46 ++++++++++++
.../execution/datasources/orc/OrcFilterSuite.scala | 84 ++++++++++++++++++++--
2 files changed, 126 insertions(+), 4 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala
index 2d129b2ca8a1..84a816a1ad2b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.orc
+import java.sql.Timestamp
import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period}
import org.apache.hadoop.hive.common.`type`.HiveDecimal
@@ -266,12 +267,57 @@ private[sql] object OrcFilters extends OrcFiltersBase {
def canPushValue(attribute: String): Boolean =
dataTypeMap.contains(attribute) && acceptsValue(attribute)
+ // A TimestampType column reads the stored ORC timestamps truncated to microseconds, while ORC
+ // compares a pushed down value with the stored nanoseconds, e.g. of a file written by another
+ // engine. So a value `v` is read from the stored values from `v` to `v` + 999 nanoseconds,
+ // and `=`, `<=>`, `<=`, `>`, and `IN` compare with that range. `<` and `>=` are exact with `v`.
+ def isTimestamp(attribute: String): Boolean =
+ canPushValue(attribute) && dataTypeMap(attribute).fieldType == TimestampType
+
+ // Returns the casted `value` and the largest stored value that is read as it.
+ def timestampRange(value: Any): (Any, Any) = castLiteralValue(value, TimestampType) match {
+ case ts: Timestamp =>
+ val upper = new Timestamp(ts.getTime)
+ upper.setNanos(ts.getNanos + 999)
+ (ts, upper)
+ case other => (other, other)
+ }
+
import org.apache.spark.sql.sources._
// NOTE: For all case branches dealing with leaf predicates below, the additional `startAnd()`
// call is mandatory. ORC `SearchArgument` builder requires that all leaf predicates must be
// wrapped by a "parent" predicate (`And`, `Or`, or `Not`).
expression match {
+ case EqualTo(name, value) if isTimestamp(name) =>
+ val (lower, upper) = timestampRange(value)
+ Some(builder.startAnd()
+ .between(dataTypeMap(name).fieldName, getType(name), lower, upper).end())
+
+ // Unlike BETWEEN, `<=>` is never null, so `NOT (ts <=> v)` holds for the null values. A null
+ // `v` stays with the case below, which ORC does not use to skip anything.
+ case EqualNullSafe(name, value) if isTimestamp(name) && value != null =>
+ val (lower, upper) = timestampRange(value)
+ Some(builder.startAnd()
+ .startNot().isNull(dataTypeMap(name).fieldName, getType(name)).end()
+ .between(dataTypeMap(name).fieldName, getType(name), lower, upper).end())
+
+ case LessThanOrEqual(name, value) if isTimestamp(name) =>
+ val (_, upper) = timestampRange(value)
+ Some(builder.startAnd()
+ .lessThanEquals(dataTypeMap(name).fieldName, getType(name), upper).end())
+
+ case GreaterThan(name, value) if isTimestamp(name) =>
+ val (_, upper) = timestampRange(value)
+ Some(builder.startNot()
+ .lessThanEquals(dataTypeMap(name).fieldName, getType(name), upper).end())
+
+ case In(name, values) if isTimestamp(name) =>
+ Some(values.foldLeft(builder.startOr()) { (b, value) =>
+ val (lower, upper) = timestampRange(value)
+ b.between(dataTypeMap(name).fieldName, getType(name), lower, upper)
+ }.end())
+
case EqualTo(name, value) if canPushValue(name) =>
val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType)
Some(builder.startAnd()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
index 3a48b77f3334..7781b18914f3 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.orc
+import java.io.File
import java.math.MathContext
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}
@@ -24,8 +25,11 @@ import java.time.{Duration, Instant, LocalDateTime, LocalTime, Period, ZoneOffse
import scala.jdk.CollectionConverters._
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector
import org.apache.hadoop.hive.ql.io.sarg.{PredicateLeaf, SearchArgument, SearchArgumentImpl}
import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory.newBuilder
+import org.apache.orc.{OrcFile, TypeDescription}
import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, Column, DataFrame, Row}
@@ -41,6 +45,7 @@ import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.tags.ExtendedSQLTest
import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.Utils
/**
* A test suite that tests Apache ORC filter API based filter pushdown optimization.
@@ -308,17 +313,17 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
val timestamps = input.map(Literal(_))
checkFilterPredicate($"_1".isNull, PredicateLeaf.Operator.IS_NULL)
- checkFilterPredicate($"_1" === timestamps(0), PredicateLeaf.Operator.EQUALS)
- checkFilterPredicate($"_1" <=> timestamps(0), PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+ checkFilterPredicate($"_1" === timestamps(0), PredicateLeaf.Operator.BETWEEN)
+ checkFilterPredicate($"_1" <=> timestamps(0), PredicateLeaf.Operator.BETWEEN)
checkFilterPredicate($"_1" < timestamps(1), PredicateLeaf.Operator.LESS_THAN)
checkFilterPredicate($"_1" > timestamps(2), PredicateLeaf.Operator.LESS_THAN_EQUALS)
checkFilterPredicate($"_1" <= timestamps(0), PredicateLeaf.Operator.LESS_THAN_EQUALS)
checkFilterPredicate($"_1" >= timestamps(3), PredicateLeaf.Operator.LESS_THAN)
- checkFilterPredicate(Literal(timestamps(0)) === $"_1", PredicateLeaf.Operator.EQUALS)
+ checkFilterPredicate(Literal(timestamps(0)) === $"_1", PredicateLeaf.Operator.BETWEEN)
checkFilterPredicate(
- Literal(timestamps(0)) <=> $"_1", PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+ Literal(timestamps(0)) <=> $"_1", PredicateLeaf.Operator.BETWEEN)
checkFilterPredicate(Literal(timestamps(1)) > $"_1", PredicateLeaf.Operator.LESS_THAN)
checkFilterPredicate(
Literal(timestamps(2)) < $"_1",
@@ -333,6 +338,77 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
}
}
+ test("SPARK-60449: filter pushdown - timestamp with sub-microsecond digits") {
+ // Another engine (e.g. Hive) can write sub-microsecond digits to an ORC timestamp column, and
+ // Spark reads the column as TimestampType by truncating every value to microseconds. So the
+ // pushed down predicates must not skip the stripes whose truncated values satisfy them.
+ withTempPath { dir =>
+ val schema = TypeDescription.fromString("struct<ts:timestamp>")
+ Seq(
+ "part-0.orc" -> Seq("2020-05-25 10:11:12.123456789", null),
+ // The last stored value that is read as `a` and the first one after it, each alone in its
+ // file, so a pushed down range that misses either edge skips a matching file.
+ "part-1.orc" -> Seq("2020-05-25 10:11:12.123456999"),
+ "part-2.orc" -> Seq("2020-05-25 10:11:12.123457"),
+ "part-3.orc" -> Seq("2020-05-25 10:11:13.000000789")
+ ).foreach { case (name, values) =>
+ val file = new Path(new File(dir, name).getCanonicalPath)
+ val writerOptions =
+ OrcFile.writerOptions(spark.sessionState.newHadoopConf()).setSchema(schema)
+ Utils.tryWithResource(OrcFile.createWriter(file, writerOptions)) { writer =>
+ val batch = schema.createRowBatch()
+ val tsCol = batch.cols(0).asInstanceOf[TimestampColumnVector]
+ values.zipWithIndex.foreach { case (v, i) =>
+ tsCol.set(i, Option(v).map(Timestamp.valueOf(_)).orNull)
+ }
+ batch.size = values.length
+ writer.addRowBatch(batch)
+ }
+ }
+ val path = dir.getCanonicalPath
+ // The stored values are read as `a`, null, `a`, `next`, and `b`.
+ val a = "TIMESTAMP'2020-05-25 10:11:12.123456'"
+ val b = "TIMESTAMP'2020-05-25 10:11:13'"
+ val next = "TIMESTAMP'2020-05-25 10:11:12.123457'"
+ checkAnswer(
+ spark.read.orc(path),
+ sql(s"SELECT $a UNION ALL SELECT null UNION ALL SELECT $a UNION ALL SELECT $next " +
+ s"UNION ALL SELECT $b"))
+
+ Seq(false, true).foreach { java8Api =>
+ withSQLConf(SQLConf.DATETIME_JAVA8API_ENABLED.key -> java8Api.toString) {
+ withAllNativeOrcReaders {
+ val df = spark.read.orc(path)
+ Seq(
+ s"ts = $b" -> 1,
+ s"ts = $a" -> 2,
+ s"ts = $next" -> 1,
+ s"ts <=> $a" -> 2,
+ s"ts <= $a" -> 2,
+ s"ts <= $b" -> 4,
+ s"ts IN ($a, $next)" -> 3,
+ s"ts IN ($a, $b)" -> 3,
+ s"ts < $next" -> 2,
+ s"ts >= $a" -> 4,
+ s"ts > $a" -> 2,
+ s"ts > $b" -> 0,
+ s"ts < $a" -> 0,
+ s"ts != $a" -> 2,
+ s"NOT (ts <=> $a)" -> 3,
+ s"ts NOT IN ($a, $next)" -> 1
+ ).foreach { case (condition, expected) =>
+ assert(df.where(condition).count() === expected, condition)
+ }
+ // The pushed down predicates still skip the files without the matching values.
+ assert(stripSparkFilter(df.where(s"ts = $a")).count() === 3)
+ assert(stripSparkFilter(df.where(s"ts > $a")).count() === 2)
+ assert(stripSparkFilter(df.where(s"ts = $b")).count() === 1)
+ }
+ }
+ }
+ }
+ }
+
test("SPARK-36357: filter pushdown - timestamp_ntz") {
val localDateTimes = Seq(
LocalDateTime.of(1000, 1, 1, 1, 2, 3, 456000000),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org