stevomitric commented on code in PR #57681:
URL: https://github.com/apache/spark/pull/57681#discussion_r3753490779
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala:
##########
@@ -398,6 +399,61 @@ class OrcFilterSuite extends OrcTest with
SharedSparkSession {
}
}
+ test("SPARK-57823: filter pushdown - nanosecond timestamp") {
+ // Wall clocks with sub-microsecond digits so the nanosecond fraction
actually participates in
+ // the pushed-down search argument (a micro-only value would not exercise
the nanos path).
Review Comment:
reworded both comments
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala:
##########
@@ -1029,6 +1030,62 @@ abstract class OrcQuerySuite extends OrcQueryTest with
SharedSparkSession {
}
}
+ test("SPARK-57823: ORC predicate pushdown returns correct results for nanos
timestamps") {
+ // One wall clock per second across a minute so each row lands in a
distinct stripe below, all
+ // carrying sub-microsecond digits so the nanosecond fraction participates
in the comparison.
+ val numRows = 60
+ val wallClocks = (0 until numRows).map { s =>
+ LocalDateTime.of(2020, 5, 25, 10, 0, s, 123456789)
+ }
+ def checkNanosPushdown(): Unit = {
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val inputDf = nanosTimestampDf(nanosType, wallClocks)
+ // A boundary literal (row 30) expressed as the nanos external
type, so a `< threshold`
+ // predicate keeps exactly the first 30 rows.
+ val boundary = nanosType match {
+ case _: TimestampNTZNanosType => Literal.create(wallClocks(30),
nanosType)
+ case _: TimestampLTZNanosType =>
+ Literal.create(wallClocks(30).toInstant(ZoneOffset.UTC),
nanosType)
+ }
+ // Rebuilt against each DataFrame's own `ts` attribute so the
filter resolves cleanly.
+ def keepFirstHalf(df: DataFrame): Column = df("ts") <
Column(boundary)
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ // Repartition so the rows are spread over several ORC
files/stripes, giving the
+ // pushed-down search argument something to skip.
+ inputDf.repartition(numRows).write.orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(
+ SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString,
+ SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ val read = spark.read.schema(new StructType().add("ts",
nanosType)).orc(path)
+
+ // Results are correct: the pushdown must not drop or
corrupt matching rows.
+ checkAnswer(
+ read.where(keepFirstHalf(read)),
inputDf.where(keepFirstHalf(inputDf)))
+
+ // Pushdown actually skips data: with the Spark-side filter
removed, fewer than
+ // all rows survive, proving the ORC search argument pruned
stripes.
+
assert(stripSparkFilter(read.where(keepFirstHalf(read))).count() < numRows)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ // The LTZ column stores/compares timestamps in the JVM default zone, so a
literal built in the
Review Comment:
reworded both comments
--
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]
