This is an automated email from the ASF dual-hosted git repository.
uros-b 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 55600bf547b8 [SPARK-57823][SQL] Support ORC predicate pushdown for
nanosecond-precision timestamps
55600bf547b8 is described below
commit 55600bf547b85ba7b95b9ea4d97144fb7f7c9e69
Author: Stevo Mitric <[email protected]>
AuthorDate: Tue Aug 11 10:47:27 2026 +0000
[SPARK-57823][SQL] Support ORC predicate pushdown for nanosecond-precision
timestamps
### What changes were proposed in this pull request?
Enable ORC predicate pushdown for nanosecond-precision timestamp columns
(`TIMESTAMP_LTZ(p)` / `TIMESTAMP_NTZ(p)`, `p` in [7, 9]).
`OrcFilters.getPredicateLeafType` / `castLiteralValue` route framework
types through `OrcTypeOps`, but both nanos ops inherited the default
`predicateLeafType = None`. A pushed filter on a nanos column therefore reached
the throwing arm of `getPredicateLeafType` and failed the query. This wires the
two nanos ops (`TimestampLTZNanosOrcOps` / `TimestampNTZNanosOrcOps`) into the
pushdown seam:
- `predicateLeafType` now returns `Some(PredicateLeaf.Type.TIMESTAMP)`,
matching the physical ORC category the types are stored in (`TIMESTAMP` for
NTZ, `TIMESTAMP_INSTANT` for LTZ).
- `castFilterLiteral` converts the external filter literal to a plain
`java.sql.Timestamp` whose local wall clock equals the value's nominal wall
clock, so the pushed value compares consistently against the ORC min/max
statistics:
- NTZ (`java.time.LocalDateTime`) -> `Timestamp.valueOf(ldt)`.
- LTZ (`java.time.Instant`) ->
`Timestamp.valueOf(LocalDateTime.ofInstant(i, UTC))`.
Two ORC-specific details drove the literal shape:
1. ORC's `SearchArgument` builder checks the literal's class by exact
equality (`value.getClass() == Type.getValueClass()`), so it rejects the
`OrcTimestamp` subclass used on the write path; the literal must be a plain
`java.sql.Timestamp`.
2. ORC evaluates a `TIMESTAMP` / `TIMESTAMP_INSTANT` predicate against the
stored value shifted into the JVM default zone (the default
`useUTCTimestamp=false` read mode Spark uses), so the literal must carry the
nominal wall clock in the local zone rather than the raw epoch instant. This
keeps pushdown correct regardless of the JVM time zone.
Stale comments in `TimestampNanosOrcOps` / `OrcTypeOps` / `OrcFilters` that
stated the nanos types had no pushdown are updated.
### Why are the changes needed?
Sub-task of SPARK-56822. Without this, a filter on a nanosecond-precision
timestamp ORC column throws `unsupportedOperationForDataTypeError` during
planning instead of pushing the predicate down (or safely skipping it).
### Does this PR introduce any user-facing change?
Yes. Predicates on nanosecond-precision timestamp ORC columns now push down
to ORC search arguments (enabling stripe/row-group skipping) instead of failing.
### How was this patch tested?
- New `OrcFilterSuite` test "SPARK-57823: filter pushdown - nanosecond
timestamp" asserts the expected `PredicateLeaf.Operator` is generated for all
comparison operators, both nanos types, and all precisions.
- New `OrcQuerySuite` test "SPARK-57823: ORC predicate pushdown returns
correct results for nanos timestamps" verifies end-to-end that results are
correct (`checkAnswer`) and that stripes are actually skipped
(`stripSparkFilter` count), across V1/V2, vectorized/non-vectorized readers,
both nanos types, and all precisions.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57681 from stevomitric/stevomitric/spark-57823-orc-nanos-pushdown.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 1a9fb51adb5009560e99c7e2cea8f77afefe5a6c)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../sql/execution/datasources/orc/OrcFilters.scala | 6 +-
.../datasources/orc/types/ops/OrcTypeOps.scala | 6 +-
.../orc/types/ops/TimestampNanosOrcOps.scala | 50 ++++++++++-
.../execution/datasources/orc/OrcFilterSuite.scala | 64 +++++++++++++-
.../execution/datasources/orc/OrcQuerySuite.scala | 98 +++++++++++++++++++++-
5 files changed, 212 insertions(+), 12 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 4a092a792a79..6ca2511b3316 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
@@ -149,9 +149,9 @@ private[sql] object OrcFilters extends OrcFiltersBase {
case DateType => PredicateLeaf.Type.DATE
case TimestampType => PredicateLeaf.Type.TIMESTAMP
case _: DecimalType => PredicateLeaf.Type.DECIMAL
- // Framework types (e.g. TimeType) supply their own predicate-leaf type. A
framework type whose
- // predicateLeafType is None (like the nanos-timestamp types), or any
other unmapped type,
- // reaches the same unsupported-type error as before this change.
+ // Framework types (e.g. TimeType, the nanosecond-timestamp types) supply
their own
+ // predicate-leaf type. A framework type whose predicateLeafType is None,
or any other unmapped
+ // type, reaches the same unsupported-type error as before this change.
case dt => OrcTypeOps(dt).flatMap(_.predicateLeafType)
.getOrElse(throw
QueryExecutionErrors.unsupportedOperationForDataTypeError(dt))
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/OrcTypeOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/OrcTypeOps.scala
index 357d3b18bad8..b5a2a6010368 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/OrcTypeOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/OrcTypeOps.scala
@@ -145,9 +145,9 @@ private[orc] trait OrcTypeOps extends Serializable {
* conversion, override castFilterLiteral); returning None is only safe for
a type that can never
* reach getPredicateLeafType.
*
- * TimeType returns Some(LONG). The nanosecond-timestamp types return None:
they had no
- * getPredicateLeafType arm before this change either, so the behavior
(throw if ever reached) is
- * preserved, not newly introduced.
+ * TimeType and the nanosecond-timestamp types return Some (LONG and
TIMESTAMP respectively) so
+ * their pushed filters convert to search arguments. A type that returns
None (there are none
+ * today) would reach the throw arm if ever pushed.
*/
def predicateLeafType: Option[PredicateLeaf.Type] = None
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/TimestampNanosOrcOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/TimestampNanosOrcOps.scala
index 87a4bd252a88..5d5923b5b571 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/TimestampNanosOrcOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/types/ops/TimestampNanosOrcOps.scala
@@ -17,6 +17,10 @@
package org.apache.spark.sql.execution.datasources.orc.types.ops
+import java.sql.Timestamp
+import java.time.{Instant, LocalDateTime, ZoneOffset}
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf
import org.apache.hadoop.io.WritableComparable
import org.apache.orc.TypeDescription
import org.apache.orc.mapred.OrcTimestamp
@@ -35,8 +39,14 @@ import org.apache.spark.unsafe.types.TimestampNanosVal
* ORC does not reuse the OrcTimestamp object (the conversion is already
expensive), so the
* serializer runs with reuse disabled regardless of the reuseObj hint.
*
- * No predicate pushdown: the inline OrcFilters code never added a nanos arm,
so predicateLeafType
- * stays None (inherited default).
+ * Predicate pushdown uses PredicateLeaf.Type.TIMESTAMP (matching the physical
category). ORC's
+ * SearchArgument builder requires the literal's class to be exactly
java.sql.Timestamp (it rejects
+ * the OrcTimestamp subclass), and ORC compares against a TIMESTAMP_INSTANT
column by first shifting
+ * the stored UTC value into the JVM default zone (the default
useUTCTimestamp=false read mode Spark
+ * uses). So the filter literal (an external java.time.Instant, already
truncated to the column
+ * precision upstream) is cast to a java.sql.Timestamp whose *local* wall
clock equals the instant's
+ * UTC wall clock, via Timestamp.valueOf(LocalDateTime.ofInstant(instant,
UTC)). This keeps the
+ * pushed predicate consistent with the ORC min/max statistics regardless of
the JVM time zone.
*
* @since 5.0.0
*/
@@ -69,6 +79,19 @@ case class TimestampLTZNanosOrcOps(t: TimestampLTZNanosType)
extends OrcTypeOps
val instant = ts.toInstant
set(ordinal, DateTimeUtils.instantToTimestampNanos(instant, t.precision))
}
+
+ // The physical ORC category is a timestamp, so the search argument uses the
TIMESTAMP leaf type.
+ override def predicateLeafType: Option[PredicateLeaf.Type] =
Some(PredicateLeaf.Type.TIMESTAMP)
+
+ // The filter literal is an external java.time.Instant (see
CatalystTypeConverters). ORC evaluates
+ // a TIMESTAMP_INSTANT predicate against the stored UTC value shifted into
the JVM default zone,
+ // so the literal must be a java.sql.Timestamp whose local wall clock equals
the instant's UTC
+ // wall clock; Timestamp.valueOf(LocalDateTime at UTC) produces exactly
that. Any non-Instant
+ // value is passed through unchanged.
+ override def castFilterLiteral(value: Any): Any = value match {
+ case i: Instant => Timestamp.valueOf(LocalDateTime.ofInstant(i,
ZoneOffset.UTC))
+ case other => other
+ }
}
/**
@@ -79,7 +102,14 @@ case class TimestampLTZNanosOrcOps(t:
TimestampLTZNanosType) extends OrcTypeOps
* read from the CATALYST_TYPE_ATTRIBUTE_NAME attribute. As with the LTZ
variant, the OrcTimestamp
* object is not reused.
*
- * No predicate pushdown (predicateLeafType stays None).
+ * Predicate pushdown uses PredicateLeaf.Type.TIMESTAMP (matching the physical
category). The filter
+ * literal is an external java.time.LocalDateTime (already truncated to the
column precision
+ * upstream); it is cast to a plain java.sql.Timestamp via Timestamp.valueOf,
the same conversion
+ * the write path applies before wrapping the result in an OrcTimestamp. ORC's
SearchArgument
+ * builder requires the literal's class to be exactly java.sql.Timestamp (it
rejects the
+ * OrcTimestamp subclass), while the stored OrcTimestamp compares as its
java.sql.Timestamp
+ * super-value, so the two are value-equal and the pushed predicate is
consistent with the ORC
+ * min/max statistics.
*
* @since 5.0.0
*/
@@ -97,7 +127,7 @@ case class TimestampNTZNanosOrcOps(t: TimestampNTZNanosType)
extends OrcTypeOps
(getter: SpecializedGetters, ordinal: Int) => {
val v = getter.get(ordinal, t).asInstanceOf[TimestampNanosVal]
val localDateTime = DateTimeUtils.timestampNanosToLocalDateTime(v)
- val ts = java.sql.Timestamp.valueOf(localDateTime)
+ val ts = Timestamp.valueOf(localDateTime)
val result = new OrcTimestamp(ts.getTime)
result.setNanos(ts.getNanos)
result
@@ -112,4 +142,16 @@ case class TimestampNTZNanosOrcOps(t:
TimestampNTZNanosType) extends OrcTypeOps
val localDateTime = ts.toLocalDateTime
set(ordinal, DateTimeUtils.localDateTimeToTimestampNanos(localDateTime,
t.precision))
}
+
+ // The physical ORC category is a timestamp, so the search argument uses the
TIMESTAMP leaf type.
+ override def predicateLeafType: Option[PredicateLeaf.Type] =
Some(PredicateLeaf.Type.TIMESTAMP)
+
+ // The filter literal is an external java.time.LocalDateTime (see
CatalystTypeConverters); convert
+ // it to a plain java.sql.Timestamp via the same Timestamp.valueOf the
serializer uses, so it is
+ // value-equal to the written OrcTimestamp. Any non-LocalDateTime value is
passed through
+ // unchanged.
+ override def castFilterLiteral(value: Any): Any = value match {
+ case ldt: LocalDateTime => Timestamp.valueOf(ldt)
+ case other => other
+ }
}
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 3168075cc9d9..a7eb37e95278 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
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.datasources.orc
import java.math.MathContext
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}
-import java.time.{Duration, LocalDateTime, LocalTime, Period}
+import java.time.{Duration, LocalDateTime, LocalTime, Period, ZoneOffset}
import scala.jdk.CollectionConverters._
@@ -32,6 +32,7 @@ import org.apache.spark.sql.{AnalysisException, Column,
DataFrame, Row}
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
import org.apache.spark.sql.execution.datasources.v2.ExtractV2Scan
import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
import org.apache.spark.sql.functions.col
@@ -398,6 +399,67 @@ class OrcFilterSuite extends OrcTest with
SharedSparkSession {
}
}
+ test("SPARK-57823: filter pushdown - nanosecond timestamp") {
+ // Wall clocks carry sub-microsecond digits. The literal is explicitly
nanos-typed, so even a
+ // microsecond-aligned value would still exercise the nanos pushdown path;
the extra digits
+ // exercise value preservation and comparison beyond microsecond precision.
+ val wallClocks = Seq(
+ LocalDateTime.of(1000, 1, 1, 1, 2, 3, 456789123),
+ LocalDateTime.of(1582, 10, 1, 0, 11, 22, 456789123),
+ LocalDateTime.of(1900, 1, 1, 23, 59, 59, 456789123),
+ LocalDateTime.of(2020, 5, 25, 10, 11, 12, 456789123))
+
+ // Builds a nanos-typed literal so the comparison against the nanos column
needs no type
+ // coercion: NTZ uses the wall clock as a LocalDateTime, LTZ as the same
instant at UTC.
+ def nanosLiteral(nanosType: DataType, wallClock: LocalDateTime):
Expression = {
+ val external: Any = nanosType match {
+ case _: TimestampNTZNanosType => wallClock
+ case _: TimestampLTZNanosType => wallClock.toInstant(ZoneOffset.UTC)
+ }
+ Literal.create(external, nanosType)
+ }
+
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val ts = wallClocks.map(nanosLiteral(nanosType, _))
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ nanosTimestampDf(nanosType, wallClocks).write.orc(path)
+ // Read back without an explicit schema: the nanos type is
recovered from the ORC
+ // catalyst-type attribute, so the pushed-down column is
nanos-typed.
+ readFile(path) { implicit df =>
+ assert(df("ts").expr.dataType === nanosType)
+
+ checkFilterPredicate($"ts".isNull,
PredicateLeaf.Operator.IS_NULL)
+
+ checkFilterPredicate($"ts" === ts(0),
PredicateLeaf.Operator.EQUALS)
+ checkFilterPredicate($"ts" <=> ts(0),
PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+
+ checkFilterPredicate($"ts" < ts(1),
PredicateLeaf.Operator.LESS_THAN)
+ checkFilterPredicate($"ts" > ts(2),
PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate($"ts" <= ts(0),
PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate($"ts" >= ts(3),
PredicateLeaf.Operator.LESS_THAN)
+
+ checkFilterPredicate(ts(0) === $"ts",
PredicateLeaf.Operator.EQUALS)
+ checkFilterPredicate(ts(0) <=> $"ts",
PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+ checkFilterPredicate(ts(1) > $"ts",
PredicateLeaf.Operator.LESS_THAN)
+ checkFilterPredicate(ts(2) < $"ts",
PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate(ts(0) >= $"ts",
PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate(ts(3) <= $"ts",
PredicateLeaf.Operator.LESS_THAN)
+
+ // In covers the per-value castLiteralValue path
(values.map(...)) in
+ // buildLeafSearchArgument, exercising the nanos literal cast
for every element.
+ checkFilterPredicate(
+ In($"ts", Seq(ts(0), ts(2))), PredicateLeaf.Operator.IN)
+ }
+ }
+ }
+ }
+ }
+ }
+
test("filter pushdown - combinations with logical operators") {
withOrcDataFrame((1 to 4).map(i => Tuple1(Option(i)))) { implicit df =>
checkFilterPredicate(
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 909ebf5daf39..b2821c3d5fad 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
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.datasources.orc
import java.io.File
import java.nio.charset.StandardCharsets
import java.sql.Timestamp
-import java.time.LocalDateTime
+import java.time.{LocalDateTime, ZoneOffset}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
@@ -35,6 +35,7 @@ import org.apache.orc.mapreduce.OrcInputFormat
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.expressions.Literal
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
import org.apache.spark.sql.execution.FileSourceScanExec
@@ -1029,6 +1030,101 @@ abstract class OrcQuerySuite extends OrcQueryTest with
SharedSparkSession {
}
}
+ test("SPARK-57823: ORC predicate pushdown returns correct results for nanos
timestamps") {
+ // One wall clock per second across a minute so each row lands in a
distinct stripe below, all
+ // carrying sub-microsecond digits so the nanosecond fraction participates
in the comparison.
+ val numRows = 60
+ val wallClocks = (0 until numRows).map { s =>
+ LocalDateTime.of(2020, 5, 25, 10, 0, s, 123456789)
+ }
+ def checkNanosPushdown(): Unit = {
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val inputDf = nanosTimestampDf(nanosType, wallClocks)
+ // A boundary literal (row 30) expressed as the nanos external
type, so a `< threshold`
+ // predicate keeps exactly the first 30 rows.
+ val boundary = nanosType match {
+ case _: TimestampNTZNanosType => Literal.create(wallClocks(30),
nanosType)
+ case _: TimestampLTZNanosType =>
+ Literal.create(wallClocks(30).toInstant(ZoneOffset.UTC),
nanosType)
+ }
+ // Rebuilt against each DataFrame's own `ts` attribute so the
filter resolves cleanly.
+ def keepFirstHalf(df: DataFrame): Column = df("ts") <
Column(boundary)
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ // Repartition so the rows are spread over several ORC
files/stripes, giving the
+ // pushed-down search argument something to skip.
+ inputDf.repartition(numRows).write.orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(
+ SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString,
+ SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ val read = spark.read.schema(new StructType().add("ts",
nanosType)).orc(path)
+
+ // Results are correct: the pushdown must not drop or
corrupt matching rows.
+ checkAnswer(
+ read.where(keepFirstHalf(read)),
inputDf.where(keepFirstHalf(inputDf)))
+
+ // Pushdown actually skips data: with the Spark-side filter
removed, fewer than
+ // all rows survive, proving the ORC search argument pruned
stripes.
+
assert(stripSparkFilter(read.where(keepFirstHalf(read))).count() < numRows)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ // The LTZ column stores the UTC instant; the JVM default zone enters only
at comparison time,
+ // when ORC evaluates the pushed predicate against the non-UTC timestamp
statistics. So a
+ // literal built in the wrong frame would only diverge from the stripe
stats outside UTC. Run
+ // under zones with both offset signs so timezone handling is exercised,
not just UTC where such
+ // a bug is masked.
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ Seq(DateTimeTestUtils.UTC, DateTimeTestUtils.LA,
DateTimeTestUtils.JST).foreach { zoneId =>
+ DateTimeTestUtils.withDefaultTimeZone(zoneId)(checkNanosPushdown())
+ }
+ }
+ }
+
+ test("SPARK-57823: LTZ nanos pushdown stays correct across a DST
spring-forward gap") {
+ // Timestamp.valueOf maps a wall clock inside the JVM zone's
spring-forward gap one hour ahead,
+ // so for instants whose UTC wall clock lands in that gap the literal
frame is non-monotonic.
+ // Under America/Los_Angeles the gap is 2020-03-08 02:00..03:00, so lay
data one minute apart
+ // straddling it (UTC wall clocks 01:30..03:29) with a boundary at 02:45.
checkAnswer must hold:
+ // in the gap the pushed predicate may only turn conservative (skip fewer
stripes), never prune
+ // a stripe that holds matching rows.
+ val numRows = 120
+ val base = LocalDateTime.of(2020, 3, 8, 1, 30, 0, 123456789)
+ val wallClocks = (0 until numRows).map(m => base.plusMinutes(m))
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.LA) {
+ foreachNanosPrecision { precision =>
+ val nanosType = TimestampLTZNanosType(precision)
+ val inputDf = nanosTimestampDf(nanosType, wallClocks)
+ val boundary = Literal.create(
+ LocalDateTime.of(2020, 3, 8, 2, 45, 0,
0).toInstant(ZoneOffset.UTC), nanosType)
+ def keepBeforeGapBoundary(df: DataFrame): Column = df("ts") <
Column(boundary)
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ inputDf.repartition(numRows).write.orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(
+ SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString,
+ SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ val read = spark.read.schema(new StructType().add("ts",
nanosType)).orc(path)
+ checkAnswer(
+ read.where(keepBeforeGapBoundary(read)),
+ inputDf.where(keepBeforeGapBoundary(inputDf)))
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
test("SPARK-57455: ORC round-trips nanos timestamps in nested/complex
types") {
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
val wallClocks = Seq(LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
