You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/08/11 10:46:46 UTC
(spark) branch master updated: [SPARK-57822][SQL] Support Parquet predicate pushdown for nanosecond-precision timestamps
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6dfac55e3beb [SPARK-57822][SQL] Support Parquet predicate pushdown for nanosecond-precision timestamps
6dfac55e3beb is described below
commit 6dfac55e3beb5693f158e4a59ec57b595b283a78
Author: Stevo Mitric <st...@gmail.com>
AuthorDate: Tue Aug 11 10:46:33 2026 +0000
[SPARK-57822][SQL] Support Parquet predicate pushdown for nanosecond-precision timestamps
### What changes were proposed in this pull request?
Add `TIMESTAMP(NANOS)` arms to `ParquetFilters` so equality/range/in predicates on nanosecond-precision timestamp columns (`TIMESTAMP_NTZ(p)`/`TIMESTAMP_LTZ(p)`, p in [7, 9]) are pushed down as Parquet row-group filters. LTZ (`java.time.Instant`) and NTZ (`java.time.LocalDateTime`) get separate schema-type constants and value converters;
values are encoded to signed INT64 epoch-nanoseconds using the same encoding `TimestampNanosParquetOps` writes (not truncated to micros). `valueCanMakeFilterOn` gates on int64-representability, so an out-of-range value falls back to a full scan instead of throwing or mis-skipping row groups. Part of SPARK-56822.
### Why are the changes needed?
There was no nanos arm in `ParquetFilters`, so predicates on nanosecond timestamp columns were silently never pushed down and every row group was scanned.
### Does this PR introduce _any_ user-facing change?
Yes. Filters on nanosecond timestamp columns are now pushed to Parquet and prune row groups (preview `spark.sql.timestampNanosTypes.enabled`); results are unchanged.
### How was this patch tested?
`ParquetFilterSuite` (new nanos NTZ/LTZ pushdown cases) and `ParquetTimestampNanosSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57430 from stevomitric/stevomitric/spark-57822-nanos.
Authored-by: Stevo Mitric <st...@gmail.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../datasources/parquet/ParquetFilters.scala | 9 +-
.../parquet/types/ops/ParquetTypeOps.scala | 5 +-
.../types/ops/TimestampNanosParquetOps.scala | 59 ++++++++
.../datasources/parquet/ParquetFilterSuite.scala | 149 ++++++++++++++++++++-
.../parquet/ParquetTimestampNanosSuite.scala | 57 ++++++++
.../types/ops/TimestampNanosParquetOpsSuite.scala | 129 +++++++++++++++++-
6 files changed, 404 insertions(+), 4 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
index f60ced3eb597..50f90048b8c8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
@@ -800,8 +800,15 @@ class ParquetFilters(
createFilterHelper(pred, canPartialPushDownConjuncts = false)
.map(FilterApi.not)
+ // Every value must be pushable, not just the head: both branches below convert *all* the
+ // values (per-element `makeEq` when under the threshold, `makeInPredicate` otherwise), so a
+ // head-only check lets a non-pushable tail element reach the converter. For a value-range-
+ // sensitive type (e.g. nanosecond timestamps, whose encoder throws outside the int64
+ // epoch-nanos range) that would crash filter creation instead of falling back to a full
+ // scan. `forall` short-circuits and, for type-only `valueCanMakeFilterOn` checks, is
+ // equivalent to the previous head check (all `In` values share the coerced column type).
case sources.In(name, values) if pushDownInFilterThreshold > 0 && values.nonEmpty &&
- canMakeFilterOn(name, values.head) =>
+ values.forall(canMakeFilterOn(name, _)) =>
val fieldType = nameToParquetField(name).fieldType
val fieldNames = nameToParquetField(name).fieldNames
if (values.length <= pushDownInFilterThreshold) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
index 7c2091c8c081..f839e6c90043 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
@@ -306,5 +306,8 @@ private[parquet] object ParquetTypeOps {
* predicate pushdown lists its [[ParquetFilterOps]] here. This is what `filterOpsFor`
* scans, so a new type participates in pushdown by adding its ops to this Seq.
*/
- private val filterOpsList: Seq[ParquetFilterOps] = Seq(TimeTypeParquetOps.filterOps)
+ private val filterOpsList: Seq[ParquetFilterOps] = Seq(
+ TimeTypeParquetOps.filterOps,
+ TimestampNanosParquetOps.ltzFilterOps,
+ TimestampNanosParquetOps.ntzFilterOps)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
index 6eb19f5f34b0..04f6633afd12 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
@@ -17,6 +17,9 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import java.lang.{Long => JLong}
+import java.time.{Instant, LocalDateTime}
+
import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
import org.apache.parquet.io.api.{Converter, RecordConsumer}
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
@@ -185,6 +188,62 @@ private[ops] object TimestampNanosParquetOps {
case ts: TimestampLogicalTypeAnnotation => ts.getUnit == TimeUnit.NANOS
case _ => false
})
+
+ // Repacks an externalized nanos filter value into the signed INT64 epoch-nanoseconds the write
+ // path produces. Conversion is at precision 9 (a lossless repack): the literal has already been
+ // floored to the column precision upstream, so no sub-microsecond digits are dropped here. The
+ // single-arg `timestampNanosToEpochNanos` throws `ArithmeticException` outside the int64 range;
+ // callers reach it only after `acceptsValue` has cleared the value (see [[epochNanosInRange]]).
+ private def instantToEpochNanos(v: Instant): JLong =
+ DateTimeUtils.timestampNanosToEpochNanos(
+ DateTimeUtils.instantToTimestampNanos(v, TimestampLTZNanosType.NANOS_PRECISION))
+
+ private def localDateTimeToEpochNanos(v: LocalDateTime): JLong =
+ DateTimeUtils.timestampNanosToEpochNanos(
+ DateTimeUtils.localDateTimeToTimestampNanos(v, TimestampNTZNanosType.NANOS_PRECISION))
+
+ // SPARK-46092-style guard: only push down when the value is representable as int64
+ // epoch-nanoseconds. An out-of-range value would throw in the encoder, and -- worse -- a
+ // wrapped/truncated encoding could silently mis-skip row groups; rejecting it falls back to a
+ // full scan, which is always correct.
+ private def epochNanosInRange(encode: => JLong): Boolean =
+ try { encode; true } catch { case _: ArithmeticException => false }
+
+ /**
+ * Parquet filter-pushdown ops for the nanosecond timestamp types, registered in
+ * [[ParquetTypeOps.filterOpsList]]. Filter dispatch is keyed on the file's on-disk encoding, so
+ * each type gets its own ops: both are stored as INT64 TIMESTAMP(NANOS) and differ only in the
+ * `isAdjustedToUTC` flag (LTZ = true, NTZ = false), which also fixes the externalized filter
+ * value (`java.time.Instant` for LTZ, `java.time.LocalDateTime` for NTZ). Values are encoded to
+ * the same signed INT64 epoch-nanoseconds `TimestampNanosParquetOps` writes, never truncated to
+ * micros. This replaces the inline nanos arms once carried in `ParquetFilters`, matching how
+ * TimeType routes its pushdown through [[TimeTypeParquetOps.filterOps]].
+ */
+ private[ops] val ltzFilterOps: ParquetFilterOps = new LongParquetFilterOps {
+ override val logicalTypeAnnotation: LogicalTypeAnnotation =
+ LogicalTypeAnnotation.timestampType(true, TimeUnit.NANOS)
+
+ override def acceptsValue(value: Any): Boolean = value match {
+ case i: Instant => epochNanosInRange(instantToEpochNanos(i))
+ case _ => false
+ }
+
+ override protected def toLong(value: Any): JLong =
+ instantToEpochNanos(value.asInstanceOf[Instant])
+ }
+
+ private[ops] val ntzFilterOps: ParquetFilterOps = new LongParquetFilterOps {
+ override val logicalTypeAnnotation: LogicalTypeAnnotation =
+ LogicalTypeAnnotation.timestampType(false, TimeUnit.NANOS)
+
+ override def acceptsValue(value: Any): Boolean = value match {
+ case ldt: LocalDateTime => epochNanosInRange(localDateTimeToEpochNanos(ldt))
+ case _ => false
+ }
+
+ override protected def toLong(value: Any): JLong =
+ localDateTimeToEpochNanos(value.asInstanceOf[LocalDateTime])
+ }
}
/**
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
index ae977e17755d..54c5b279f6ce 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
@@ -22,7 +22,7 @@ import java.lang.{Double => JDouble, Float => JFloat, Long => JLong}
import java.math.{BigDecimal => JBigDecimal}
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}
-import java.time.{Duration, LocalDate, LocalDateTime, LocalTime, Period, ZoneId}
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime, Period, ZoneId}
import java.util.HashSet
import scala.reflect.ClassTag
@@ -2275,6 +2275,153 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession {
}
}
}
+
+ test("SPARK-57822: filter pushdown - nanosecond timestamps") {
+ // The value that reaches `ParquetFilters.createFilter` for a nanosecond timestamp column is
+ // the externalized Java time value (Instant for LTZ, LocalDateTime for NTZ), NOT a
+ // TimestampNanosVal. Both convert to the signed INT64 epoch-nanoseconds that
+ // `TimestampNanosParquetOps` writes, so every operator must produce a `long`-column filter.
+ // Threshold 1 so a 2-element `In` exceeds it and exercises the `makeInPredicate` nanos arm
+ // (below the threshold, `In` expands to an `Or` of `Eq`, which is covered by the `Eq` arm).
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> "1") {
+ Seq(7, 8, 9).foreach { p =>
+ // NTZ column: filter values are LocalDateTime. Use a sub-microsecond fraction so the
+ // epoch-nanos encoding (not truncated micros) is exercised.
+ val ntzSchema = new SparkToParquetSchemaConverter(conf)
+ .convert(new StructType().add("c", TimestampNTZNanosType(p)))
+ val ntzFilters = createParquetFilters(ntzSchema)
+ val ntz = LocalDateTime.parse("2020-01-01T12:34:56.000000123")
+ val ntzHi = LocalDateTime.parse("2020-01-01T12:34:57.000000123")
+ assert(ntzFilters.createFilter(sources.IsNull("c")).exists(_.isInstanceOf[Eq[_]]))
+ assert(ntzFilters.createFilter(sources.IsNotNull("c")).exists(_.isInstanceOf[NotEq[_]]))
+ assert(ntzFilters.createFilter(sources.EqualTo("c", ntz)).exists(_.isInstanceOf[Eq[_]]))
+ assert(ntzFilters.createFilter(sources.EqualNullSafe("c", ntz))
+ .exists(_.isInstanceOf[Eq[_]]))
+ assert(ntzFilters.createFilter(sources.Not(sources.EqualTo("c", ntz)))
+ .exists(_.isInstanceOf[NotEq[_]]))
+ assert(ntzFilters.createFilter(sources.LessThan("c", ntz)).exists(_.isInstanceOf[Lt[_]]))
+ assert(ntzFilters.createFilter(sources.LessThanOrEqual("c", ntz))
+ .exists(_.isInstanceOf[LtEq[_]]))
+ assert(ntzFilters.createFilter(sources.GreaterThan("c", ntz)).exists(_.isInstanceOf[Gt[_]]))
+ assert(ntzFilters.createFilter(sources.GreaterThanOrEqual("c", ntz))
+ .exists(_.isInstanceOf[GtEq[_]]))
+ assert(ntzFilters.createFilter(sources.In("c", Array[Any](ntz, ntzHi)))
+ .exists(_.isInstanceOf[FilterIn[_]]))
+
+ // LTZ column: filter values are Instant.
+ val ltzSchema = new SparkToParquetSchemaConverter(conf)
+ .convert(new StructType().add("c", TimestampLTZNanosType(p)))
+ val ltzFilters = createParquetFilters(ltzSchema)
+ val ltz = Instant.parse("2020-01-01T12:34:56.000000123Z")
+ val ltzHi = Instant.parse("2020-01-01T12:34:57.000000123Z")
+ assert(ltzFilters.createFilter(sources.IsNull("c")).exists(_.isInstanceOf[Eq[_]]))
+ assert(ltzFilters.createFilter(sources.IsNotNull("c")).exists(_.isInstanceOf[NotEq[_]]))
+ assert(ltzFilters.createFilter(sources.EqualTo("c", ltz)).exists(_.isInstanceOf[Eq[_]]))
+ assert(ltzFilters.createFilter(sources.EqualNullSafe("c", ltz))
+ .exists(_.isInstanceOf[Eq[_]]))
+ assert(ltzFilters.createFilter(sources.Not(sources.EqualTo("c", ltz)))
+ .exists(_.isInstanceOf[NotEq[_]]))
+ assert(ltzFilters.createFilter(sources.LessThan("c", ltz)).exists(_.isInstanceOf[Lt[_]]))
+ assert(ltzFilters.createFilter(sources.LessThanOrEqual("c", ltz))
+ .exists(_.isInstanceOf[LtEq[_]]))
+ assert(ltzFilters.createFilter(sources.GreaterThan("c", ltz)).exists(_.isInstanceOf[Gt[_]]))
+ assert(ltzFilters.createFilter(sources.GreaterThanOrEqual("c", ltz))
+ .exists(_.isInstanceOf[GtEq[_]]))
+ assert(ltzFilters.createFilter(sources.In("c", Array[Any](ltz, ltzHi)))
+ .exists(_.isInstanceOf[FilterIn[_]]))
+ }
+ }
+ }
+
+ test("SPARK-57822: don't push down nanosecond timestamp filters that would overflow int64") {
+ // Values outside the int64 epoch-nanoseconds range (~1677-09-21 .. 2262-04-11) must NOT be
+ // pushed down: encoding them would throw, and a truncated encoding could silently mis-skip a
+ // row group. Falling back to a full scan is always correct (SPARK-46092 rationale).
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val ntzSchema = new SparkToParquetSchemaConverter(conf)
+ .convert(new StructType().add("c", TimestampNTZNanosType(9)))
+ val ntzFilters = createParquetFilters(ntzSchema)
+ val ltzSchema = new SparkToParquetSchemaConverter(conf)
+ .convert(new StructType().add("c", TimestampLTZNanosType(9)))
+ val ltzFilters = createParquetFilters(ltzSchema)
+
+ // Out of range: must not push down.
+ val ntzOverflow = LocalDateTime.parse("2300-01-01T00:00:00")
+ val ltzOverflow = Instant.parse("2300-01-01T00:00:00Z")
+ Seq(
+ sources.LessThan("c", ntzOverflow),
+ sources.LessThanOrEqual("c", ntzOverflow),
+ sources.GreaterThan("c", ntzOverflow),
+ sources.GreaterThanOrEqual("c", ntzOverflow),
+ sources.EqualTo("c", ntzOverflow),
+ sources.EqualNullSafe("c", ntzOverflow),
+ sources.Not(sources.EqualTo("c", ntzOverflow)),
+ sources.In("c", Array[Any](ntzOverflow))
+ ).foreach { filter =>
+ assert(ntzFilters.createFilter(filter).isEmpty,
+ s"Row group filter $filter shouldn't be pushed down.")
+ }
+ Seq(
+ sources.LessThan("c", ltzOverflow),
+ sources.LessThanOrEqual("c", ltzOverflow),
+ sources.GreaterThan("c", ltzOverflow),
+ sources.GreaterThanOrEqual("c", ltzOverflow),
+ sources.EqualTo("c", ltzOverflow),
+ sources.EqualNullSafe("c", ltzOverflow),
+ sources.Not(sources.EqualTo("c", ltzOverflow)),
+ sources.In("c", Array[Any](ltzOverflow))
+ ).foreach { filter =>
+ assert(ltzFilters.createFilter(filter).isEmpty,
+ s"Row group filter $filter shouldn't be pushed down.")
+ }
+
+ // In range: must push down.
+ assert(ntzFilters.createFilter(
+ sources.LessThan("c", LocalDateTime.parse("2020-01-01T00:00:00"))).isDefined)
+ assert(ltzFilters.createFilter(
+ sources.LessThan("c", Instant.parse("2020-01-01T00:00:00Z"))).isDefined)
+ }
+ }
+
+ test("SPARK-57822: an In list mixing in-range and out-of-range nanos values isn't pushed down") {
+ // A mixed In list (in-range head, out-of-range tail) must fall back to a full scan, not crash
+ // filter creation. The In arm converts *every* element - per-element `makeEq` under the
+ // pushdown threshold, `makeInPredicate` above it - so a head-only guard would let the
+ // out-of-range tail reach the encoder, which throws (SPARK-46092: never throw, full scan
+ // instead). Both branches are exercised by setting the threshold below and above the list
+ // size. The list must lead with an in-range value so the fix is what rejects it, not the old
+ // head check.
+ Seq("1", "10").foreach { threshold =>
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> threshold) {
+ val ntzFilters = createParquetFilters(new SparkToParquetSchemaConverter(conf)
+ .convert(new StructType().add("c", TimestampNTZNanosType(9))))
+ val ltzFilters = createParquetFilters(new SparkToParquetSchemaConverter(conf)
+ .convert(new StructType().add("c", TimestampLTZNanosType(9))))
+ val ntzIn = sources.In("c", Array[Any](
+ LocalDateTime.parse("2020-01-01T00:00:00"), LocalDateTime.parse("2300-01-01T00:00:00")))
+ val ltzIn = sources.In("c", Array[Any](
+ Instant.parse("2020-01-01T00:00:00Z"), Instant.parse("2300-01-01T00:00:00Z")))
+ // Must not throw, and must not push down (a single out-of-range element disqualifies the
+ // whole In).
+ assert(ntzFilters.createFilter(ntzIn).isEmpty,
+ s"Mixed-range NTZ In (threshold=$threshold) shouldn't be pushed down.")
+ assert(ltzFilters.createFilter(ltzIn).isEmpty,
+ s"Mixed-range LTZ In (threshold=$threshold) shouldn't be pushed down.")
+ // An all-in-range In of the same size still pushes down (the fix rejects only on a
+ // non-pushable element, it doesn't disable In pushdown wholesale).
+ assert(ntzFilters.createFilter(sources.In("c", Array[Any](
+ LocalDateTime.parse("2020-01-01T00:00:00"),
+ LocalDateTime.parse("2021-01-01T00:00:00")))).isDefined)
+ assert(ltzFilters.createFilter(sources.In("c", Array[Any](
+ Instant.parse("2020-01-01T00:00:00Z"),
+ Instant.parse("2021-01-01T00:00:00Z")))).isDefined)
+ }
+ }
+ }
}
@ExtendedSQLTest
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
index e62d952a9eb9..2ce156602725 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
@@ -309,6 +309,63 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
}
}
+ test("SPARK-57822: nanos timestamp filter pushdown prunes row groups with identical results") {
+ // End-to-end: write many rows across multiple row groups, filter on the nanos column, and
+ // assert that (a) the filter is pushed to Parquet and skips row groups (a stripped-Spark-filter
+ // scan reads more than the matching rows but fewer than all rows) and (b) the full query result
+ // equals reading the same predicate with pushdown disabled. Record-level filtering is disabled
+ // so ONLY row-group-level skipping is exercised; the non-vectorized reader is used so
+ // `stripSparkFilter` reflects exactly what Parquet returned.
+ withNanosEnabled {
+ Seq(7, 8, 9).foreach { p =>
+ val frac = "000000123".take(p)
+ Seq("ntz", "ltz").foreach { kind =>
+ val typ = if (kind == "ntz") "TIMESTAMP_NTZ" else "TIMESTAMP_LTZ"
+ val expectedType =
+ if (kind == "ntz") TimestampNTZNanosType(p) else TimestampLTZNanosType(p)
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ // 1024 rows one second apart, monotonically increasing so consecutive row groups hold
+ // disjoint value ranges (statistics-based skipping is effective). A small block size
+ // forces multiple row groups.
+ spark.sql(
+ s"""SELECT
+ | $typ '2020-01-01 00:00:00.$frac' + make_dt_interval(0,0,0,id) AS c
+ |FROM range(0, 1024)""".stripMargin)
+ .coalesce(1)
+ .write.option("parquet.block.size", 512).parquet(path)
+ assert(spark.read.parquet(path).schema("c").dataType === expectedType)
+
+ // Matches the single row at id = 1000, which lives only in a late row group.
+ val predicate = s"c = $typ '2020-01-01 00:16:40.$frac'"
+
+ // Row-group-level skipping: with record filtering off, a stripped scan returns whole
+ // surviving row groups. Fewer than all rows (skipping happened) but more than the one
+ // matching row (record filtering is off).
+ withSQLConf(
+ SQLConf.PARQUET_RECORD_FILTER_ENABLED.key -> "false",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val df = spark.read.parquet(path).filter(predicate)
+ val actual = stripSparkFilter(df).collect().length
+ assert(actual > 1 && actual < 1024,
+ s"Expected row-group skipping (p=$p, $kind) but scanned $actual of 1024 rows.")
+ }
+
+ // Results must be identical whether or not the filter is pushed to Parquet.
+ val expected = withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false") {
+ spark.read.parquet(path).filter(predicate).collect().toSeq
+ }
+ assert(expected.length === 1)
+ withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ checkAnswer(spark.read.parquet(path).filter(predicate), expected)
+ }
+ }
+ }
+ }
+ }
+ }
+
test("SPARK-57102: nanos timestamps round-trip via the V2 file source") {
withNanosEnabled {
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
index febeb7c5fb1e..f0063f86d111 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
@@ -17,11 +17,15 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import java.time.{Instant, LocalDateTime, ZoneOffset}
+
import org.apache.parquet.column.ColumnDescriptor
+import org.apache.parquet.filter2.predicate.FilterApi
+import org.apache.parquet.filter2.predicate.SparkFilterApi.longColumn
import org.apache.parquet.io.api.PrimitiveConverter
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.LogicalTypeAnnotation.{TimestampLogicalTypeAnnotation, TimeUnit}
-import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.{INT32, INT64}
import org.apache.parquet.schema.Type.Repetition.REQUIRED
import org.apache.spark.{SparkArithmeticException, SparkFunSuite, SparkRuntimeException}
@@ -136,6 +140,129 @@ class TimestampNanosParquetOpsSuite extends SparkFunSuite {
}
}
+ // ---------- filter-pushdown ops ----------
+
+ test("filterOps accepts the matching Java time value and rejects others") {
+ // LTZ pushes down java.time.Instant; NTZ pushes down java.time.LocalDateTime. Each rejects
+ // the other type (and non-temporal values) so a mismatched literal falls through to no
+ // pushdown rather than a ClassCastException in the converter.
+ val ltzOps = TimestampNanosParquetOps.ltzFilterOps
+ val ntzOps = TimestampNanosParquetOps.ntzFilterOps
+ assert(ltzOps.acceptsValue(Instant.parse("2020-01-01T00:00:00Z")))
+ assert(!ltzOps.acceptsValue(LocalDateTime.parse("2020-01-01T00:00:00")))
+ assert(!ltzOps.acceptsValue(java.lang.Long.valueOf(1L)))
+ assert(ntzOps.acceptsValue(LocalDateTime.parse("2020-01-01T00:00:00")))
+ assert(!ntzOps.acceptsValue(Instant.parse("2020-01-01T00:00:00Z")))
+ assert(!ntzOps.acceptsValue("2020-01-01"))
+ }
+
+ test("filterOps rejects values outside the INT64 epoch-nanos range (falls back to full scan)") {
+ // Year 2300 is past the ~2262 int64 epoch-nanos cutoff: encoding would overflow, so
+ // acceptsValue must reject it (SPARK-46092-style guard) instead of throwing during filter
+ // creation. An in-range value is accepted.
+ val ltzOps = TimestampNanosParquetOps.ltzFilterOps
+ val ntzOps = TimestampNanosParquetOps.ntzFilterOps
+ assert(!ltzOps.acceptsValue(Instant.parse("2300-01-01T00:00:00Z")))
+ assert(!ntzOps.acceptsValue(LocalDateTime.parse("2300-01-01T00:00:00")))
+ assert(ltzOps.acceptsValue(Instant.parse("2020-01-01T00:00:00Z")))
+ assert(ntzOps.acceptsValue(LocalDateTime.parse("2020-01-01T00:00:00")))
+ }
+
+ test("filterOps make* throws on an out-of-range value - why acceptsValue must gate every In " +
+ "element, not just the head") {
+ // The encoder is exact (Math.addExact/multiplyExact), so building a predicate directly from an
+ // out-of-range value throws rather than clamping. This is the contract behind the
+ // ParquetFilters In-arm gating every element through acceptsValue: a makeIn set (or per-element
+ // makeEq) that includes an out-of-range tail would crash filter creation. `acceptsValue`
+ // (tested above) is the guard that keeps make* off such values on the pushdown path.
+ val path = Array("c")
+ val ltzOps = TimestampNanosParquetOps.ltzFilterOps
+ val ntzOps = TimestampNanosParquetOps.ntzFilterOps
+ val ltzOverflow = Instant.parse("2300-01-01T00:00:00Z")
+ val ntzOverflow = LocalDateTime.parse("2300-01-01T00:00:00")
+ intercept[ArithmeticException](ltzOps.makeEq(path, ltzOverflow))
+ intercept[ArithmeticException](ntzOps.makeEq(path, ntzOverflow))
+ // A mixed set (in-range head + out-of-range tail) is the crash the head-only guard let through.
+ val ltzInRange = Instant.parse("2020-01-01T00:00:00Z")
+ val ntzInRange = LocalDateTime.parse("2020-01-01T00:00:00")
+ intercept[ArithmeticException](ltzOps.makeIn(path, Array[Any](ltzInRange, ltzOverflow)))
+ intercept[ArithmeticException](ntzOps.makeIn(path, Array[Any](ntzInRange, ntzOverflow)))
+ }
+
+ test("filterOps declares the canonical nanos-timestamp Parquet encoding") {
+ // LTZ is isAdjustedToUTC=true, NTZ is false; both INT64 TIMESTAMP(NANOS). These are the keys
+ // the ParquetFilters reverse lookup matches against the file schema.
+ val ltzOps = TimestampNanosParquetOps.ltzFilterOps
+ val ntzOps = TimestampNanosParquetOps.ntzFilterOps
+ assert(ltzOps.primitiveTypeName === INT64)
+ assert(ltzOps.logicalTypeAnnotation ===
+ LogicalTypeAnnotation.timestampType(true, TimeUnit.NANOS))
+ assert(ntzOps.primitiveTypeName === INT64)
+ assert(ntzOps.logicalTypeAnnotation ===
+ LogicalTypeAnnotation.timestampType(false, TimeUnit.NANOS))
+ }
+
+ test("filterOps builds predicates converting to signed INT64 epoch-nanoseconds") {
+ val path = Array("c")
+ val col = longColumn(path)
+
+ // LTZ: Instant -> epoch-nanos. Sub-microsecond digits are preserved (not truncated to micros).
+ val ltzOps = TimestampNanosParquetOps.ltzFilterOps
+ val instant = Instant.parse("2020-01-01T12:34:56.000000789Z")
+ val ltzNanos = java.lang.Long.valueOf(
+ instant.getEpochSecond * DateTimeConstants.NANOS_PER_SECOND + instant.getNano)
+ assert(ltzOps.makeEq(path, instant) === FilterApi.eq(col, ltzNanos))
+ assert(ltzOps.makeNotEq(path, instant) === FilterApi.notEq(col, ltzNanos))
+ assert(ltzOps.makeLt(path, instant) === FilterApi.lt(col, ltzNanos))
+ assert(ltzOps.makeLtEq(path, instant) === FilterApi.ltEq(col, ltzNanos))
+ assert(ltzOps.makeGt(path, instant) === FilterApi.gt(col, ltzNanos))
+ assert(ltzOps.makeGtEq(path, instant) === FilterApi.gtEq(col, ltzNanos))
+
+ // NTZ: LocalDateTime (interpreted at UTC) -> epoch-nanos.
+ val ntzOps = TimestampNanosParquetOps.ntzFilterOps
+ val ldt = LocalDateTime.parse("2020-01-01T12:34:56.000000789")
+ val ntzInstant = ldt.toInstant(ZoneOffset.UTC)
+ val ntzNanos = java.lang.Long.valueOf(
+ ntzInstant.getEpochSecond * DateTimeConstants.NANOS_PER_SECOND + ntzInstant.getNano)
+ assert(ntzOps.makeEq(path, ldt) === FilterApi.eq(col, ntzNanos))
+ assert(ntzOps.makeIn(path, Array[Any](ldt)) === {
+ val set = new java.util.HashSet[java.lang.Long]()
+ set.add(ntzNanos)
+ FilterApi.in(col, set)
+ })
+ }
+
+ test("filterOps eq/notEq/in tolerate a null value (IsNull / IsNotNull)") {
+ val path = Array("c")
+ val col = longColumn(path)
+ val nullLong = null.asInstanceOf[java.lang.Long]
+ // null value -> null Long comparand; used by ParquetFilters for IsNull / IsNotNull.
+ Seq(TimestampNanosParquetOps.ltzFilterOps, TimestampNanosParquetOps.ntzFilterOps).foreach {
+ ops =>
+ assert(ops.makeEq(path, null) === FilterApi.eq(col, nullLong))
+ assert(ops.makeNotEq(path, null) === FilterApi.notEq(col, nullLong))
+ val set = new java.util.HashSet[java.lang.Long]()
+ set.add(null)
+ assert(ops.makeIn(path, Array[Any](null)) === FilterApi.in(col, set))
+ }
+ }
+
+ test("ParquetTypeOps.filterOpsFor resolves each nanos encoding and nothing else") {
+ // The LTZ and NTZ encodings differ only in isAdjustedToUTC; each resolves to its own ops.
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timestampType(true, TimeUnit.NANOS), INT64)
+ .contains(TimestampNanosParquetOps.ltzFilterOps))
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timestampType(false, TimeUnit.NANOS), INT64)
+ .contains(TimestampNanosParquetOps.ntzFilterOps))
+ // MICROS unit, or an INT32 primitive, is not a nanos-timestamp encoding (pushdown falls
+ // through to no framework ops).
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timestampType(true, TimeUnit.MICROS), INT64).isEmpty)
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timestampType(false, TimeUnit.NANOS), INT32).isEmpty)
+ }
+
// ---------- helpers ----------
private def nanosField(isAdjustedToUTC: Boolean): Type =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org