Uh oh!
There was an error while loading. Please reload this page.
[SPARK-55817][SQL] Enable Parquet row-group skipping for shredded variant - #54598
Closed
qlong wants to merge 1 commit into
Closed
[SPARK-55817][SQL] Enable Parquet row-group skipping for shredded variant#54598qlong wants to merge 1 commit into
qlong wants to merge 1 commit into
Conversation
…iant When PushVariantIntoScan rewrites variant_get() calls into struct field accesses, the rewritten predicates reference logical paths like "v.`0`" that ParquetFilters cannot resolve to any physical column, so they are dropped and row-group skipping is disabled for all shredded variant queries. This change adds variantExtractionSchema to ParquetFilters, and resolves the logical path to the corresponding typed_value leaf in the physical Parquet schema.The resolved entries allow predicates on shredded variant to participate in row-group skipping. Array-index paths and fields absent from a file's physical schema are skipped.
qlongforce-pushed
the
SPARK-55817-row-group-skipping
branch
from
March 3, 2026 21:41
079b01f to
ef8bc2fCompareqlong
commented
Mar 4, 2026
ContributorAuthor
@chenhao-db Can you take a look at this? It is a follow up to your PR #49235 |
This was referenced Apr 28, 2026
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable. |
qlong
commented
Jun 13, 2026
ContributorAuthor
I think this PR is still needed to improve variant read performance |
viirya added a commit
that referenced
this pull request
Aug 23, 2026
…iant columns ### 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) Closes#58050 from viirya/SPARK-55817-variant-shredded-rowgroup-skipping. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>
viirya added a commit
that referenced
this pull request
Aug 23, 2026
…iant columns ### 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) Closes#58050 from viirya/SPARK-55817-variant-shredded-rowgroup-skipping. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com> (cherry picked from commit fa6f713) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
When PushVariantIntoScan rewrites variant_get() calls into struct field accesses, the rewritten predicates reference logical paths like "v.
0" that ParquetFilters cannot resolve to any physical column, so they are dropped and row-group skipping is disabled for all shredded variant queries.This change adds variantExtractionSchema to ParquetFilters, and resolves the logical path to the corresponding typed_value leaf in the physical Parquet schema.The resolved entries allow predicates on shredded variant to participate in row-group skipping.
Array-index paths and fields absent from a file's physical schema are skipped.
Jira: https://issues.apache.org/jira/browse/SPARK-55817
Why are the changes needed?
Performance improvement. The shreded variant predicates are pushed down to participate row group filtering.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Was this patch authored or co-authored using generative AI tooling?
co-authorized with Claude 4.6 Sonnet.