This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fa6f71301587 [SPARK-55817][SQL] Enable Parquet row-group skipping for
shredded Variant columns
fa6f71301587 is described below
commit fa6f71301587395d3576acb09b7bba2d8d3afc74
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Sat Aug 22 22:55:09 2026 -0700
[SPARK-55817][SQL] Enable Parquet row-group skipping for shredded Variant
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 [...]
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/ [...]
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-in [...]
- 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 `ParquetS [...]
### 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 larg [...]
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, `deferCastErr [...]
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 <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 27 ++
...dedPredicatePushdownBenchmark-jdk21-results.txt | 35 ++
...dedPredicatePushdownBenchmark-jdk25-results.txt | 35 ++
...tShreddedPredicatePushdownBenchmark-results.txt | 35 ++
.../datasources/parquet/ParquetFileFormat.scala | 15 +-
.../datasources/parquet/ParquetFilters.scala | 406 ++++++++++++++++++-
.../v2/parquet/ParquetScanBuilder.scala | 9 +
...VariantShreddedPredicatePushdownBenchmark.scala | 166 ++++++++
.../datasources/parquet/ParquetFilterSuite.scala | 413 ++++++++++++++++++-
.../VariantShreddingFilterPushdownSuite.scala | 445 +++++++++++++++++++++
10 files changed, 1567 insertions(+), 19 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 0ed4399361d0..2a8d23eaceba 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -7102,6 +7102,33 @@ 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 a predicate on the physical shredded typed_value leaf column,
guarded so that a row " +
+ "group is skipped only when the leaf min/max cannot match AND every
value for the path " +
+ "is provably in the typed leaf (either every untyped residual value
column along the " +
+ "path is entirely null, or the leaf column itself has no nulls). 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. It
also does not fire " +
+ "for a strict cast to a non-string type when " +
+ "spark.sql.variant.pushVariantIntoScan.deferCastError is true (the
extraction is wrapped " +
+ "in UnwrapVariantCastError and is not translated to a pushable
filter); try_variant_get " +
+ "and string targets are unaffected. Results are unaffected either way;
this only " +
+ "controls whether row groups can be skipped.")
+ .version("4.4.0")
+ // Physical scan optimization only: it changes which Parquet row groups
are read, not the
+ // resolved plan of a view/UDF/procedure body, so it does not
participate in binding.
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val LEGACY_CSV_ENABLE_DATE_TIME_PARSING_FALLBACK =
buildConf("spark.sql.legacy.csv.enableDateTimeParsingFallback")
.internal()
diff --git
a/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-jdk21-results.txt
b/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-jdk21-results.txt
new file mode 100644
index 000000000000..f91bddaedecb
--- /dev/null
+++
b/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-jdk21-results.txt
@@ -0,0 +1,35 @@
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip all row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 1948 2008
87 10.8 92.9 1.0X
+With shredded predicate pushdown 92 107
10 227.1 4.4 21.1X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip some row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2030 2048
10 10.3 96.8 1.0X
+With shredded predicate pushdown 108 115
5 194.1 5.2 18.8X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip no row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2440 2468
28 8.6 116.3 1.0X
+With shredded predicate pushdown 2500 2530
21 8.4 119.2 1.0X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip no row groups (default block size): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2166 2196
23 9.7 103.3 1.0X
+With shredded predicate pushdown 2175 2211
27 9.6 103.7 1.0X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip some row groups (partial object): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2347 2355
10 8.9 111.9 1.0X
+With shredded predicate pushdown 164 174
9 127.6 7.8 14.3X
+
diff --git
a/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-jdk25-results.txt
b/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-jdk25-results.txt
new file mode 100644
index 000000000000..71a0f414335e
--- /dev/null
+++
b/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-jdk25-results.txt
@@ -0,0 +1,35 @@
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip all row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 1515 1574
79 13.8 72.3 1.0X
+With shredded predicate pushdown 67 76
8 312.8 3.2 22.6X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip some row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 1507 1526
18 13.9 71.9 1.0X
+With shredded predicate pushdown 84 91
8 250.7 4.0 18.0X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip no row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 1876 1893
17 11.2 89.5 1.0X
+With shredded predicate pushdown 1940 1958
19 10.8 92.5 1.0X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip no row groups (default block size): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 1658 1695
38 12.6 79.1 1.0X
+With shredded predicate pushdown 1659 1666
10 12.6 79.1 1.0X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip some row groups (partial object): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 1772 1789
20 11.8 84.5 1.0X
+With shredded predicate pushdown 116 123
7 180.2 5.5 15.2X
+
diff --git
a/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-results.txt
b/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-results.txt
new file mode 100644
index 000000000000..78372cd131f1
--- /dev/null
+++ b/sql/core/benchmarks/VariantShreddedPredicatePushdownBenchmark-results.txt
@@ -0,0 +1,35 @@
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip all row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2118 2175
132 9.9 101.0 1.0X
+With shredded predicate pushdown 91 103
10 229.9 4.3 23.2X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip some row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2133 2162
29 9.8 101.7 1.0X
+With shredded predicate pushdown 112 124
9 186.8 5.4 19.0X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip no row groups: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2562 2583
21 8.2 122.2 1.0X
+With shredded predicate pushdown 2637 2663
26 8.0 125.7 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip no row groups (default block size): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2261 2296
40 9.3 107.8 1.0X
+With shredded predicate pushdown 2260 2284
21 9.3 107.8 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Can skip some row groups (partial object): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------
+Without shredded predicate pushdown 2461 2475
9 8.5 117.3 1.0X
+With shredded predicate pushdown 156 161
5 134.3 7.4 15.8X
+
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index a97cd7d57e41..c108e74eca0f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -210,6 +210,18 @@ class ParquetFileFormat
val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
val pushDownInFilterThreshold =
sqlConf.parquetFilterPushDownInFilterThreshold
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+ // When shredded-variant predicate pushdown is enabled and
`requiredSchema` actually carries a
+ // variant-extraction struct produced by PushVariantIntoScan, passing it
lets ParquetFilters map
+ // logical paths like "v.`0`" to the physical shredded columns for
row-group skipping. The
+ // `isVariantStruct` check keeps the per-file ParquetFilters construction
free of any shredded
+ // traversal for the common case of scans with no variant-extraction
columns.
+ val variantExtractionSchema =
+ if (sqlConf.getConf(SQLConf.VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED)
&&
+ requiredSchema.existsRecursively(VariantMetadata.isVariantStruct)) {
+ Some(requiredSchema)
+ } else {
+ None
+ }
val parquetOptions = new ParquetOptions(options, sqlConf)
val datetimeRebaseModeInRead = parquetOptions.datetimeRebaseModeInRead
val int96RebaseModeInRead = parquetOptions.int96RebaseModeInRead
@@ -266,7 +278,8 @@ class ParquetFileFormat
pushDownStringPredicate,
pushDownInFilterThreshold,
isCaseSensitive,
- datetimeRebaseSpec)
+ datetimeRebaseSpec,
+ variantExtractionSchema = variantExtractionSchema)
filters
// Collects all converted Parquet filter predicates. Notice that
not all predicates
// can be converted (`ParquetFilters.createFilter` returns an
`Option`). That's why
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
index 50f90048b8c8..0adece577c58 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
@@ -37,11 +37,14 @@ import
org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName._
import org.apache.parquet.schema.Type.Repetition
+import org.apache.spark.sql.catalyst.expressions.variant.{ObjectExtraction,
VariantPathParser}
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils,
IntervalUtils}
import
org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays,
rebaseGregorianToJulianMicros, RebaseSpec}
+import org.apache.spark.sql.execution.datasources.VariantMetadata
import
org.apache.spark.sql.execution.datasources.parquet.types.ops.{ParquetFilterOps,
ParquetTypeOps}
import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.sources
+import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.ArrayImplicits._
@@ -56,25 +59,18 @@ class ParquetFilters(
pushDownStringPredicate: Boolean,
pushDownInFilterThreshold: Int,
caseSensitive: Boolean,
- datetimeRebaseSpec: RebaseSpec) {
+ datetimeRebaseSpec: RebaseSpec,
+ variantExtractionSchema: Option[StructType] = None) {
+ // Shredded-variant physical field-name constants.
+ private val TYPED_VALUE = "typed_value"
+ private val VALUE = "value"
+
// A map which contains parquet field name and data type, if predicate push
down applies.
//
// Each key in `nameToParquetField` represents a column; `dots` are used as
separators for
// nested columns. If any part of the names contains `dots`, it is quoted to
avoid confusion.
// See `org.apache.spark.sql.connector.catalog.quote` for implementation
details.
private val nameToParquetField : Map[String, ParquetPrimitiveField] = {
- 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
- }
- }
-
// Recursively traverse the parquet schema to get primitive fields that
can be pushed-down.
// `parentFieldNames` is used to keep track of the current nested level
when traversing.
def getPrimitiveFields(
@@ -128,6 +124,278 @@ 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,
which is what the
+ * pushed predicate's guard checks (see
`makeShreddedFilter`).
+ * Spark's writer
(`VariantShreddingWriter.castShredded`) always shreds
+ * an object field that is in the shredding schema
and routes only
+ * non-schema keys into a level's own `value`, so
with Spark-written
+ * files a value can only fall back to the leaf's
own-level residual;
+ * the ancestor-level residuals guard against
writers that legitimately
+ * decline to shred an intermediate level.
+ */
+ 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 a guarded predicate
that skips a row group
+ // only when the leaf cannot match AND 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, always case-sensitively. Returns the
child type together
+ // with its actual physical name so callers build paths from the on-disk
names.
+ //
+ // The match is always exact-case because every caller resolves either a
variant object key or a
+ // structural `typed_value`/`value` name. Variant object keys 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`
+ // separately, in `shreddedVariantEntries`.
+ private def findChild(group: GroupType, name: String): Option[Type] = {
+ group.getFields.asScala.find(_.getName == 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).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) 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)) 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) match {
+ case Some(p: PrimitiveType) if p.getRepetition != Repetition.REPEATED =>
+ val leafType =
+ ParquetSchemaType(getNormalizedLogicalType(p),
p.getPrimitiveTypeName, p.getTypeLength)
+ // Accept the leaf when the extraction's target type maps to it
exactly, or when the leaf is
+ // a narrower signed integer than the target (safe widening).
Narrowing must be rejected:
+ // for a narrower extraction such as smallint over an int leaf, the
leaf min/max is over int
+ // values, so a row group holding only out-of-int16-range values
(residuals null) would be
+ // skipped, changing an eager INVALID_VARIANT_CAST into an empty
result. Widening is sound:
+ // every value in a narrower leaf casts to the wider target losslessly
and ordering is
+ // preserved, so the leaf stats bound the target predicate;
`valueMatchesParquetType` still
+ // rejects a literal outside the leaf's representable range at push
time.
+ if (!expectedLeafType(targetType).contains(leafType) &&
+ !isSafeIntegerWidening(leafType, targetType)) {
+ None
+ } else {
+ val leaf = ParquetPrimitiveField(namePath :+ p.getName, leafType)
+ Some(ShreddedVariantField(leaf, residuals.toSeq))
+ }
+ case _ => None
+ }
+ }
+
+ // Whether an extraction of `targetType` may be pushed against a physical
`leafType` that is a
+ // narrower signed integer (e.g. bigint extraction over an
int/smallint/tinyint leaf). Only the
+ // integer family widens soundly here.
+ private def isSafeIntegerWidening(
+ leafType: ParquetSchemaType, targetType: DataType): Boolean = {
+ def intRank(t: ParquetSchemaType): Option[Int] = t match {
+ case ParquetByteType => Some(0)
+ case ParquetShortType => Some(1)
+ case ParquetIntegerType => Some(2)
+ case ParquetLongType => Some(3)
+ case _ => None
+ }
+ (intRank(leafType), expectedLeafType(targetType).flatMap(intRank)) match {
+ case (Some(leafRank), Some(targetRank)) => leafRank < targetRank
+ case _ => false
+ }
+ }
+
+ // 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 {
+ case BooleanType => Some(ParquetBooleanType)
+ case ByteType => Some(ParquetByteType)
+ case ShortType => Some(ParquetShortType)
+ case IntegerType => Some(ParquetIntegerType)
+ case LongType => Some(ParquetLongType)
+ case FloatType => Some(ParquetFloatType)
+ case DoubleType => Some(ParquetDoubleType)
+ case _: StringType => Some(ParquetStringType)
+ case BinaryType => Some(ParquetBinaryType)
+ case DateType => Some(ParquetDateType)
+ case d: DecimalType if DecimalType.is32BitDecimalType(d) =>
+ Some(ParquetSchemaType(LogicalTypeAnnotation.decimalType(d.scale,
d.precision), INT32, 0))
+ case d: DecimalType if DecimalType.is64BitDecimalType(d) =>
+ Some(ParquetSchemaType(LogicalTypeAnnotation.decimalType(d.scale,
d.precision), INT64, 0))
+ case d: DecimalType =>
+ Some(ParquetSchemaType(LogicalTypeAnnotation.decimalType(d.scale,
d.precision),
+ FIXED_LEN_BYTE_ARRAY, Decimal.minBytesForPrecision(d.precision)))
+ case _ => None
+ }
+
+ // Walk the variant-extraction schema alongside the physical Parquet group,
collecting
+ // logicalName -> ShreddedVariantField entries for shredded scalar object
paths that this file
+ // actually shreds. Only object-extraction, scalar-leaf paths are eligible;
array-index paths and
+ // synthetic (empty / placeholder / companion / full-variant passthrough)
paths resolve to None.
+ //
+ // `logicalParentNames` accumulates the logical field names (used to build
the map key that the
+ // pushed filter references); `physParentNames` accumulates the on-disk
field names (used to build
+ // the physical Parquet column paths). They differ only in case under
case-insensitive matching.
+ private def shreddedVariantEntries(
+ variantFields: Seq[StructField],
+ physGroup: GroupType,
+ logicalParentNames: Array[String],
+ physParentNames: Array[String]): Seq[(String, ShreddedVariantField)] = {
+ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ variantFields.flatMap { field =>
+ val physChildOpt = physGroup.getFields.asScala.collectFirst {
+ case g: GroupType if
+ (if (caseSensitive) g.getName == field.name
+ else g.getName.equalsIgnoreCase(field.name)) => g
+ }
+ physChildOpt match {
+ case None => Nil
+ case Some(physChild) =>
+ val logicalColPath = logicalParentNames :+ field.name
+ val physColPath = physParentNames :+ physChild.getName
+ field.dataType match {
+ // Variant struct: each child is a requested extraction carrying
VariantMetadata.
+ case s: StructType if VariantMetadata.isVariantStruct(s) =>
+ s.fields.toSeq.flatMap { extraction =>
+ if
(!extraction.metadata.contains(VariantMetadata.METADATA_KEY)) {
+ Nil
+ } else {
+ val meta = VariantMetadata.fromMetadata(extraction.metadata)
+ // `VariantPathParser.parse` returns None for an unparseable
path; use it directly
+ // rather than `parsedPath()` (which throws) so a bad path
is a clean no-push and
+ // we don't swallow unrelated exceptions.
+ VariantPathParser.parse(meta.path) match {
+ case None => Nil
+ case Some(segments) =>
+ // Only scalar object-extraction paths are eligible.
Reject array-index paths
+ // (a mix of ObjectExtraction and ArrayExtraction fails
this check) and empty
+ // paths ("$" / passthrough / companion), which yield no
keys.
+ val keys = segments.collect { case o: ObjectExtraction
=> o.key }
+ if (keys.isEmpty || keys.length != segments.length) {
+ Nil
+ } else {
+ // `physChild` is this variant column's physical
group; `physColPath` holds
+ // its on-disk name path. Navigate the shredding
layout from there. The
+ // extraction's target type must match the physical
leaf type exactly.
+ resolveShredded(physChild, physColPath, keys,
extraction.dataType) match {
+ case None => Nil
+ case Some(shredded) =>
+ val logicalName =
+ (logicalColPath :+
extraction.name).toImmutableArraySeq.quoted
+ Seq(logicalName -> shredded)
+ }
+ }
+ }
+ }
+ }
+ // Ordinary struct: recurse into nested fields.
+ case s: StructType if !VariantMetadata.isVariantStruct(s) =>
+ shreddedVariantEntries(s.fields.toSeq, physChild,
logicalColPath, physColPath)
+ case _ => Nil
+ }
+ }
+ }
+ }
+
private case class ParquetSchemaType(
logicalTypeAnnotation: LogicalTypeAnnotation,
primitiveTypeName: PrimitiveTypeName,
@@ -624,6 +892,9 @@ class ParquetFilters(
} else {
Some(sources.Or(leftResultOptional.get, rightResultOptional.get))
}
+ // A negated shredded-variant predicate cannot be pushed soundly (see
the matching guard in
+ // `createFilterHelper`), so it is not convertible either.
+ case sources.Not(pred) if referencesShreddedName(pred) => None
case sources.Not(pred) =>
val resultOptional = convertibleFiltersHelper(pred, canPartialPushDown
= false)
resultOptional.map(sources.Not)
@@ -647,7 +918,12 @@ class ParquetFilters(
// Parquet's type in the given file should be matched to the value's type
// in the pushed filter in order to push down the filter to Parquet.
private def valueCanMakeFilterOn(name: String, value: Any): Boolean = {
- value == null || (nameToParquetField(name).fieldType match {
+ valueMatchesParquetType(nameToParquetField(name).fieldType, value)
+ }
+
+ // The value's type must match the field's on-disk Parquet type for a filter
to be pushed.
+ private def valueMatchesParquetType(fieldType: ParquetSchemaType, value:
Any): Boolean = {
+ value == null || (fieldType match {
case ParquetBooleanType => value.isInstanceOf[JBoolean]
case ParquetIntegerType if value.isInstanceOf[Period] => true
case ParquetByteType | ParquetShortType | ParquetIntegerType => value
match {
@@ -692,6 +968,74 @@ 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,
and(anyResidualNotNull,
+ // isNull(leaf)))`, and parquet-mr's LogicalInverseRewriter pushes
`not(...)` inside to
+ // `and(not(leaf), or(and(eq(residual, null)...), notEq(leaf, null)))`,
which is row-group-
+ // droppable as soon as some residual has no nulls AND the leaf is entirely
NULL -- exactly an
+ // all-fallback row group, whose matching values are all in a 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, and(anyResidualNotNull, isNull(leaf)))
+ // where `anyResidualNotNull` is `or(notEq(residual_0, null), ...,
notEq(residual_n, null))` and
+ // `isNull(leaf)` is `eq(leaf, null)`.
+ //
+ // Parquet's statistics drop logic: `or(a, b)` is row-group-droppable iff
BOTH `a` and `b` are
+ // droppable; `and(a, b)` iff EITHER is; `notEq(col, null)` (IS NOT NULL)
iff the column is
+ // entirely NULL (no non-nulls); `eq(col, null)` (IS NULL) iff the column
has no nulls. So the
+ // whole `or` drops the row group iff the leaf min/max cannot match AND
(every residual is
+ // entirely NULL OR the leaf column has no nulls). The second arm is what
makes this sound and
+ // still effective: a value for the path can be outside the typed leaf only
on a row where the
+ // leaf is NULL, so a leaf with zero nulls means every value is provably in
the typed leaf and the
+ // leaf min/max is a complete summary -- regardless of what the residual
columns hold (they may be
+ // non-null because a sibling key outside the shredding schema landed in the
level's `value`,
+ // which is the normal layout for real Variant data). Per record it still
keeps every row that
+ // could match: a row whose value fell back to a residual has a NULL leaf
and a non-null residual,
+ // so `and(anyResidualNotNull, isNull(leaf))` holds for it.
+ //
+ // 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. The
+ // earlier flat `or(leaf, isNotNull(residual)...)` was sound but could never
drop a row group once
+ // any residual was non-null (e.g. a partial object), i.e. it paid the
pushdown cost without ever
+ // skipping on that common layout.
+ //
+ // `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
+ .map(n => FilterApi.notEq(binaryColumn(n), null.asInstanceOf[Binary]))
+ .reduceLeftOption[FilterPredicate](FilterApi.or) match {
+ case None =>
+ // No residuals exist in this file for the path, so the leaf is a
complete summary.
+ leafPredicate
+ case Some(anyResidualNotNull) =>
+ // `makeLeaf` returned Some, so the leaf type is one `makeEq` also
covers (its case list
+ // is a superset of the comparison ops), hence `makeEq.lift` is
defined here.
+ val leafIsNull =
makeEq.lift(field.leaf.fieldType).get(field.leaf.fieldNames, null)
+ FilterApi.or(leafPredicate, FilterApi.and(anyResidualNotNull,
leafIsNull))
+ }
+ }
+ }
+
/**
* @param predicate the input filter predicates. Not all the predicates can
be pushed down.
* @param canPartialPushDownConjuncts whether a subset of conjuncts of
predicates can be pushed
@@ -718,6 +1062,36 @@ class ParquetFilters(
// Probably I missed something and obviously this should be changed.
predicate match {
+ // Shredded-variant paths (e.g. "v.`0`"). Only comparison predicates
that use min/max
+ // statistics are eligible. Each pushes the guarded predicate built by
`makeShreddedFilter`.
+ // IS NULL / IS NOT NULL on the logical variant field are intentionally
out of scope: "the
+ // extracted field is null" is not the same as "typed_value is null", so
we must not conflate
+ // them.
+ case sources.EqualTo(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, value)))
+ case sources.EqualNullSafe(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, value)))
+ case sources.LessThan(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeLt.lift(t).map(_(n, value)))
+ case sources.LessThanOrEqual(name, value) if
canMakeShreddedFilterOn(name, value) =>
+ makeShreddedFilter(name, (t, n) => makeLtEq.lift(t).map(_(n, value)))
+ case sources.GreaterThan(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeGt.lift(t).map(_(n, value)))
+ case sources.GreaterThanOrEqual(name, value) if
canMakeShreddedFilterOn(name, value) =>
+ makeShreddedFilter(name, (t, n) => makeGtEq.lift(t).map(_(n, value)))
+ case sources.In(name, values) if pushDownInFilterThreshold > 0 &&
values.nonEmpty &&
+ values.forall(v => canMakeShreddedFilterOn(name, v)) =>
+ // Build the leaf predicate once (an OR of per-value equalities under
the threshold, or a
+ // single FilterApi.in above it), then OR the residual isNotNull
guards on once via
+ // `makeShreddedFilter`. Mirrors the regular `In` path below:
threshold on `values.length`,
+ // and `makeInPredicate`/`FilterApi.in` for large lists so those still
get skipping.
+ makeShreddedFilter(name, (t, n) =>
+ if (values.length <= pushDownInFilterThreshold) {
+ values.distinct.flatMap(v => makeEq.lift(t).map(_(n,
v))).reduceLeftOption(FilterApi.or)
+ } else {
+ makeInPredicate.lift(t).map(_(n, values))
+ })
+
case sources.IsNull(name) if canMakeFilterOn(name, null) =>
makeEq.lift(nameToParquetField(name).fieldType)
.map(_(nameToParquetField(name).fieldNames, null))
@@ -796,6 +1170,10 @@ class ParquetFilters(
rhsFilter <- createFilterHelper(rhs, canPartialPushDownConjuncts)
} yield FilterApi.or(lhsFilter, rhsFilter)
+ // Refuse to push a negated predicate that references a shredded-variant
path: not() of the
+ // guarded shredded predicate is rewritten by parquet-mr into a form
that can drop an
+ // all-fallback row group (see `referencesShreddedName`).
+ case sources.Not(pred) if referencesShreddedName(pred) => None
case sources.Not(pred) =>
createFilterHelper(pred, canPartialPushDownConjuncts = false)
.map(FilterApi.not)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
index 149d7e6f0b72..e702cfc3ad25 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
@@ -66,6 +66,15 @@ case class ParquetScanBuilder(
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
val parquetSchema =
new
SparkToParquetSchemaConverter(sparkSession.sessionState.conf).convert(readDataSchema())
+ // 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.
val parquetFilters = new ParquetFilters(
parquetSchema,
pushDownDate,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/VariantShreddedPredicatePushdownBenchmark.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/VariantShreddedPredicatePushdownBenchmark.scala
new file mode 100644
index 000000000000..388a555c7d0c
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/VariantShreddedPredicatePushdownBenchmark.scala
@@ -0,0 +1,166 @@
+/*
+ * 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.benchmark
+
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Synthetic benchmark for row-group skipping on shredded Variant columns
(SPARK-55817).
+ *
+ * The optimization pushes a predicate on a shredded Variant field to the
physical typed_value leaf
+ * (guarded so residual fallbacks are never skipped), letting Parquet skip row
groups the leaf
+ * min/max cannot match. The lift depends on the layout: the field must be
shredded, the predicate a
+ * literal comparison, the data sorted on that field, and a file must hold
many row groups. This
+ * benchmark writes such a layout (sorted on the shredded field, small block
size so a single file
+ * has many row groups) and compares scan time with the optimization on vs off.
+ *
+ * To run this benchmark:
+ * {{{
+ * 1. without sbt:
+ * bin/spark-submit --class <this class>
+ * --jars <spark core test jar>,<spark catalyst test jar> <sql core
test jar>
+ * 2. build/sbt "sql/Test/runMain <this class>"
+ * 3. generate result:
+ * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this
class>"
+ * Results will be written to
+ * "benchmarks/VariantShreddedPredicatePushdownBenchmark-results.txt".
+ * }}}
+ */
+object VariantShreddedPredicatePushdownBenchmark extends SqlBasedBenchmark {
+
+ private val N = 20 * 1024 * 1024
+ private val NUMBER_OF_ITER = 10
+
+ // A single-column shredded Variant dataset with an object field `a` sorted
ascending, so that a
+ // literal predicate on `a` maps to a contiguous range of row groups.
+ private val df: DataFrame = spark
+ .range(0, N, 1, 1)
+ .selectExpr("parse_json('{\"a\":' || id || '}') AS v")
+
+ // Same, but every row also carries a key `z` outside the shredding schema,
so the whole partial
+ // object lands in the top-level residual `v.value` (non-null on every row).
`a` is still fully
+ // shredded into the typed leaf. This is the normal layout for real Variant
data (the inferred
+ // shredding schema is capped, so extra keys are common), and it is where
the flat
+ // `or(leaf, isNotNull(residual)...)` guard could never skip -- the tighter
+ // `or(leaf, and(anyResidualNotNull, isNull(leaf)))` guard skips via the
"leaf has no nulls" arm.
+ private val dfPartialObject: DataFrame = spark
+ .range(0, N, 1, 1)
+ .selectExpr("parse_json('{\"a\":' || id || ', \"z\":\"outside\"}') AS v")
+
+ // Confs to write the Variant column shredded, forcing `a` to a bigint typed
leaf. A small block
+ // size makes the writer emit many row groups per file.
+ private val writeConf = Seq(
+ SQLConf.VARIANT_WRITE_SHREDDING_ENABLED.key -> "true",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true",
+ SQLConf.VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key -> "a bigint")
+
+ private def addCase(
+ benchmark: Benchmark,
+ inputPath: String,
+ enablePushdown: String,
+ name: String,
+ withFilter: DataFrame => DataFrame): Unit = {
+ val loadDF = spark.read.parquet(inputPath).selectExpr("variant_get(v,
'$.a', 'bigint') AS a")
+ benchmark.addCase(name) { _ =>
+ withSQLConf(
+ SQLConf.VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED.key ->
enablePushdown,
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true") {
+ withFilter(loadDF).noop()
+ }
+ }
+ }
+
+ // A tiny block size makes one file hold many row groups, which is the
layout row-group skipping
+ // needs. `blockSize = None` uses the Parquet default (one row group for
this dataset), used to
+ // show the skip-none overhead disappears when a file is not deliberately
dense.
+ private def createAndRunBenchmark(
+ name: String,
+ withFilter: DataFrame => DataFrame,
+ data: DataFrame = df,
+ blockSize: Option[Int] = Some(128 * 1024)): Unit = {
+ withTempPath { tempDir =>
+ val outputPath = tempDir.getCanonicalPath
+ withSQLConf(writeConf: _*) {
+ val writer = data.write.mode(SaveMode.Overwrite)
+ blockSize.foreach(bs => writer.option("parquet.block.size",
bs.toString))
+ writer.parquet(outputPath)
+ }
+ val benchmark = new Benchmark(name, N, NUMBER_OF_ITER, output = output)
+ addCase(benchmark, outputPath, enablePushdown = "false",
+ "Without shredded predicate pushdown", withFilter)
+ addCase(benchmark, outputPath, enablePushdown = "true",
+ "With shredded predicate pushdown", withFilter)
+ benchmark.run()
+ }
+ }
+
+ /**
+ * Filter that matches nothing, so the leaf min/max lets Parquet skip every
row group when the
+ * optimization is on.
+ */
+ def runSkipAllRowGroups(): Unit = {
+ createAndRunBenchmark("Can skip all row groups", _.filter("a < 0"))
+ }
+
+ /**
+ * Highly selective filter matching only the last few row groups of the
sorted data.
+ */
+ def runSkipSomeRowGroups(): Unit = {
+ createAndRunBenchmark("Can skip some row groups", _.filter(s"a > ${(N *
0.99).toLong}"))
+ }
+
+ /**
+ * Filter that matches the whole range, so no row group can be skipped --
measures the overhead
+ * of building and evaluating the pushed predicate when it never helps.
Written with the tiny
+ * block size (many row groups), so this is the worst case for the overhead:
it is paid per row
+ * group. Compare with `runSkipNoRowGroupsDefaultBlockSize`.
+ */
+ def runSkipNoRowGroups(): Unit = {
+ createAndRunBenchmark("Can skip no row groups", _.filter(s"a >= 0 and a <=
$N"))
+ }
+
+ /**
+ * Same skip-none filter but written at the default Parquet block size (one
row group for this
+ * dataset), the layout a default-configured writer produces. The
per-row-group overhead
+ * effectively vanishes here -- it scales with row-group count, the same
knob as the benefit.
+ */
+ def runSkipNoRowGroupsDefaultBlockSize(): Unit = {
+ createAndRunBenchmark("Can skip no row groups (default block size)",
+ _.filter(s"a >= 0 and a <= $N"), blockSize = None)
+ }
+
+ /**
+ * Same selective filter as `runSkipSomeRowGroups`, but on data whose
objects carry a key outside
+ * the shredding schema (so the top-level residual is non-null on every
row). This is the layout
+ * where the earlier flat OR guard could never skip; the tighter guard still
skips here.
+ */
+ def runSkipSomeRowGroupsPartialObject(): Unit = {
+ createAndRunBenchmark("Can skip some row groups (partial object)",
+ _.filter(s"a > ${(N * 0.99).toLong}"), data = dfPartialObject)
+ }
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ runSkipAllRowGroups()
+ runSkipSomeRowGroups()
+ runSkipNoRowGroups()
+ runSkipNoRowGroupsDefaultBlockSize()
+ runSkipSomeRowGroupsPartialObject()
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
index 54c5b279f6ce..f023e8390f15 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
@@ -34,7 +34,7 @@ import org.apache.parquet.filter2.predicate.FilterApi._
import org.apache.parquet.filter2.predicate.Operators.{Column => _, Eq, Gt,
GtEq, In => FilterIn, Lt, LtEq, NotEq, UserDefinedByInstance}
import org.apache.parquet.hadoop.{ParquetFileReader, ParquetInputFormat,
ParquetOutputFormat}
import org.apache.parquet.hadoop.util.HadoopInputFile
-import org.apache.parquet.schema.MessageType
+import org.apache.parquet.schema.{MessageType, MessageTypeParser}
import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException}
import org.apache.spark.sql._
@@ -45,7 +45,7 @@ import
org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath
import org.apache.spark.sql.execution.ExplainMode
-import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
HadoopFsRelation, LogicalRelationWithTable, PushableColumnAndNestedColumn}
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
HadoopFsRelation, LogicalRelationWithTable, PushableColumnAndNestedColumn,
VariantMetadata}
import org.apache.spark.sql.execution.datasources.v2.ExtractV2Scan
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
import org.apache.spark.sql.functions._
@@ -82,13 +82,15 @@ abstract class ParquetFilterSuite extends ParquetTest with
SharedSparkSession {
protected def createParquetFilters(
schema: MessageType,
caseSensitive: Option[Boolean] = None,
- datetimeRebaseSpec: RebaseSpec =
RebaseSpec(LegacyBehaviorPolicy.CORRECTED)
+ datetimeRebaseSpec: RebaseSpec =
RebaseSpec(LegacyBehaviorPolicy.CORRECTED),
+ variantExtractionSchema: Option[StructType] = None
): ParquetFilters =
new ParquetFilters(schema, conf.parquetFilterPushDownDate,
conf.parquetFilterPushDownTimestamp,
conf.parquetFilterPushDownDecimal,
conf.parquetFilterPushDownStringPredicate,
conf.parquetFilterPushDownInFilterThreshold,
caseSensitive.getOrElse(conf.caseSensitiveAnalysis),
- datetimeRebaseSpec)
+ datetimeRebaseSpec,
+ variantExtractionSchema = variantExtractionSchema)
override def beforeEach(): Unit = {
super.beforeEach()
@@ -2422,6 +2424,409 @@ abstract class ParquetFilterSuite extends ParquetTest
with SharedSparkSession {
}
}
}
+
+ //
----------------------------------------------------------------------------------------------
+ // Shredded-variant filter pushdown (SPARK-55817).
+ //
+ // PushVariantIntoScan rewrites variant_get(v, '$.a', 'bigint') > 999 into a
struct-field access
+ // "v.`0`" > 999 where "0" carries VariantMetadata for path "$.a".
ParquetFilters maps that
+ // logical path to the physical shredded leaf v.typed_value.a.typed_value
and, for soundness,
+ // guards it so a row group is skipped only when the leaf cannot match AND
every value for the
+ // path is provably in the leaf (see `makeShreddedFilter`).
+ //
----------------------------------------------------------------------------------------------
+
+ /** A variant-extraction StructField named by ordinal, carrying
VariantMetadata for `path`. */
+ private def variantField(name: String, dt: DataType, path: String):
StructField =
+ StructField(name, dt, metadata = VariantMetadata(path, failOnError = true,
"UTC").toMetadata)
+
+ /**
+ * The variantExtractionSchema PushVariantIntoScan produces for a top-level
variant column
+ * `colName` with the given extraction fields (each an ordinal-named field
with VariantMetadata).
+ */
+ private def variantExtractionSchema(colName: String, fields: StructField*):
StructType =
+ StructType(Seq(StructField(colName, StructType(fields))))
+
+ test("shredded variant filter: single-level bigint resolves to leaf with
residual guards") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ val filter = pf.createFilter(sources.GreaterThan("v.`0`", 999L))
+ assert(filter.isDefined, "Expected shredded variant predicate to be
created")
+ val s = filter.get.toString
+ // Guarded shape: or(gt(leaf), and(or(notEq(residual)...), eq(leaf,
null))).
+ assert(s.contains("v.typed_value.a.typed_value"),
+ s"Expected leaf column v.typed_value.a.typed_value in $s")
+ assert(s.contains("v.typed_value.a.value"), s"Expected L1 residual guard
in $s")
+ assert(s.contains("v.value"), s"Expected top-level residual guard in $s")
+ // The leaf-is-null arm must be present (this is what keeps the guard
sound and effective).
+ assert(s.contains("eq(v.typed_value.a.typed_value, null)"),
+ s"Expected an isNull(leaf) guard arm in $s")
+ }
+
+ test("shredded variant filter: without variantExtractionSchema the logical
path is unknown") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val pf =
createParquetFilters(MessageTypeParser.parseMessageType(parquetSchema))
+ assert(pf.createFilter(sources.GreaterThan("v.`0`", 999L)).isEmpty)
+ }
+
+ test("shredded variant filter: string leaf and all comparison operators") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group b {
+ | optional binary value;
+ | optional binary typed_value (STRING);
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0",
StringType, "$.b"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ Seq(
+ sources.EqualTo("v.`0`", "str"),
+ sources.LessThan("v.`0`", "str"),
+ sources.LessThanOrEqual("v.`0`", "str"),
+ sources.GreaterThan("v.`0`", "str"),
+ sources.GreaterThanOrEqual("v.`0`", "str")).foreach { f =>
+ val filter = pf.createFilter(f)
+ assert(filter.isDefined, s"Expected $f to push down")
+ val s = filter.get.toString
+ assert(s.contains("v.typed_value.b.typed_value"), s"Expected leaf column
in $s for $f")
+ assert(s.contains("v.typed_value.b.value") && s.contains("v.value"),
+ s"Expected residual guards in $s for $f")
+ }
+ }
+
+ test("shredded variant filter: In pushes OR of guarded equalities") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ val filter = pf.createFilter(sources.In("v.`0`", Array[Any](1L, 2L, 3L)))
+ assert(filter.isDefined, "Expected In on shredded column to push down")
+ val s = filter.get.toString
+ assert(s.contains("v.typed_value.a.typed_value"), s"Expected leaf column
in $s")
+ assert(s.contains("v.typed_value.a.value") && s.contains("v.value"),
+ s"Expected residual guards in $s")
+ }
+
+ test("shredded variant filter: multi-level path resolves with a residual per
level") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group b {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a.b"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ val filter = pf.createFilter(sources.GreaterThan("v.`0`", 5L))
+ assert(filter.isDefined, "Expected multi-level shredded predicate to be
created")
+ val s = filter.get.toString
+ assert(s.contains("v.typed_value.a.typed_value.b.typed_value"),
+ s"Expected multi-level leaf column in $s")
+ // Three residual guards: L0, L1 (a), and leaf-level sibling (b).
+ assert(s.contains("v.value"), s"Expected L0 residual guard in $s")
+ assert(s.contains("v.typed_value.a.value"), s"Expected L1 residual guard
in $s")
+ assert(s.contains("v.typed_value.a.typed_value.b.value"),
+ s"Expected leaf-level residual guard in $s")
+ }
+
+ test("shredded variant filter: array-index path is rejected") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a[0]"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ assert(pf.createFilter(sources.GreaterThan("v.`0`", 999L)).isEmpty,
+ "Array-index paths must not resolve to a shredded leaf")
+ }
+
+ test("shredded variant filter: synthetic fields (placeholder / companion)
resolve to None") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val placeholder = variantField("0", BooleanType, "$.__placeholder_field__")
+ val pfPlaceholder = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ variantExtractionSchema = Some(variantExtractionSchema("v",
placeholder)))
+ assert(pfPlaceholder.createFilter(sources.EqualTo("v.`0`", true)).isEmpty,
+ "Placeholder field must not resolve to a shredded leaf")
+
+ // Full-variant passthrough path "$" yields no keys -> None.
+ val passthrough = variantField("0", LongType, "$")
+ val pfPassthrough = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ variantExtractionSchema = Some(variantExtractionSchema("v",
passthrough)))
+ assert(pfPassthrough.createFilter(sources.GreaterThan("v.`0`",
1L)).isEmpty,
+ "Full-variant passthrough must not resolve to a shredded leaf")
+ }
+
+ test("shredded variant filter: absent shredded field resolves to None") {
+ // The physical schema does not shred `a` (no typed_value.a subtree); the
value lives entirely
+ // in the opaque residual. Nothing should be pushed.
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group c {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ assert(pf.createFilter(sources.GreaterThan("v.`0`", 999L)).isEmpty,
+ "A path not shredded in this file must not be pushed")
+ }
+
+ test("shredded variant filter: case-insensitive column name resolves; keys
stay exact-case") {
+ // The top-level variant column name is a Spark identifier, matched
case-insensitively.
+ // The object key is variant data, matched exact-case, so the physical key
must equal the
+ // requested path's key exactly.
+ val parquetSchema =
+ """message spark_schema {
+ | optional group V {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ // Logical column name `v` differs in case from physical `V`; key `a`
matches exactly.
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ caseSensitive = Some(false), variantExtractionSchema = Some(extraction))
+ val filter = pf.createFilter(sources.GreaterThan("v.`0`", 999L))
+ assert(filter.isDefined, "Case-insensitive column matching should resolve
the shredded leaf")
+
assert(filter.get.toString.toLowerCase(java.util.Locale.ROOT).contains("typed_value"),
+ s"Expected a typed_value leaf predicate, got ${filter.get}")
+ }
+
+ test("shredded variant filter: object key is matched case-sensitively even
when case-" +
+ "insensitive analysis") {
+ // Physical schema shreds a key `A` (uppercase). A request for `$.a`
(lowercase) must NOT bind
+ // to `A`, because variant keys are data resolved exact-case by the
reader. Binding to `A`
+ // would be unsound.
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group A {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ caseSensitive = Some(false), variantExtractionSchema = Some(extraction))
+ assert(pf.createFilter(sources.GreaterThan("v.`0`", 999L)).isEmpty,
+ "Key `$.a` must not case-insensitively bind to the physical `A` subtree")
+ }
+
+ test("shredded variant filter: negated predicate is not 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.
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
variantExtractionSchema = Some(extraction))
+ // != arrives as Not(EqualTo(...)); NOT IN as Not(In(...));
Not(GreaterThan(...)) likewise.
+ assert(pf.createFilter(sources.Not(sources.EqualTo("v.`0`",
700L))).isEmpty,
+ "Negated EqualTo on a shredded path must not be pushed")
+ assert(pf.createFilter(sources.Not(sources.In("v.`0`", Array[Any](1L,
2L)))).isEmpty,
+ "Negated In on a shredded path must not be pushed")
+ assert(pf.createFilter(sources.Not(sources.GreaterThan("v.`0`",
700L))).isEmpty,
+ "Negated GreaterThan on a shredded path must not be pushed")
+ // Still pushed inside an AND alongside a negated shredded predicate? The
AND can push the
+ // non-negated side, but the negated shredded conjunct itself must be
dropped.
+ val andFilter = pf.createFilter(
+ sources.And(sources.GreaterThan("v.`0`", 700L),
sources.Not(sources.EqualTo("v.`0`", 900L))))
+ assert(andFilter.isDefined, "AND should still push its non-negated
shredded conjunct")
+ assert(!andFilter.get.toString.contains("not("),
+ s"AND must not contain a negated shredded predicate, got
${andFilter.get}")
+ }
+
+ test("shredded variant filter: extraction type narrower than the leaf is not
pushed") {
+ // Physical leaf is a plain int (INT32). A smallint extraction must NOT be
pushed: the leaf
+ // min/max is over int values, so a row group holding only
out-of-int16-range values would be
+ // wrongly skipped, turning an eager INVALID_VARIANT_CAST into an empty
result.
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int32 typed_value (INTEGER(32,true));
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val narrower = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ variantExtractionSchema = Some(variantExtractionSchema("v",
variantField("0", ShortType,
+ "$.a"))))
+ assert(narrower.createFilter(sources.GreaterThan("v.`0`",
5.toShort)).isEmpty,
+ "A smallint extraction against an int leaf must not be pushed")
+ // The exact type (int extraction against int leaf) is pushed.
+ val exact = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ variantExtractionSchema = Some(variantExtractionSchema("v",
variantField("0", IntegerType,
+ "$.a"))))
+ assert(exact.createFilter(sources.GreaterThan("v.`0`", 5)).isDefined,
+ "An int extraction against an int leaf should be pushed")
+ }
+
+ test("shredded variant filter: extraction type wider than the leaf is pushed
(safe widening)") {
+ // Physical leaf is a plain int (INT32); a bigint extraction is sound to
push because every int
+ // value casts to bigint losslessly and ordering is preserved.
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int32 typed_value (INTEGER(32,true));
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ variantExtractionSchema = Some(variantExtractionSchema("v",
variantField("0", LongType,
+ "$.a"))))
+ // A long literal within int range widens to the int leaf and is pushed.
+ val filter = pf.createFilter(sources.GreaterThan("v.`0`", 5L))
+ assert(filter.isDefined, "A bigint extraction over an int leaf should push
(safe widening)")
+ assert(filter.get.toString.contains("v.typed_value.a.typed_value"),
+ s"Expected the int leaf column in ${filter.get}")
+ // A long literal outside int range is still rejected by
valueMatchesParquetType.
+ assert(pf.createFilter(sources.GreaterThan("v.`0`", Int.MaxValue.toLong +
1L)).isEmpty,
+ "An out-of-int-range literal must not be pushed against an int leaf")
+ }
+
+ test("shredded variant filter: large In above threshold still pushes") {
+ val parquetSchema =
+ """message spark_schema {
+ | optional group v {
+ | optional binary value;
+ | optional group typed_value {
+ | optional group a {
+ | optional binary value;
+ | optional int64 typed_value;
+ | }
+ | }
+ | }
+ |}""".stripMargin
+ val extraction = variantExtractionSchema("v", variantField("0", LongType,
"$.a"))
+ withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> "2") {
+ val pf = createParquetFilters(
+ MessageTypeParser.parseMessageType(parquetSchema),
+ variantExtractionSchema = Some(extraction))
+ // 3 values > threshold 2: the regular path uses FilterApi.in; the
shredded path must too,
+ // OR-ed with the residual guards, rather than returning None.
+ val filter = pf.createFilter(sources.In("v.`0`", Array[Any](1L, 2L, 3L)))
+ assert(filter.isDefined, "A large In on a shredded path should still
push down")
+ val s = filter.get.toString
+ assert(s.contains("v.typed_value.a.typed_value"), s"Expected leaf column
in $s")
+ assert(s.contains("v.typed_value.a.value") && s.contains("v.value"),
+ s"Expected residual guards in $s")
+ }
+ }
}
@ExtendedSQLTest
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala
new file mode 100644
index 000000000000..9fd3b2f63e88
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala
@@ -0,0 +1,445 @@
+/*
+ * 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.SparkException
+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 guards it so a row group is skipped only when the
leaf cannot match AND
+ * every value for the path is provably in the leaf (see `makeShreddedFilter`).
+ *
+ * Scope: the optimization fires on the DSv1 read path only. DSv2 does rewrite
variant extractions
+ * into `v.`0`` struct accesses, but only after filter pushdown has run, so
the filters offered to
+ * the Parquet scan builder are still `variant_get(v, ...)` predicates and
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 = true): 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
+ }
+ }
+
+ // Assert that a row group whose only match for `$.a > 999` is a residual
fallback (a value the
+ // int64 leaf cannot hold, so it lands in v.typed_value.a.value with the
leaf NULL) is not
+ // dropped. `fallbackJson` is the JSON for the fallback row. The leaf
min/max is 0..49, so only
+ // the guard keeps the row group; a leaf-only predicate would drop it and
lose the row (#54598).
+ private def checkResidualFallbackNotDropped(fallbackJson: String): Unit = {
+ withTempDir { dir =>
+ val jsonExpr =
+ "case when id = 50 then '" + fallbackJson + "' 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")
+ val expected = baseline(read)
+ assert(expected == Seq(Row(1500L)), s"baseline should return the
fallback row, got $expected")
+
+ forEachReader { (dsv1, vectorized) =>
+ // The row group's only match is in the residual with a NULL leaf, so
the guard must keep
+ // it: results include the fallback row and the row group is not
skipped.
+ checkAnswer(read, expected)
+ if (dsv1 && vectorized) {
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a")
+ assert(countRowGroupsRead(read) == countRowGroupsRead(all),
+ "Row group whose only match is a residual fallback must NOT be
skipped")
+ }
+ }
+ }
+ }
+
+ test("residual fallback beyond the leaf's min/max is not dropped") {
+ // Different fallback encodings that all miss the int64 leaf: a
non-integral decimal that
+ // 1500.5 rounds to 1500, and a string "1500". Both are read back as
bigint 1500.
+ checkResidualFallbackNotDropped("{\"a\":1500.5}")
+ checkResidualFallbackNotDropped("{\"a\":\"1500\"}")
+ }
+
+ test("negated predicate over an all-fallback row group is not dropped") {
+ withTempDir { dir =>
+ // `a` shredded as bigint. Both rows are non-integral decimals the int64
leaf cannot hold, so
+ // both land in the residual (typed leaf entirely NULL, residual has no
nulls). The path is
+ // still pushable (bigint extraction over a bigint leaf). A naive
negated push would rewrite
+ // `!= 700` into and(notEq(leaf), eq(residual, null)) and skip the row
group -- losing both
+ // rows. The negation guard must prevent pushing, so {500, 600} come
back.
+ val jsonExpr = "case when id = 0 then '{\"a\":500.5}' else
'{\"a\":600.5}' end"
+ writeShredded(dir, "a bigint", jsonExpr, numRows = 2, blockSize = 1024 *
1024)
+
+ Seq(
+ "try_variant_get(v, '$.a', 'bigint') != 700" -> Seq(Row(500L),
Row(600L)),
+ "try_variant_get(v, '$.a', 'bigint') NOT IN (700, 800)" ->
Seq(Row(500L), Row(600L))
+ ).foreach { case (predicate, want) =>
+ def read: DataFrame = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a")
+ .where(predicate)
+ assert(baseline(read).sortBy(_.getLong(0)) == want, s"baseline for
$predicate")
+ forEachReader { (_, _) =>
+ checkAnswer(read, want)
+ }
+ }
+ }
+ }
+
+ test("type-mismatch fallback: string values in a numeric-shredded field are
not dropped") {
+ withTempDir { dir =>
+ // `a` shredded as bigint. Even rows -> a is a number (typed); odd rows
-> a is a string
+ // (type mismatch -> residual, typed_value NULL). Use try_variant_get so
the string rows
+ // resolve to NULL (filtered out) rather than raising a strict-cast
error, and assert the
+ // matching numeric rows are still returned.
+ val jsonExpr =
+ "case when id % 2 = 0 then '{\"a\":' || (id + 1000) || '}' " +
+ "else '{\"a\":\"str' || id || '\"}' end"
+ writeShredded(dir, "a bigint", jsonExpr, numRows = 20, blockSize = 1024
* 1024)
+
+ def read: DataFrame = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a")
+ .where("a > 1005")
+ val expected = baseline(read)
+ assert(expected.nonEmpty, "baseline should return the matching numeric
rows")
+
+ forEachReader { (_, _) =>
+ checkAnswer(read, expected)
+ }
+ }
+ }
+
+ test("file without the shredded path: value read from residual, predicate
not pushed") {
+ withTempDir { dir =>
+ // Force a shredding schema that does NOT contain `a`; `$.a` lives
entirely in the opaque
+ // top-level residual. Nothing is pushed for `$.a`; results must still
be correct.
+ val jsonExpr = "'{\"a\":' || id || '}'"
+ writeShredded(dir, "b bigint", jsonExpr, numRows = 20, blockSize = 1024
* 1024)
+
+ def read: DataFrame = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a")
+ .where("a > 9")
+ val expected = baseline(read)
+ assert(expected == (10L to 19L).map(Row(_)), s"unexpected baseline:
$expected")
+
+ forEachReader { (_, _) =>
+ checkAnswer(read.orderBy("a"), expected)
+ }
+ }
+ }
+
+ test("residual-null happy path: a row group is skipped (DSv1) and results
are correct") {
+ withTempDir { dir =>
+ // Homogeneous typed data across two row groups. All values shred
cleanly (residuals all
+ // NULL), so the optimization fires and one row group is skipped on DSv1.
+ val jsonExpr = "'{\"a\":' || id || '}'"
+ // Small block size -> at least two row groups: [0,999] and [1000,1999].
+ writeShredded(dir, "a bigint", jsonExpr, numRows = 2000, blockSize = 512)
+
+ forEachReader { (dsv1, vectorized) =>
+ val filtered = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a")
+ .where("a > 999")
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a")
+ checkAnswer(filtered.orderBy("a"), (1000L to 1999L).map(Row(_)))
+ if (dsv1 && vectorized) {
+ assert(countRowGroupsRead(filtered) < countRowGroupsRead(all),
+ "Expected at least one row group to be skipped by the shredded
leaf statistics")
+ }
+ }
+ }
+ }
+
+ test("partial object with a non-shredded sibling key still skips (leaf has
no nulls)") {
+ withTempDir { dir =>
+ // Every row also carries a key `z` outside the shredding schema, so the
whole partial object
+ // lands in the top-level residual v.value -- it is non-null on every
row. `a` is still fully
+ // shredded into the typed leaf (no nulls). The flat OR guard could
never skip here (v.value
+ // never all-null); the tighter guard skips via the "leaf has no nulls"
arm. Sorted on `a`
+ // across two row groups so `a > 999` can drop the first.
+ val jsonExpr = "'{\"a\":' || id || ', \"z\":\"outside\"}'"
+ writeShredded(dir, "a bigint", jsonExpr, numRows = 2000, blockSize = 512)
+
+ forEachReader { (dsv1, vectorized) =>
+ val filtered = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a").where("a > 999")
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a")
+ checkAnswer(filtered.orderBy("a"), (1000L to 1999L).map(Row(_)))
+ if (dsv1 && vectorized) {
+ assert(countRowGroupsRead(filtered) < countRowGroupsRead(all),
+ "Expected skipping despite a non-null top-level residual (partial
object)")
+ }
+ }
+ }
+ }
+
+ test("multi-level $.a.b: skip fires on DSv1 and results are correct") {
+ withTempDir { dir =>
+ // `a` shredded as struct<b bigint>. Homogeneous nested typed data
across two row groups so
+ // the skip fires on the nested leaf
`v.typed_value.a.typed_value.b.typed_value`.
+ val typedJson = "'{\"a\":{\"b\":' || id || '}}'"
+ writeShredded(dir, "a struct<b bigint>", typedJson, numRows = 2000,
blockSize = 512)
+
+ forEachReader { (dsv1, vectorized) =>
+ val filtered = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a.b', 'bigint') AS b")
+ .where("b > 999")
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a.b', 'bigint') AS b")
+ checkAnswer(filtered.orderBy("b"), (1000L to 1999L).map(Row(_)))
+ if (dsv1 && vectorized) {
+ assert(countRowGroupsRead(filtered) < countRowGroupsRead(all),
+ "Expected a row group to be skipped by the nested shredded leaf
statistics")
+ }
+ }
+ }
+ }
+
+ test("multi-level $.a.b: nested residual fallback beyond the leaf's min/max
is not dropped") {
+ withTempDir { dir =>
+ // `a` shredded as struct<b bigint>. Rows 0..19 shred cleanly, so the
nested leaf
+ // v.typed_value.a.typed_value.b.typed_value 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 -- this makes the
+ // nested leaf-level residual load-bearing (a leaf-only predicate would
drop it).
+ val jsonExpr =
+ "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)
+
+ def read: DataFrame = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a.b', 'bigint') AS b")
+ .where("b > 999")
+ assert(baseline(read) == Seq(Row(1500L)), "baseline should return the
nested fallback row")
+
+ forEachReader { (dsv1, vectorized) =>
+ checkAnswer(read, Seq(Row(1500L)))
+ if (dsv1 && vectorized) {
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a.b', 'bigint') AS b")
+ assert(countRowGroupsRead(read) == countRowGroupsRead(all),
+ "Row group whose only match is a nested residual fallback must NOT
be skipped")
+ }
+ }
+ }
+ }
+
+ test("unannotated variant layout: skip and fallback still work") {
+ // The suite writes the production-default annotated layout everywhere
else; this test covers
+ // the unannotated physical layout explicitly. Both a skip-eligible query
and a residual
+ // fallback must behave correctly.
+ withTempDir { dir =>
+ writeShredded(dir, "a bigint", "'{\"a\":' || id || '}'", numRows = 2000,
blockSize = 512,
+ annotate = false)
+ forEachReader { (dsv1, vectorized) =>
+ val filtered = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a").where("a > 999")
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a")
+ checkAnswer(filtered.orderBy("a"), (1000L to 1999L).map(Row(_)))
+ if (dsv1 && vectorized) {
+ assert(countRowGroupsRead(filtered) < countRowGroupsRead(all),
+ "Expected a row group to be skipped on the unannotated layout")
+ }
+ }
+ }
+
+ withTempDir { dir =>
+ // Residual fallback under the unannotated layout: the row whose only
match is in the residual
+ // (NULL leaf) must survive.
+ writeShredded(dir, "a bigint",
+ "case when id = 50 then '{\"a\":1500.5}' else '{\"a\":' || id || '}'
end",
+ numRows = 51, blockSize = 1024 * 1024, annotate = false)
+ def read: DataFrame = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a").where("a >
999")
+ assert(baseline(read) == Seq(Row(1500L)))
+ forEachReader { (_, _) =>
+ checkAnswer(read, Seq(Row(1500L)))
+ }
+ }
+ }
+
+ test("deferCastError=true: strict non-string cast does not fire;
try_variant_get still does") {
+ // With deferCastError, a strict cast to a non-string, non-variant type is
rewritten into
+ // UnwrapVariantCastError, which is not translated to a pushable filter --
so shredded pushdown
+ // does not fire for it (results still correct). try_variant_get
(failOnError=false) and string
+ // targets are unaffected and still fire. Results must be correct in every
combination.
+ withTempDir { dir =>
+ writeShredded(dir, "a bigint", "'{\"a\":' || id || '}'", numRows = 2000,
blockSize = 512)
+ Seq("true", "false").foreach { defer =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
+ SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> defer,
+ SQLConf.VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true") {
+ withClue(s"(deferCastError=$defer) ") {
+ // Strict cast: correct either way (does not fire when defer=true).
+ val strict = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'bigint') AS a").where("a >
999")
+ checkAnswer(strict.orderBy("a"), (1000L to 1999L).map(Row(_)))
+ // try_variant_get: unaffected by deferCastError and still skips a
row group on DSv1.
+ val tryGet = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a").where("a
> 999")
+ val all = spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("try_variant_get(v, '$.a', 'bigint') AS a")
+ checkAnswer(tryGet.orderBy("a"), (1000L to 1999L).map(Row(_)))
+ assert(countRowGroupsRead(tryGet) < countRowGroupsRead(all),
+ "try_variant_get should still skip regardless of deferCastError")
+ }
+ }
+ }
+ }
+ }
+
+ test("strict variant_get preserves INVALID_VARIANT_CAST on a residual
fallback (not empty)") {
+ // The worst failure mode: a thrown cast error silently becoming an empty
result. `a` shredded
+ // as int; row 50 stores 3000000000, which overflows int32 so it lands in
the residual with the
+ // leaf NULL (extraction type matches the leaf exactly, so the path is
pushed). With
+ // deferCastError=false (default) the scan casts eagerly, so strict
`variant_get(v,'$.a','int')`
+ // must raise INVALID_VARIANT_CAST -- a leaf-only push would drop the row
group (leaf max 49)
+ // and return empty instead. The guard keeps the row group, so the error
is preserved.
+ withTempDir { dir =>
+ val jsonExpr = "case when id = 50 then '{\"a\":3000000000}' else
'{\"a\":' || id || '}' end"
+ writeShredded(dir, "a int", jsonExpr, numRows = 51, blockSize = 1024 *
1024)
+ Seq("parquet", "").foreach { useV1 =>
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> useV1,
+ SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "false",
+ 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"(useV1='$useV1', vectorized=$vectorized) ") {
+ val e = intercept[SparkException] {
+ spark.read.parquet(dir.getAbsolutePath)
+ .selectExpr("variant_get(v, '$.a', 'int') AS a").where("a >
999").collect()
+ }
+ assert(findCause(e, "INVALID_VARIANT_CAST"),
+ s"Expected INVALID_VARIANT_CAST to be preserved, got: $e")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ // Walk an exception's cause chain for a Spark error condition (message
substring).
+ private def findCause(e: Throwable, condition: String): Boolean = {
+ var cur: Throwable = e
+ while (cur != null) {
+ if (Option(cur.getMessage).exists(_.contains(condition))) return true
+ cur = cur.getCause
+ }
+ false
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
