peter-toth commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3829099810
##########
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. In DSv2, variant extraction is pushed through the
separate
+ // SupportsPushDownVariantExtractions mechanism, and the filter reaching
this builder stays a
+ // `variant_get(v, ...)` predicate -- it is never rewritten into a
struct-field access like
Review Comment:
**Finding 14.** The conclusion is right but the reason isn't: the `` v.`0`
`` rewrite is not DSv1-only.
`V2ScanRelationPushDown.buildScanWithPushedVariants` runs the *same* rewrite on
the filters —
```scala
// V2ScanRelationPushDown.scala:953
val rewrittenFilterExprs = filters.map(variants.rewriteExpr(_, attributeMap))
```
— where `variants` is a `VariantInRelation`, the very class
`PushVariantIntoScan` uses, and the result goes into a `Filter` above the scan.
So `` v.`0` `` predicates do exist on the DSv2 path.
What actually keeps them out of `pushDataFilters` is rule ordering: in
`V2ScanRelationPushDown.apply` the `pushdownRules` list runs `pushDownFilters`
(`:54`) before `pushDownVariants` (`:60`) and `buildScanWithPushedVariants`
(`:64`), so when this method is called the predicate is still `variant_get(v,
...)`, which never translates to a source `Filter`. Worth stating it that way,
because the invariant then reads as what it is — an ordering property that a
later reordering or a second filter-push pass after the variant rewrite would
silently break — rather than something true by construction:
```scala
// Shredded-variant predicate pushdown (SPARK-55817) is not wired
here: it applies to the
// DSv1 path only. DSv2 does rewrite variant extractions into `v.`0``
struct accesses, but
// only in `V2ScanRelationPushDown.buildScanWithPushedVariants`, which
runs *after*
// `pushDownFilters`. So the filters reaching this method are still
`variant_get(v, ...)`
// predicates, which do not translate to a source `Filter` at all --
there is no
// shredded-variant logical name for `ParquetFilters` to resolve here,
and nothing would be
// reported convertible even with a `variantExtractionSchema`. DSv2
reads remain correct (the
// variant filter is applied post-scan); they just do not get
row-group skipping on shredded
// columns.
```
The same "never rewritten" claim is in the PR description and in
`VariantShreddingFilterPushdownSuite`'s scaladoc (`:40`), so those two want the
same correction.
##########
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 = {
Review Comment:
**Finding 16.** This case is the number the open default-on discussion is
resting on, and it is measured on a layout no default-configured writer
produces: `createAndRunBenchmark` writes with `parquet.block.size = 128KB`,
which for this dataset puts **2721 row groups** in one file. I re-ran the same
case at 128KB and at the Parquet default block size on the head:
```
rowGroups off best/avg on
best/avg delta
Skip none, 128KB blocks (as shipped) 2721 914 / 923 953 /
956 +4.3% / +3.6%
Skip none, DEFAULT block size 1 891 / 894 883 /
891 -0.9% / -0.3%
```
(20M rows, 5 iterations, JDK 21, Apple M4 Max; the 128KB row reproduces the
~3-4% in the checked-in `*-results.txt`, so the setup matches.)
So the overhead scales with row-group count — the same knob that produces
the 15-22x wins — and a file has to be deliberately tuned dense before it can
pay anything at all. That answers @qlong's "with default parquet block size,
the overhead could be lower (worth a testing)": measured, it goes to zero,
which argues for keeping the default on. It also sharpens the earlier "within
run-to-run noise" framing: at 2721 row groups the delta is consistent and
outside stdev on all three checked-in JDK runs (+57ms/±7, +64ms/±21, +82ms/±14
on best time), so it is real there and *absent* at the default layout, rather
than noisy in both.
Either adding a default-block-size skip-none case here, or a sentence in
this scaladoc saying the figure is specific to the 128KB layout, would stop the
number being read as a general default-on cost.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2422,6 +2424,407 @@ 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, isNotNull(residual))) is rewritten by parquet-mr into an
unsound
+ // and(notEq(leaf), eq(residual, null)), so a negated shredded predicate
must not be pushed.
Review Comment:
**Finding 15.** 0e10378 updated six descriptions of the guard, but this is a
seventh — and it sits on the test that protects the negation refusal, so it is
the one where a stale derivation costs the most. The shipped predicate is
`or(leaf, and(anyResidualNotNull, isNull(leaf)))`, whose parquet-mr inverse is
`and(not(leaf), or(and(eq(residual, null)...), notEq(leaf, null)))` — droppable
as soon as some residual has no nulls **and** the leaf is entirely NULL, which
is exactly the `{"a":500.5}` / `{"a":600.5}` all-fallback row group this
suite's e2e counterpart builds. `referencesShreddedName` in
`ParquetFilters.scala` already carries the corrected derivation; only this copy
is behind.
```suggestion
// 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.
```
--
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]
