This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new c3f6abd07ef6 [SPARK-56660][SQL] Decompose struct equality into field
predicates at the data source pushdown layer
c3f6abd07ef6 is described below
commit c3f6abd07ef61b3028ecb2c77e13387e6dc54762
Author: Anupam Yadav <[email protected]>
AuthorDate: Thu Jul 2 13:42:51 2026 +0800
[SPARK-56660][SQL] Decompose struct equality into field predicates at the
data source pushdown layer
### What changes were proposed in this pull request?
At the data source pushdown/translation layer, decompose a struct-level
equality (`s = struct_literal` or `s <=> struct_literal`) in a `Filter` into
field-level equality predicates (`s.f1 = v1 AND s.f2 = v2 ...`) offered to the
source as *additional* pushdown candidates. The original struct predicate is
left in the plan and retained as a post-scan filter, so the pushed field
predicates only need to be a sound over-approximation.
Implemented as `DataSourceStrategy.expandStructPredicatesForPushdown`,
wired into both pushdown paths:
- **V1**: `FileSourceStrategy`.
- **V2**: `FileScanBuilder.pushFilters` (the
`SupportsPushDownCatalystFilters` interface that all V2 file sources -
Parquet/ORC/CSV/JSON/Avro - use). Only the translate-and-push loop sees the
expanded filters; the returned post-scan filters remain the original struct
predicates.
Gated by `spark.sql.sources.structPredicateDecompose.enabled` (default
true), bounded by `spark.sql.sources.structPredicateDecompose.maxFields`
(default 100, counting leaf fields).
Soundness: since the field predicates are only pushdown hints (the original
struct filter still runs post-scan), they must over-approximate. For a struct
literal with a NULL-valued field we do not push `s.f = null` (which would
incorrectly drop rows); only non-null fields are pushed. Both `=` and `<=>` are
handled (field predicates use `=`).
### Why are the changes needed?
Struct-literal equality is opaque to data source filter pushdown (which
only understands scalar predicates), so `struct_col = <literal>` cannot drive
file pruning (Parquet row-group skipping, partition pruning, etc.) even though
the equivalent per-field predicates could. Decomposing at the pushdown layer
produces those pushable field predicates while - unlike a logical rewrite -
preserving the whole-struct predicate for sources that can push it natively,
and without requiring exact NU [...]
### Does this PR introduce _any_ user-facing change?
No. It only adds pushdown candidates; results are unchanged (the original
predicate is always retained post-scan).
### How was this patch tested?
`StructPredicatePushdownSuite` (sql/core): decomposed field predicates
reach the scan's pushed filters; the original struct filter is retained
post-scan; NULL-valued literal fields are not pushed (soundness); nested
structs decompose to the exact leaf-field set; `maxFields` bounds the
expansion; the conf toggles the behavior; and results are identical rule-on vs
rule-off across whole-null / all-null-fields / non-null rows for both `=` and
`<=>`. Both the V1 (`FileSourceScanExec`) and [...]
### Was this patch authored or co-authored using generative AI tooling?
Authored with assistance by Claude Opus 4.8.
Closes #56244 from yadavay-amzn/fix/SPARK-56660-struct-predicate-decompose.
Authored-by: Anupam Yadav <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 28 ++
.../execution/datasources/DataSourceStrategy.scala | 100 +++++
.../execution/datasources/FileSourceStrategy.scala | 8 +-
.../execution/datasources/v2/FileScanBuilder.scala | 8 +-
.../datasources/StructPredicatePushdownSuite.scala | 433 +++++++++++++++++++++
5 files changed, 574 insertions(+), 3 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 91635d346605..8be23cdaef00 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -5499,6 +5499,30 @@ object SQLConf {
.stringConf
.createWithDefault("parquet,orc")
+ val STRUCT_PREDICATE_DECOMPOSE_ENABLED =
+ buildConf("spark.sql.sources.structPredicateDecompose.enabled")
+ .doc("When true, struct equality predicates (= and <=>) are decomposed
into " +
+ "field-level equality predicates for filter pushdown to data sources.
The " +
+ "decomposed predicates are pushed as additional hints for data
skipping (e.g. " +
+ "Parquet row-group filtering). The original struct predicate is always
retained " +
+ "as a post-scan filter for correctness.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
+ val STRUCT_PREDICATE_DECOMPOSE_MAX_FIELDS =
+ buildConf("spark.sql.sources.structPredicateDecompose.maxFields")
+ .internal()
+ .doc("The maximum number of leaf fields a struct type may have for its
equality " +
+ "predicates to be decomposed into field-level predicates for pushdown.
Structs " +
+ "exceeding this limit are not decomposed.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .intConf
+ .checkValue(_ > 0, "The threshold must be positive.")
+ .createWithDefault(100)
+
val SERIALIZER_NESTED_SCHEMA_PRUNING_ENABLED =
buildConf("spark.sql.optimizer.serializer.nestedSchemaPruning.enabled")
.internal()
@@ -8746,6 +8770,10 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def avoidDoubleFilterEval: Boolean = getConf(AVOID_DOUBLE_FILTER_EVAL)
+ def structPredicateDecomposeEnabled: Boolean =
getConf(STRUCT_PREDICATE_DECOMPOSE_ENABLED)
+
+ def structPredicateDecomposeMaxFields: Int =
getConf(STRUCT_PREDICATE_DECOMPOSE_MAX_FIELDS)
+
def readSideCharPadding: Boolean = getConf(SQLConf.READ_SIDE_CHAR_PADDING)
def cliPrintHeader: Boolean = getConf(SQLConf.CLI_PRINT_HEADER)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
index 7aff4ed1e3de..05bc14dfb700 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
@@ -868,6 +868,106 @@ object DataSourceStrategy
sortOrders.flatMap(translateSortOrder)
}
+ /**
+ * Expands struct equality predicates into additional field-level equality
predicates
+ * suitable for data source pushdown. The original predicates are preserved
(not replaced)
+ * so they serve as post-scan correctness filters.
+ *
+ * For `struct_col = struct_literal` or `struct_col <=> struct_literal`,
generates:
+ * GetStructField(struct_col, i) = literal_field_i
+ * for each non-null leaf field in the struct literal. Fields whose literal
value is null
+ * are NOT decomposed because pushing `field = null` would incorrectly
filter out rows
+ * (SQL `= null` is always false/null). Skipping null-valued fields is
sound: the pushed
+ * predicates form a weaker (superset) filter, and the original struct
predicate retained
+ * post-scan guarantees exact correctness.
+ *
+ * @param filters The original filter expressions.
+ * @param conf The active SQLConf.
+ * @return original filters ++ decomposed field-level equality filters.
+ */
+ protected[sql] def expandStructPredicatesForPushdown(
+ filters: Seq[Expression],
+ conf: SQLConf): Seq[Expression] = {
+ if (!conf.structPredicateDecomposeEnabled) {
+ return filters
+ }
+ val maxFields = conf.structPredicateDecomposeMaxFields
+ val additional = mutable.ArrayBuffer.empty[Expression]
+ filters.foreach {
+ case expressions.EqualTo(left, right) =>
+ decomposeStructEquality(left, right, maxFields).foreach(additional ++=
_)
+ case expressions.EqualNullSafe(left, right) =>
+ decomposeStructEquality(left, right, maxFields).foreach(additional ++=
_)
+ case _ =>
+ }
+ if (additional.isEmpty) filters else filters ++ additional
+ }
+
+ /**
+ * For a struct equality (either `=` or `<=>`), decomposes into field-level
`EqualTo`
+ * predicates for non-null literal fields. Returns None if the predicate is
not
+ * decomposable (not a struct equality against a foldable literal, or
exceeds maxFields).
+ */
+ private def decomposeStructEquality(
+ left: Expression,
+ right: Expression,
+ maxFields: Int): Option[Seq[Expression]] = {
+ val (col, lit) = (left, right) match {
+ case (l, r) if r.foldable && r.dataType.isInstanceOf[StructType] &&
+ l.dataType.isInstanceOf[StructType] => (l, r)
+ case (l, r) if l.foldable && l.dataType.isInstanceOf[StructType] &&
+ r.dataType.isInstanceOf[StructType] => (r, l)
+ case _ => return None
+ }
+ val st = lit.dataType.asInstanceOf[StructType]
+ if (totalLeafFields(st) > maxFields) return None
+ val litValue = lit.eval(EmptyRow)
+ if (litValue == null) return None // whole struct is null, nothing to push
+ val litRow = litValue.asInstanceOf[InternalRow]
+ val fieldPreds = extractFieldPredicates(col, litRow, st)
+ if (fieldPreds.isEmpty) None else Some(fieldPreds)
+ }
+
+ /**
+ * Recursively extracts field-level EqualTo predicates for all leaf
(non-struct) fields
+ * whose literal value is non-null.
+ */
+ private def extractFieldPredicates(
+ col: Expression,
+ litRow: InternalRow,
+ st: StructType): Seq[Expression] = {
+ val buf = mutable.ArrayBuffer.empty[Expression]
+ st.fields.indices.foreach { i =>
+ val field = st.fields(i)
+ val fieldExpr = GetStructField(col, i)
+ field.dataType match {
+ case nested: StructType =>
+ if (!litRow.isNullAt(i)) {
+ val nestedRow = litRow.getStruct(i, nested.length)
+ buf ++= extractFieldPredicates(fieldExpr, nestedRow, nested)
+ }
+ // if litRow.isNullAt(i), skip entire nested struct (sound
over-approximation)
+ case dt =>
+ if (!litRow.isNullAt(i)) {
+ val litVal = litRow.get(i, dt)
+ buf += expressions.EqualTo(fieldExpr, Literal(litVal, dt))
+ }
+ // null-valued field: do NOT push `field = null` (unsound)
+ }
+ }
+ buf.toSeq
+ }
+
+ /** Counts total leaf (non-struct) fields recursively. */
+ private def totalLeafFields(st: StructType): Int = {
+ st.fields.iterator.map { f =>
+ f.dataType match {
+ case s: StructType => totalLeafFields(s)
+ case _ => 1
+ }
+ }.sum
+ }
+
/**
* Convert RDD of Row into RDD of InternalRow with objects in catalyst types
*/
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
index 396375890c24..b7a1736bc2e9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
@@ -204,7 +204,11 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
val supportNestedPredicatePushdown =
DataSourceUtils.supportNestedPredicatePushdown(fsRelation)
- val pushedFilters = dataFilters
+ // Expand struct equality predicates into field-level predicates for
pushdown.
+ // The original struct predicates remain in afterScanFilters for
correctness.
+ val expandedDataFilters =
DataSourceStrategy.expandStructPredicatesForPushdown(
+ dataFilters, fsRelation.sparkSession.sessionState.conf)
+ val pushedFilters = expandedDataFilters
.flatMap(DataSourceStrategy.translateFilter(_,
supportNestedPredicatePushdown))
logInfo(log"Pushed Filters: ${MDC(PUSHED_FILTERS,
pushedFilters.mkString(","))}")
@@ -332,7 +336,7 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
partitionKeyFilters.toSeq,
bucketSet,
None,
- rebindFileSourceMetadataAttributesInFilters(dataFilters),
+ rebindFileSourceMetadataAttributesInFilters(expandedDataFilters),
table.map(_.identifier))
// extra Project node: wrap flat metadata columns to a metadata struct
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
index 7e0bc25a9a1e..46f6702003ff 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
@@ -78,8 +78,14 @@ abstract class FileScanBuilder(
!SubqueryExpression.hasSubquery(f) &&
!f.exists(_.isInstanceOf[PythonUDF])
}
this.dataFilters = dataFilters
+ // Expand struct equality predicates into field-level predicates for
pushdown only.
+ // The original struct predicates stay in `dataFilters` (returned below as
post-scan
+ // filters), so the expanded field predicates only need to be a sound
over-approximation.
+ val expandedDataFilters =
+ DataSourceStrategy.expandStructPredicatesForPushdown(
+ dataFilters, sparkSession.sessionState.conf)
val translatedFilters = mutable.ArrayBuffer.empty[sources.Filter]
- for (filterExpr <- dataFilters) {
+ for (filterExpr <- expandedDataFilters) {
val translated = DataSourceStrategy.translateFilter(filterExpr, true)
if (translated.nonEmpty) {
translatedFilters += translated.get
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/StructPredicatePushdownSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/StructPredicatePushdownSuite.scala
new file mode 100644
index 000000000000..ac463d602d5a
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/StructPredicatePushdownSuite.scala
@@ -0,0 +1,433 @@
+/*
+ * 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
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{EqualTo, Expression,
GetStructField, Literal}
+import org.apache.spark.sql.execution.{FileSourceScanExec, FilterExec}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.sources.{EqualTo => SourcesEqualTo, Filter =>
SourcesFilter}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+/**
+ * Tests for struct equality predicate decomposition at the pushdown layer.
+ *
+ * Validates that:
+ * - Decomposed field-level predicates reach the scan
(PushedFilters/DataFilters)
+ * - The original struct predicate is retained as a post-scan filter
(correctness)
+ * - Null-valued literal fields are NOT pushed as `= null` (soundness)
+ * - maxFields bound is respected
+ * - Conf on/off behavior works correctly
+ * - Results are identical with rule on vs off (parity)
+ */
+class StructPredicatePushdownSuite extends QueryTest with SharedSparkSession {
+
+ private def getFileSourceScanDataFilters(df: DataFrame): Seq[Expression] = {
+ val plan = df.queryExecution.executedPlan
+ plan.collect { case scan: FileSourceScanExec => scan.dataFilters }.flatten
+ }
+
+ private def hasPostScanFilter(df: DataFrame): Boolean = {
+ val plan = df.queryExecution.executedPlan
+ plan.collect { case f: FilterExec => f }.nonEmpty
+ }
+
+ // Pushed filters on the DSv2 parquet scan (BatchScanExec -> ParquetScan).
+ private def getV2ParquetPushedFilters(df: DataFrame): Seq[SourcesFilter] = {
+ val plan = df.queryExecution.executedPlan
+ plan.collect {
+ case scan: BatchScanExec =>
+ scan.scan match {
+ case p: ParquetScan => p.pushedFilters.toSeq
+ case _ => Seq.empty[SourcesFilter]
+ }
+ }.flatten
+ }
+
+ test("field-level predicates are pushed for struct equality") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id + 1 as string)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', 5, 'b', '6')")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ // Should contain field-level EqualTo predicates for s.a and s.b
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), Literal(_, _)) => eq
+ case eq @ EqualTo(Literal(_, _), GetStructField(_, _, _)) => eq
+ }
+ assert(fieldPredicates.nonEmpty,
+ s"Expected field-level predicates in DataFilters but got:
$dataFilters")
+
+ // Verify results are correct
+ checkAnswer(df, spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id + 1 as string)) as s"
+ ).where("s.a = 5 AND s.b = '6'"))
+ }
+ }
+
+ test("original struct filter is retained post-scan") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id + 1 as int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', 3, 'b', 4)")
+
+ // The FilterExec above the scan should still contain the original
struct predicate
+ assert(hasPostScanFilter(df),
+ "Expected post-scan FilterExec containing the original struct
predicate")
+
+ checkAnswer(df, Row(Row(3, 4)))
+ }
+ }
+
+ test("null-valued literal fields are NOT pushed as field = null
(soundness)") {
+ withTempPath { path =>
+ // Create data with some null struct fields
+ val data = Seq(
+ Row(Row(1, null)),
+ Row(Row(2, "hello")),
+ Row(null),
+ Row(Row(1, "world"))
+ )
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(
+ StructField("a", IntegerType, nullable = false),
+ StructField("b", StringType, nullable = true)
+ )), nullable = true)
+ ))
+ spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+ .write.parquet(path.getAbsolutePath)
+
+ // Filter where the literal has a null field: s = struct(1, null)
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', 1, 'b', cast(null as string))")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ // Should push s.a = 1 but NOT s.b = null
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, ordinal, _), lit: Literal) =>
(ordinal, lit)
+ case eq @ EqualTo(lit: Literal, GetStructField(_, ordinal, _)) =>
(ordinal, lit)
+ }
+ // s.a = 1 should be pushed
+ assert(fieldPredicates.exists { case (_, lit) => lit.value != null },
+ s"Expected non-null field predicates but got: $fieldPredicates")
+ // s.b = null should NOT be pushed
+ assert(!fieldPredicates.exists { case (_, lit) => lit.value == null },
+ s"Should not push null field predicate but found one in:
$fieldPredicates")
+
+ // Result correctness: s = struct(1, null) matches the row [1, null]
because
+ // Spark's struct equality treats null=null field-by-field as equal.
+ checkAnswer(df, Row(Row(1, null)))
+ }
+ }
+
+ test("only non-null literal fields are pushed for struct with all-null
fields") {
+ withTempPath { path =>
+ val data = Seq(Row(Row(null, null)), Row(Row(1, 2)), Row(null))
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(
+ StructField("a", IntegerType, nullable = true),
+ StructField("b", IntegerType, nullable = true)
+ )), nullable = true)
+ ))
+ spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+ .write.parquet(path.getAbsolutePath)
+
+ // All fields in literal are null - no field predicates should be
generated
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', cast(null as int), 'b', cast(null as
int))")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), _) => eq
+ case eq @ EqualTo(_, GetStructField(_, _, _)) => eq
+ }
+ // No field predicates should be pushed since all literal fields are null
+ assert(fieldPredicates.isEmpty,
+ s"Expected no field predicates for all-null literal but got:
$fieldPredicates")
+ }
+ }
+
+ test("EqualNullSafe (<=>) also triggers struct decomposition") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id * 2 as int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s <=> named_struct('a', 3, 'b', 6)")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), _) => eq
+ case eq @ EqualTo(_, GetStructField(_, _, _)) => eq
+ }
+ assert(fieldPredicates.nonEmpty,
+ s"Expected field-level predicates for <=> but got: $dataFilters")
+
+ checkAnswer(df, Row(Row(3, 6)))
+ }
+ }
+
+ test("maxFields bound is respected - wide struct is not decomposed") {
+ withTempPath { path =>
+ // Create a struct with 5 fields
+ spark.range(5).selectExpr(
+ "named_struct('f1', cast(id as int), 'f2', cast(id as int), " +
+ "'f3', cast(id as int), 'f4', cast(id as int), 'f5', cast(id as
int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ // Set maxFields to 3 so a 5-field struct is not decomposed
+ withSQLConf(SQLConf.STRUCT_PREDICATE_DECOMPOSE_MAX_FIELDS.key -> "3") {
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('f1', 2, 'f2', 2, 'f3', 2, 'f4', 2, 'f5',
2)")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), _) => eq
+ case eq @ EqualTo(_, GetStructField(_, _, _)) => eq
+ }
+ // Should NOT have field-level predicates since struct exceeds
maxFields
+ assert(fieldPredicates.isEmpty,
+ s"Expected no field predicates (maxFields=3) but got:
$fieldPredicates")
+ }
+
+ // With default maxFields (100), the 5-field struct IS decomposed
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('f1', 2, 'f2', 2, 'f3', 2, 'f4', 2, 'f5', 2)")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), _) => eq
+ case eq @ EqualTo(_, GetStructField(_, _, _)) => eq
+ }
+ assert(fieldPredicates.length == 5,
+ s"Expected 5 field predicates but got: $fieldPredicates")
+
+ checkAnswer(df, Row(Row(2, 2, 2, 2, 2)))
+ }
+ }
+
+ test("conf disabled - no decomposition") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id + 1 as int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ withSQLConf(SQLConf.STRUCT_PREDICATE_DECOMPOSE_ENABLED.key -> "false") {
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', 5, 'b', 6)")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), _) => eq
+ case eq @ EqualTo(_, GetStructField(_, _, _)) => eq
+ }
+ // No field-level predicates when conf is disabled
+ assert(fieldPredicates.isEmpty,
+ s"Expected no field predicates when conf disabled but got:
$fieldPredicates")
+
+ // Results should still be correct (just without the pushdown
optimization)
+ checkAnswer(df, Row(Row(5, 6)))
+ }
+ }
+ }
+
+ test("nested struct decomposition") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('inner', named_struct('x', cast(id as int), " +
+ "'y', cast(id + 1 as int)), 'z', cast(id * 10 as int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('inner', named_struct('x', 3, 'y', 4), 'z',
30)")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(g: GetStructField, _: Literal) => eq
+ case eq @ EqualTo(_: Literal, g: GetStructField) => eq
+ }
+ // Should have predicates for s.inner.x, s.inner.y, and s.z (3 leaf
fields)
+ assert(fieldPredicates.length == 3,
+ s"Expected 3 field predicates but got ${fieldPredicates.length}:
$dataFilters")
+
+ checkAnswer(df, Row(Row(Row(3, 4), 30)))
+ }
+ }
+
+ test("parity: results identical with decomposition on vs off") {
+ withTempPath { path =>
+ // Include rows with: normal values, null struct, struct with null fields
+ val data = Seq(
+ Row(Row(1, "a")),
+ Row(Row(2, "b")),
+ Row(Row(1, null)),
+ Row(null),
+ Row(Row(3, "a"))
+ )
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(
+ StructField("a", IntegerType, nullable = false),
+ StructField("b", StringType, nullable = true)
+ )), nullable = true)
+ ))
+ spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+ .write.parquet(path.getAbsolutePath)
+
+ val queries = Seq(
+ "s = named_struct('a', 1, 'b', 'a')",
+ "s = named_struct('a', 1, 'b', cast(null as string))",
+ "s <=> named_struct('a', 1, 'b', 'a')",
+ "s <=> named_struct('a', 1, 'b', cast(null as string))"
+ )
+
+ for (query <- queries) {
+ val resultOn = withSQLConf(
+ SQLConf.STRUCT_PREDICATE_DECOMPOSE_ENABLED.key -> "true") {
+ spark.read.parquet(path.getAbsolutePath).where(query).collect()
+ }
+ val resultOff = withSQLConf(
+ SQLConf.STRUCT_PREDICATE_DECOMPOSE_ENABLED.key -> "false") {
+ spark.read.parquet(path.getAbsolutePath).where(query).collect()
+ }
+ assert(resultOn.toSeq.sortBy(_.toString) ==
resultOff.toSeq.sortBy(_.toString),
+ s"Results differ for query '$query': on=$resultOn, off=$resultOff")
+ }
+ }
+ }
+
+ test("whole-null struct rows are correctly filtered out") {
+ withTempPath { path =>
+ val data = Seq(
+ Row(Row(1, 2)),
+ Row(null),
+ Row(Row(3, 4))
+ )
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("b", IntegerType)
+ )), nullable = true)
+ ))
+ spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+ .write.parquet(path.getAbsolutePath)
+
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', 1, 'b', 2)")
+
+ // Post-scan filter ensures null struct rows are filtered out
+ checkAnswer(df, Row(Row(1, 2)))
+ }
+ }
+
+ test("literal on left side of equality") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id + 1 as int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ // Literal on the left side
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("named_struct('a', 5, 'b', 6) = s")
+
+ val dataFilters = getFileSourceScanDataFilters(df)
+ val fieldPredicates = dataFilters.collect {
+ case eq @ EqualTo(GetStructField(_, _, _), _) => eq
+ case eq @ EqualTo(_, GetStructField(_, _, _)) => eq
+ }
+ assert(fieldPredicates.nonEmpty,
+ s"Expected field predicates for literal-on-left but got: $dataFilters")
+
+ checkAnswer(df, Row(Row(5, 6)))
+ }
+ }
+
+ test("field-level predicates are pushed on the DSv2 file-source path") {
+ // Force parquet to resolve to the DSv2 reader (FileScanBuilder /
+ // SupportsPushDownCatalystFilters) rather than the V1 FileSourceStrategy.
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ withTempPath { path =>
+ spark.range(10).selectExpr(
+ "named_struct('a', cast(id as int), 'b', cast(id + 1 as int)) as s"
+ ).write.parquet(path.getAbsolutePath)
+
+ val df = spark.read.parquet(path.getAbsolutePath)
+ .where("s = named_struct('a', 5, 'b', 6)")
+
+ // Sanity: this must actually be the V2 path.
+ assert(df.queryExecution.executedPlan.collect {
+ case b: BatchScanExec => b
+ }.nonEmpty, "Expected a DSv2 BatchScanExec in the plan")
+
+ // The decomposed field predicates (s.a = 5, s.b = 6) must reach the
V2 scan.
+ val pushed = getV2ParquetPushedFilters(df)
+ val fieldPredicates = pushed.collect {
+ case eq @ SourcesEqualTo(attr, _) if attr.contains(".") => eq
+ }
+ assert(fieldPredicates.nonEmpty,
+ s"Expected decomposed field predicates in V2 pushed filters but got:
$pushed")
+
+ checkAnswer(df, Row(Row(5, 6)))
+ }
+ }
+ }
+
+ test("DSv2 parity: results identical with decomposition on vs off") {
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ withTempPath { path =>
+ val data = Seq(
+ Row(Row(1, "a")), Row(Row(2, "b")), Row(Row(1, null)), Row(null),
Row(Row(3, "a")))
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(
+ StructField("a", IntegerType, nullable = false),
+ StructField("b", StringType, nullable = true)
+ )), nullable = true)))
+ spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+ .write.parquet(path.getAbsolutePath)
+
+ val queries = Seq(
+ "s = named_struct('a', 1, 'b', 'a')",
+ "s = named_struct('a', 1, 'b', cast(null as string))",
+ "s <=> named_struct('a', 1, 'b', 'a')")
+
+ for (query <- queries) {
+ val resultOn = withSQLConf(
+ SQLConf.STRUCT_PREDICATE_DECOMPOSE_ENABLED.key -> "true") {
+ spark.read.parquet(path.getAbsolutePath).where(query).collect()
+ }
+ val resultOff = withSQLConf(
+ SQLConf.STRUCT_PREDICATE_DECOMPOSE_ENABLED.key -> "false") {
+ spark.read.parquet(path.getAbsolutePath).where(query).collect()
+ }
+ assert(resultOn.toSeq.sortBy(_.toString) ==
resultOff.toSeq.sortBy(_.toString),
+ s"DSv2 results differ for query '$query': on=$resultOn,
off=$resultOff")
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
