dongjoon-hyun commented on code in PR #59606:
URL: https://github.com/apache/spark/pull/59606#discussion_r4319357102
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala:
##########
@@ -62,6 +64,14 @@ private[parquet] trait ParquetFilterOps {
/** Whether `value` (a non-null external filter value) is pushable for this
type. */
def acceptsValue(value: Any): Boolean
+ /**
+ * Whether a predicate with a value is pushable for a column of this
encoding that is read as
+ * `readType`, the Spark type in the read schema. Parquet evaluates a pushed
down predicate
+ * against the stored values, so this is false when the reader changes them
for `readType`, e.g.
+ * truncates them to a lower precision. IsNull and IsNotNull are pushed down
regardless.
+ */
Review Comment:
Thank you for the review, @MaxGekk . Yes, `TIMESTAMP(NANOS)` has the same
issue. I am already working on it in a new JIRA issue, including the test
coverage that you mentioned.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2309,6 +2311,91 @@ abstract class ParquetFilterSuite extends ParquetTest
with SharedSparkSession {
}
}
+ test("SPARK-60358: filter pushdown - time read with a lower precision than
the stored value") {
+ // A TIME(6) value 12:00:00.1005 read as TIME(3) is truncated to
12:00:00.100, while Parquet
+ // evaluates a pushed down predicate against the stored value. For
example, a pushed down
+ // `t = 12:00:00.100` would skip the row group that holds the matching row.
+ import testImplicits._
+ def time(s: String): LocalTime = LocalTime.parse(s)
+ def time3(s: String): String = s"CAST('$s' AS TIME(3))"
+ def toTime3(dt: DataType): DataType = dt match {
+ case s: StructType => StructType(s.map(f => f.copy(dataType =
toTime3(f.dataType))))
+ case _: TimeType => TimeType(3)
+ case other => other
+ }
+ val inputDF = Seq(Some(time("12:00:00.1005")), Some(time("12:00:01")),
None).toDF()
+ withNestedDataFrame(inputDF).foreach { case (df, colName, resultFun) =>
+ withTempPath { dir =>
+ df.coalesce(1).write.parquet(dir.getCanonicalPath)
+ val readDF =
spark.read.schema(toTime3(df.schema).asInstanceOf[StructType])
+ .parquet(dir.getCanonicalPath)
+ def check(condition: String, expected: Seq[String]): Unit = {
+ checkAnswer(readDF.where(condition), expected.map(s =>
Row(resultFun(time(s)))))
+ }
+ withAllParquetReaders {
+ check(s"$colName = ${time3("12:00:00.1")}", Seq("12:00:00.1"))
+ check(s"$colName <=> ${time3("12:00:00.1")}", Seq("12:00:00.1"))
+ check(s"$colName != ${time3("12:00:00.1")}", Seq("12:00:01"))
+ check(s"$colName < ${time3("12:00:01")}", Seq("12:00:00.1"))
+ check(s"$colName <= ${time3("12:00:00.1")}", Seq("12:00:00.1"))
+ check(s"$colName > ${time3("12:00:00.1")}", Seq("12:00:01"))
+ check(s"$colName >= ${time3("12:00:00.1")}", Seq("12:00:00.1",
"12:00:01"))
+ Seq(1, 10).foreach { threshold =>
+ withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key
-> s"$threshold") {
+ check(s"$colName IN (${time3("12:00:00.1")},
${time3("12:00:02")})",
+ Seq("12:00:00.1"))
+ }
+ }
+ checkAnswer(readDF.where(s"$colName IS NULL"), Row(resultFun(null)))
+ check(s"$colName IS NOT NULL", Seq("12:00:00.1", "12:00:01"))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60358: TIME(MICROS) predicates with a value are not pushed down
for TIME(0-5)") {
Review Comment:
Yes, the new JIRA issue for `TIMESTAMP(NANOS)` will include these test cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
