viirya commented on code in PR #58050:
URL: https://github.com/apache/spark/pull/58050#discussion_r3806186410
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -692,6 +914,69 @@ 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 = {
Review Comment:
Good point -- with the default `deferCastError=false` the eager strict cast
makes this observable, so I treated it as a real result change. Fixed in
20ff8d1: `resolveShredded` now requires the extraction target type to map to
the *exact* physical leaf type (`expectedLeafType`), so a narrower extraction
such as smallint against an int leaf is no longer pushed and results stay
identical. Timestamps are conservatively not pushed for now. Added a unit test
(narrower not pushed, exact pushed).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:
##########
@@ -210,6 +210,15 @@ class ParquetFileFormat
val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
val pushDownInFilterThreshold =
sqlConf.parquetFilterPushDownInFilterThreshold
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+ // When shredded-variant predicate pushdown is enabled, `requiredSchema`
may carry the
+ // variant-extraction structs produced by PushVariantIntoScan. Passing it
lets ParquetFilters
+ // map logical paths like "v.`0`" to the physical shredded columns for
row-group skipping.
+ val variantExtractionSchema =
Review Comment:
Fixed in 20ff8d1: `ParquetFileFormat` passes `Some(requiredSchema)` only
when `requiredSchema.existsRecursively(VariantMetadata.isVariantStruct)`, so
non-variant DSv1 scans do no shredded traversal (and no `CaseInsensitiveMap`
wrapping) per file.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:
##########
@@ -2422,6 +2424,315 @@ 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,
+ // conjoins IS NULL on every residual `value` column along the path.
Review Comment:
Fixed in 20ff8d1 -- the comment now says the implementation OR-s an IS NOT
NULL guard on every residual, matching `makeShreddedFilter`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.
+ private def writeConf(forceSchema: String): 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,
+ // Keep the physical group unannotated so the schema is a plain shredded
struct.
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> "false")
Review Comment:
Added in 20ff8d1: an annotated-layout run (`annotateLogicalType` left at its
default `true`) of both the skip test and the overflow-fallback test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -718,6 +1003,37 @@ class ParquetFilters(
// Probably I missed something and obviously this should be changed.
predicate match {
+ // Shredded-variant paths (e.g. "v.`0`"). Only comparison predicates
that use min/max
+ // statistics are eligible. Each pushes or(leafPredicate,
isNotNull(residual)...) over every
+ // residual `value` column along the path (see `makeShreddedFilter`). IS
NULL / IS NOT NULL on
+ // the logical variant field are intentionally out of scope: "the
extracted field is null" is
+ // not the same as "typed_value is null", so we must not conflate them.
+ case sources.EqualTo(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, value)))
+ case sources.EqualNullSafe(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, value)))
+ case sources.LessThan(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeLt.lift(t).map(_(n, value)))
+ case sources.LessThanOrEqual(name, value) if
canMakeShreddedFilterOn(name, value) =>
+ makeShreddedFilter(name, (t, n) => makeLtEq.lift(t).map(_(n, value)))
+ case sources.GreaterThan(name, value) if canMakeShreddedFilterOn(name,
value) =>
+ makeShreddedFilter(name, (t, n) => makeGt.lift(t).map(_(n, value)))
+ case sources.GreaterThanOrEqual(name, value) if
canMakeShreddedFilterOn(name, value) =>
+ makeShreddedFilter(name, (t, n) => makeGtEq.lift(t).map(_(n, value)))
+ case sources.In(name, values) if pushDownInFilterThreshold > 0 &&
values.nonEmpty &&
+ values.forall(v => canMakeShreddedFilterOn(name, v)) =>
+ // Convert `In` to the OR of per-value equalities, each already OR-ed
with the residual
+ // isNotNull guards, then combine. Reuses the same soundness guard as
the comparison
+ // predicates (the repeated residual disjuncts are harmless).
+ val distinct = values.distinct
+ if (distinct.length <= pushDownInFilterThreshold) {
+ distinct.flatMap { v =>
+ makeShreddedFilter(name, (t, n) => makeEq.lift(t).map(_(n, v)))
+ }.reduceLeftOption(FilterApi.or)
+ } else {
+ None
Review Comment:
Fixed in 20ff8d1: large IN lists above the threshold now push via
`FilterApi.in` (`or(in(leaf, set), isNotNull(residual)...)`), the threshold is
measured on `values.length` like the regular path, and the residual guards are
appended once instead of once per value.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantShreddingFilterPushdownSuite.scala:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.
+ private def writeConf(forceSchema: String): 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,
+ // Keep the physical group unannotated so the schema is a plain shredded
struct.
+ SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> "false")
+
+ /**
+ * 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): Unit = {
+ withSQLConf(writeConf(forceSchema): _*) {
+ 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",
Review Comment:
Added in 20ff8d1: the grid now also varies `deferCastError`, asserting
results stay correct when it is on (the optimization silently does not fire).
Also added a sentence to the config's `.doc()` noting this.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala:
##########
@@ -692,6 +914,69 @@ 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.
+ private def referencesShreddedName(predicate: sources.Filter): Boolean =
predicate match {
Review Comment:
Done in 20ff8d1 -- `referencesShreddedName` is now
`predicate.references.exists(nameToShreddedVariantField.contains)`. Thanks,
that removes the fragile hand-enumeration.
--
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]
