You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2025/04/02 09:08:04 UTC
(spark) branch master updated: [SPARK-51687][SQL] Pushdown filters with TIME values to parquet
This is an automated email from the ASF dual-hosted git repository.
maxgekk 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 beb509fe6d79 [SPARK-51687][SQL] Pushdown filters with TIME values to parquet
beb509fe6d79 is described below
commit beb509fe6d79ada842e14ee7c4a7c55d874fd8b7
Author: Max Gekk <ma...@gmail.com>
AuthorDate: Wed Apr 2 12:07:46 2025 +0300
[SPARK-51687][SQL] Pushdown filters with TIME values to parquet
### What changes were proposed in this pull request?
In the PR, I propose to support pushdown filters over TIME columns in the parquet datasource. In particular, pushdown the following operators: `===, <=>, =!=, isNull, isNotNull, >, >=, <, <=`.
### Why are the changes needed?
The PR https://github.com/apache/spark/pull/50389 supported the TIME type in the Parquet datasource as INT64 but filters with such values are not pushed down. So, comparing to primitive types, the TIME type can suffer from worse performance in read. This PR aims to solve the issue, and achieve feature parity with other types.
### Does this PR introduce _any_ user-facing change?
No, the changes might influence on performance of the parquet datasource only.
### How was this patch tested?
By running new tests in `ParquetFilterSuite`:
```
$ build/sbt clean "test:testOnly *ParquetV1FilterSuite"
$ build/sbt clean "test:testOnly *ParquetV2FilterSuite"
```
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #50473 from MaxGekk/parquet-time-filter.
Authored-by: Max Gekk <ma...@gmail.com>
Signed-off-by: Max Gekk <ma...@gmail.com>
---
.../datasources/parquet/ParquetFilters.scala | 37 ++++++++++++++++-
.../datasources/parquet/ParquetFilterSuite.scala | 48 +++++++++++++++++++++-
2 files changed, 82 insertions(+), 3 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 3dc9ddf386f1..565742671b9c 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
@@ -21,7 +21,7 @@ import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float =
import java.math.{BigDecimal => JBigDecimal}
import java.nio.charset.StandardCharsets.UTF_8
import java.sql.{Date, Timestamp}
-import java.time.{Duration, Instant, LocalDate, Period}
+import java.time.{Duration, Instant, LocalDate, LocalTime, Period}
import java.util.HashSet
import java.util.Locale
@@ -149,6 +149,8 @@ class ParquetFilters(
ParquetSchemaType(LogicalTypeAnnotation.timestampType(true, TimeUnit.MICROS), INT64, 0)
private val ParquetTimestampMillisType =
ParquetSchemaType(LogicalTypeAnnotation.timestampType(true, TimeUnit.MILLIS), INT64, 0)
+ private val ParquetTimeType =
+ ParquetSchemaType(LogicalTypeAnnotation.timeType(false, TimeUnit.MICROS), INT64, 0)
private def dateToDays(date: Any): Int = {
val gregorianDays = date match {
@@ -173,6 +175,10 @@ class ParquetFilters(
}
}
+ private def localTimeToMicros(v: Any): JLong = {
+ DateTimeUtils.localTimeToMicros(v.asInstanceOf[LocalTime])
+ }
+
private def decimalToInt32(decimal: JBigDecimal): Integer = decimal.unscaledValue().intValue()
private def decimalToInt64(decimal: JBigDecimal): JLong = decimal.unscaledValue().longValue()
@@ -207,6 +213,7 @@ class ParquetFilters(
private def toLongValue(v: Any): JLong = v match {
case d: Duration => IntervalUtils.durationToMicros(d)
+ case lt: LocalTime => DateTimeUtils.localTimeToMicros(lt)
case l => l.asInstanceOf[JLong]
}
@@ -244,6 +251,10 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.eq(
longColumn(n),
Option(v).map(timestampToMillis).orNull)
+ case ParquetTimeType =>
+ (n: Array[String], v: Any) => FilterApi.eq(
+ longColumn(n),
+ Option(v).map(localTimeToMicros).orNull)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], v: Any) => FilterApi.eq(
@@ -293,6 +304,10 @@ class ParquetFilters(
(n: Array[String], v: Any) => FilterApi.notEq(
longColumn(n),
Option(v).map(timestampToMillis).orNull)
+ case ParquetTimeType =>
+ (n: Array[String], v: Any) => FilterApi.notEq(
+ longColumn(n),
+ Option(v).map(localTimeToMicros).orNull)
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], v: Any) => FilterApi.notEq(
@@ -333,6 +348,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 ParquetTimeType =>
+ (n: Array[String], v: Any) => FilterApi.lt(longColumn(n), localTimeToMicros(v))
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -370,6 +387,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 ParquetTimeType =>
+ (n: Array[String], v: Any) => FilterApi.ltEq(longColumn(n), localTimeToMicros(v))
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -407,6 +426,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 ParquetTimeType =>
+ (n: Array[String], v: Any) => FilterApi.gt(longColumn(n), localTimeToMicros(v))
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -444,6 +465,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 ParquetTimeType =>
+ (n: Array[String], v: Any) => FilterApi.gtEq(longColumn(n), localTimeToMicros(v))
case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], v: Any) =>
@@ -533,6 +556,14 @@ class ParquetFilters(
}
FilterApi.in(longColumn(n), set)
+ case ParquetTimeType =>
+ (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 ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
(n: Array[String], values: Array[Any]) =>
val set = new HashSet[Integer]()
@@ -620,7 +651,8 @@ class ParquetFilters(
case v: JLong => v.longValue() >= Int.MinValue && v.longValue() <= Int.MaxValue
case _ => false
}
- case ParquetLongType => value.isInstanceOf[JLong] || value.isInstanceOf[Duration]
+ case ParquetLongType =>
+ value.isInstanceOf[JLong] || value.isInstanceOf[Duration]
case ParquetFloatType => value.isInstanceOf[JFloat]
case ParquetDoubleType => value.isInstanceOf[JDouble]
case ParquetStringType => value.isInstanceOf[String]
@@ -629,6 +661,7 @@ class ParquetFilters(
value.isInstanceOf[Date] || value.isInstanceOf[LocalDate]
case ParquetTimestampMicrosType | ParquetTimestampMillisType =>
value.isInstanceOf[Timestamp] || value.isInstanceOf[Instant]
+ case ParquetTimeType => value.isInstanceOf[LocalTime]
case ParquetSchemaType(decimalType: DecimalLogicalTypeAnnotation, INT32, _) =>
isDecimalMatched(value, decimalType)
case ParquetSchemaType(decimalType: DecimalLogicalTypeAnnotation, INT64, _) =>
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 6080a5e8e4bb..6b73cc8618d1 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, Period, ZoneId}
+import java.time.{Duration, LocalDate, LocalDateTime, LocalTime, Period, ZoneId}
import java.util.HashSet
import scala.reflect.ClassTag
@@ -2230,6 +2230,52 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
}
}
}
+
+ test("SPARK-51687: filter pushdown - time") {
+ def secs(s: Int): LocalTime = LocalTime.ofSecondOfDay(s)
+ def secsLit(s: Int): Literal = Literal(secs(s))
+ val data = (1 to 4).map(i => Tuple1(secs(i)))
+ withNestedParquetDataFrame(data) { case (inputDF, colName, resultFun) =>
+ implicit val df: DataFrame = inputDF
+
+ val iAttr = df(colName).expr
+ assert(df(colName).expr.dataType === TimeType())
+
+ checkFilterPredicate(iAttr.isNull, classOf[Eq[_]], Seq.empty[Row])
+ checkFilterPredicate(iAttr.isNotNull, classOf[NotEq[_]],
+ (1 to 4).map(i => Row.apply(resultFun(secs(i)))))
+
+ checkFilterPredicate(iAttr === secsLit(1), classOf[Eq[_]], resultFun(secs(1)))
+ checkFilterPredicate(iAttr <=> secsLit(1), classOf[Eq[_]], resultFun(secs(1)))
+ checkFilterPredicate(iAttr =!= secsLit(1), classOf[NotEq[_]],
+ (2 to 4).map(i => Row.apply(resultFun(secs(i)))))
+
+ checkFilterPredicate(iAttr < secsLit(2), classOf[Lt[_]], resultFun(secs(1)))
+ checkFilterPredicate(iAttr > secsLit(3), classOf[Gt[_]], resultFun(secs(4)))
+ checkFilterPredicate(iAttr <= secsLit(1), classOf[LtEq[_]], resultFun(secs(1)))
+ checkFilterPredicate(iAttr >= secsLit(4), classOf[GtEq[_]], resultFun(secs(4)))
+
+ checkFilterPredicate(secsLit(1) === iAttr, classOf[Eq[_]], resultFun(secs(1)))
+ checkFilterPredicate(secsLit(1) <=> iAttr, classOf[Eq[_]], resultFun(secs(1)))
+ checkFilterPredicate(secsLit(2) > iAttr, classOf[Lt[_]], resultFun(secs(1)))
+ checkFilterPredicate(secsLit(3) < iAttr, classOf[Gt[_]], resultFun(secs(4)))
+ checkFilterPredicate(secsLit(1) >= iAttr, classOf[LtEq[_]], resultFun(secs(1)))
+ checkFilterPredicate(secsLit(4) <= iAttr, classOf[GtEq[_]], resultFun(secs(4)))
+
+ checkFilterPredicate(!(iAttr < secsLit(4)), classOf[GtEq[_]], resultFun(secs(4)))
+ checkFilterPredicate(iAttr < secsLit(2) || iAttr > secsLit(3), classOf[Operators.Or],
+ Seq(Row(resultFun(secs(1))), Row(resultFun(secs(4)))))
+
+ Seq(3, 20).foreach { threshold =>
+ withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> s"$threshold") {
+ checkFilterPredicate(
+ In(iAttr, Array(2, 3, 4, 5, 6, 7).map(secsLit).toImmutableArraySeq),
+ if (threshold == 3) classOf[FilterIn[_]] else classOf[Operators.Or],
+ Seq(Row(resultFun(secs(2))), Row(resultFun(secs(3))), Row(resultFun(secs(4)))))
+ }
+ }
+ }
+ }
}
@ExtendedSQLTest
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org