viirya commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3815262886
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7037,6 +7037,31 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED =
+ buildConf("spark.sql.variant.shreddedPredicatePushdown.enabled")
+ .internal()
+ .doc("When true, comparison predicates on shredded Variant fields
produced by " +
+ "PushVariantIntoScan (e.g. variant_get(v, '$.a', 'bigint') > 999) are
pushed to Parquet " +
+ "as the predicate on the physical shredded typed_value leaf column
OR-ed with an " +
+ "IS NOT NULL check on every untyped residual value column along the
path, so that a row " +
+ "group is skipped only when the leaf cannot match and every residual
is entirely null " +
+ "(i.e. the whole path is provably in the typed leaf). This enables
row-group skipping " +
+ "for shredded Variant columns while never dropping rows that fall back
to an untyped " +
+ "residual. The benefit depends on the data layout, like any Parquet
min/max skipping: it " +
+ "helps most when the data is sorted on the filtered field (so each row
group covers a " +
+ "narrow value range) and a file holds many row groups; unsorted data
or a single row " +
+ "group per file gains little. Has no effect unless the Parquet column
is shredded and " +
+ "spark.sql.variant.pushVariantIntoScan is also true, and it does not
fire when " +
+ "spark.sql.variant.pushVariantIntoScan.deferCastError is true (the
extraction is " +
+ "rewritten into a form that is not translated to a pushable filter).
Results are " +
+ "unaffected either way; this only controls whether row groups can be
skipped.")
+ .version("4.3.0")
Review Comment:
Fixed in 4fce3cf -- `branch-4.3` is already cut, so this ships first in
`branch-4.x` (4.4.0). Changed to `.version("4.4.0")`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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
Review Comment:
Fixed in 4fce3cf. Rewrote the fallback tests so the guard is load-bearing:
`a bigint` leaf with a fallback the int64 leaf can't hold (`1500.5` /
`"1500"`), so the leaf min/max can't match and only the guard keeps the row
group. I verified your check -- reducing `makeShreddedFilter` to the leaf-only
predicate now fails `residual fallback beyond the leaf's min/max is not
dropped` (data loss), and neutering the `Not` guard fails the negation test.
Used your exact `1500.5` case, thanks.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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. `annotate` controls
whether the physical
+ // variant group carries the VARIANT logical-type annotation (the production
default is true).
+ private def writeConf(forceSchema: String, annotate: Boolean): 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,
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> annotate.toString)
+
+ /**
+ * 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,
+ annotate: Boolean = false): Unit = {
+ withSQLConf(writeConf(forceSchema, annotate): _*) {
+ 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",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString,
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true") {
+ withClue(s"(dsv1=$dsv1, vectorized=$vectorized) ") {
+ block(dsv1, vectorized)
+ }
+ }
+ }
+ }
+ }
+
+ // Read the same query with pushdown disabled: the baseline that must never
lose rows.
+ private def baseline(read: => DataFrame): Seq[Row] = {
+ withSQLConf(
+ SQLConf.VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED.key -> "false",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true") {
+ read.collect().toSeq
+ }
+ }
+
+ test("overflow fallback: matching row in residual is not dropped") {
+ withTempDir { dir =>
+ // `a` shredded as tinyint. id 0..49 -> a=id (fits, typed). id 50 ->
a=1500 (overflows
+ // tinyint, stored in the residual with typed_value NULL). All 51 rows
fit one row group
+ // (blockSize large enough), so the typed leaf stats are min=0,max=49; a
leaf-only `a > 999`
+ // would drop the row group and lose the 1500 row.
+ val jsonExpr =
+ "case when id = 50 then '{\"a\":1500}' else '{\"a\":' || id || '}' end"
+ writeShredded(dir, "a tinyint", jsonExpr, numRows = 51, blockSize = 1024
* 1024)
Review Comment:
Fixed in 4fce3cf, same change as finding 2 -- the `a tinyint` + `bigint`
mismatch that made these push nothing is gone; they now use a `bigint` leaf
with a non-int-representable fallback (and safe widening from finding 5 also
removes that mismatch class). The negation test uses
`{"a":500.5}`/`{"a":600.5}` so both rows are all-residual with a pushable path,
and it fails with the `Not` guards removed.
##########
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:
Great catch -- adopted your tighter guard in 4fce3cf: `or(leaf,
and(anyResidualNotNull, isNull(leaf)))`. I re-derived the soundness (a value is
outside the typed leaf only on a row where the leaf is NULL, so zero leaf nulls
=> the leaf min/max is a complete summary regardless of residual contents) and
added a partial-object test (`{"a": id, "z": "..."}`) that skips on DSv1 where
the flat OR couldn't. Agreed this makes keeping the default on the better call.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -128,6 +126,256 @@ 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`.
+ //
+ // Lazy so it is computed after the `Parquet*Type` vals below are
initialized (resolution reads
+ // them via `expectedLeafType`); a strict val here would see them as null
under Scala's
+ // declaration-order initialization.
+ private lazy 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
+ }
+
+ // Shared by both the `nameToParquetField` traversal and the shredded leaf
resolution so the two
+ // pushdown paths normalize physical types identically.
+ private def getNormalizedLogicalType(p: PrimitiveType):
LogicalTypeAnnotation = {
+ // SPARK-40280: Signed 64 bits on an INT64 and signed 32 bits on an INT32
are optional, but
+ // the rest of the code here assumes they are not set, so normalize them
to not being set.
+ (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],
+ targetType: DataType): 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 leafType =
+ ParquetSchemaType(getNormalizedLogicalType(p),
p.getPrimitiveTypeName, p.getTypeLength)
+ // Require the extraction's target type to map to the exact physical
leaf type. Comparing on
+ // representation alone (as `valueMatchesParquetType` does for the
literal) would push a
+ // narrower extraction such as smallint against an int leaf: the leaf
min/max is over int
+ // values, so a row group holding only out-of-range values (residuals
null) would be
+ // skipped, changing an eager INVALID_VARIANT_CAST into an empty
result. Requiring an exact
+ // type match keeps the optimization result-preserving.
+ if (!expectedLeafType(targetType).contains(leafType)) {
+ None
+ } else {
+ val leaf = ParquetPrimitiveField(namePath :+ p.getName, leafType)
+ Some(ShreddedVariantField(leaf, residuals.toSeq))
+ }
+ case _ => None
+ }
+ }
+
+ // The physical Parquet leaf type a shredded scalar of `targetType` is
written as, matching
+ // `SparkShreddingUtils.variantShreddingSchema` (which writes the scalar's
natural type) and the
+ // `Parquet*Type` normalization used for the leaf. Returns None for types
that are not shredded as
+ // a comparable scalar leaf (or that this pushdown does not handle), so the
path is not pushed.
+ private def expectedLeafType(targetType: DataType):
Option[ParquetSchemaType] = targetType match {
Review Comment:
Fixed in 4fce3cf. `resolveShredded` now accepts a narrower signed-integer
leaf than the target (`isSafeIntegerWidening`: byte->short/int/long,
short->int/long, int->long); narrowing stays rejected and out-of-range literals
are still refused by `valueMatchesParquetType`. Added a widening unit test.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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. `annotate` controls
whether the physical
+ // variant group carries the VARIANT logical-type annotation (the production
default is true).
+ private def writeConf(forceSchema: String, annotate: Boolean): 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,
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> annotate.toString)
+
+ /**
+ * 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,
+ annotate: Boolean = false): Unit = {
Review Comment:
Fixed in 4fce3cf -- `writeShredded`'s `annotate` now defaults to `true` (the
production layout), and one test passes `annotate = false` explicitly for the
unannotated case.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -128,6 +126,256 @@ 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`.
+ //
+ // Lazy so it is computed after the `Parquet*Type` vals below are
initialized (resolution reads
+ // them via `expectedLeafType`); a strict val here would see them as null
under Scala's
+ // declaration-order initialization.
+ private lazy 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] = {
Review Comment:
Fixed in 4fce3cf -- dropped the `exact` parameter; `findChild` now always
compares with `==`, so the exact-case rule holds by construction. The top-level
column name is still matched by `caseSensitive` in `shreddedVariantEntries`.
--
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]
