This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 6db4ab94cf05 [SPARK-55444][SQL] Route TimeType Parquet filter pushdown
through the Types Framework
6db4ab94cf05 is described below
commit 6db4ab94cf05de2c9b03f07a65a86e77d19a1c76
Author: Stevo Mitric <[email protected]>
AuthorDate: Wed Jul 8 01:07:03 2026 +0800
[SPARK-55444][SQL] Route TimeType Parquet filter pushdown through the Types
Framework
### What changes were proposed in this pull request?
Routes TimeType's Parquet predicate pushdown through the Types Framework
instead of the inline `ParquetTimeMicrosType` handling — the last Parquet
integration point still hardcoded in `ParquetFilters`
(schema/write/row-read/vectorized-read already moved to `ParquetTypeOps`).
- New `ParquetFilterOps` trait: the Parquet encoding a framework type owns
(primitive + logical annotation), value acceptance, and the 7 predicate
builders (eq/notEq/lt/ltEq/gt/gtEq/in).
- `TimeTypeParquetOps.filterOps` (LocalTime → micros-of-day Long),
registered in `ParquetTypeOps.filterOpsList` and resolved via the reverse
`filterOpsFor` lookup.
- Replace the scattered `ParquetTimeMicrosType` arms in `ParquetFilters` (7
`make*` + `valueCanMakeFilterOn`) with a `FrameworkFilterOps` extractor.
Dispatch is keyed on the Parquet file's on-disk encoding (reverse lookup),
not the Spark type, because filter pushdown binds predicates to physical
columns and the value converter depends on the physical unit — matching the
existing physical-schema dispatch.
### Why are the changes needed?
Framework types get filter pushdown with no per-type changes to
`ParquetFilters`, keeping the per-type filter knowledge with the type. No
`ParquetFilters` constructor change.
### Does this PR introduce _any_ user-facing change?
No. The extractor matches only the canonical INT64 TIME(MICROS,
isAdjustedToUTC=false) encoding that `ParquetTimeMicrosType` matched; behavior
is identical. NANOS pushdown remains unsupported.
### How was this patch tested?
`TimeTypeParquetOpsSuite` (+5 unit tests: 4 covering `filterOps`, 1
covering the `filterOpsFor` reverse lookup);
`ParquetV1FilterSuite`/`ParquetV2FilterSuite` "SPARK-51687: filter pushdown -
time" pass unchanged.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56965 from stevomitric/stevomitric/parquet-tf-filter-pushdown-fw.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../datasources/parquet/ParquetFilters.scala | 57 +++++----
.../parquet/types/ops/ParquetFilterOps.scala | 134 +++++++++++++++++++++
.../parquet/types/ops/ParquetTypeOps.scala | 33 ++++-
.../parquet/types/ops/TimeTypeParquetOps.scala | 24 ++++
.../types/ops/TimeTypeParquetOpsSuite.scala | 70 +++++++++++
5 files changed, 290 insertions(+), 28 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 4a9b17bf98e5..f60ced3eb597 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
@@ -39,6 +39,7 @@ import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils,
IntervalUtils}
import
org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays,
rebaseGregorianToJulianMicros, RebaseSpec}
+import
org.apache.spark.sql.execution.datasources.parquet.types.ops.{ParquetFilterOps,
ParquetTypeOps}
import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.sources
import org.apache.spark.unsafe.types.UTF8String
@@ -150,8 +151,21 @@ class ParquetFilters(
ParquetSchemaType(LogicalTypeAnnotation.timestampType(true,
TimeUnit.MICROS), INT64, 0)
private val ParquetTimestampMillisType =
ParquetSchemaType(LogicalTypeAnnotation.timestampType(true,
TimeUnit.MILLIS), INT64, 0)
- private val ParquetTimeMicrosType =
- ParquetSchemaType(LogicalTypeAnnotation.timeType(false, TimeUnit.MICROS),
INT64, 0)
+
+ /**
+ * Extractor that maps a Parquet field's schema to its Types Framework
filter ops, if the
+ * field's on-disk encoding belongs to a framework-managed type. Defined
here, not in the
+ * ops package, because it pattern-matches on the private
[[ParquetSchemaType]]. A `Some`
+ * routes the field's predicates through the framework ops; `None` falls
through to the
+ * built-in cases below. Framework types use Parquet encodings distinct from
the built-in
+ * cases, so the extractor never shadows them. This replaces the inline
TimeType handling
+ * (TIME(MICROS) -> micros Long), which now lives in
TimeTypeParquetOps.filterOps.
+ */
+ private object FrameworkFilterOps {
+ def unapply(parquetSchemaType: ParquetSchemaType):
Option[ParquetFilterOps] =
+ ParquetTypeOps.filterOpsFor(
+ parquetSchemaType.logicalTypeAnnotation,
parquetSchemaType.primitiveTypeName)
+ }
private def dateToDays(date: Any): Int = {
val gregorianDays = date match {
@@ -252,10 +266,8 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.eq(
longColumn(n),
Option(v).map(timestampToMillis).orNull)
- case ParquetTimeMicrosType =>
- (n: Array[String], v: Any) => FilterApi.eq(
- longColumn(n),
- Option(v).map(localTimeToMicros).orNull)
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], v: Any) => ops.makeEq(n, v)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], v: Any) => FilterApi.eq(
@@ -305,10 +317,8 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.notEq(
longColumn(n),
Option(v).map(timestampToMillis).orNull)
- case ParquetTimeMicrosType =>
- (n: Array[String], v: Any) => FilterApi.notEq(
- longColumn(n),
- Option(v).map(localTimeToMicros).orNull)
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], v: Any) => ops.makeNotEq(n, v)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], v: Any) => FilterApi.notEq(
@@ -349,8 +359,8 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.lt(longColumn(n),
timestampToMicros(v))
case ParquetTimestampMillisType if pushDownTimestamp =>
(n: Array[String], v: Any) => FilterApi.lt(longColumn(n),
timestampToMillis(v))
- case ParquetTimeMicrosType =>
- (n: Array[String], v: Any) => FilterApi.lt(longColumn(n),
localTimeToMicros(v))
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], v: Any) => ops.makeLt(n, v)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -388,8 +398,8 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.ltEq(longColumn(n),
timestampToMicros(v))
case ParquetTimestampMillisType if pushDownTimestamp =>
(n: Array[String], v: Any) => FilterApi.ltEq(longColumn(n),
timestampToMillis(v))
- case ParquetTimeMicrosType =>
- (n: Array[String], v: Any) => FilterApi.ltEq(longColumn(n),
localTimeToMicros(v))
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], v: Any) => ops.makeLtEq(n, v)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -427,8 +437,8 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.gt(longColumn(n),
timestampToMicros(v))
case ParquetTimestampMillisType if pushDownTimestamp =>
(n: Array[String], v: Any) => FilterApi.gt(longColumn(n),
timestampToMillis(v))
- case ParquetTimeMicrosType =>
- (n: Array[String], v: Any) => FilterApi.gt(longColumn(n),
localTimeToMicros(v))
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], v: Any) => ops.makeGt(n, v)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -466,8 +476,8 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.gtEq(longColumn(n),
timestampToMicros(v))
case ParquetTimestampMillisType if pushDownTimestamp =>
(n: Array[String], v: Any) => FilterApi.gtEq(longColumn(n),
timestampToMillis(v))
- case ParquetTimeMicrosType =>
- (n: Array[String], v: Any) => FilterApi.gtEq(longColumn(n),
localTimeToMicros(v))
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], v: Any) => ops.makeGtEq(n, v)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -557,13 +567,8 @@ class ParquetFilters(
}
FilterApi.in(longColumn(n), set)
- case ParquetTimeMicrosType =>
- (n: Array[String], values: Array[Any]) =>
- val set = new HashSet[JLong]()
- for (value <- values) {
- set.add(Option(value).map(localTimeToMicros).orNull)
- }
- FilterApi.in(longColumn(n), set)
+ case FrameworkFilterOps(ops) =>
+ (n: Array[String], values: Array[Any]) => ops.makeIn(n, values)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if
pushDownDecimal =>
(n: Array[String], values: Array[Any]) =>
@@ -662,7 +667,7 @@ class ParquetFilters(
value.isInstanceOf[Date] || value.isInstanceOf[LocalDate]
case ParquetTimestampMicrosType | ParquetTimestampMillisType =>
value.isInstanceOf[Timestamp] || value.isInstanceOf[Instant]
- case ParquetTimeMicrosType => value.isInstanceOf[LocalTime]
+ case FrameworkFilterOps(ops) => ops.acceptsValue(value)
case ParquetSchemaType(decimalType: DecimalLogicalTypeAnnotation, INT32,
_) =>
isDecimalMatched(value, decimalType)
case ParquetSchemaType(decimalType: DecimalLogicalTypeAnnotation, INT64,
_) =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala
new file mode 100644
index 000000000000..8894dbfe6671
--- /dev/null
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet.types.ops
+
+import java.lang.{Long => JLong}
+import java.util.HashSet
+
+import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate}
+import org.apache.parquet.filter2.predicate.Operators.{Column, SupportsLtGt}
+import org.apache.parquet.filter2.predicate.SparkFilterApi.longColumn
+import org.apache.parquet.schema.LogicalTypeAnnotation
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+
+/**
+ * Optional Parquet filter-pushdown support for a Types Framework type.
+ *
+ * A framework type that wants its predicates pushed down to Parquet provides a
+ * [[ParquetFilterOps]] and registers it in [[ParquetTypeOps.filterOpsList]];
types that don't
+ * support pushdown add nothing and are read without filtering.
+ *
+ * Dispatch is keyed off the Parquet file's column encoding, not the requested
Spark type,
+ * because filter pushdown matches the on-disk schema. The ops therefore
declares the
+ * Parquet primitive + logical annotation it owns ([[primitiveTypeName]] /
+ * [[logicalTypeAnnotation]]); `ParquetFilters` reverse-looks-up the ops for a
field via
+ * [[ParquetTypeOps.filterOpsFor]] and routes that field's predicates here.
This keeps the
+ * per-type filter knowledge (value conversion, predicate construction) with
the type
+ * instead of scattered across `ParquetFilters`.
+ *
+ * Implementations build the parquet-mr [[FilterPredicate]] for each
comparison directly, so they
+ * own the choice of physical column and the external-value -> physical-value
conversion. The
+ * eq/notEq/in builders must tolerate a null `value` (used for IsNull /
IsNotNull); the ordered
+ * builders (lt/ltEq/gt/gtEq) are only invoked with non-null values. Rather
than implement the
+ * builders directly, a type should extend [[TypedParquetFilterOps]] (or its
INT64 alias
+ * [[LongParquetFilterOps]]), which writes the boilerplate and the
null-handling split once.
+ *
+ * @see TimeTypeParquetOps.filterOps for a reference implementation
(INT64-backed TimeType)
+ * @since 4.3.0
+ */
+private[parquet] trait ParquetFilterOps {
+
+ /** The Parquet logical type annotation of the column this ops handles (may
be null). */
+ def logicalTypeAnnotation: LogicalTypeAnnotation
+
+ /** The Parquet primitive type of the column this ops handles. */
+ def primitiveTypeName: PrimitiveTypeName
+
+ /** Whether `value` (a non-null external filter value) is pushable for this
type. */
+ def acceptsValue(value: Any): Boolean
+
+ def makeEq(columnPath: Array[String], value: Any): FilterPredicate
+ def makeNotEq(columnPath: Array[String], value: Any): FilterPredicate
+ def makeLt(columnPath: Array[String], value: Any): FilterPredicate
+ def makeLtEq(columnPath: Array[String], value: Any): FilterPredicate
+ def makeGt(columnPath: Array[String], value: Any): FilterPredicate
+ def makeGtEq(columnPath: Array[String], value: Any): FilterPredicate
+ def makeIn(columnPath: Array[String], values: Array[Any]): FilterPredicate
+}
+
+/**
+ * Base [[ParquetFilterOps]] for a type stored in an ordered Parquet primitive
column of physical
+ * type `T` (e.g. `java.lang.Long` for INT64, `java.lang.Integer` for INT32,
`Binary` for BINARY).
+ * Implements all seven predicate builders once against the parquet-mr
`FilterApi`, so a concrete
+ * type supplies only the encoding it owns ([[logicalTypeAnnotation]] /
[[primitiveTypeName]]), the
+ * [[column]] accessor for its physical type, [[acceptsValue]], and the
[[toPhysical]] conversion.
+ * The null-handling split (eq/notEq/in tolerate a null value for IsNull /
IsNotNull; the ordered
+ * builders never receive null) and the `makeIn` set construction live here
once, so an
+ * implementer can't get them wrong. The column must support ordering
(`SupportsLtGt`, which
+ * extends `SupportsEqNotEq`); every framework-relevant physical type does
(only `BooleanColumn`
+ * is eq-only, and no framework type is boolean-backed).
+ *
+ * `T` is confined to this base: the public [[ParquetFilterOps]] the registry,
`filterOpsFor`,
+ * and the `ParquetFilters` extractor consume stays non-generic.
+ */
+private[parquet] abstract class TypedParquetFilterOps[T <: Comparable[T]]
extends ParquetFilterOps {
+
+ /** The physical column accessor for `T` (e.g. `SparkFilterApi.longColumn`).
*/
+ protected def column(columnPath: Array[String]): Column[T] with SupportsLtGt
+
+ /** Converts a non-null pushable `value` to the physical value `T` stored in
the file. */
+ protected def toPhysical(value: Any): T
+
+ private def toPhysicalOrNull(value: Any): T =
+ if (value == null) null.asInstanceOf[T] else toPhysical(value)
+
+ override def makeEq(columnPath: Array[String], value: Any): FilterPredicate =
+ FilterApi.eq(column(columnPath), toPhysicalOrNull(value))
+ override def makeNotEq(columnPath: Array[String], value: Any):
FilterPredicate =
+ FilterApi.notEq(column(columnPath), toPhysicalOrNull(value))
+ override def makeLt(columnPath: Array[String], value: Any): FilterPredicate =
+ FilterApi.lt(column(columnPath), toPhysical(value))
+ override def makeLtEq(columnPath: Array[String], value: Any):
FilterPredicate =
+ FilterApi.ltEq(column(columnPath), toPhysical(value))
+ override def makeGt(columnPath: Array[String], value: Any): FilterPredicate =
+ FilterApi.gt(column(columnPath), toPhysical(value))
+ override def makeGtEq(columnPath: Array[String], value: Any):
FilterPredicate =
+ FilterApi.gtEq(column(columnPath), toPhysical(value))
+ override def makeIn(columnPath: Array[String], values: Array[Any]):
FilterPredicate = {
+ val set = new HashSet[T]()
+ values.foreach(v => set.add(toPhysicalOrNull(v)))
+ FilterApi.in(column(columnPath), set)
+ }
+}
+
+/**
+ * [[TypedParquetFilterOps]] specialized to an INT64 (`longColumn`) physical
column. A concrete
+ * type supplies only [[logicalTypeAnnotation]], [[acceptsValue]], and the
[[toLong]] conversion.
+ */
+private[parquet] abstract class LongParquetFilterOps extends
TypedParquetFilterOps[JLong] {
+
+ override val primitiveTypeName: PrimitiveTypeName = PrimitiveTypeName.INT64
+
+ override protected def column(columnPath: Array[String]): Column[JLong] with
SupportsLtGt =
+ longColumn(columnPath)
+
+ /** Converts a non-null pushable `value` to the INT64 physical value stored
in the file. */
+ protected def toLong(value: Any): JLong
+
+ override protected def toPhysical(value: Any): JLong = toLong(value)
+}
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 63209572c94d..7c2091c8c081 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
@@ -21,7 +21,8 @@ import java.time.ZoneId
import org.apache.parquet.column.ColumnDescriptor
import org.apache.parquet.io.api.{Converter, RecordConsumer}
-import org.apache.parquet.schema.Type
+import org.apache.parquet.schema.{LogicalTypeAnnotation, Type}
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
@@ -46,7 +47,9 @@ import org.apache.spark.sql.types.{DataType, StructType,
TimestampLTZNanosType,
* - Type gates: declaring Parquet support (supportDataType)
* - Schema clipping: declaring internal struct schema for column pruning
*
- * NOT yet on the trait (deferred to follow-ups): filter-pushdown predicates.
+ * Filter pushdown is handled separately, in the companion object rather than
on this trait,
+ * because it is keyed on the Parquet file's on-disk encoding (a reverse
lookup), not on the
+ * Spark DataType: see [[ParquetTypeOps.filterOpsFor]] and
[[ParquetFilterOps]].
*
* DISPATCH PATTERN: Framework FIRST at all integration sites. Each Parquet
infrastructure
* method wraps itself with:
@@ -278,4 +281,30 @@ private[parquet] object ParquetTypeOps {
dt: DataType, descriptor: ColumnDescriptor): java.lang.Boolean =
apply(dt).map(o =>
java.lang.Boolean.valueOf(o.supportsLazyDictionaryDecoding(descriptor)))
.orNull
+
+ /**
+ * Reverse lookup for filter pushdown: given a Parquet field's logical
annotation and
+ * primitive type (from the file schema), returns the framework filter ops
that owns that
+ * encoding, if any. Used by `ParquetFilters` (via its FrameworkFilterOps
extractor) so
+ * framework types participate in predicate pushdown with no per-type
changes there.
+ *
+ * Only the primitive name + logical annotation are matched, not the type
length. That is
+ * sufficient today because every registered ops is a fixed-width primitive
(getTypeLength == 0),
+ * so length carries no information. A future FIXED_LEN_BYTE_ARRAY-backed
ops would need length
+ * added to the key to disambiguate widths.
+ */
+ private[parquet] def filterOpsFor(
+ logicalTypeAnnotation: LogicalTypeAnnotation,
+ primitiveTypeName: PrimitiveTypeName): Option[ParquetFilterOps] =
+ filterOpsList.find { ops =>
+ ops.primitiveTypeName == primitiveTypeName &&
+ ops.logicalTypeAnnotation == logicalTypeAnnotation
+ }
+
+ /**
+ * Registration point for filter pushdown: every framework type that
supports Parquet
+ * 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)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
index 8e6fcee925d7..3639320ab7e7 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
@@ -17,6 +17,10 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import java.lang.{Long => JLong}
+import java.time.LocalTime
+import java.time.temporal.ChronoField.MICRO_OF_DAY
+
import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
import org.apache.parquet.io.api.{Converter, RecordConsumer}
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
@@ -132,6 +136,26 @@ case class TimeTypeParquetOps(t: TimeType) extends
ParquetTypeOps {
private[ops] object TimeTypeParquetOps {
+ /**
+ * Parquet filter-pushdown ops for TimeType, registered in
[[ParquetTypeOps.filterOpsList]].
+ * Filter dispatch is keyed on the file's on-disk encoding (not the Spark
precision), so this
+ * single instance targets only the MICROS encoding: TimeType is stored as
INT64
+ * TIME(MICROS, isAdjustedToUTC=false) for precision 0..6 and TIME(NANOS)
for precision 7..9,
+ * and only MICROS is pushed down here (filter values are
java.time.LocalTime converted to
+ * micros-of-day Longs). A TIME(NANOS) column resolves to no framework ops
and falls through
+ * to no pushdown. This matches the inline TimeType handling in
ParquetFilters before filter
+ * pushdown was routed through the framework, so pushdown behavior is
unchanged.
+ */
+ private[ops] val filterOps: ParquetFilterOps = new LongParquetFilterOps {
+ override val logicalTypeAnnotation: LogicalTypeAnnotation =
+ LogicalTypeAnnotation.timeType(false, TimeUnit.MICROS)
+
+ override def acceptsValue(value: Any): Boolean =
value.isInstanceOf[LocalTime]
+
+ override protected def toLong(value: Any): JLong =
+ value.asInstanceOf[LocalTime].getLong(MICRO_OF_DAY)
+ }
+
/**
* Whether the Parquet field is an INT64 TIME(NANOS) column. The
isAdjustedToUTC flag is
* intentionally ignored: Spark's TimeType is zone-less, so a TIME(NANOS)
value decodes to the
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
index 70091c6379bc..7e0b39c07da2 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
@@ -17,7 +17,12 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import java.time.LocalTime
+import java.time.temporal.ChronoField.MICRO_OF_DAY
+
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.schema.{LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.{INT32, INT64}
@@ -42,6 +47,9 @@ import org.apache.spark.sql.types.{IntegerType, TimeType}
* TimeType is zone-less local time, so the flag carries no extra information
on read and the
* raw time-of-day value decodes identically either way. This keeps the
framework read path
* consistent with both the legacy row-based reader and the vectorized reader.
+ *
+ * Also covers the filter-pushdown ops ([[TimeTypeParquetOps.filterOps]]) and
the
+ * [[ParquetTypeOps.filterOpsFor]] reverse lookup that resolves them.
*/
class TimeTypeParquetOpsSuite extends SparkFunSuite {
@@ -181,6 +189,68 @@ class TimeTypeParquetOpsSuite extends SparkFunSuite {
assert(ParquetTypeOps.supportsLazyDictionaryDecodingOrNull(IntegerType,
null) == null)
}
+ // ---------- filter pushdown ops ----------
+
+ test("filterOps accepts LocalTime values and rejects others") {
+ val ops = TimeTypeParquetOps.filterOps
+ assert(ops.acceptsValue(LocalTime.of(1, 2, 3)))
+ assert(!ops.acceptsValue(java.lang.Long.valueOf(1L)))
+ assert(!ops.acceptsValue("12:00:00"))
+ }
+
+ test("filterOps declares the canonical TimeType Parquet encoding") {
+ val ops = TimeTypeParquetOps.filterOps
+ assert(ops.primitiveTypeName === INT64)
+ assert(ops.logicalTypeAnnotation ===
+ LogicalTypeAnnotation.timeType(false, TimeUnit.MICROS))
+ }
+
+ test("filterOps builds predicates for LocalTime, converting to
micros-of-day") {
+ val ops = TimeTypeParquetOps.filterOps
+ val path = Array("c")
+ val col = longColumn(path)
+ val t = LocalTime.of(23, 59, 59, 123456000)
+ // parquet-mr operators implement value equality, so we can pin the exact
pushed-down value:
+ // LocalTime -> micros-of-day Long, the same conversion the removed inline
TimeType arms used.
+ val micros = java.lang.Long.valueOf(t.getLong(MICRO_OF_DAY))
+ assert(ops.makeEq(path, t) === FilterApi.eq(col, micros))
+ assert(ops.makeNotEq(path, t) === FilterApi.notEq(col, micros))
+ assert(ops.makeLt(path, t) === FilterApi.lt(col, micros))
+ assert(ops.makeLtEq(path, t) === FilterApi.ltEq(col, micros))
+ assert(ops.makeGt(path, t) === FilterApi.gt(col, micros))
+ assert(ops.makeGtEq(path, t) === FilterApi.gtEq(col, micros))
+ val set = new java.util.HashSet[java.lang.Long]()
+ set.add(micros)
+ set.add(null)
+ assert(ops.makeIn(path, Array[Any](t, null)) === FilterApi.in(col, set))
+ }
+
+ test("filterOps eq/notEq/in tolerate a null value (IsNull / IsNotNull)") {
+ val ops = TimeTypeParquetOps.filterOps
+ val path = Array("c")
+ val col = longColumn(path)
+ // null value -> null Long comparand; used by ParquetFilters for IsNull /
IsNotNull.
+ assert(ops.makeEq(path, null) === FilterApi.eq(col,
null.asInstanceOf[java.lang.Long]))
+ assert(ops.makeNotEq(path, null) === FilterApi.notEq(col,
null.asInstanceOf[java.lang.Long]))
+ 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 the TimeType encoding and nothing
else") {
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timeType(false, TimeUnit.MICROS), INT64).isDefined)
+ // A different unit, isAdjustedToUTC=true, primitive, or annotation kind
is not the
+ // TimeType encoding, so no framework filter ops is returned (pushdown
falls through).
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timeType(false, TimeUnit.NANOS), INT64).isEmpty)
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timeType(true, TimeUnit.MICROS), INT64).isEmpty)
+ assert(ParquetTypeOps.filterOpsFor(
+ LogicalTypeAnnotation.timeType(false, TimeUnit.MICROS), INT32).isEmpty)
+ assert(ParquetTypeOps.filterOpsFor(null, INT64).isEmpty)
+ }
+
// ---------- helper ----------
private def assertRejects(sparkType: TimeType, field: Type): Unit = {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
