peter-toth commented on code in PR #58050: URL: https://github.com/apache/spark/pull/58050#discussion_r3814611364
########## 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: **Finding 2.** The residual `IS NOT NULL` guard is the whole soundness argument of this PR, and nothing in the suite makes it decide anything. I patched `makeShreddedFilter` on this head down to the leaf-only predicate -- exactly the #54598 shape this PR exists to fix: ```scala makeLeaf(field.leaf.fieldType, field.leaf.fieldNames) ``` and re-ran: all 9 tests here and all 13 new `shredded variant filter:` tests in `ParquetFilterSuite` pass. Three of them push nothing at all (finding 3). `type-mismatch fallback` and `multi-level $.a.b: fallback at an intermediate level` do push, but in both the leaf min/max can match the literal on its own (leaf max 1018 for `a > 1005`; 5000 for `b > 999`), so the row group is kept whether or not the guard is there. For the guard to be load-bearing the fallback row has to be the only match *and* the leaf stats must be unable to match the literal. Measured on this head: ```scala test("residual fallback beyond the leaf's min/max is not dropped") { withTempDir { dir => // `a` shredded as bigint. id 0..49 -> a = id (typed leaf, min 0 max 49). id 50 -> a = 1500.5, // a decimal the int64 leaf cannot hold, so it lands in v.typed_value.a.value with typed_value // NULL. One row group: `a > 999` matches only that row and the leaf min/max cannot match it, // so only the IS NOT NULL guard keeps the row group. val jsonExpr = "case when id = 50 then '{\"a\":1500.5}' else '{\"a\":' || id || '}' end" writeShredded(dir, "a bigint", jsonExpr, numRows = 51, blockSize = 1024 * 1024) def read: DataFrame = spark.read.parquet(dir.getAbsolutePath) .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a") .where("a > 999") forEachReader { (_, _) => checkAnswer(read, Seq(Row(1500L))) } } } ``` With the guard: `[1500]`, 1 row group read. Leaf-only: `[]`, 0 row groups read. A double (`1.5005e3`) or a string (`"1500"`) fallback behaves the same. Note `1500.0` does *not* work -- an integral decimal gets shredded into the int64 leaf, so the leaf max becomes 1500 and the row group survives anyway. ########## 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: **Finding 1.** `.version("4.3.0")` doesn't match any branch this can first ship in. `branch-4.3` is already cut and sits at `4.3.0-SNAPSHOT`, `branch-4.x` is `4.4.0-SNAPSHOT`, and `master` is `5.0.0-SNAPSHOT`. A new improvement on `master` that gets the usual backport first ships in 4.4.0: ```suggestion .version("4.4.0") ``` (5.0.0 if you mean this to be master-only, but that seems unlikely for a perf-only change.) ########## 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: **Finding 3.** `a tinyint` shreds `$.a` as `optional int32 typed_value (INTEGER(8,true))` while the extraction here is `bigint`. Since 20ff8d1 `resolveShredded` requires `expectedLeafType(targetType)` to equal the physical leaf type exactly, and `expectedLeafType(LongType)` is `ParquetLongType` = `(null, INT64, 0)`, not `ParquetByteType` = `(INT(8,true), INT32, 0)` -- so the path resolves to nothing and no filter is pushed. Verified on this head by building `ParquetFilters` over the written file's footer schema: ``createFilter(GreaterThan("v.`0`", 999L))`` returns `None`. So this test, `negated predicate over an all-fallback row group is not dropped`, and the overflow half of `annotated variant layout` all assert about a scan with no pushed predicate -- `countRowGroupsRead(read) == countRowGroupsRead(all)` is trivially true. The negation one matters most: it is the only end-to-end coverage of the `Not` refusal, and removing `referencesShreddedName` plus both `Not` guards leaves it green. These predate the type gate (they were written against @dongjoon-hyun's original repro). The fix is to make the extraction type match the shredded type -- keep `a tinyint` and query `variant_get(v, '$.a', 'tinyint')`, or shred `a bigint` and use a fallback the int64 leaf can't hold (finding 2). For the negation test, `a bigint` with both rows stored as decimals (`{"a":500.5}` / `{"a":600.5}`) should reproduce the original all-residual row group with a pushable path; worth confirming 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: **Finding 4.** This flat OR of `isNotNull(residual)` can never drop a row group once any row's object carries a key outside the shredding schema. `VariantShreddingWriter.castShredded` puts the non-shredded keys of a level into that level's own `value` as a partial object (`result.addVariantValue(...)`), so `v.value` is non-null on every row and `notEq(v.value, null)` is never droppable. That is the normal layout for real Variant data -- the inferred shredding schema is capped at `spark.sql.variant.shredding.maxSchemaWidth` fields, and one extra key anywhere along the path is enough. Measured on this head (DSv1, vectorized, 2000 rows, `parquet.block.size=512`, `a bigint`, `variant_get(v, '$.a', 'bigint') > 999`): | data | row groups read, filtered / all | |---|---| | `{"a": id}` | 10 / 20 | | `{"a": id, "z": "xyzxyzxyz"}` | **20 / 20** | So on that layout the optimization is on by default, cannot skip anything, and still pays the pushed-predicate cost -- the ~3% @qlong measured and the `Can skip no row groups` row of the new benchmark. Some of that cost is more than statistics evaluation: `SpecificParquetRecordReaderBase.java:292` reads row groups through `reader.readNextFilteredRowGroup()`, so with a filter present parquet-mr also loads the ColumnIndex/OffsetIndex for the leaf *and every residual column* per row group and computes row ranges. A tighter guard is sound and fixes it. A value for the path can only be outside the typed leaf on a row where the leaf is NULL, so conjoin the residual guards with `isNull(leaf)` instead of OR-ing them in flat: ```scala val field = nameToShreddedVariantField(name) makeLeaf(field.leaf.fieldType, field.leaf.fieldNames).map { leafPredicate => val guards: Seq[FilterPredicate] = field.residualFieldNames.map { n => FilterApi.notEq(binaryColumn(n), null.asInstanceOf[Binary]) } val leafIsNull = makeEq.lift(field.leaf.fieldType).map(_(field.leaf.fieldNames, null)) (guards.reduceLeftOption[FilterPredicate](FilterApi.or(_, _)), leafIsNull) match { case (None, _) => leafPredicate case (Some(anyResidual), Some(isNull)) => FilterApi.or(leafPredicate, FilterApi.and(anyResidual, isNull)) case (Some(_), None) => guards.foldLeft(leafPredicate)(FilterApi.or(_, _)) } } ``` `and(x, y)` is droppable iff *either* side is, so this drops iff the leaf min/max cannot match AND (every residual is entirely NULL OR the leaf column has zero nulls). The second arm is the new one: zero nulls in the leaf means every row's value for the path is in the leaf, so nothing can hide regardless of what the residuals hold. Per record it still evaluates true for every row that could match -- a row whose value is in a residual has a non-null residual and a NULL leaf, so both conjuncts hold. Measured with that patch applied to this head: `{"a": id, "z": ...}` goes to 10 / 20, `{"a": id}` stays at 10 / 20, the finding-2 fallback row still comes back as `[1500]` for all four fallback encodings, and all 79 tests in `VariantShreddingFilterPushdownSuite` + `ParquetV1FilterSuite` pass. With this in, keeping the config on by default reads much better than the on/off choice you and @qlong were weighing: the cost stops landing on queries that structurally can't benefit. ########## 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: **Finding 5.** Requiring exact type identity also rejects the *widening* direction, which is sound and is the shape users actually write: `variant_get(v, '$.a', 'bigint')` against a file that shreds `a` as `int` (or `smallint`/`tinyint`) resolves to nothing. Only narrowing has the `INVALID_VARIANT_CAST`-suppression problem @dongjoon-hyun described -- every value in a narrower leaf casts to a wider target without error, and the leaf's ordering is preserved. `valueMatchesParquetType` already refuses a literal outside the leaf's range (`case v: JLong => v.longValue() >= Int.MinValue && ...`), so an out-of-range comparison still isn't pushed. Allowing the leaf to be narrower within the integer family would recover it: ```scala // in resolveShredded, replacing the exact `contains` check val target = expectedLeafType(targetType) if (!target.contains(leafType) && !isSafeWidening(leafType, target)) None else ... ``` with `isSafeWidening` accepting `ParquetByteType -> Short/Integer/Long`, `ParquetShortType -> Integer/Long`, `ParquetIntegerType -> Long`. Non-blocking -- it is a lost optimization, not a correctness issue -- but it is also what makes the three tests in finding 3 vacuous, so it is worth deciding deliberately rather than by accident. ########## 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: **Finding 6.** `annotate` defaults to `false` here while `spark.sql.parquet.variant.annotateLogicalType.enabled` defaults to `true`, so 8 of the 9 tests write a layout no default-configured writer produces. Flipping this default to `true` and passing `annotate = false` explicitly in one test would put the coverage the right way round. ########## 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: **Finding 7.** `exact` is `true` at all four call sites (`residualIn`, and both `findChild` calls in `resolveShredded`), so `|| caseSensitive` is unreachable and the second half of the comment ("otherwise it honors `caseSensitive`") describes behaviour that can't happen -- the top-level column name is matched by the separate inline predicate in `shreddedVariantEntries`, not here. Dropping the parameter and comparing with `==` would make the exact-case rule you documented unconditional by construction. -- 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]
