viirya commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3806187616
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -128,6 +138,210 @@ 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. When `exact` is true the match is
always case-sensitive,
+ // regardless of `caseSensitive`; otherwise it honors `caseSensitive`.
Returns the child type
+ // together with its actual physical name so callers build paths from the
on-disk names.
+ //
+ // Variant object keys must be matched `exact = true`: they are data, not
Spark identifiers, and
+ // the reader resolves them case-sensitively (VariantSchema.objectSchemaMap
and
+ // Variant.getFieldByKey use exact equals). A file may legally shred sibling
keys differing only
+ // in case (e.g. `A` and `a`), so a case-insensitive first-match could bind
the predicate to the
+ // wrong physical subtree and skip a row group that holds matching rows --
silent data loss. The
+ // top-level variant column name is a Spark identifier and is matched by
`caseSensitive` (in
+ // `shreddedVariantEntries`); the structural `typed_value`/`value` names are
fixed, so `exact` is
+ // used for them too.
+ private def findChild(group: GroupType, name: String, exact: Boolean):
Option[Type] = {
+ group.getFields.asScala.find { f =>
+ if (exact || 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. `value` is a fixed structural
name; matched exact.
+ private def residualIn(group: GroupType): Option[String] =
+ findChild(group, VALUE, exact = true).collect {
+ case p: PrimitiveType if p.getRepetition != Repetition.REPEATED =>
p.getName
+ }
+
+ // Copy of `getNormalizedLogicalType` from the `nameToParquetField` closure,
needed here for the
Review Comment:
Done in 20ff8d1: hoisted `getNormalizedLogicalType` to a shared class-scope
`def` so both pushdown paths use the same normalization.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -128,6 +138,210 @@ 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. When `exact` is true the match is
always case-sensitive,
+ // regardless of `caseSensitive`; otherwise it honors `caseSensitive`.
Returns the child type
+ // together with its actual physical name so callers build paths from the
on-disk names.
+ //
+ // Variant object keys must be matched `exact = true`: they are data, not
Spark identifiers, and
+ // the reader resolves them case-sensitively (VariantSchema.objectSchemaMap
and
+ // Variant.getFieldByKey use exact equals). A file may legally shred sibling
keys differing only
+ // in case (e.g. `A` and `a`), so a case-insensitive first-match could bind
the predicate to the
+ // wrong physical subtree and skip a row group that holds matching rows --
silent data loss. The
+ // top-level variant column name is a Spark identifier and is matched by
`caseSensitive` (in
+ // `shreddedVariantEntries`); the structural `typed_value`/`value` names are
fixed, so `exact` is
+ // used for them too.
+ private def findChild(group: GroupType, name: String, exact: Boolean):
Option[Type] = {
+ group.getFields.asScala.find { f =>
+ if (exact || 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. `value` is a fixed structural
name; matched exact.
+ private def residualIn(group: GroupType): Option[String] =
+ findChild(group, VALUE, exact = true).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`). Object
keys and the structural
+ // typed_value/value names are matched case-sensitively (variant keys are
data; see `findChild`).
+ // A value for the path can only be hiding in one of these residual `value`
columns when the typed
+ // leaf is NULL, so IS NOT 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, exact = true) match {
+ case Some(g: GroupType) => g
+ case _ => return None
+ }
+ val typedName = typedChild.getName
+ // Variant object keys are data, matched case-sensitively (see
`findChild`).
+ val keyChild = findChild(typedChild, keys(idx), exact = true) match {
+ case Some(g: GroupType) => g
+ case _ => return None
+ }
+ namePath = namePath ++ Array(typedName, keyChild.getName)
+ group = keyChild
+ residualIn(group).foreach(r => residuals += (namePath :+ r))
+ idx += 1
+ }
+ // The leaf is the typed_value of the last key group.
+ findChild(group, TYPED_VALUE, exact = true) match {
+ case Some(p: PrimitiveType) if p.getRepetition != Repetition.REPEATED =>
+ val leaf = ParquetPrimitiveField(namePath :+ p.getName,
+ ParquetSchemaType(getNormalizedLogicalType(p),
p.getPrimitiveTypeName, p.getTypeLength))
+ Some(ShreddedVariantField(leaf, residuals.toSeq))
+ case _ => None
+ }
+ }
+
+ // Walk the variant-extraction schema alongside the physical Parquet group,
collecting
+ // logicalName -> ShreddedVariantField entries for shredded scalar object
paths that this file
+ // actually shreds. Only object-extraction, scalar-leaf paths are eligible;
array-index paths and
+ // synthetic (empty / placeholder / companion / full-variant passthrough)
paths resolve to None.
+ //
+ // `logicalParentNames` accumulates the logical field names (used to build
the map key that the
+ // pushed filter references); `physParentNames` accumulates the on-disk
field names (used to build
+ // the physical Parquet column paths). They differ only in case under
case-insensitive matching.
+ private def shreddedVariantEntries(
+ variantFields: Seq[StructField],
+ physGroup: GroupType,
+ logicalParentNames: Array[String],
+ physParentNames: Array[String]): Seq[(String, ShreddedVariantField)] = {
+ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ variantFields.flatMap { field =>
+ val physChildOpt = physGroup.getFields.asScala.collectFirst {
+ case g: GroupType if
+ (if (caseSensitive) g.getName == field.name
+ else g.getName.equalsIgnoreCase(field.name)) => g
+ }
+ physChildOpt match {
+ case None => Nil
+ case Some(physChild) =>
+ val logicalColPath = logicalParentNames :+ field.name
+ val physColPath = physParentNames :+ physChild.getName
+ field.dataType match {
+ // Variant struct: each child is a requested extraction carrying
VariantMetadata.
+ case s: StructType if VariantMetadata.isVariantStruct(s) =>
+ s.fields.toSeq.flatMap { extraction =>
+ if
(!extraction.metadata.contains(VariantMetadata.METADATA_KEY)) {
+ Nil
+ } else {
+ val meta = VariantMetadata.fromMetadata(extraction.metadata)
+ val segments = try { meta.parsedPath() } catch { case _:
Exception => null }
Review Comment:
Done in 20ff8d1: switched to `VariantPathParser.parse(meta.path)` directly,
dropping the catch-all and the null sentinel.
--
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]
