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/20 22:13:37 UTC
(spark) branch branch-4.3 updated: [SPARK-60396][SQL] Fix Hive ORC filter pushdown on nanosecond timestamp columns
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 2024bd8178d8 [SPARK-60396][SQL] Fix Hive ORC filter pushdown on nanosecond timestamp columns
2024bd8178d8 is described below
commit 2024bd8178d8392289fc114df5d9e711d41ade2a
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Oct 20 15:13:13 2026 -0700
[SPARK-60396][SQL] Fix Hive ORC filter pushdown on nanosecond timestamp columns
### What changes were proposed in this pull request?
This PR aims to fix Hive ORC filter pushdown (`spark.sql.orc.impl=hive`) on nanosecond timestamp columns by not pushing down predicates on them, including nested struct fields. Spark still evaluates these predicates after the scan.
### Why are the changes needed?
Since SPARK-57823, the Hive ORC reader gets these predicates with wrong values and skips matching rows.
- The Kryo round trip of the search argument truncates a `java.sql.Timestamp` value to milliseconds, e.g. `00:00:00.123456789` to `00:00:00.123`, while ORC statistics have nanosecond precision.
- A `TIMESTAMP_LTZ(p)` value is pushed down for the `timestamp with local time zone` column of the native ORC writer, but the Hive ORC writer writes a Hive `timestamp` column. So the value is shifted by the JVM time zone offset.
For example, the following returns 0 instead of 1.
```scala
spark.conf.set("spark.sql.orc.impl", "hive")
spark.conf.set("spark.sql.timestampNanosTypes.enabled", "true")
spark.sql("SELECT CAST('1970-01-01 00:00:00.123456789' AS TIMESTAMP_NTZ(9)) AS ts").write.orc(path)
spark.read.schema("ts TIMESTAMP_NTZ(9)").orc(path)
.where("ts = CAST('1970-01-01 00:00:00.123456789' AS TIMESTAMP_NTZ(9))").count()
```
The native ORC reader has the same Kryo issue, which will be handled separately.
### Does this PR introduce _any_ user-facing change?
No. Nanosecond timestamp types are not released yet.
### How was this patch tested?
Pass the CIs with the newly added test case.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59649 from dongjoon-hyun/SPARK-60396.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit b1c198fed6ea37ae13321582961b112a09afa699)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../apache/spark/sql/hive/orc/OrcFileFormat.scala | 13 +++++-
.../spark/sql/hive/orc/HiveOrcSourceSuite.scala | 48 +++++++++++++++++++++-
2 files changed, 59 insertions(+), 2 deletions(-)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
index 04857cff9d9b..7bcad9fa48f5 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
@@ -139,7 +139,7 @@ case class OrcFileFormat() extends FileFormat
if (getSqlConf(sparkSession).orcFilterPushDown) {
// Sets pushed predicates
- OrcFilters.createFilter(requiredSchema, filters).foreach { f =>
+ OrcFilters.createFilter(withoutNanosTimestamps(requiredSchema), filters).foreach { f =>
hadoopConf.set(OrcFileFormat.SARG_PUSHDOWN, toKryo(f))
hadoopConf.setBoolean("hive.optimize.index.filter", true)
}
@@ -211,6 +211,17 @@ case class OrcFileFormat() extends FileFormat
case _ => false
}
+ // Removes the nanosecond timestamp fields, so that no predicate on them is pushed down. The
+ // search argument reaches Hive's ORC reader through a Kryo round trip (see `toKryo`) that
+ // truncates a java.sql.Timestamp value to milliseconds, while ORC compares it with nanosecond
+ // statistics. Besides, the TIMESTAMP_LTZ(p) value that `OrcFilters` pushes down is for the
+ // `timestamp with local time zone` column of the native ORC writer, while this format writes a
+ // Hive `timestamp` column.
+ private def withoutNanosTimestamps(schema: StructType): StructType = StructType(schema.collect {
+ case f @ StructField(_, st: StructType, _, _) => f.copy(dataType = withoutNanosTimestamps(st))
+ case f if !f.dataType.isInstanceOf[AnyTimestampNanoType] => f
+ })
+
// HIVE-11253 moved `toKryo` from `SearchArgument` to `storage-api` module.
// This is copied from Hive 1.2's SearchArgumentImpl.toKryo().
private def toKryo(sarg: SearchArgument): String = {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
index 96ac15893d07..f90b3037c3bd 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.hive.orc
import java.io.File
-import java.time.LocalDateTime
+import java.time.{Instant, LocalDateTime}
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.TestingUDT.{IntervalData, IntervalUDT}
@@ -453,6 +453,52 @@ class HiveOrcSourceSuite extends OrcSuite with TestHiveSingleton {
}
}
+ test("SPARK-60396: Hive ORC filter pushdown on nanosecond timestamps") {
+ // A pushed down predicate on a nanosecond timestamp reaches the Hive ORC reader with a wrong
+ // value: the Kryo round trip of the search argument truncates it to milliseconds, and a
+ // TIMESTAMP_LTZ(p) value is also shifted by the offset of the JVM time zone. For example,
+ // `ts = 1970-01-01 00:00:00.123456789` skips the stripe that holds the matching row.
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.ORC_IMPLEMENTATION.key -> "hive",
+ SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ Seq(
+ "TIMESTAMP_LTZ" -> ((s: String) => Instant.parse(s"1970-01-01T${s}Z")),
+ "TIMESTAMP_NTZ" -> ((s: String) => LocalDateTime.parse(s"1970-01-01T$s"))
+ ).foreach { case (tsType, toValue) =>
+ foreachNanosPrecision { p =>
+ withTempPath { dir =>
+ sql(s"SELECT ts, named_struct('ts', ts) AS s FROM (SELECT CAST(c AS $tsType($p)) " +
+ "AS ts FROM VALUES ('1970-01-01 00:00:00.123456789'), ('1970-01-01 00:00:01'), " +
+ "(NULL) AS t(c))").coalesce(1).write.orc(dir.getCanonicalPath)
+ val df = spark.read.schema(s"ts $tsType($p), s STRUCT<ts: $tsType($p)>")
+ .orc(dir.getCanonicalPath)
+ // The stored value, e.g. 00:00:00.1234567 for p = 7.
+ val v = "00:00:00." + "123456789".take(p)
+ def lit(s: String): String = s"CAST('1970-01-01 $s' AS $tsType($p))"
+ def check(condition: String, expected: Seq[String]): Unit = {
+ checkAnswer(df.where(condition).select("ts"), expected.map(s => Row(toValue(s))))
+ }
+ // Both the top-level column and the nested field.
+ Seq("ts", "s.ts").foreach { c =>
+ check(s"$c = ${lit(v)}", Seq(v))
+ check(s"$c <=> ${lit(v)}", Seq(v))
+ check(s"$c != ${lit(v)}", Seq("00:00:01"))
+ check(s"$c < ${lit("00:00:00.1239")}", Seq(v))
+ check(s"$c <= ${lit(v)}", Seq(v))
+ check(s"$c > ${lit(v)}", Seq("00:00:01"))
+ check(s"$c >= ${lit(v)}", Seq(v, "00:00:01"))
+ check(s"$c IN (${lit(v)}, ${lit("00:00:02")})", Seq(v))
+ checkAnswer(df.where(s"$c IS NULL").select("ts"), Row(null))
+ check(s"$c IS NOT NULL", Seq(v, "00:00:01"))
+ }
+ }
+ }
+ }
+ }
+ }
+
test("SPARK-31580: Read a file written before ORC-569") {
// Test ORC file came from ORC-621
val df = readResourceOrcFile("test-data/TestStringDictionary.testRowIndex.orc")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org