peter-toth commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3816013473
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7037,6 +7037,31 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED =
+ buildConf("spark.sql.variant.shreddedPredicatePushdown.enabled")
+ .internal()
+ .doc("When true, comparison predicates on shredded Variant fields
produced by " +
+ "PushVariantIntoScan (e.g. variant_get(v, '$.a', 'bigint') > 999) are
pushed to Parquet " +
+ "as the predicate on the physical shredded typed_value leaf column
OR-ed with an " +
+ "IS NOT NULL check on every untyped residual value column along the
path, so that a row " +
Review Comment:
**Finding 8.** 4fce3cf changed the pushed predicate from the flat `or(leaf,
isNotNull(residual)...)` to `or(leaf, and(anyResidualNotNull, isNull(leaf)))`,
but every prose description of it still states the old shape and the old skip
rule. This doc line is the user-visible one: with the second arm a row group is
also skipped when the leaf column has **zero nulls**, regardless of what the
residuals hold. That is the whole point of the change (it is what makes the
partial-object layout skip) and it is exactly what a reader needs in order to
judge soundness, so it should not be the one sentence that is missing.
The other five places:
- the PR description — "To stay sound, the pushed predicate is:
`or(leafPredicate, isNotNull(residual_0), isNotNull(residual_1), ...)`" and "a
row group is skipped only when the leaf min/max cannot match **and** every
residual is entirely NULL". Both now false.
- `ParquetFilters.scala:1068` — "Each pushes or(leafPredicate,
isNotNull(residual)...) over every residual `value` column along the path".
- `ParquetFilters.scala:976` — `referencesShreddedName`'s rationale derives
the unsound negation from the flat shape. With the current shape
`LogicalInverseRewriter` produces `and(not(leafPred), or(and(eq(residual_i,
null)...), notEq(leaf, null)))`, which is droppable as soon as some residual
has no nulls **and** the leaf is entirely NULL — precisely the all-fallback row
group in the new negation test, which is why removing the guard fails it. The
guard is right; the derivation shown is not the one that applies.
- `ParquetFilterSuite.scala:2434` — "OR-s an IS NOT NULL guard on every
residual `value` column along the path".
- `VariantShreddingFilterPushdownSuite.scala:35` — same wording.
`ParquetFilterSuite.scala:2434` is the line @dongjoon-hyun flagged for
describing a shape the code did not build, and 20ff8d1 fixed it; 4fce3cf
re-broke it. Worth fixing all six in one pass so the next reader does not have
to work out which description is current.
While in the description: "How was this patch tested?" still lists only the
round-1 test set. The negation, safe-widening, narrowing-rejection, large-`In`,
partial-object, unannotated-layout and `deferCastError` tests, and the new
benchmark, are all missing from it.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,387 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.AccumulatorContext
+
+/**
+ * End-to-end tests for row-group skipping on shredded Variant columns in
Parquet (SPARK-55817).
+ *
+ * When a Variant column is written with shredding enabled, each extracted
scalar field is stored
+ * as a typed Parquet leaf column (e.g. `v.typed_value.a.typed_value` for
`$.a`) carrying min/max
+ * statistics. On the DSv1 path, PushVariantIntoScan rewrites
+ * `variant_get(v, '$.a', 'bigint') > 999` into a struct-field access `v.`0` >
999`, and (when
+ * `spark.sql.variant.shreddedPredicatePushdown.enabled` is true)
ParquetFilters maps `v.`0`` to
+ * the physical leaf and OR-s in an IS NOT NULL guard on every untyped
residual `value` column
+ * along the path.
+ *
+ * Scope: the optimization fires on the DSv1 read path only. On the DSv2 path
variant extraction is
+ * pushed through the separate SupportsPushDownVariantExtractions mechanism,
and the filter is never
+ * rewritten into `v.`0``, so it cannot be pushed for row-group skipping (see
the comment in
+ * ParquetScanBuilder). DSv2 reads remain correct -- the variant filter is
applied post-scan -- they
+ * just do not skip row groups. These tests therefore assert skipping only on
DSv1, and assert
+ * correctness on both DSv1 and DSv2.
+ *
+ * The central correctness concern is soundness under fallback: shredding is
per-row and per-file
+ * best-effort, so values that don't fit the shredded type (overflow / type
mismatch) or that are
+ * in a file that doesn't shred the path are stored in an opaque residual with
`typed_value` NULL.
+ * Parquet min/max excludes NULLs, so a naive leaf-only predicate could skip a
row group that still
+ * holds a matching row. These tests mix typed and fallback rows in a single
row group and assert
+ * that no matching row is ever dropped and results equal the no-pushdown
baseline.
+ */
+class VariantShreddingFilterPushdownSuite extends QueryTest with ParquetTest
+ with SharedSparkSession {
+
+ // Base configs to write shredded Variant Parquet files. `annotate` controls
whether the physical
+ // variant group carries the VARIANT logical-type annotation (the production
default is true).
+ private def writeConf(forceSchema: String, annotate: Boolean): Seq[(String,
String)] = Seq(
+ SQLConf.VARIANT_WRITE_SHREDDING_ENABLED.key -> "true",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true",
+ SQLConf.VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key -> forceSchema,
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> annotate.toString)
+
+ /**
+ * Counts how many Parquet row groups are actually read by the given
DataFrame, using the
+ * accumulator technique from ParquetFilterSuite. Only meaningful with the
vectorized reader,
+ * which reports the row-group count into a registered NumRowGroupsAcc.
+ */
+ private def countRowGroupsRead(df: DataFrame): Int = {
+ val accu = new NumRowGroupsAcc
+ sparkContext.register(accu)
+ try {
+ df.foreachPartition((it: Iterator[Row]) => it.foreach(_ => accu.add(0)))
+ accu.value
+ } finally {
+ AccumulatorContext.remove(accu.id)
+ }
+ }
+
+ /**
+ * Writes a JSON-per-row Variant Parquet file coalesced to a single
partition with a tiny block
+ * size so the writer emits multiple row groups. `jsonExpr` is the SQL
expression producing the
+ * JSON string per `id` in `range(0, numRows, 1, 1)`.
+ */
+ private def writeShredded(
+ dir: File,
+ forceSchema: String,
+ jsonExpr: String,
+ numRows: Int,
+ blockSize: Int = 512,
+ annotate: Boolean = 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
+ }
+ }
+
+ test("residual fallback beyond the leaf's min/max is not dropped") {
+ // The guard is load-bearing only when (a) the fallback row is the sole
match and (b) the leaf
+ // min/max cannot match the literal on its own. `a` is shredded as bigint;
id 0..49 -> a = id
+ // (typed leaf, min 0 max 49). id 50 -> a = 1500.5, a decimal the int64
leaf cannot hold, so it
+ // lands in v.typed_value.a.value with typed_value NULL. One row group: `a
> 999` matches only
+ // that row, and the leaf min/max (<=49) cannot match it, so only the
guard keeps the row group.
+ // A leaf-only predicate would drop it and lose the row (the #54598 bug).
+ Seq(
+ // Different fallback encodings that all miss the int64 leaf.
+ "'{\"a\":1500.5}'" -> Row(1500L), // non-integral decimal
+ "'{\"a\":\"1500\"}'" -> Row(1500L) // string
+ ).foreach { case (fallbackJson, want) =>
+ withTempDir { dir =>
+ val jsonExpr = s"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(want), 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("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: fallback at an intermediate level is not dropped") {
Review Comment:
**Finding 2 (round 1).** The single-level fallback tests are load-bearing
now — I verified your check and it fails as you describe. This is the part of
finding 2 that is left: this test still decides nothing.
Measured on this head, three ways:
- reduce `makeShreddedFilter` to `leafPredicate` (the #54598 shape): this
test passes (only the single-level fallback test and the fallback half of
`unannotated variant layout` fail);
- keep only the leaf-level residual (`residuals.toSeq.takeRight(1)`,
dropping `v.value` and `v.typed_value.a.value`): the whole suite passes, 10/10;
- revert to the flat OR: this test passes.
The premise in the comment is what breaks it. Row 6 stores
`{"a":{"b":5000}}`, so the nested leaf's max is 5000 and `b > 999` can never be
dropped by leaf statistics — the row group is kept with or without any guard.
Row 5's `{"a":9999}` cannot be what the guard saves either, because `$.a.b`
over a scalar `a` is NULL and so can never match `b > 999`.
More generally, `VariantShreddingWriter.castShredded` always shreds an
object field that is in the shredding schema, routing only *non-schema* keys
into that level's own `value`. So with Spark's writer a value for `$.a.b` can
never sit behind a NULL `a.typed_value`, and the ancestor-level guards are
defensive against writers that legitimately decline to shred a level — real,
but not reachable from Spark. That belongs in the `residualFieldNames` doc at
`ParquetFilters.scala:134` rather than being asserted by a test that cannot
show it.
What this test *can* cover load-bearingly is the nested leaf-level residual,
by moving the fallback down onto `b`:
```scala
// `a` shredded as struct<b bigint>. Rows 0..19 shred cleanly, so the
nested leaf is min 0 /
// max 19. Row 20 stores `b` as a non-integral decimal the int64 leaf
cannot hold, so it lands
// in v.typed_value.a.typed_value.b.value with the leaf NULL. `b >
999` matches only that row
// and the leaf min/max cannot match it, so only the guard keeps the
row group.
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)
```
with `try_variant_get(v, '$.a.b', 'bigint') > 999` and the same "row group
not skipped" assertion as the single-level test.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/VariantShreddedPredicatePushdownBenchmark.scala:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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()
+ }
+ }
+ }
+
+ private def createAndRunBenchmark(
+ name: String,
+ withFilter: DataFrame => DataFrame,
+ data: DataFrame = df): Unit = {
+ withTempPath { tempDir =>
+ val outputPath = tempDir.getCanonicalPath
+ withSQLConf(writeConf: _*) {
+ data.write.mode(SaveMode.Overwrite)
+ .option("parquet.block.size", (128 * 1024).toString)
+ .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.
+ */
+ def runSkipNoRowGroups(): Unit = {
+ createAndRunBenchmark("Can skip no row groups", _.filter(s"a >= 0 and a <=
$N"))
+ }
+
+ /**
+ * 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()
+ runSkipSomeRowGroupsPartialObject()
Review Comment:
**Finding 9.** The three checked-in `*-results.txt` are stale in two
independent ways.
1. They were generated at cef54f6 / cbb5644 / e005f3c, all *before* 4fce3cf
changed the pushed predicate from `or(leaf, isNotNull(residual)...)` to
`or(leaf, and(anyResidualNotNull, isNull(leaf)))`. None of the numbers — the
20-23x wins or the skip-none overhead — describe the predicate this PR ships.
2. This line adds a fourth case, but each results file holds only three
(`grep -c 'row groups:'` → 3).
That is more than bookkeeping here, because @qlong's default-on question is
still open and it turns on two numbers: the skip-none overhead (~3.5% in the
current files: 2480→2557, 1846→1911, 2396→2479 ms) and how often the
optimization can fire at all. The partial-object case is what answers the
second half — it is the layout that read 20/20 row groups under the flat OR and
skips under the new guard — and its row is exactly the one missing. Re-running
the benchmark action gives you both the corrected overhead and that row.
Related: your 2026-08-19 reply to @qlong argues the overhead cannot be
reduced because "`or(leaf, isNotNull(residual)...)` is a few disjuncts across a
few columns ... there isn't a part of it I can drop without giving that up".
4fce3cf makes that out of date in the direction that helps your case, so it is
worth restating with the new shape and the regenerated numbers.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7037,6 +7037,31 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val VARIANT_SHREDDED_PREDICATE_PUSHDOWN_ENABLED =
+ buildConf("spark.sql.variant.shreddedPredicatePushdown.enabled")
+ .internal()
+ .doc("When true, comparison predicates on shredded Variant fields
produced by " +
+ "PushVariantIntoScan (e.g. variant_get(v, '$.a', 'bigint') > 999) are
pushed to Parquet " +
+ "as the predicate on the physical shredded typed_value leaf column
OR-ed with an " +
+ "IS NOT NULL check on every untyped residual value column along the
path, so that a row " +
+ "group is skipped only when the leaf cannot match and every residual
is entirely null " +
+ "(i.e. the whole path is provably in the typed leaf). This enables
row-group skipping " +
+ "for shredded Variant columns while never dropping rows that fall back
to an untyped " +
+ "residual. The benefit depends on the data layout, like any Parquet
min/max skipping: it " +
+ "helps most when the data is sorted on the filtered field (so each row
group covers a " +
+ "narrow value range) and a file holds many row groups; unsorted data
or a single row " +
+ "group per file gains little. Has no effect unless the Parquet column
is shredded and " +
+ "spark.sql.variant.pushVariantIntoScan is also true, and it does not
fire when " +
Review Comment:
**Finding 10.** "it does not fire when
`spark.sql.variant.pushVariantIntoScan.deferCastError` is true" holds only for
a strict cast to a non-string, non-variant type.
`VariantInRelation.shouldWrapCastError` is `field.path.failOnError &&
deferCastErrorEnabled` with `VariantType | StringType` excluded up front, so
with `deferCastError = true`:
- `try_variant_get(v, '$.a', 'bigint') > 999` — `failOnError = false`, no
companion wrap, the filter stays a bare `GetStructField` and is pushed;
- `variant_get(v, '$.a', 'string') = 'x'` — `StringType`, same.
The test at line 365 only exercises strict `bigint`, so its name
(`deferCastError=true: optimization does not fire`) over-claims for the same
reason. Suggest scoping both — something like "does not fire for a strict cast
to a non-string type, where the extraction is wrapped in
`UnwrapVariantCastError` and is not translated to a pushable filter;
`try_variant_get` and string targets are unaffected" — and adding a
`try_variant_get` row to that test asserting skipping still happens with
`deferCastError = true`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -692,6 +964,80 @@ class ParquetFilters(
nameToParquetField.contains(name) && valueCanMakeFilterOn(name, value)
}
+ // Whether `name` is a shredded-variant logical path whose typed leaf
accepts `value`. `value`
+ // must be non-null: shredded pushdown only handles comparison predicates.
+ private def canMakeShreddedFilterOn(name: String, value: Any): Boolean = {
+ value != null && nameToShreddedVariantField.get(name).exists { f =>
+ valueMatchesParquetType(f.leaf.fieldType, value)
+ }
+ }
+
+ // Whether `predicate` references a shredded-variant logical path anywhere.
Used to refuse
+ // conversion under negation: the shredded predicate is `or(leaf,
isNotNull(residual)...)`, and
+ // `not(...)` of it is rewritten by parquet-mr's LogicalInverseRewriter into
+ // `and(notEq(leaf), eq(residual, null))`, whose `eq(residual, null)`
conjunct makes an AND
+ // row-group-droppable whenever the residual has no nulls -- unsound (drops
a row group whose
+ // matching values are all in the residual). Since a negated shredded
predicate cannot be
+ // expressed soundly with row-group statistics, we do not push it at all.
+ //
+ // `sources.Filter.references` already recurses through And/Or/Not and every
leaf filter, so this
+ // stays correct if new Filter subtypes are added.
+ private def referencesShreddedName(predicate: sources.Filter): Boolean =
+ predicate.references.exists(nameToShreddedVariantField.contains)
+
+ // Build the sound shredded-variant predicate:
+ // or(leafPredicate, 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
=>
+ val anyResidualNotNull = field.residualFieldNames
+ .map(n => FilterApi.notEq(binaryColumn(n), null.asInstanceOf[Binary]))
+ .reduceLeftOption[FilterPredicate](FilterApi.or)
+ val leafIsNull = makeEq.lift(field.leaf.fieldType)
+ .map(_(field.leaf.fieldNames, null))
+ (anyResidualNotNull, leafIsNull) match {
+ case (Some(residual), Some(isNull)) =>
+ FilterApi.or(leafPredicate, FilterApi.and(residual, isNull))
+ case (Some(residual), None) =>
Review Comment:
**Finding 12.** This branch is unreachable. `makeLeaf` returning `Some`
means the leaf type is matched by one of `makeEq` / `makeLt` / `makeLtEq` /
`makeGt` / `makeGtEq` / `makeInPredicate`, and `makeEq`'s case list is a
superset of the others — same types, same `pushDownDate` / `pushDownDecimal`
guards, plus `ParquetBooleanType` — so `makeEq.lift(field.leaf.fieldType)` is
defined whenever we get here. Every type `expectedLeafType` can return is in
`makeEq` as well.
Suggest dropping the branch (and its comment) so only `case (Some(residual),
Some(isNull))` and `case (None, _)` remain, or turning `leafIsNull` into a
direct `makeEq(field.leaf.fieldType)(...)` application to make the invariant
explicit rather than silently unexercised.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,387 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.AccumulatorContext
+
+/**
+ * End-to-end tests for row-group skipping on shredded Variant columns in
Parquet (SPARK-55817).
+ *
+ * When a Variant column is written with shredding enabled, each extracted
scalar field is stored
+ * as a typed Parquet leaf column (e.g. `v.typed_value.a.typed_value` for
`$.a`) carrying min/max
+ * statistics. On the DSv1 path, PushVariantIntoScan rewrites
+ * `variant_get(v, '$.a', 'bigint') > 999` into a struct-field access `v.`0` >
999`, and (when
+ * `spark.sql.variant.shreddedPredicatePushdown.enabled` is true)
ParquetFilters maps `v.`0`` to
+ * the physical leaf and OR-s in an IS NOT NULL guard on every untyped
residual `value` column
+ * along the path.
+ *
+ * Scope: the optimization fires on the DSv1 read path only. On the DSv2 path
variant extraction is
+ * pushed through the separate SupportsPushDownVariantExtractions mechanism,
and the filter is never
+ * rewritten into `v.`0``, so it cannot be pushed for row-group skipping (see
the comment in
+ * ParquetScanBuilder). DSv2 reads remain correct -- the variant filter is
applied post-scan -- they
+ * just do not skip row groups. These tests therefore assert skipping only on
DSv1, and assert
+ * correctness on both DSv1 and DSv2.
+ *
+ * The central correctness concern is soundness under fallback: shredding is
per-row and per-file
+ * best-effort, so values that don't fit the shredded type (overflow / type
mismatch) or that are
+ * in a file that doesn't shred the path are stored in an opaque residual with
`typed_value` NULL.
+ * Parquet min/max excludes NULLs, so a naive leaf-only predicate could skip a
row group that still
+ * holds a matching row. These tests mix typed and fallback rows in a single
row group and assert
+ * that no matching row is ever dropped and results equal the no-pushdown
baseline.
+ */
+class VariantShreddingFilterPushdownSuite extends QueryTest with ParquetTest
+ with SharedSparkSession {
+
+ // Base configs to write shredded Variant Parquet files. `annotate` controls
whether the physical
+ // variant group carries the VARIANT logical-type annotation (the production
default is true).
+ private def writeConf(forceSchema: String, annotate: Boolean): Seq[(String,
String)] = Seq(
+ SQLConf.VARIANT_WRITE_SHREDDING_ENABLED.key -> "true",
+ SQLConf.VARIANT_ALLOW_READING_SHREDDED.key -> "true",
+ SQLConf.VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key -> forceSchema,
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> annotate.toString)
+
+ /**
+ * Counts how many Parquet row groups are actually read by the given
DataFrame, using the
+ * accumulator technique from ParquetFilterSuite. Only meaningful with the
vectorized reader,
+ * which reports the row-group count into a registered NumRowGroupsAcc.
+ */
+ private def countRowGroupsRead(df: DataFrame): Int = {
+ val accu = new NumRowGroupsAcc
+ sparkContext.register(accu)
+ try {
+ df.foreachPartition((it: Iterator[Row]) => it.foreach(_ => accu.add(0)))
+ accu.value
+ } finally {
+ AccumulatorContext.remove(accu.id)
+ }
+ }
+
+ /**
+ * Writes a JSON-per-row Variant Parquet file coalesced to a single
partition with a tiny block
+ * size so the writer emits multiple row groups. `jsonExpr` is the SQL
expression producing the
+ * JSON string per `id` in `range(0, numRows, 1, 1)`.
+ */
+ private def writeShredded(
+ dir: File,
+ forceSchema: String,
+ jsonExpr: String,
+ numRows: Int,
+ blockSize: Int = 512,
+ annotate: Boolean = 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
+ }
+ }
+
+ test("residual fallback beyond the leaf's min/max is not dropped") {
Review Comment:
**Finding 11.** Every fallback test uses `try_variant_get`, so the property
the `resolveShredded` comment leans on to justify rejecting narrowing —
"changing an eager `INVALID_VARIANT_CAST` into an empty result" — has no test.
That is the worst failure mode this feature has (a thrown error silently
becoming an empty result), and it is reachable through the residual path with
an exactly-matching leaf type too, not only through narrowing.
Concretely: shred `a` as `int`, rows 0..49 as `{"a": id}`, row 50 as `{"a":
3000000000}` — `tryTypedShred`'s `INT` case rejects it (`value != (int)
value`), so it lands in `v.typed_value.a.value` with the leaf NULL — then
strict `variant_get(v, '$.a', 'int') > 999`. The extraction type matches the
leaf exactly, so the path is pushed. With `deferCastError = false` (the
default) the scan casts eagerly, so the baseline raises `INVALID_VARIANT_CAST`;
the leaf min/max is 0..49, so a leaf-only push would drop the row group and
return an empty result instead. A `checkError` case beside the existing
`1500.5` one would pin it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -56,25 +59,20 @@ class ParquetFilters(
pushDownStringPredicate: Boolean,
pushDownInFilterThreshold: Int,
caseSensitive: Boolean,
- datetimeRebaseSpec: RebaseSpec) {
+ datetimeRebaseSpec: RebaseSpec,
+ variantExtractionSchema: Option[StructType] = None) {
+ // Shredded-variant physical field-name constants. Declared first so they
are initialized before
Review Comment:
**Finding 13.** `nameToShreddedVariantField` is `lazy`, so it is not built
during construction and declaration order of these two is irrelevant (the
reason was true only before the map became lazy). Suggest dropping the
sentence, or replacing it with the constraint that does still bind — the map
has to be `lazy` because resolution reads the `Parquet*Type` vals declared
further down, which is already noted at its own definition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
