dongjoon-hyun commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3802245192
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -692,6 +914,69 @@ 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 = {
Review Comment:
`valueMatchesParquetType` only checks representation compatibility, so a
predicate whose extraction target type is *narrower* than the shredded leaf is
still pushed (e.g. a `JShort` literal against a plain INT32 leaf when the file
shreds `a` as `int` but the query asks `variant_get(v, '$.a', 'smallint')`;
same for decimals, where only the scale is checked).
This makes row-group skipping observably change results in one case: with
the default `spark.sql.variant.pushVariantIntoScan.deferCastError=false`, the
scan-side strict cast raises `INVALID_VARIANT_CAST` eagerly, even for rows the
filter would reject. If a row group holds only out-of-range typed values
(residuals all null), e.g. `{"a":100000}` with `WHERE
variant_get(v,'$.a','smallint') = 5S`, the leaf min/max excludes the literal
and the row group is skipped, so the query returns empty where it previously
threw.
To be fair, a pushed filter on a *different* regular column could already
skip the same row group and suppress the same error pre-PR, so this may be
acceptable-by-design — but since the PR description claims "query results are
identical", it seems worth either requiring the extraction type to match the
leaf type width, or explicitly documenting this as accepted behavior.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2422,6 +2424,315 @@ abstract class ParquetFilterSuite extends ParquetTest
with SharedSparkSession {
}
}
}
+
+ //
----------------------------------------------------------------------------------------------
+ // Shredded-variant filter pushdown (SPARK-55817).
+ //
+ // PushVariantIntoScan rewrites variant_get(v, '$.a', 'bigint') > 999 into a
struct-field access
+ // "v.`0`" > 999 where "0" carries VariantMetadata for path "$.a".
ParquetFilters maps that
+ // logical path to the physical shredded leaf v.typed_value.a.typed_value
and, for soundness,
+ // conjoins IS NULL on every residual `value` column along the path.
Review Comment:
This header comment says the implementation "conjoins IS NULL on every
residual `value` column" — but that is exactly the `and(leaf,
isNull(residual))` shape that the comment in
`ParquetFilters.makeShreddedFilter` proves unsound. The actual implementation
ORs `IS NOT NULL` guards: `or(leaf, isNotNull(residual)...)`. Wrong connective
and wrong polarity; a future reader "aligning" code to this description would
reintroduce the exact flaw of #54598. The suite-level doc in
`VariantShreddingFilterPushdownSuite` has the correct wording.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:
##########
@@ -210,6 +210,15 @@ class ParquetFileFormat
val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
val pushDownInFilterThreshold =
sqlConf.parquetFilterPushDownInFilterThreshold
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+ // When shredded-variant predicate pushdown is enabled, `requiredSchema`
may carry the
+ // variant-extraction structs produced by PushVariantIntoScan. Passing it
lets ParquetFilters
+ // map logical paths like "v.`0`" to the physical shredded columns for
row-group skipping.
+ val variantExtractionSchema =
Review Comment:
`Some(requiredSchema)` is passed whenever the config is on (the default),
with no check that the schema contains any variant-extraction struct. Since
`nameToShreddedVariantField` is a strict `val`, `shreddedVariantEntries` walks
`requiredSchema` against the physical group during `ParquetFilters`
construction — per file, per task — for every DSv1 Parquet scan, the
overwhelming majority of which have no variant columns. Under case-insensitive
analysis the empty result is also wrapped in `CaseInsensitiveMap`, which
lowercases the name on every comparison-predicate lookup.
Could we compute once on the driver whether `requiredSchema` actually
contains a `VariantMetadata.isVariantStruct` struct (e.g. via
`existsRecursively`) and pass `None` otherwise? That keeps the per-file cost at
zero for the common case.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.AccumulatorContext
+
+/**
+ * End-to-end tests for row-group skipping on shredded Variant columns in
Parquet (SPARK-55817).
+ *
+ * When a Variant column is written with shredding enabled, each extracted
scalar field is stored
+ * as a typed Parquet leaf column (e.g. `v.typed_value.a.typed_value` for
`$.a`) carrying min/max
+ * statistics. On the DSv1 path, PushVariantIntoScan rewrites
+ * `variant_get(v, '$.a', 'bigint') > 999` into a struct-field access `v.`0` >
999`, and (when
+ * `spark.sql.variant.shreddedPredicatePushdown.enabled` is true)
ParquetFilters maps `v.`0`` to
+ * the physical leaf and OR-s in an IS NOT NULL guard on every untyped
residual `value` column
+ * along the path.
+ *
+ * Scope: the optimization fires on the DSv1 read path only. On the DSv2 path
variant extraction is
+ * pushed through the separate SupportsPushDownVariantExtractions mechanism,
and the filter is never
+ * rewritten into `v.`0``, so it cannot be pushed for row-group skipping (see
the comment in
+ * ParquetScanBuilder). DSv2 reads remain correct -- the variant filter is
applied post-scan -- they
+ * just do not skip row groups. These tests therefore assert skipping only on
DSv1, and assert
+ * correctness on both DSv1 and DSv2.
+ *
+ * The central correctness concern is soundness under fallback: shredding is
per-row and per-file
+ * best-effort, so values that don't fit the shredded type (overflow / type
mismatch) or that are
+ * in a file that doesn't shred the path are stored in an opaque residual with
`typed_value` NULL.
+ * Parquet min/max excludes NULLs, so a naive leaf-only predicate could skip a
row group that still
+ * holds a matching row. These tests mix typed and fallback rows in a single
row group and assert
+ * that no matching row is ever dropped and results equal the no-pushdown
baseline.
+ */
+class VariantShreddingFilterPushdownSuite extends QueryTest with ParquetTest
+ with SharedSparkSession {
+
+ // Base configs to write shredded Variant Parquet files.
+ private def writeConf(forceSchema: String): Seq[(String, String)] = Seq(
+ SQLConf.VARIANT_WRITE_SHREDDING_ENABLED.key -> "true",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true",
+ SQLConf.VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key -> forceSchema,
+ // Keep the physical group unannotated so the schema is a plain shredded
struct.
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> "false")
Review Comment:
Every write in this suite forces
`spark.sql.parquet.variant.annotateLogicalType.enabled=false`, but that config
defaults to `true` — so the end-to-end path (resolution + actual row-group
skip) is never exercised against the annotated variant layout that
default-configured writers produce. If the annotated group ever behaves
differently in the resolver, skipping would silently stop firing (or misfire)
in exactly the default production layout while this suite stays green. Could we
add at least one annotated-layout run of the skip + fallback tests?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.AccumulatorContext
+
+/**
+ * End-to-end tests for row-group skipping on shredded Variant columns in
Parquet (SPARK-55817).
+ *
+ * When a Variant column is written with shredding enabled, each extracted
scalar field is stored
+ * as a typed Parquet leaf column (e.g. `v.typed_value.a.typed_value` for
`$.a`) carrying min/max
+ * statistics. On the DSv1 path, PushVariantIntoScan rewrites
+ * `variant_get(v, '$.a', 'bigint') > 999` into a struct-field access `v.`0` >
999`, and (when
+ * `spark.sql.variant.shreddedPredicatePushdown.enabled` is true)
ParquetFilters maps `v.`0`` to
+ * the physical leaf and OR-s in an IS NOT NULL guard on every untyped
residual `value` column
+ * along the path.
+ *
+ * Scope: the optimization fires on the DSv1 read path only. On the DSv2 path
variant extraction is
+ * pushed through the separate SupportsPushDownVariantExtractions mechanism,
and the filter is never
+ * rewritten into `v.`0``, so it cannot be pushed for row-group skipping (see
the comment in
+ * ParquetScanBuilder). DSv2 reads remain correct -- the variant filter is
applied post-scan -- they
+ * just do not skip row groups. These tests therefore assert skipping only on
DSv1, and assert
+ * correctness on both DSv1 and DSv2.
+ *
+ * The central correctness concern is soundness under fallback: shredding is
per-row and per-file
+ * best-effort, so values that don't fit the shredded type (overflow / type
mismatch) or that are
+ * in a file that doesn't shred the path are stored in an opaque residual with
`typed_value` NULL.
+ * Parquet min/max excludes NULLs, so a naive leaf-only predicate could skip a
row group that still
+ * holds a matching row. These tests mix typed and fallback rows in a single
row group and assert
+ * that no matching row is ever dropped and results equal the no-pushdown
baseline.
+ */
+class VariantShreddingFilterPushdownSuite extends QueryTest with ParquetTest
+ with SharedSparkSession {
+
+ // Base configs to write shredded Variant Parquet files.
+ private def writeConf(forceSchema: String): Seq[(String, String)] = Seq(
+ SQLConf.VARIANT_WRITE_SHREDDING_ENABLED.key -> "true",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true",
+ SQLConf.VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key -> forceSchema,
+ // Keep the physical group unannotated so the schema is a plain shredded
struct.
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> "false")
+
+ /**
+ * Counts how many Parquet row groups are actually read by the given
DataFrame, using the
+ * accumulator technique from ParquetFilterSuite. Only meaningful with the
vectorized reader,
+ * which reports the row-group count into a registered NumRowGroupsAcc.
+ */
+ private def countRowGroupsRead(df: DataFrame): Int = {
+ val accu = new NumRowGroupsAcc
+ sparkContext.register(accu)
+ try {
+ df.foreachPartition((it: Iterator[Row]) => it.foreach(_ => accu.add(0)))
+ accu.value
+ } finally {
+ AccumulatorContext.remove(accu.id)
+ }
+ }
+
+ /**
+ * Writes a JSON-per-row Variant Parquet file coalesced to a single
partition with a tiny block
+ * size so the writer emits multiple row groups. `jsonExpr` is the SQL
expression producing the
+ * JSON string per `id` in `range(0, numRows, 1, 1)`.
+ */
+ private def writeShredded(
+ dir: File,
+ forceSchema: String,
+ jsonExpr: String,
+ numRows: Int,
+ blockSize: Int = 512): Unit = {
+ withSQLConf(writeConf(forceSchema): _*) {
+ spark.sql(
+ s"""SELECT parse_json($jsonExpr) AS v
+ |FROM range(0, $numRows, 1, 1)""".stripMargin)
+ .coalesce(1)
+ .write
+ .option("parquet.block.size", blockSize)
+ .mode("overwrite")
+ .parquet(dir.getAbsolutePath)
+ }
+ }
+
+ // Run `block` with pushdown enabled, across the {DSv1, DSv2} x {vectorized,
non-vectorized} grid.
+ // `dsv1` is passed so a test can assert row-group skipping only on the DSv1
path.
+ private def forEachReader(block: (Boolean, Boolean) => Unit): Unit = {
+ Seq("parquet" -> true, "" -> false).foreach { case (useV1, dsv1) =>
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> useV1,
+ SQLConf.VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true",
Review Comment:
One more untested config interaction: with
`spark.sql.variant.pushVariantIntoScan.deferCastError=true`, strict
`variant_get` is rewritten into `UnwrapVariantCastError(...)`, which
`PushableColumnBase` cannot translate into a source `Filter` — so the
optimization silently never fires for that combination. Results stay correct
(performance-only), but the grid here never varies `deferCastError`, so nothing
would catch a regression either way. Worth a test, and perhaps a sentence in
the new config's `.doc()`.
##########
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:
Rather than keeping a verbatim copy, could we hoist the closure-local
`getNormalizedLogicalType` out of the `nameToParquetField` initializer to class
scope and share it? A `def` has no initialization-order constraint, so the
closure can call it directly. Two copies of the SPARK-40280 normalization can
drift silently — if one gains a new case, shredded and regular pushdown would
disagree on the same physical type with no error and no failing test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -692,6 +914,69 @@ 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.
+ private def referencesShreddedName(predicate: sources.Filter): Boolean =
predicate match {
Review Comment:
`sources.Filter` already exposes `references`, which recurses through
`And`/`Or`/`Not`, so this whole match can be:
```scala
private def referencesShreddedName(predicate: sources.Filter): Boolean =
predicate.references.exists(nameToShreddedVariantField.contains)
```
Since this is the soundness guard against pushing negated shredded
predicates, the hand-enumerated list is a bit fragile: a `Filter` subtype added
later (or missed today) silently falls into `case _ => false`. Today a miss
happens to be saved by the fact that shredded logical names never appear in
`nameToParquetField`, but nothing states or tests that invariant.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -718,6 +1003,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) =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, value)))
+ case sources.EqualNullSafe(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, value)))
+ case sources.LessThan(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeLt.lift(t).map(_(n, value)))
+ case sources.LessThanOrEqual(name, value) if
canMakeShreddedFilterOn(name, value) =>
+ makeShreddedFilter(name, (t, n) => makeLtEq.lift(t).map(_(n, value)))
+ case sources.GreaterThan(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeGt.lift(t).map(_(n, value)))
+ case sources.GreaterThanOrEqual(name, value) if
canMakeShreddedFilterOn(name, value) =>
+ makeShreddedFilter(name, (t, n) => makeGtEq.lift(t).map(_(n, value)))
+ case sources.In(name, values) if pushDownInFilterThreshold > 0 &&
values.nonEmpty &&
+ values.forall(v => canMakeShreddedFilterOn(name, v)) =>
+ // Convert `In` to the OR of per-value equalities, each already OR-ed
with the residual
+ // isNotNull guards, then combine. Reuses the same soundness guard as
the comparison
+ // predicates (the repeated residual disjuncts are harmless).
+ val distinct = values.distinct
+ if (distinct.length <= pushDownInFilterThreshold) {
+ distinct.flatMap { v =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, v)))
+ }.reduceLeftOption(FilterApi.or)
+ } else {
+ None
Review Comment:
Two asymmetries with the regular `In` path below:
1. Above the threshold this returns `None`, while the regular path falls
back to `makeInPredicate` (`FilterApi.in`). `or(in(leaf, set),
isNotNull(residual)...)` would be equally sound, so large IN lists — the
workloads that benefit most from skipping — currently get nothing. Also, the
threshold here is measured on `distinct.length` while the regular path uses
`values.length`.
2. Under the threshold, each of the N values re-folds the residual
`isNotNull` guards, producing N×R redundant `notEq` nodes. Building the OR of
plain leaf equalities first and appending the R guards once is semantically
identical under Parquet's OR-droppability rule and keeps the predicate tree
(and pushed-filter EXPLAIN output) small.
##########
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:
nit: `VariantPathParser.parse(meta.path)` already returns an `Option`, and
`parsedPath()` is just that plus a throw. Using it directly avoids the
catch-all (`case _: Exception => null`), which would also swallow unrelated
exceptions into a silent no-push, and drops the `null` sentinel:
```scala
VariantPathParser.parse(meta.path) match {
case None => Nil
case Some(segments) => ...
}
```
--
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]
