This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 ef1ec9dc146a [SPARK-57571][SQL] Support TIME predicate pushdown to ORC
ef1ec9dc146a is described below
commit ef1ec9dc146af4671afb80bfeeeb780dd954f269
Author: Maxim Gekk <[email protected]>
AuthorDate: Fri Jun 26 23:40:46 2026 +0200
[SPARK-57571][SQL] Support TIME predicate pushdown to ORC
### What changes were proposed in this pull request?
This PR adds predicate pushdown for the `TIME` (`TimeType`) data type to
the ORC reader, matching the existing Parquet TIME filter pushdown
(SPARK-51687).
Changes:
- `OrcFilters`:
- `getPredicateLeafType` now maps `TimeType` to `PredicateLeaf.Type.LONG`.
- `castLiteralValue` converts a `java.time.LocalTime` literal to
nanos-of-day via `DateTimeUtils.localTimeToNanos`, which is how ORC physically
stores `TIME` values (ORC `LONG`, see `OrcSerializer`/`OrcUtils`).
- The existing leaf handlers (`EqualTo`, `EqualNullSafe`, `LessThan`,
`LessThanOrEqual`, `GreaterThan`, `GreaterThanOrEqual`, `In`, `IsNull`) then
work for `TIME` unchanged.
- Add a `localTimeToLiteral` implicit to the Catalyst `dsl` package,
mirroring the existing `timestampNTZToLiteral`, so `TIME` literals can be used
in dsl predicates (used by the test).
- Add a test in `OrcFilterSuite` (also exercised by `OrcV1FilterSuite`).
### Why are the changes needed?
Native ORC read/write for `TIME` is already supported (SPARK-54472), but
`OrcFilters.getPredicateLeafType` had no `TimeType` case, so `TIME` predicates
were not pushed down to the ORC reader. This closes that gap and brings ORC to
parity with Parquet, avoiding unnecessary row scans for filters on `TIME`
columns.
### Does this PR introduce _any_ user-facing change?
No behavior change in query results. Filters on `TIME` columns over ORC are
now pushed down to the reader, which can improve scan performance.
### How was this patch tested?
Added `SPARK-57571: filter pushdown - time` in `OrcFilterSuite`, covering
EQ / NULL_SAFE_EQUALS / LT / LTE / GT / GTE / IsNull in both `col op literal`
and `literal op col` forms. Ran:
```
build/sbt 'sql/testOnly *OrcFilterSuite *OrcV1FilterSuite'
```
All 40 tests passed. Also ran `./dev/scalastyle` (passed) and scalafmt on
the connect/api modules.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (Claude Opus 4.8)
Closes #56823 from MaxGekk/time-orc-pred-pushdown.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit 82498147a0c2f18f35612d7376395de5d974e318)
Signed-off-by: Max Gekk <[email protected]>
---
.../apache/spark/sql/catalyst/dsl/package.scala | 3 +-
.../sql/execution/datasources/orc/OrcFilters.scala | 8 +++--
.../execution/datasources/orc/OrcFilterSuite.scala | 35 +++++++++++++++++++++-
3 files changed, 41 insertions(+), 5 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala
index 06f4489c4ab4..3ba538c0e216 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.catalyst
import java.sql.{Date, Timestamp}
-import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period}
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime,
Period}
import scala.language.implicitConversions
@@ -176,6 +176,7 @@ package object dsl extends SQLConfHelper {
implicit def decimalToLiteral(d: Decimal): Literal = Literal(d)
implicit def timestampToLiteral(t: Timestamp): Literal = Literal(t)
implicit def timestampNTZToLiteral(l: LocalDateTime): Literal = Literal(l)
+ implicit def localTimeToLiteral(t: LocalTime): Literal = Literal(t)
implicit def instantToLiteral(i: Instant): Literal = Literal(i)
implicit def binaryToLiteral(a: Array[Byte]): Literal = Literal(a)
implicit def periodToLiteral(p: Period): Literal = Literal(p)
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 4bb1c187c45f..66770870dabe 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,7 +17,7 @@
package org.apache.spark.sql.execution.datasources.orc
-import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period}
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime,
Period}
import org.apache.hadoop.hive.common.`type`.HiveDecimal
import org.apache.hadoop.hive.ql.io.sarg.{PredicateLeaf, SearchArgument}
@@ -25,7 +25,7 @@ import
org.apache.hadoop.hive.ql.io.sarg.SearchArgument.Builder
import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory.newBuilder
import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable
-import org.apache.spark.sql.catalyst.util.DateTimeUtils.{instantToMicros,
localDateTimeToMicros, localDateToDays, toJavaDate, toJavaTimestamp}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils.{instantToMicros,
localDateTimeToMicros, localDateToDays, localTimeToNanos, toJavaDate,
toJavaTimestamp}
import org.apache.spark.sql.catalyst.util.IntervalUtils
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
@@ -142,7 +142,7 @@ private[sql] object OrcFilters extends OrcFiltersBase {
def getPredicateLeafType(dataType: DataType): PredicateLeaf.Type = dataType
match {
case BooleanType => PredicateLeaf.Type.BOOLEAN
case ByteType | ShortType | IntegerType | LongType |
- _: AnsiIntervalType | TimestampNTZType => PredicateLeaf.Type.LONG
+ _: AnsiIntervalType | TimestampNTZType | _: TimeType =>
PredicateLeaf.Type.LONG
case FloatType | DoubleType => PredicateLeaf.Type.FLOAT
case StringType => PredicateLeaf.Type.STRING
case DateType => PredicateLeaf.Type.DATE
@@ -170,6 +170,8 @@ private[sql] object OrcFilters extends OrcFiltersBase {
toJavaTimestamp(instantToMicros(value.asInstanceOf[Instant]))
case _: TimestampNTZType if value.isInstanceOf[LocalDateTime] =>
localDateTimeToMicros(value.asInstanceOf[LocalDateTime])
+ case _: TimeType if value.isInstanceOf[LocalTime] =>
+ localTimeToNanos(value.asInstanceOf[LocalTime])
case _: YearMonthIntervalType =>
IntervalUtils.periodToMonths(value.asInstanceOf[Period]).longValue()
case _: DayTimeIntervalType =>
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 46bd619d4cab..3168075cc9d9 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, Period}
+import java.time.{Duration, LocalDateTime, LocalTime, Period}
import scala.jdk.CollectionConverters._
@@ -365,6 +365,39 @@ class OrcFilterSuite extends OrcTest with
SharedSparkSession {
}
}
+ test("SPARK-57571: filter pushdown - time") {
+ val times = Seq(
+ LocalTime.of(0, 0, 0),
+ LocalTime.of(1, 2, 3, 456000000),
+ LocalTime.of(12, 30, 45, 123456000),
+ LocalTime.of(23, 59, 59, 999999000))
+ withOrcFile(times.map(Tuple1(_))) { path =>
+ readFile(path) { implicit df =>
+ checkFilterPredicate($"_1".isNull, PredicateLeaf.Operator.IS_NULL)
+
+ checkFilterPredicate($"_1" === times(0), PredicateLeaf.Operator.EQUALS)
+ checkFilterPredicate($"_1" <=> times(0),
PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+
+ checkFilterPredicate($"_1" < times(1),
PredicateLeaf.Operator.LESS_THAN)
+ checkFilterPredicate($"_1" > times(2),
PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate($"_1" <= times(0),
PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate($"_1" >= times(3),
PredicateLeaf.Operator.LESS_THAN)
+
+ checkFilterPredicate(Literal(times(0)) === $"_1",
PredicateLeaf.Operator.EQUALS)
+ checkFilterPredicate(
+ Literal(times(0)) <=> $"_1", PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+ checkFilterPredicate(Literal(times(1)) > $"_1",
PredicateLeaf.Operator.LESS_THAN)
+ checkFilterPredicate(
+ Literal(times(2)) < $"_1",
+ PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate(
+ Literal(times(0)) >= $"_1",
+ PredicateLeaf.Operator.LESS_THAN_EQUALS)
+ checkFilterPredicate(Literal(times(3)) <= $"_1",
PredicateLeaf.Operator.LESS_THAN)
+ }
+ }
+ }
+
test("filter pushdown - combinations with logical operators") {
withOrcDataFrame((1 to 4).map(i => Tuple1(Option(i)))) { implicit df =>
checkFilterPredicate(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
