MaxGekk commented on code in PR #59641:
URL: https://github.com/apache/spark/pull/59641#discussion_r4326481644
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala:
##########
@@ -320,6 +320,16 @@ private[ops] object TimestampNanosParquetOps {
private def epochNanosInRange(encode: => JLong): Boolean =
try { encode; true } catch { case _: ArithmeticException => false }
+ // Below nanosecond precision, the reader truncates the stored
epoch-nanoseconds to the read
+ // precision (see nanosConverter and TimestampNanosVectorUpdater), while a
pushed down predicate
+ // compares the stored epoch-nanoseconds. E.g. a stored 00:00:00.123456789
is read as
+ // 00:00:00.1234567 for TIMESTAMP_NTZ(7), so a pushed down `ts =
00:00:00.1234567` would skip it.
+ private def acceptsNanosReadType(readType: DataType): Boolean = readType
match {
Review Comment:
Thanks for the quick follow-up. Same point as on #59606, but it costs more
here: this withholds every value predicate below precision 9, although with `B
= 10^(9-p)` and `read(s) = floor(s / B) * B`, `<` and `>=` on a literal on the
read grid are exact against the stored values, and `>` / `!=` are only
over-inclusive, which the retained post-scan Filter trims. Only `=`, `<=>`,
`<=` and `IN` produce false negatives.
The regression is measurable: the row-group skipping test in
`ParquetTimestampNanosSuite` now asserts a full scan (`actual === 1024`) for a
Spark-written TIMESTAMP_NTZ(7) file inferred back at precision 7, i.e. the
default read without a user schema, where the stored values already sit on the
read grid.
A bound-widening rewrite keeps the skipping and is exact for foreign files
too: `eq X` as `gtEq X AND lt X+B`, `ltEq X` as `lt X+B`, `gt X` as `gtEq X+B`,
`lt` / `gtEq` unchanged, `notEq` / `In` as the complement / OR of such ranges,
with `X+B` guarded by the existing `epochNanosInRange`. At minimum,
`acceptsReadType` could take the operator so the four sound comparisons stay
pushed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala:
##########
@@ -339,6 +349,8 @@ private[ops] object TimestampNanosParquetOps {
case _ => false
}
+ override def acceptsReadType(readType: DataType): Boolean =
acceptsNanosReadType(readType)
Review Comment:
nit: this and `TimeTypeParquetOps.filterOps.acceptsReadType` are the same
rule, "the read precision must equal the storage precision", with a constant
swapped, and the trait default `acceptsReadType = true` means the next
truncating type silently inherits the wrong-results bug if the override is
forgotten. A `storagePrecision`-style hook on `ParquetFilterOps`, compared
generically against the read type's precision in `ParquetFilters`, would
implement it once and make the gate opt-out rather than opt-in.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2557,6 +2561,86 @@ abstract class ParquetFilterSuite extends ParquetTest
with SharedSparkSession {
}
}
+ test("SPARK-60380: filter pushdown - nanosecond timestamp read with a lower
precision") {
Review Comment:
`withAllParquetReaders` toggles the vectorized vs row readers, not V1 vs V2,
so the `ParquetScanBuilder.convertibleFilters` ->
`ParquetPartitionReaderFactory` path with a lower-precision read schema is not
exercised end to end. The SPARK-57102 test above shows the `USE_V1_SOURCE_LIST
-> ""` pattern; wrapping this test in both modes would close the gap.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2416,11 +2416,14 @@ abstract class ParquetFilterSuite extends ParquetTest
with SharedSparkSession {
withSQLConf(
SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> "1") {
+ // The columns are read with precision 9. A lower read precision
truncates the values, so
+ // only IsNull and IsNotNull are pushed down then.
Seq(7, 8, 9).foreach { p =>
Review Comment:
With the read schema hard-coded to precision 9 and `convertToParquetType`
ignoring `p` (always INT64 TIMESTAMP(NANOS)), the three iterations of this loop
are now byte-identical. Either drop the loop, since the new test below iterates
the read precision with `isDefined === (p == 9)`, or make `p` the read
precision and fold the two tests together.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2557,6 +2561,86 @@ abstract class ParquetFilterSuite extends ParquetTest
with SharedSparkSession {
}
}
+ test("SPARK-60380: filter pushdown - nanosecond timestamp read with a lower
precision") {
+ // A TIMESTAMP(9) value 1970-01-01 00:00:00.123456789 read as TIMESTAMP(7)
is truncated to
+ // 00:00:00.1234567, while Parquet evaluates a pushed down predicate
against the stored value.
Review Comment:
nit: only read precision 7 is exercised end to end here; precision 8 (and
8-written / 7-read) is covered only at the `createFilter` level, which checks
`isDefined` only. Looping `Seq(7, 8)` for the read precision, as
`ParquetTimestampNanosSuite` does, is a one-line change.
--
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]
