dongjoon-hyun commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3798555083
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -718,6 +962,37 @@ class ParquetFilters(
// Probably I missed something and obviously this should be changed.
predicate match {
+ // Shredded-variant paths (e.g. "v.`0`"). Only comparison predicates
that use min/max
+ // statistics are eligible. Each pushes or(leafPredicate,
isNotNull(residual)...) over every
+ // residual `value` column along the path (see `makeShreddedFilter`). IS
NULL / IS NOT NULL on
+ // the logical variant field are intentionally out of scope: "the
extracted field is null" is
+ // not the same as "typed_value is null", so we must not conflate them.
+ case sources.EqualTo(name, value) if canMakeShreddedFilterOn(name,
value) =>
Review Comment:
These shredded cases are reachable through the pre-existing generic `case
sources.Not(pred)` recursion below, and under negation the pushed predicate
becomes unsound.
A `!=` predicate arrives as ``sources.Not(EqualTo("v.`0`", 700))``. The
guarded `Not(EqualTo)` fast path doesn't match (the shredded logical name is
not in `nameToParquetField`), so the generic `Not` case recurses into the
shredded `EqualTo` case here and wraps the result in `FilterApi.not`.
parquet-mr's `LogicalInverseRewriter` then rewrites
```
not(or(eq(leaf, 700), notEq(residual, null)))
```
into
```
and(notEq(leaf, 700), eq(residual, null))
```
which is exactly the `and(..., isNull(residual))` shape the comment on
`makeShreddedFilter` proves unsound: `StatisticsFilter` drops an AND row group
if **any** conjunct is droppable, and `eq(residual, null)` is droppable
whenever the residual column has zero nulls.
Concrete repro: shred with `a tinyint`, one row group whose rows are
`{"a":500}` and `{"a":600}` (both overflow tinyint, so both stored in the
residual; residual nullCount = 0, typed leaf entirely null). `WHERE
variant_get(v, '$.a', 'bigint') != 700` skips the row group and returns an
empty result instead of {500, 600} — silent data loss. The same hole exists for
`NOT IN` and `Not(EqualNullSafe/GreaterThan/...)`.
Since `not(or(leaf, isNotNull(residual)))` cannot be expressed soundly with
row-group statistics, the shredded conversion must refuse to be produced under
negation — e.g. have the generic `sources.Not` case return `None` when the
child predicate references a shredded-variant logical name. It would also be
good to add a `!=` / `NOT IN` test with an all-fallback row group; the PR
currently has no test covering a negated predicate.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -128,6 +138,198 @@ class ParquetFilters(
fieldNames: Array[String],
fieldType: ParquetSchemaType)
+ /**
+ * Holds the mapping from a logical shredded-variant path (e.g. "v.`0`") to
the physical
+ * shredded columns needed to push a sound row-group-skipping predicate.
+ *
+ * @param leaf the physical `typed_value` scalar leaf carrying min/max
statistics
+ * @param residualFieldNames the untyped `value` residual columns along the
path, from the
+ * top-level residual down to the leaf's own-level
sibling. Each is a
+ * physical field-name array. Only residuals that
exist in this file's
+ * schema are included; a value for the path can
only be hiding in one
+ * of these residuals when the typed leaf is NULL,
so the pushed
+ * predicate OR-s in an IS NOT NULL guard on each
(see
+ * `makeShreddedFilter`).
+ */
+ private case class ShreddedVariantField(
+ leaf: ParquetPrimitiveField,
+ residualFieldNames: Seq[Array[String]])
+
+ // Maps logical shredded-variant paths produced by PushVariantIntoScan (e.g.
"v.`0`") to the
+ // physical shredded columns. Populated only when `variantExtractionSchema`
is provided and the
+ // physical file schema actually shreds the requested path.
+ //
+ // Soundness: shredding is per-row and per-file best-effort. A row whose
value does not fit the
+ // shredded type (type mismatch or overflow), or whose field is not shredded
in this file, is
+ // stored in an untyped `value` residual with `typed_value` NULL. Parquet
min/max excludes NULLs,
+ // so pushing the predicate on the typed leaf alone could skip a row group
that still holds a
+ // matching row in a residual. To stay sound we push `or(leafPredicate,
isNotNull(residual)...)`
+ // over every residual `value` column along the path: Parquet drops the row
group only when the
+ // leaf cannot match AND every residual is entirely NULL, so a row group is
skipped only when
+ // every value for the path is provably in the typed leaf. See
`makeShreddedFilter`.
+ private val nameToShreddedVariantField: Map[String, ShreddedVariantField] = {
+ variantExtractionSchema match {
+ case Some(variantSchema) =>
+ val entries = shreddedVariantEntries(
+ variantSchema.fields.toSeq, schema.asGroupType(), Array.empty,
Array.empty)
+ if (caseSensitive) {
+ entries.toMap
+ } else {
+ // Mirror `nameToParquetField`: drop names that are ambiguous under
case-insensitive
+ // matching rather than risk pushing a filter on the wrong physical
column.
+ val dedup = entries
+ .groupBy(_._1.toLowerCase(Locale.ROOT))
+ .filter(_._2.size == 1)
+ .transform((_, v) => v.head._2)
+ CaseInsensitiveMap(dedup)
+ }
+ case None => Map.empty
+ }
+ }
+
+ // Look up a child of `group` by name, honoring `caseSensitive`. Returns the
child type together
+ // with its actual physical name so callers build paths from the on-disk
names (needed for
+ // correct case-insensitive matching, where the requested key case may
differ from the file's).
+ private def findChild(group: GroupType, name: String): Option[Type] = {
+ group.getFields.asScala.find { f =>
+ if (caseSensitive) f.getName == name else
f.getName.equalsIgnoreCase(name)
+ }
+ }
+
+ // Look up the untyped `value` residual sibling in `group`, if it exists as
a non-REPEATED
+ // primitive. Returns the physical field name.
+ private def residualIn(group: GroupType): Option[String] = findChild(group,
VALUE).collect {
+ case p: PrimitiveType if p.getRepetition != Repetition.REPEATED =>
p.getName
+ }
+
+ // Copy of `getNormalizedLogicalType` from the `nameToParquetField` closure,
needed here for the
+ // shredded leaf resolution which runs outside that closure.
+ private def getNormalizedLogicalType(p: PrimitiveType):
LogicalTypeAnnotation = {
+ (p.getPrimitiveTypeName, p.getLogicalTypeAnnotation) match {
+ case (INT32, intType: IntLogicalTypeAnnotation)
+ if intType.getBitWidth() == 32 && intType.isSigned() => null
+ case (INT64, intType: IntLogicalTypeAnnotation)
+ if intType.getBitWidth() == 64 && intType.isSigned() => null
+ case (_, otherType) => otherType
+ }
+ }
+
+ // Navigate the regular shredding layout from a variant column's physical
group, resolving both
+ // the typed leaf and the residual `value` columns along the path. The
layout is:
+ // <col> / typed_value / k0 / typed_value / ... / kN / typed_value (leaf)
+ // <col> / value (L0
residual)
+ // <col> / typed_value / k0 / value (L1
residual)
+ // ...
+ // <col> / typed_value / k0 / ... / kN / value
(leaf-level residual)
+ // Paths are built from the on-disk field names (via `findChild`) so
case-insensitive matching
+ // uses the file's actual names. A value for the path can only be hiding in
one of these residual
+ // `value` columns when the typed leaf is NULL, so IS NULL on all of them is
the soundness guard.
+ // Residuals absent in this file's schema are skipped (that level cannot
hold a fallback here).
+ // Returns None if the file does not shred this path down to a non-REPEATED
scalar leaf (nothing
+ // is pushed and the row group is simply read).
+ private def resolveShredded(
+ physCol: GroupType,
+ physColPath: Array[String],
+ keys: Array[String]): Option[ShreddedVariantField] = {
+ if (keys.isEmpty) return None
+ val residuals = scala.collection.mutable.ArrayBuffer.empty[Array[String]]
+ // L0: the variant column's own residual.
+ residualIn(physCol).foreach(r => residuals += (physColPath :+ r))
+ // Descend key by key: <group>/typed_value/<key>. Collect each level's
residual sibling.
+ var group = physCol
+ var namePath = physColPath
+ var idx = 0
+ while (idx < keys.length) {
+ val typedChild = findChild(group, TYPED_VALUE) match {
+ case Some(g: GroupType) => g
+ case _ => return None
+ }
+ val typedName = typedChild.getName
+ val keyChild = findChild(typedChild, keys(idx)) match {
Review Comment:
The variant object-key segments (`keys(idx)`) are matched here through
`findChild`, which honors `spark.sql.caseSensitiveAnalysis` — but variant field
extraction is always exact-case at read time
(`SparkShreddingUtils.getFieldsToExtract` uses
`schema.objectSchemaMap.get(key)`, and the residual fallback uses
`Variant.getFieldByKey`, both plain `equals`). Variant keys are data, not Spark
identifiers, so applying the identifier case-sensitivity config to them can
bind the predicate to the wrong physical subtree.
Concrete unsound scenario with the default `caseSensitive=false`: a file
legally shreds **both** keys `A` and `a` (variant keys are case-distinct, and
Parquet allows sibling fields differing only in case — producible by an
external spec-compliant writer or a forced shredding schema), with schema order
`[A, a]`. For `variant_get(v, '$.a', 'bigint') > 999`, `findChild` first-match
binds the leaf and the residual guards to the `A` subtree. In a row group where
every row is fully shredded under `a` (all guarded residuals null, `A` leaf max
<= 999), every disjunct is droppable, so the row group is skipped while
`v.typed_value.a.typed_value` holds matching rows — silent data loss. Note the
case-insensitive dedup in `nameToShreddedVariantField` doesn't help: it dedups
logical map keys, not case-colliding physical siblings inside `typed_value`.
The key segments should be compared exact-case regardless of the config
(mirroring `objectSchemaMap`); keeping case-insensitive matching for the
top-level column-name segments and the structural `typed_value`/`value` names
is fine.
--
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]
