MaxGekk commented on code in PR #59606:
URL: https://github.com/apache/spark/pull/59606#discussion_r4319319876
##########
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:
The same truncation-on-read exists for `TIMESTAMP(NANOS)`:
`TimestampNanosParquetOps.nanosConverter` and
`TimestampNanosVectorUpdater(precision)` floor the stored nanos to the read
precision via `epochNanosToTimestampNanos(value, p)` (see
`ParquetTimestampNanosSuite` "SPARK-57102: explicit lower-precision read schema
truncates sub-precision nanos"), but `ltzFilterOps`/`ntzFilterOps` still
inherit this default. So a file storing `00:00:00.123456789` read as
`TIMESTAMP_NTZ(7)` with `WHERE ts = CAST('... 00:00:00.1234567' AS
TIMESTAMP_NTZ(7))` pushes `Eq(123456700)` against the stored `123456789` and
returns no rows (same for `<=`, `<=>`, `IN`, V1 and V2, both readers).
Could we wire the nanos ops here too (or in a follow-up)?
```scala
override def acceptsReadType(readType: DataType): Boolean = readType match {
case t: TimestampLTZNanosType => t.precision >= 9
case t: TimestampNTZNanosType => t.precision >= 9
case _ => true
}
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -125,6 +126,27 @@ class ParquetFilters(
fieldNames: Array[String],
fieldType: ParquetSchemaType)
+ // A map which contains the field names in `readSchema` and the Spark data
types that the fields
+ // are read as, keyed like `nameToParquetField`. A data type can differ from
the type that the
+ // file was written with, e.g. under a user-specified schema. Lazy because
it is used only for a
+ // predicate with a value on a Types Framework field, see
`valueCanMakeFilterOn`.
+ private lazy val nameToReadType: Map[String, DataType] = {
+ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ def getLeafFields(
+ fields: Seq[StructField],
+ parentFieldNames: Seq[String] = Seq.empty): Seq[(String, DataType)] = {
+ fields.flatMap { f =>
+ val fieldNames = parentFieldNames :+ f.name
+ f.dataType match {
+ case s: StructType => getLeafFields(s.fields.toImmutableArraySeq,
fieldNames)
+ case dataType => Seq(fieldNames.quoted -> dataType)
+ }
+ }
+ }
+ val readTypes = readSchema.toSeq.flatMap(s =>
getLeafFields(s.fields.toImmutableArraySeq)).toMap
+ if (caseSensitive) readTypes else CaseInsensitiveMap(readTypes)
Review Comment:
nit: `nameToParquetField` drops names that are ambiguous under case folding,
while `CaseInsensitiveMap(readTypes)` keeps whichever entry comes last.
Near-zero likelihood since the analyzer rejects such schemas, but it would be
nice for the two maps to apply the same dedup rule.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -918,8 +940,15 @@ class ParquetFilters(
// Parquet's type in the given file should be matched to the value's type
// in the pushed filter in order to push down the filter to Parquet.
+ // Also, the type that a Types Framework field is read as should not make
the reader change the
+ // stored values, e.g. truncate TIME(MICROS) values to TIME(3), because
Parquet evaluates the
+ // pushed filter against the stored values. Otherwise, only IsNull and
IsNotNull are pushed down.
private def valueCanMakeFilterOn(name: String, value: Any): Boolean = {
- valueMatchesParquetType(nameToParquetField(name).fieldType, value)
+ val fieldType = nameToParquetField(name).fieldType
+ valueMatchesParquetType(fieldType, value) && (value == null || (fieldType
match {
+ case FrameworkFilterOps(ops) =>
nameToReadType.get(name).forall(ops.acceptsReadType)
Review Comment:
nit: `FrameworkFilterOps.unapply` is a linear `filterOpsList.find`, and it
now runs twice per non-null value (here and in `valueMatchesParquetType`), so
`IN (v1..vN)` pays 2N lookups. Matching `FrameworkFilterOps(ops)` once and
checking `ops.acceptsValue(value) && readType.forall(ops.acceptsReadType)` from
the same binding would avoid that.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -61,7 +61,8 @@ class ParquetFilters(
pushDownInFilterThreshold: Int,
caseSensitive: Boolean,
datetimeRebaseSpec: RebaseSpec,
- variantExtractionSchema: Option[StructType] = None) {
+ variantExtractionSchema: Option[StructType] = None,
+ readSchema: Option[StructType] = None) {
Review Comment:
This defaults fail-open: with `readSchema = None`, or a filter column
missing from the read schema, `nameToReadType.get(name).forall(...)` pushes the
unsafe predicate. All three production callers pass it now, but a future caller
that omits it silently reinstates the wrong-results bug with no compile-time
signal. Could we make it a required parameter, and/or treat a column absent
from `nameToReadType` as not pushable for a value predicate?
##########
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:
Test gap related to the `TIMESTAMP(NANOS)` comment above: nothing checks
that a `TIMESTAMP(NANOS)` column read as precision 7 or 8 withholds value
predicates. A case mirroring this `TIME(0-5)` matrix for `TIMESTAMP(NANOS,
true/false)` with `TimestampLTZNanosType`/`TimestampNTZNanosType(7|8)`
expecting `createFilter(EqualTo(...)).isEmpty`, plus an end-to-end read of a
9-digit stored value at precision 7, would guard it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala:
##########
@@ -163,6 +163,15 @@ private[ops] object TimeTypeParquetOps {
override def acceptsValue(value: Any): Boolean =
value.isInstanceOf[LocalTime] &&
isMicrosResolution(value.asInstanceOf[LocalTime])
+ // Below microsecond precision, the reader truncates the stored micros
to the read precision
+ // (see TimeTypeParquetOps.newConverter and TimeVectorUpdater), while a
pushed down predicate
+ // compares the stored micros. E.g. a stored 12:00:00.1005 is read as
12:00:00.100 for
+ // TIME(3), so a pushed down `t = 12:00:00.100` would skip it.
+ override def acceptsReadType(readType: DataType): Boolean = readType
match {
Review Comment:
This withholds every value predicate for `TIME(0-5)`, but only `=`, `<=>`,
`<=` and `IN` actually produce false negatives. For a literal `v` on the read
grid (step = 10^(6-p) micros), a stored `s` truncates below `v` iff `s < v`, so
`<` and `>=` are exact against the stored micros, and `>` / `!=` are only
over-inclusive, which is harmless since Spark re-evaluates pushed Parquet
filters after the scan. E.g. `t < CAST('12:00:01' AS TIME(3))` is a correct
row-group-skipping predicate that this now drops.
Since `acceptsReadType` sees neither the operator nor the value, maybe pass
the operator (or decide per `make*` method) and keep the four safe ones?
Optionally `=` could be rewritten to `[v, v+step)` and `<=` to `< v+step` so
nothing is lost.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala:
##########
@@ -85,7 +85,8 @@ case class ParquetScanBuilder(
isCaseSensitive,
// The rebase mode doesn't matter here because the filters are used to
determine
// whether they is convertible.
- RebaseSpec(LegacyBehaviorPolicy.CORRECTED))
+ RebaseSpec(LegacyBehaviorPolicy.CORRECTED),
+ readSchema = Some(readDataSchema()))
Review Comment:
nit: `readDataSchema()` is now computed twice in `pushDataFilters` (also for
`SparkToParquetSchemaConverter.convert` above). It rebuilds a `StructType` each
call, so binding it to a local `val` and passing it to both would be cheaper.
--
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]
