qlong commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3823419768
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -692,6 +948,56 @@ class ParquetFilters(
nameToParquetField.contains(name) && valueCanMakeFilterOn(name, value)
}
+ // Whether `name` is a shredded-variant logical path whose typed leaf
accepts `value`. `value`
+ // must be non-null: shredded pushdown only handles comparison predicates.
+ private def canMakeShreddedFilterOn(name: String, value: Any): Boolean = {
+ value != null && nameToShreddedVariantField.get(name).exists { f =>
+ valueMatchesParquetType(f.leaf.fieldType, value)
+ }
+ }
+
+ // Whether `predicate` references a shredded-variant logical path anywhere.
Used to refuse
+ // conversion under negation: the shredded predicate is `or(leaf,
isNotNull(residual)...)`, and
+ // `not(...)` of it is rewritten by parquet-mr's LogicalInverseRewriter into
+ // `and(notEq(leaf), eq(residual, null))`, whose `eq(residual, null)`
conjunct makes an AND
+ // row-group-droppable whenever the residual has no nulls -- unsound (drops
a row group whose
+ // matching values are all in the residual). Since a negated shredded
predicate cannot be
+ // expressed soundly with row-group statistics, we do not push it at all.
+ //
+ // `sources.Filter.references` already recurses through And/Or/Not and every
leaf filter, so this
+ // stays correct if new Filter subtypes are added.
+ private def referencesShreddedName(predicate: sources.Filter): Boolean =
+ predicate.references.exists(nameToShreddedVariantField.contains)
+
+ // Build the sound shredded-variant predicate:
+ // or(leafPredicate, isNotNull(residual_0), ..., isNotNull(residual_n))
+ // where each isNotNull is `notEq(residual, null)`.
+ //
+ // Parquet's statistics drop logic is: `or(a, b)` is row-group-droppable iff
BOTH `a` and `b` are
+ // droppable, and `notEq(col, null)` (IS NOT NULL) is droppable iff the
column is entirely NULL in
+ // the row group (no non-nulls). So the whole `or` drops the row group iff
the leaf predicate is
+ // droppable (leaf min/max cannot match) AND every residual is entirely NULL
(no value for the
+ // path is hiding in a residual). If any residual holds a non-null, its
isNotNull conjunct is not
+ // droppable, so the row group is kept -- we never drop a row group that
could contain a matching
+ // residual value.
+ //
+ // The naive `and(leafPredicate, isNull(residual))` is UNSOUND: `and` drops
iff EITHER conjunct is
+ // droppable, so the leaf predicate alone would drop the row group
regardless of the residual.
+ //
+ // `makeLeaf` produces the leaf predicate from the leaf's field-name array;
it returns None if the
+ // leaf type has no comparison encoding.
+ private def makeShreddedFilter(
+ name: String,
+ makeLeaf: (ParquetSchemaType, Array[String]) => Option[FilterPredicate]
+ ): Option[FilterPredicate] = {
+ val field = nameToShreddedVariantField(name)
+ makeLeaf(field.leaf.fieldType, field.leaf.fieldNames).map { leafPredicate
=>
+ field.residualFieldNames.foldLeft(leafPredicate) { (acc, residualNames)
=>
Review Comment:
+1 on the new guard with isNull(leaf) suggested by @peter-toth, especially
true for the strict inferenced schema generated by Spark.
With regard to default on/off, I think your benchmark shows the worst case
for overhead due to the large number of rowgroups. With default parquet block
size, the overhead could be lower (worth a testing). But even with the new
guard, the chance this optimization can be triggered still requires delicated
tuning.
--
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]
