viirya commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3832121832
##########
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:
Good catch, corrected in 82c5b599c40. You're right that DSv2 does run the
same `VariantInRelation.rewriteExpr` (in `buildScanWithPushedVariants`); it's
the rule ordering -- `pushDownFilters` runs before it -- that leaves the
filters here as `variant_get(v, ...)`. Rewrote the `ParquetScanBuilder` comment
to your wording, and fixed the same "never rewritten" claim in the suite
scaladoc and the PR description.
##########
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:
Fixed in 82c5b599c40 -- updated the comment to the shipped shape:
`not(or(leaf, and(anyResidualNotNull, isNull(leaf))))` rewrites to
`and(not(leaf), or(and(eq(residual, null)...), notEq(leaf, null)))`, droppable
once some residual has no nulls AND the leaf is entirely NULL. That was the
seventh spot the earlier doc pass missed.
##########
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:
Agreed -- the ~3-4% is a 128KB / 2721-row-group artifact. Added a "Can skip
no row groups (default block size)" case in 82c5b599c40 (one row group for this
dataset); locally it measures ~-2.5% (ON slightly faster), matching your
numbers. So the overhead scales with row-group count, the same knob as the
benefit. Re-triggered the benchmark Action on JDK 17/21/25 (create-commit) to
regenerate the golden files with the new case; will confirm once they land.
--
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]
