viirya commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3825575679
##########
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:
Fixed in 0e10378 -- updated all six descriptions to the current `or(leaf,
and(anyResidualNotNull, isNull(leaf)))` shape and the "leaf min/max can't match
AND (every residual null OR leaf has no nulls)" skip rule: the config doc, the
`referencesShreddedName` rationale (with the corrected negation derivation you
gave), the `createFilterHelper` branch comment, the `Not`-guard comment, and
both test headers. Also expanded "How was this patch tested?" in the PR
description with the negation / widening / narrowing-rejection / large-In /
partial-object / unannotated / deferCastError / error-preservation tests and
the benchmark.
##########
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:
Regenerating now -- I re-triggered the benchmark GitHub Action on 0e10378
for JDK 17/21/25 (create-commit), so the golden files will reflect the current
predicate and include the partial-object case (the row that answers the
default-on question). Will confirm once the results commits land.
##########
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:
You're right, verified -- the old version passed with the whole guard
removed. Fixed in 0e10378 using your suggestion: `a struct<b bigint>`, rows
0..19 clean, row 20 stores `b` as `1500.5` (int64 leaf can't hold it -> nested
leaf-level residual, leaf NULL), `try_variant_get(v,'$.a.b','bigint') > 999`.
Now it fails with the guard removed. Also moved the "Spark's writer never puts
a value behind a NULL intermediate leaf; the ancestor-level guards are for
writers that decline to shred a level" note into the `residualFieldNames` doc,
as you suggested.
##########
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:
Fixed in 0e10378. Scoped the config doc to "a strict cast to a non-string
type" and noted `try_variant_get` and string targets are unaffected. The test
now also asserts `try_variant_get(v,'$.a','bigint')` still fires and skips a
row group with `deferCastError=true`.
##########
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:
Added in 0e10378, using your repro: `a int`, row 50 = `3000000000`
(overflows int32 -> residual, leaf NULL), strict `variant_get(v,'$.a','int') >
999` with `deferCastError=false`. Asserts `INVALID_VARIANT_CAST` is raised (not
an empty result) across DSv1/DSv2 x vectorized/non-vectorized -- a leaf-only
push would drop the row group and return empty.
##########
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:
Fixed in 0e10378 -- dropped the unreachable branch; `leafIsNull` is now a
direct `makeEq.lift(...).get(...)` application, making the "`makeEq` covers
every leaf type the comparison ops do" invariant explicit.
##########
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:
Fixed in 0e10378 -- dropped the comment; the map is `lazy`, so the ordering
of these two constants no longer matters (the real constraint, that it must be
lazy because it reads the `Parquet*Type` vals below, is already noted at its
own definition).
--
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]
