cloud-fan commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3503668827
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -69,7 +69,12 @@ object PushDownUtils extends Logging {
// Catalyst filter expression that can't be translated to data source
filters.
val untranslatableExprs = mutable.ArrayBuffer.empty[Expression]
- for (filterExpr <- filters) {
+ // Expand struct equality predicates into field-level predicates for
pushdown.
+ // Original struct predicates remain in the list and end up as
post-scan filters.
+ val expandedFilters =
DataSourceStrategy.expandStructPredicatesForPushdown(
Review Comment:
**The V2 wiring is on the wrong `pushFilters` branch, so it does not reach
any V2 file source.**
`expandedFilters` is only consumed by the `SupportsPushDownFilters` arm (the
loop just below). But all V2 file sources — Parquet, ORC, CSV, JSON, Avro —
extend `FileScanBuilder`, which implements `SupportsPushDownCatalystFilters`,
handled by the *separate* branch at lines 141-143:
```scala
case r: SupportsPushDownCatalystFilters =>
val postScanFilters = r.pushFilters(filters) // <- un-expanded
`filters`, not `expandedFilters`
```
`SupportsPushDownFilters` is implemented only by `PythonScanBuilder` (and
the V1Scan fallback in `V2ScanRelationPushDown`), neither of which does
Parquet/ORC row-group pruning. So for the sources that would actually benefit
from field-level pruning, the decomposition never runs — the V2 half of the
feature the PR describes is inert.
The natural home for V2 file sources is `FileScanBuilder.pushFilters`
(expand `dataFilters` before the translate loop there), or the
`SupportsPushDownCatalystFilters` branch here. Worth confirming which interface
the intended V2 sources actually use before landing.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/StructPredicatePushdownSuite.scala:
##########
@@ -0,0 +1,354 @@
+/*
+ * 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.internal.SQLConf
+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 {
Review Comment:
**No test covers the V2 datasource pushdown path.** Every test here uses
`spark.read.parquet(...)` under the default `spark.sql.sources.useV1SourceList`
(which includes `parquet`), so all of them exercise the **V1**
`FileSourceStrategy` path. None routes through the DSv2 reader.
This matters because it's exactly the gap that hides the design issue on the
V2 wiring (see the `PushDownUtils.scala` comment): a test on the V2 path would
assert field predicates reach the scan and find that none do. Consider adding a
case that forces the V2 path (e.g. `withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key
-> "")` so parquet resolves to DSv2) and checks the pushed filters — once the
wiring reaches V2 file sources.
--
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]
