dongjoon-hyun commented on code in PR #59606:
URL: https://github.com/apache/spark/pull/59606#discussion_r4319480500
##########
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:
Thank you. Updated to fail closed. A Types Framework column without a read
type, e.g. without a read schema or missing from it, pushes down only `IsNull`
and `IsNotNull` now. The test harness and the existing test cases pass the read
schema.
##########
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:
Updated to skip the ambiguous names like `nameToParquetField`. With the
fail-closed change, such a column pushes down only `IsNull` and `IsNotNull`.
##########
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:
Updated to match `FrameworkFilterOps` once in `valueMatchesParquetType`.
##########
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:
Updated.
--
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]
