[SPARK-55817][SQL] Enable Parquet row-group skipping for shredded Variant columns

[viirya]

What changes were proposed in this pull request?

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. Today ParquetFilters cannot map v.0`` to a physical column, so the predicate is dropped and no row-group skipping happens for shredded-variant queries.

This PR maps such predicates to the physical shredded leaf and enables row-group skipping, as a pure performance win with no behavior change.

The subtlety: shredding is per-row and per-file best-effort. Values that don't fit the shredded type (overflow / type mismatch) or fields not shredded in a given file are stored in opaque untyped value residual columns with typed_value NULL. Parquet min/max excludes NULLs, so pushing the predicate on the typed leaf alone would let a row group be skipped while it still contains matching rows in a residual -- silently dropping data. (A prior attempt, #54598, did exactly this and was unsound.)

To stay sound, the pushed predicate is:

or(leafPredicate, and(anyResidualNotNull, isNull(leaf)))

where anyResidualNotNull is or(notEq(residual_0, null), ..., notEq(residual_n, null)) over the leaf's sibling value and every ancestor level's value up to the top-level v.value, and isNull(leaf) is eq(leaf, null). Parquet drops a row group for an or only when both sides are droppable; an and when either is; notEq(col, null) only when the column is entirely NULL; eq(col, null) only when the column has no nulls. So a row group is skipped only when the leaf min/max cannot match and (every residual is entirely NULL or the leaf column has no nulls) -- i.e. the whole path is provably in the typed leaf. The "leaf has no nulls" arm is what lets the common partial-object layout skip (a key outside the shredding schema leaves the top-level residual non-null on every row, yet the field is still fully shredded). Otherwise the row group is kept: worst case we lose the optimization, never correctness.

Details:

• New internal config spark.sql.variant.shreddedPredicatePushdown.enabled (version 4.4.0, default on), gating the optimization; it can be turned off if a soundness edge case surfaces.
• ParquetFilters gains an optional variantExtractionSchema parameter; when set, it resolves eligible scalar object-extraction paths ($.a, $.a.b) to the physical shredded leaf and residual chain, and emits the guarded predicate for Gt/GtEq/Lt/LtEq/Eq/EqualNullSafe/In. The extraction type must map to the leaf type or a safe integer widening of it (e.g. bigint over an int leaf); narrowing is rejected so an eager INVALID_VARIANT_CAST never becomes an empty result. Array-index paths and empty/$/companion/placeholder paths resolve to nothing. IsNull/IsNotNull on the logical field, and negated predicates, are out of scope.
• Wired on the DSv1 path (ParquetFileFormat).

Scope: DSv1 only. DSv2 does rewrite variant extractions into v.0`` struct accesses (V2ScanRelationPushDown.buildScanWithPushedVariants), but that runs after filter pushdown, so the filters offered to the Parquet scan builder are still `variant_get(v, ...)` predicates and cannot be pushed for row-group skipping. DSv2 reads remain correct (the variant filter is applied post-scan); they just don't skip row groups on shredded columns. This is noted in a code comment in `ParquetScanBuilder`.

Why are the changes needed?

Predicates on shredded Variant fields currently get no row-group skipping, so queries scan all row groups even when the shredded leaf statistics prove a group cannot match. This adds that skipping soundly, improving scan performance for selective filters on shredded Variant columns.

Does this PR introduce any user-facing change?

No result change. The optimization is enabled by default and only affects which Parquet row groups are read -- query results are identical. It is gated behind an internal config (spark.sql.variant.shreddedPredicatePushdown.enabled) that can be turned off.

How was this patch tested?

New unit tests in ParquetFilterSuite (both ParquetV1FilterSuite and ParquetV2FilterSuite): single/multi-level resolution, the guarded predicate shape (leaf + residual guards + isNull(leaf) arm), array-index rejection, synthetic (placeholder/companion/passthrough) fields resolving to nothing, absent-field, case-insensitive column matching with exact-case variant keys, negated predicate not pushed, exact type + safe widening pushed / narrowing and out-of-range rejected, and large In above the threshold.

New integration suite VariantShreddingFilterPushdownSuite, running across DSv1/DSv2 and vectorized/non-vectorized readers: residual fallback beyond the leaf's min/max not dropped (single- and multi-level, load-bearing), negated predicate over an all-fallback row group not dropped, type-mismatch fallback, file without the shredded path, residual-null happy-path skip, partial object with a non-shredded sibling key still skips (the isNull(leaf) arm), unannotated layout, deferCastError=true (strict non-string cast does not fire; try_variant_get still does), and strict variant_get preserving INVALID_VARIANT_CAST on a residual fallback rather than returning empty.

A benchmark, VariantShreddedPredicatePushdownBenchmark, measures skip-all / skip-some / skip-none / skip-some-on-partial-object; results generated by the GitHub Actions benchmark workflow.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

[dongjoon-hyun]

These shredded cases are reachable through the pre-existing generic case sources.Not(pred) recursion below, and under negation the pushed predicate becomes unsound.

A != predicate arrives as sources.Not(EqualTo("v.`0`", 700)). The guarded Not(EqualTo) fast path doesn't match (the shredded logical name is not in nameToParquetField), so the generic Not case recurses into the shredded EqualTo case here and wraps the result in FilterApi.not. parquet-mr's LogicalInverseRewriter then rewrites

not(or(eq(leaf, 700), notEq(residual, null)))

into

and(notEq(leaf, 700), eq(residual, null))

which is exactly the and(..., isNull(residual)) shape the comment on makeShreddedFilter proves unsound: StatisticsFilter drops an AND row group if any conjunct is droppable, and eq(residual, null) is droppable whenever the residual column has zero nulls.

Concrete repro: shred with a tinyint, one row group whose rows are {"a":500} and {"a":600} (both overflow tinyint, so both stored in the residual; residual nullCount = 0, typed leaf entirely null). WHERE variant_get(v, '$.a', 'bigint') != 700 skips the row group and returns an empty result instead of {500, 600} — silent data loss. The same hole exists for NOT IN and Not(EqualNullSafe/GreaterThan/...).

Since not(or(leaf, isNotNull(residual))) cannot be expressed soundly with row-group statistics, the shredded conversion must refuse to be produced under negation — e.g. have the generic sources.Not case return None when the child predicate references a shredded-variant logical name. It would also be good to add a != / NOT IN test with an all-fallback row group; the PR currently has no test covering a negated predicate.

[viirya]

Good catch, confirmed -- thank you. Not(EqualTo("v.0", 700)) recursed through the generic Not case into the shredded branch, and not(or(eq(leaf), notEq(residual))) gets rewritten by LogicalInverseRewriter into and(notEq(leaf), eq(residual, null)), which drops the row group whenever the residual has no nulls -- the exact unsound AND shape. Reproduced with your a tinyint / {500, 600} case.

Fixed in a91401f: a negated predicate that references a shredded-variant path is no longer pushed. Added referencesShreddedName and guard the generic Not in both createFilterHelper and convertibleFiltersHelper to return None; a non-negated shredded conjunct inside an AND still pushes. Added a unit test (!= / NOT IN / Not(Gt) not pushed, and the AND case) and an integration test that runs your != and NOT IN repro over an all-fallback row group and asserts {500, 600} come back.

[dongjoon-hyun]

The variant object-key segments (keys(idx)) are matched here through findChild, which honors spark.sql.caseSensitiveAnalysis — but variant field extraction is always exact-case at read time (SparkShreddingUtils.getFieldsToExtract uses schema.objectSchemaMap.get(key), and the residual fallback uses Variant.getFieldByKey, both plain equals). Variant keys are data, not Spark identifiers, so applying the identifier case-sensitivity config to them can bind the predicate to the wrong physical subtree.

Concrete unsound scenario with the default caseSensitive=false: a file legally shreds both keys A and a (variant keys are case-distinct, and Parquet allows sibling fields differing only in case — producible by an external spec-compliant writer or a forced shredding schema), with schema order [A, a]. For variant_get(v, '$.a', 'bigint') > 999, findChild first-match binds the leaf and the residual guards to the A subtree. In a row group where every row is fully shredded under a (all guarded residuals null, A leaf max <= 999), every disjunct is droppable, so the row group is skipped while v.typed_value.a.typed_value holds matching rows — silent data loss. Note the case-insensitive dedup in nameToShreddedVariantField doesn't help: it dedups logical map keys, not case-colliding physical siblings inside typed_value.

The key segments should be compared exact-case regardless of the config (mirroring objectSchemaMap); keeping case-insensitive matching for the top-level column-name segments and the structural typed_value/value names is fine.

[viirya]

Agreed, confirmed -- variant keys are data and the reader resolves them exact-case (objectSchemaMap.get / getFieldByKey), so applying caseSensitiveAnalysis to them can bind the predicate to the wrong subtree when a file shreds sibling keys differing only in case.

Fixed in a91401f: findChild takes an exact flag; object keys and the structural typed_value/value names are matched exact-case, while the top-level variant column name still honors caseSensitive (it is a Spark identifier). Added a unit test that a $.a request does not case-insensitively bind to a physical A subtree under caseSensitive=false.

[uros-b]

Thank you @viirya and @dongjoon-hyun for review!

[dongjoon-hyun]

Thank you for working on this. The core soundness design (or(leafPredicate, isNotNull(residual)...), the Not refusal, and the exact-case variant key matching) looks solid to me. I left a few comments below.

[dongjoon-hyun]

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.

[viirya]

Good point -- with the default deferCastError=false the eager strict cast makes this observable, so I treated it as a real result change. Fixed in 20ff8d1: resolveShredded now requires the extraction target type to map to the exact physical leaf type (expectedLeafType), so a narrower extraction such as smallint against an int leaf is no longer pushed and results stay identical. Timestamps are conservatively not pushed for now. Added a unit test (narrower not pushed, exact pushed).

[dongjoon-hyun]

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.

[viirya]

Fixed in 20ff8d1: ParquetFileFormat passes Some(requiredSchema) only when requiredSchema.existsRecursively(VariantMetadata.isVariantStruct), so non-variant DSv1 scans do no shredded traversal (and no CaseInsensitiveMap wrapping) per file.

[dongjoon-hyun]

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.

[viirya]

Fixed in 20ff8d1 -- the comment now says the implementation OR-s an IS NOT NULL guard on every residual, matching makeShreddedFilter.

[dongjoon-hyun]

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?

[viirya]

Added in 20ff8d1: an annotated-layout run (annotateLogicalType left at its default true) of both the skip test and the overflow-fallback test.

[dongjoon-hyun]

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.

[viirya]

Fixed in 20ff8d1: large IN lists above the threshold now push via FilterApi.in (or(in(leaf, set), isNotNull(residual)...)), the threshold is measured on values.length like the regular path, and the residual guards are appended once instead of once per value.

[dongjoon-hyun]

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().

[viirya]

Added in 20ff8d1: the grid now also varies deferCastError, asserting results stay correct when it is on (the optimization silently does not fire). Also added a sentence to the config's .doc() noting this.

[dongjoon-hyun]

sources.Filter already exposes references, which recurses through And/Or/Not, so this whole match can be:

privatedefreferencesShreddedName(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.

[viirya]

Done in 20ff8d1 -- referencesShreddedName is now predicate.references.exists(nameToShreddedVariantField.contains). Thanks, that removes the fragile hand-enumeration.

[dongjoon-hyun]

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.

[viirya]

Done in 20ff8d1: hoisted getNormalizedLogicalType to a shared class-scope def so both pushdown paths use the same normalization.

[dongjoon-hyun]

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:

VariantPathParser.parse(meta.path) match {
caseNone=>NilcaseSome(segments) => ...
}

[viirya]

Done in 20ff8d1: switched to VariantPathParser.parse(meta.path) directly, dropping the catch-all and the null sentinel.

[qlong]

Thanks @viirya for picking this up — the soundness handling (or(leaf, isNotNull(residual)...), the negation guard, exact-case variant-key matching) fixes the holes that made the original #54598 unsound.

The reason I didn't push #54598 further was I couldn't demonstrate measurable improvement. I tested your PR using https://github.com/cloudera-labs/variant-conformance-benchmark, the results (TPC-DS SF=5 flat variant + GHA 1-day event-payload, shreddedPredicatePushdown ON vs OFF, same build) does not show meaningful difference for three reasons for those two workloads:

1. Filtered fields aren't always shredded.
2. Predicates are mostly join keys, not literals.
3. One row group per file at test scale.

I do think the theorical lift of rowgroup skipping is very high, but it requires a few things line up:

1. the filtered field is shreddred
2. literal predicate
3. data sorted on that field
4. large number of rowgroup in a file

3 & 4 would require some delicated setup or tuning. have you run a benchmark showing measurable improvement on realistic data? That would significantly strengthen the case for carrying this soundness-critical code.

[viirya]

Thanks @qlong -- and you're right that the lift only shows up when the field is shredded, the predicate is a literal, the data is sorted on that field, and a file has many row groups, which is why the join-key / one-row-group-per-file workloads didn't move.

I added a benchmark (VariantShreddedPredicatePushdownBenchmark, 60e877a) that sets up exactly that layout -- a shredded bigint field sorted ascending, small block size so one file holds many row groups -- and measures a literal predicate on/off. Results (JDK 21, Apple M4 Max, 20M rows):

Can skip all row groups: 774ms -> 36ms 21.6x
Can skip some row groups: 841ms -> 46ms 18.3x
Can skip no row groups: 1003ms -> 1044ms 1.0x

So when the layout cooperates the skip is a large win, and the "skip no row groups" case shows negligible overhead (~4%, within noise) when nothing can be skipped. It's reproducible via build/sbt "sql/Test/runMain org.apache.spark.sql.execution.benchmark.VariantShreddedPredicatePushdownBenchmark".

[qlong]

The PR looks good to me.

The new benchmark illustrates the high ceiling of benefits when tuned , and almost no cost when not. Maybe we should document this "sort + small row group" tuning guidance in this PR or in Jira?

[viirya]

Thanks @qlong! Added the "sort + many row groups per file" tuning guidance to the config's doc string in 4bacd69, framed as the general Parquet min/max dependence: 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, and gains little on unsorted data or a single row group per file.

[qlong]

Thanks @viirya for updating the doc. Looking more closely at my test run against tpcds (SF=5) workload, I also noticed around 3% performance degradation when the flag is on, which is consistent with your own benchmark test. My test also runs on my laptop so that degradation could be just noise. Given the narrow applicability of this optimization, wondering if we should disable it by default if the 3% degradation turns out to be consistent?

[viirya]

Good question. I looked into where that ~3% comes from: it's the cost of Parquet evaluating the pushed predicate against each row group's statistics, not extra data being read -- the read schema and the number of row groups read are identical with the flag on and off (no residual columns get pulled in), and record-level filtering is off under the vectorized reader, so there's no per-row cost. It's the same kind of fixed cost any pushed-but-non-pruning filter has; a plain (non-variant) column shows a comparable skip-none overhead in my measurements.

Our predicate is a bit heavier than a plain one -- or(leaf, isNotNull(residual)...) is a few disjuncts across a few columns rather than a single comparison -- but those residual guards are what make the skip sound (they're the fix for the original data-loss hole), so there isn't a part of it I can drop without giving that up.

On the default: I'm inclined to keep it on, since the cost only lands on queries that both have a shredded Variant field and a literal predicate on it while the layout doesn't allow skipping, and it's small and within run-to-run noise. But I don't feel strongly -- if you or @dongjoon-hyun would rather ship it off and let sorted-layout workloads opt in, I'm fine flipping it.

[peter-toth]

Thanks for the PR, @viirya!

The or(leafPredicate, isNotNull(residual)...) shape is the right core idea, and @dongjoon-hyun's 11 findings all look resolved on this head. I built the PR head and measured the new suites with the guard patched out and with a tightened guard, which turned up three things. The residual guard -- the whole soundness argument -- has no effective coverage: reduce makeShreddedFilter to the leaf-only predicate (exactly the #54598 shape) and all 9 VariantShreddingFilterPushdownSuite tests plus all 13 new ParquetFilterSuite tests still pass, partly because three of the fallback tests push nothing at all against a tinyint leaf. And on the default-on question you and @qlong were weighing: the guard as written can never drop a row group once any row's object carries a key outside the shredding schema, which is the normal layout for real Variant data -- I measured 20/20 row groups read where the same data without the extra key reads 10/20 -- so that ~3% is paid by queries that structurally cannot benefit. A tightened guard fixes that (measured), which I think makes keeping it on the better call.

Blocking

• 1.Config version:.version("4.3.0") matches no branch this can first ship in -- branch-4.3 is already cut, branch-4.x is 4.4.0-SNAPSHOT. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7058]
• 2.Residual guard is untested: patching makeShreddedFilter down to the leaf-only predicate leaves all 9 e2e tests and all 13 new unit tests green. A measured test that does fail without the guard is inline. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:45]
• 3.Three fallback tests never push anything:a tinyint + a bigint extraction is rejected by the exact-type gate added in 20ff8d1, so overflow fallback, negated predicate over an all-fallback row group (the only e2e coverage of the negation guard) and the fallback half of annotated variant layout all assert about a scan with no pushed filter. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:139]
• 4.Default-on, but cannot fire on partial objects: one key outside the shredding schema puts a partial object in v.value, so notEq(v.value, null) is never droppable -- measured 20/20 row groups read vs 10/20 without that key. A tighter sound guard restores skipping on that layout; measurements and the patch are inline. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:995]

Non-blocking

• 5.expectedLeafType also rejects safe widening: exact type identity blocks bigint over an int/smallint/tinyint leaf, which is sound and is the shape users write. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:289]

Minor

• 6.Test layout default:writeShredded's annotate = false default is the non-production layout, so 8 of 9 tests run it. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:90]
• 7.Dead parameter:findChild's exact is true at every call site, so || caseSensitive and half its comment are unreachable. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:194]

[peter-toth]

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:

Suggested change

	.version("4.3.0")
	.version("4.4.0")

(5.0.0 if you mean this to be master-only, but that seems unlikely for a perf-only change.)

[viirya]

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").

[peter-toth]

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:

 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:

 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.valjsonExpr="case when id = 50 then '{\"a\":1500.5}' else '{\"a\":' || id || '}' end"
writeShredded(dir, "a bigint", jsonExpr, numRows =51, blockSize =1024*1024)
defread: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.

[viirya]

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.

[peter-toth]

Finding 3.a tinyint shreds $.a as optional int32 typed_value (INTEGER(8,true)) while the extraction here is bigint. Since 20ff8d1resolveShredded 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.

[viirya]

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.

[peter-toth]

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:

valfield= nameToShreddedVariantField(name)
makeLeaf(field.leaf.fieldType, field.leaf.fieldNames).map { leafPredicate =>valguards:Seq[FilterPredicate] = field.residualFieldNames.map { n =>FilterApi.notEq(binaryColumn(n), null.asInstanceOf[Binary])
}
valleafIsNull= 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.

[viirya]

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.

[qlong]

+1 on the new guard with isNull(leaf) suggested by @peter-toth, especially true for the strict inferenced schema generated by Spark.

With regard to default on/off, I think your benchmark shows the worst case for overhead due to the large number of rowgroups. With default parquet block size, the overhead could be lower (worth a testing). But even with the new guard, the chance this optimization can be triggered still requires delicated tuning.

[peter-toth]

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:

// in resolveShredded, replacing the exact `contains` checkvaltarget= expectedLeafType(targetType)
if (!target.contains(leafType) &&!isSafeWidening(leafType, target)) Noneelse ...

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.

[viirya]

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.

[peter-toth]

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.

[viirya]

Fixed in 4fce3cf -- writeShredded's annotate now defaults to true (the production layout), and one test passes annotate = false explicitly for the unannotated case.

[peter-toth]

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.

[viirya]

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.

[peter-toth]

Re-checked through cda65dc — findings 1, 3, 4, 5, 6, 7 resolved, and the core of finding 2 with them. I re-ran the ablations on this head and the guard is now load-bearing: reducing makeShreddedFilter to the leaf-only predicate fails residual fallback beyond the leaf's min/max is not dropped and the fallback half of unannotated variant layout; reverting to the flat OR fails partial object with a non-shredded sibling key still skips; neutering the two Not guards fails negated predicate over an all-fallback row group is not dropped. On @qlong's still-open default-on question, the tighter guard does shrink the population that pays the overhead without benefit — the partial-object layout now skips — so I'd keep it on, but the checked-in numbers no longer describe the shipped predicate (finding 9).

What's left is mostly what 4fce3cf left behind: the guard shape is still described as the old flat OR in six places including the PR description, and the benchmark results predate both the guard change and the case added in cda65dc.

Blocking

• 8.Guard shape documented as the old flat OR (new): the code now also skips when the leaf has zero nulls, but the config doc, the PR description and four comments still state the old or(leaf, isNotNull(residual)...) rule. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7046]
• 9.Benchmark results predate the shipped predicate (new): all three *-results.txt were generated before 4fce3cf changed the pushed predicate, and none contains the fourth case added in cda65dc — including the partial-object row that answers the default-on question. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/VariantShreddedPredicatePushdownBenchmark.scala:147]

Non-blocking

• 2.Multi-level intermediate-fallback test is still a no-op (round 1): scoped down from Blocking — the single-level tests are load-bearing now, but this one passes with the entire residual guard removed (measured) and its comment's premise is wrong. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:304]
• 10.Config doc's deferCastError claim is not true (new): the optimization does fire with deferCastError = true for try_variant_get and for string targets, because shouldWrapCastError requires a strict cast to a non-string, non-variant type. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7054]
• 11.No test pins error preservation for a strict variant_get (new): every fallback test uses try_variant_get, so the "eager INVALID_VARIANT_CAST must not become an empty result" property that justifies rejecting narrowing is untested. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:131]

Minor

• 12.Dead fallback branch in makeShreddedFilter (new):makeEq's type coverage is a superset of the other make*, so (Some(residual), None) is unreachable. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:1028]
• 13.Stale initialization-order comment (new):nameToShreddedVariantField is lazy, so declaration order of these two constants no longer matters. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:64]

[peter-toth]

Finding 8.4fce3cf changed the pushed predicate from the flat or(leaf, isNotNull(residual)...) to or(leaf, and(anyResidualNotNull, isNull(leaf))), but every prose description of it still states the old shape and the old skip rule. This doc line is the user-visible one: with the second arm a row group is also skipped when the leaf column has zero nulls, regardless of what the residuals hold. That is the whole point of the change (it is what makes the partial-object layout skip) and it is exactly what a reader needs in order to judge soundness, so it should not be the one sentence that is missing.

The other five places:

• the PR description — "To stay sound, the pushed predicate is: or(leafPredicate, isNotNull(residual_0), isNotNull(residual_1), ...)" and "a row group is skipped only when the leaf min/max cannot match and every residual is entirely NULL". Both now false.
• ParquetFilters.scala:1068 — "Each pushes or(leafPredicate, isNotNull(residual)...) over every residual value column along the path".
• ParquetFilters.scala:976 — referencesShreddedName's rationale derives the unsound negation from the flat shape. With the current shape LogicalInverseRewriter produces and(not(leafPred), or(and(eq(residual_i, null)...), notEq(leaf, null))), which is droppable as soon as some residual has no nulls and the leaf is entirely NULL — precisely the all-fallback row group in the new negation test, which is why removing the guard fails it. The guard is right; the derivation shown is not the one that applies.
• ParquetFilterSuite.scala:2434 — "OR-s an IS NOT NULL guard on every residual value column along the path".
• VariantShreddingFilterPushdownSuite.scala:35 — same wording.

ParquetFilterSuite.scala:2434 is the line @dongjoon-hyun flagged for describing a shape the code did not build, and 20ff8d1 fixed it; 4fce3cf re-broke it. Worth fixing all six in one pass so the next reader does not have to work out which description is current.

While in the description: "How was this patch tested?" still lists only the round-1 test set. The negation, safe-widening, narrowing-rejection, large-In, partial-object, unannotated-layout and deferCastError tests, and the new benchmark, are all missing from it.

[viirya]

Fixed in 0e10378 -- updated all six descriptions to the current or(leaf, and(anyResidualNotNull, isNull(leaf))) shape and the "leaf min/max can't match AND (every residual null OR leaf has no nulls)" skip rule: the config doc, the referencesShreddedName rationale (with the corrected negation derivation you gave), the createFilterHelper branch comment, the Not-guard comment, and both test headers. Also expanded "How was this patch tested?" in the PR description with the negation / widening / narrowing-rejection / large-In / partial-object / unannotated / deferCastError / error-preservation tests and the benchmark.

[peter-toth]

Finding 9. The three checked-in *-results.txt are stale in two independent ways.

1. They were generated at cef54f6 / cbb5644 / e005f3c, all before4fce3cf changed the pushed predicate from or(leaf, isNotNull(residual)...) to or(leaf, and(anyResidualNotNull, isNull(leaf))). None of the numbers — the 20-23x wins or the skip-none overhead — describe the predicate this PR ships.
2. This line adds a fourth case, but each results file holds only three (grep -c 'row groups:' → 3).

That is more than bookkeeping here, because @qlong's default-on question is still open and it turns on two numbers: the skip-none overhead (~3.5% in the current files: 2480→2557, 1846→1911, 2396→2479 ms) and how often the optimization can fire at all. The partial-object case is what answers the second half — it is the layout that read 20/20 row groups under the flat OR and skips under the new guard — and its row is exactly the one missing. Re-running the benchmark action gives you both the corrected overhead and that row.

Related: your 2026-08-19 reply to @qlong argues the overhead cannot be reduced because "or(leaf, isNotNull(residual)...) is a few disjuncts across a few columns ... there isn't a part of it I can drop without giving that up". 4fce3cf makes that out of date in the direction that helps your case, so it is worth restating with the new shape and the regenerated numbers.

[viirya]

Regenerating now -- I re-triggered the benchmark GitHub Action on 0e10378 for JDK 17/21/25 (create-commit), so the golden files will reflect the current predicate and include the partial-object case (the row that answers the default-on question). Will confirm once the results commits land.

[peter-toth]

Finding 2 (round 1). The single-level fallback tests are load-bearing now — I verified your check and it fails as you describe. This is the part of finding 2 that is left: this test still decides nothing.

Measured on this head, three ways:

• reduce makeShreddedFilter to leafPredicate (the [SPARK-55817][SQL] Enable Parquet row-group skipping for shredded variant #54598 shape): this test passes (only the single-level fallback test and the fallback half of unannotated variant layout fail);
• keep only the leaf-level residual (residuals.toSeq.takeRight(1), dropping v.value and v.typed_value.a.value): the whole suite passes, 10/10;
• revert to the flat OR: this test passes.

The premise in the comment is what breaks it. Row 6 stores {"a":{"b":5000}}, so the nested leaf's max is 5000 and b > 999 can never be dropped by leaf statistics — the row group is kept with or without any guard. Row 5's {"a":9999} cannot be what the guard saves either, because $.a.b over a scalar a is NULL and so can never match b > 999.

More generally, VariantShreddingWriter.castShredded always shreds an object field that is in the shredding schema, routing only non-schema keys into that level's own value. So with Spark's writer a value for $.a.b can never sit behind a NULL a.typed_value, and the ancestor-level guards are defensive against writers that legitimately decline to shred a level — real, but not reachable from Spark. That belongs in the residualFieldNames doc at ParquetFilters.scala:134 rather than being asserted by a test that cannot show it.

What this test can cover load-bearingly is the nested leaf-level residual, by moving the fallback down onto b:

// `a` shredded as struct<b bigint>. Rows 0..19 shred cleanly, so the nested leaf is min 0 /// max 19. Row 20 stores `b` as a non-integral decimal the int64 leaf cannot hold, so it lands// in v.typed_value.a.typed_value.b.value with the leaf NULL. `b > 999` matches only that row// and the leaf min/max cannot match it, so only the guard keeps the row group.valjsonExpr="case when id = 20 then '{\"a\":{\"b\":1500.5}}' else '{\"a\":{\"b\":' || id || '}}' end"
writeShredded(dir, "a struct<b bigint>", jsonExpr, numRows =21, blockSize =1024*1024)

with try_variant_get(v, '$.a.b', 'bigint') > 999 and the same "row group not skipped" assertion as the single-level test.

[viirya]

You're right, verified -- the old version passed with the whole guard removed. Fixed in 0e10378 using your suggestion: a struct<b bigint>, rows 0..19 clean, row 20 stores b as 1500.5 (int64 leaf can't hold it -> nested leaf-level residual, leaf NULL), try_variant_get(v,'$.a.b','bigint') > 999. Now it fails with the guard removed. Also moved the "Spark's writer never puts a value behind a NULL intermediate leaf; the ancestor-level guards are for writers that decline to shred a level" note into the residualFieldNames doc, as you suggested.

[peter-toth]

Finding 10. "it does not fire when spark.sql.variant.pushVariantIntoScan.deferCastError is true" holds only for a strict cast to a non-string, non-variant type. VariantInRelation.shouldWrapCastError is field.path.failOnError && deferCastErrorEnabled with VariantType | StringType excluded up front, so with deferCastError = true:

• try_variant_get(v, '$.a', 'bigint') > 999 — failOnError = false, no companion wrap, the filter stays a bare GetStructField and is pushed;
• variant_get(v, '$.a', 'string') = 'x' — StringType, same.

The test at line 365 only exercises strict bigint, so its name (deferCastError=true: optimization does not fire) over-claims for the same reason. Suggest scoping both — something like "does not fire for a strict cast to a non-string type, where the extraction is wrapped in UnwrapVariantCastError and is not translated to a pushable filter; try_variant_get and string targets are unaffected" — and adding a try_variant_get row to that test asserting skipping still happens with deferCastError = true.

[viirya]

Fixed in 0e10378. Scoped the config doc to "a strict cast to a non-string type" and noted try_variant_get and string targets are unaffected. The test now also asserts try_variant_get(v,'$.a','bigint') still fires and skips a row group with deferCastError=true.

[peter-toth]

Finding 11. Every fallback test uses try_variant_get, so the property the resolveShredded comment leans on to justify rejecting narrowing — "changing an eager INVALID_VARIANT_CAST into an empty result" — has no test. That is the worst failure mode this feature has (a thrown error silently becoming an empty result), and it is reachable through the residual path with an exactly-matching leaf type too, not only through narrowing.

Concretely: shred a as int, rows 0..49 as {"a": id}, row 50 as {"a": 3000000000} — tryTypedShred's INT case rejects it (value != (int) value), so it lands in v.typed_value.a.value with the leaf NULL — then strict variant_get(v, '$.a', 'int') > 999. The extraction type matches the leaf exactly, so the path is pushed. With deferCastError = false (the default) the scan casts eagerly, so the baseline raises INVALID_VARIANT_CAST; the leaf min/max is 0..49, so a leaf-only push would drop the row group and return an empty result instead. A checkError case beside the existing 1500.5 one would pin it.

[viirya]

Added in 0e10378, using your repro: a int, row 50 = 3000000000 (overflows int32 -> residual, leaf NULL), strict variant_get(v,'$.a','int') > 999 with deferCastError=false. Asserts INVALID_VARIANT_CAST is raised (not an empty result) across DSv1/DSv2 x vectorized/non-vectorized -- a leaf-only push would drop the row group and return empty.

[peter-toth]

Finding 12. This branch is unreachable. makeLeaf returning Some means the leaf type is matched by one of makeEq / makeLt / makeLtEq / makeGt / makeGtEq / makeInPredicate, and makeEq's case list is a superset of the others — same types, same pushDownDate / pushDownDecimal guards, plus ParquetBooleanType — so makeEq.lift(field.leaf.fieldType) is defined whenever we get here. Every type expectedLeafType can return is in makeEq as well.

Suggest dropping the branch (and its comment) so only case (Some(residual), Some(isNull)) and case (None, _) remain, or turning leafIsNull into a direct makeEq(field.leaf.fieldType)(...) application to make the invariant explicit rather than silently unexercised.

[viirya]

Fixed in 0e10378 -- dropped the unreachable branch; leafIsNull is now a direct makeEq.lift(...).get(...) application, making the "makeEq covers every leaf type the comparison ops do" invariant explicit.

[peter-toth]

Finding 13.nameToShreddedVariantField is lazy, so it is not built during construction and declaration order of these two is irrelevant (the reason was true only before the map became lazy). Suggest dropping the sentence, or replacing it with the constraint that does still bind — the map has to be lazy because resolution reads the Parquet*Type vals declared further down, which is already noted at its own definition.

[viirya]

Fixed in 0e10378 -- dropped the comment; the map is lazy, so the ordering of these two constants no longer matters (the real constraint, that it must be lazy because it reads the Parquet*Type vals below, is already noted at its own definition).

[qlong]

can we add some tests to pushdown more than 1 predicates?

[peter-toth]

Re-checked through 03dc637 — findings 2, 8, 9, 10, 11, 12, 13 resolved, nothing regressed, and nothing blocking left from my side. CI is green on this head (29 success, 1 skipped).

The PR was rebased onto current master since R2, so I re-measured the ablations against the new base rather than carrying the R2 verdicts forward. With makeShreddedFilter reduced to the leaf-only predicate (the #54598 shape), 4 of the 11 VariantShreddingFilterPushdownSuite tests fail — and two of them are the tests this round rewrote: multi-level $.a.b: nested residual fallback beyond the leaf's min/max is not dropped (finding 2) and strict variant_get preserves INVALID_VARIANT_CAST on a residual fallback (finding 11). Both are load-bearing now.

On @qlong's two comments that are still open: for the "more than 1 predicates" request I measured it rather than guessing — $.a > 999 AND $.b < 500 over a file shredding a bigint, b bigint reads 5 of 20 row groups against 20 with the config off, results match the no-pushdown baseline, OR is correct too, and a residual fallback on the second path survives. So the And/Or combinators compose soundly and each guarded conjunct prunes independently; it is purely a test-coverage gap, not a correctness one. The default-block-size question is answered in finding 16 — measured, the overhead goes to zero.

Non-blocking

• 14.DSv2 scope comment gives the wrong reason (late catch): DSv2 does rewrite the filter into v.`0` — buildScanWithPushedVariants calls the same VariantInRelation.rewriteExpr. What actually keeps it out of pushDataFilters is rule ordering, and saying so protects the invariant. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala:72]
• 15.Negation test still documents the pre-4fce3cf flat OR (late catch): the six prose spots from finding 8 are fixed, but this seventh one — on the test that guards the negation refusal — still derives the unsoundness from the old shape. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:2707]
• 16.The skip-none overhead is a 2721-row-group artifact (new): the case writes 128KB blocks, so its ~3-4% is measured on 2721 row groups in one file; at the default block size the same case has 1 row group and the overhead is gone (-0.9% best / -0.3% avg). The cost scales with the same knob as the benefit, which is a positive argument for keeping the default on. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/VariantShreddedPredicatePushdownBenchmark.scala:129]

Thanks for working through all of these, @viirya.

[peter-toth]

Finding 14. The conclusion is right but the reason isn't: the v.`0` rewrite is not DSv1-only. V2ScanRelationPushDown.buildScanWithPushedVariants runs the same rewrite on the filters —

// V2ScanRelationPushDown.scala:953valrewrittenFilterExprs= filters.map(variants.rewriteExpr(_, attributeMap))

— where variants is a VariantInRelation, the very class PushVariantIntoScan uses, and the result goes into a Filter above the scan. So v.`0` predicates do exist on the DSv2 path.

What actually keeps them out of pushDataFilters is rule ordering: in V2ScanRelationPushDown.apply the pushdownRules list runs pushDownFilters (:54) before pushDownVariants (:60) and buildScanWithPushedVariants (:64), so when this method is called the predicate is still variant_get(v, ...), which never translates to a source Filter. Worth stating it that way, because the invariant then reads as what it is — an ordering property that a later reordering or a second filter-push pass after the variant rewrite would silently break — rather than something true by construction:

// Shredded-variant predicate pushdown (SPARK-55817) is not wired here: it applies to the// DSv1 path only. DSv2 does rewrite variant extractions into `v.`0`` struct accesses, but// only in `V2ScanRelationPushDown.buildScanWithPushedVariants`, which runs *after*// `pushDownFilters`. So the filters reaching this method are still `variant_get(v, ...)`// predicates, which do not translate to a source `Filter` at all -- there is no// shredded-variant logical name for `ParquetFilters` to resolve here, and nothing would be// reported convertible even with a `variantExtractionSchema`. DSv2 reads remain correct (the// variant filter is applied post-scan); they just do not get row-group skipping on shredded// columns.

The same "never rewritten" claim is in the PR description and in VariantShreddingFilterPushdownSuite's scaladoc (:40), so those two want the same correction.

[viirya]

Good catch, corrected in 82c5b59. You're right that DSv2 does run the same VariantInRelation.rewriteExpr (in buildScanWithPushedVariants); it's the rule ordering -- pushDownFilters runs before it -- that leaves the filters here as variant_get(v, ...). Rewrote the ParquetScanBuilder comment to your wording, and fixed the same "never rewritten" claim in the suite scaladoc and the PR description.

[peter-toth]

Finding 15.0e10378 updated six descriptions of the guard, but this is a seventh — and it sits on the test that protects the negation refusal, so it is the one where a stale derivation costs the most. The shipped predicate is or(leaf, and(anyResidualNotNull, isNull(leaf))), whose parquet-mr inverse is and(not(leaf), or(and(eq(residual, null)...), notEq(leaf, null))) — droppable as soon as some residual has no nulls and the leaf is entirely NULL, which is exactly the {"a":500.5} / {"a":600.5} all-fallback row group this suite's e2e counterpart builds. referencesShreddedName in ParquetFilters.scala already carries the corrected derivation; only this copy is behind.

Suggested change

	// not(or(leaf, isNotNull(residual))) is rewritten by parquet-mr into an unsound
	// and(notEq(leaf), eq(residual, null)), so a negated shredded predicate must not be pushed.
	// not(or(leaf, and(anyResidualNotNull, isNull(leaf)))) is rewritten by parquet-mr into
	// and(not(leaf), or(and(eq(residual, null)...), notEq(leaf, null))), which is droppable once
	// some residual has no nulls AND the leaf is entirely NULL -- an all-fallback row group. So a
	// negated shredded predicate must not be pushed.

[viirya]

Fixed in 82c5b59 -- updated the comment to the shipped shape: not(or(leaf, and(anyResidualNotNull, isNull(leaf)))) rewrites to and(not(leaf), or(and(eq(residual, null)...), notEq(leaf, null))), droppable once some residual has no nulls AND the leaf is entirely NULL. That was the seventh spot the earlier doc pass missed.

[peter-toth]

Finding 16. This case is the number the open default-on discussion is resting on, and it is measured on a layout no default-configured writer produces: createAndRunBenchmark writes with parquet.block.size = 128KB, which for this dataset puts 2721 row groups in one file. I re-ran the same case at 128KB and at the Parquet default block size on the head:

 rowGroups off best/avg on best/avg delta
Skip none, 128KB blocks (as shipped) 2721 914 / 923 953 / 956 +4.3% / +3.6%
Skip none, DEFAULT block size 1 891 / 894 883 / 891 -0.9% / -0.3%

(20M rows, 5 iterations, JDK 21, Apple M4 Max; the 128KB row reproduces the ~3-4% in the checked-in *-results.txt, so the setup matches.)

So the overhead scales with row-group count — the same knob that produces the 15-22x wins — and a file has to be deliberately tuned dense before it can pay anything at all. That answers @qlong's "with default parquet block size, the overhead could be lower (worth a testing)": measured, it goes to zero, which argues for keeping the default on. It also sharpens the earlier "within run-to-run noise" framing: at 2721 row groups the delta is consistent and outside stdev on all three checked-in JDK runs (+57ms/±7, +64ms/±21, +82ms/±14 on best time), so it is real there and absent at the default layout, rather than noisy in both.

Either adding a default-block-size skip-none case here, or a sentence in this scaladoc saying the figure is specific to the 128KB layout, would stop the number being read as a general default-on cost.

[viirya]

Agreed -- the ~3-4% is a 128KB / 2721-row-group artifact. Added a "Can skip no row groups (default block size)" case in 82c5b59 (one row group for this dataset); locally it measures ~-2.5% (ON slightly faster), matching your numbers. So the overhead scales with row-group count, the same knob as the benefit. Re-triggered the benchmark Action on JDK 17/21/25 (create-commit) to regenerate the golden files with the new case; will confirm once they land.

[viirya]

@dongjoon-hyun Would you like to take another look before I merge? Thanks!

[dongjoon-hyun]

+1, LGTM. Thank you, @viirya and all.

[viirya]

Merge Summary:

• merged into master fa6f713
• merged into branch-4.x 8f7ab72

Posted by merge_spark_pr.py

[viirya]

Thanks @dongjoon-hyun@uros-b@qlong@peter-toth !