dongjoon-hyun opened a new pull request, #59810:
URL: https://github.com/apache/spark/pull/59810
### What changes were proposed in this pull request?
`MergeSubplans` merges different non-correlated scalar subqueries into one,
and rewrites each reference to `GetStructField(ScalarSubquery(<merged>), i)`.
At run time, the filter pushdown paths replaced only the `ScalarSubquery` node
with its result. That left `GetStructField(Literal(<struct>), i)`, while the
filter translators need a bare literal. This PR replaces such a field read with
a literal of its own.
- Add `replaceWithLiterals` to a new companion object of the physical
`ScalarSubquery` (`org.apache.spark.sql.execution`). It replaces each scalar
subquery in an expression with a literal of its result, and a `GetStructField`
over a scalar subquery with a literal of that field. The field is read straight
from the subquery result, so no literal of the whole struct is built.
- Use it where a scalar subquery is literalized for pushdown:
- `FileSourceScanLike.pushedDownFilters`: the data filters of a V1 file
scan (SPARK-43402).
- `DataSourceV2Strategy.translateScalarSubqueryFilterV2`: the runtime
filters of a `SupportsRuntimeV2Filtering` scan (SPARK-56467).
- `PushDownUtils.unwrapRuntimeFilterExpression`: the runtime filters of a
`SupportsRuntimeCatalystFiltering` scan (SPARK-58523), and the
`PartitionPredicate`s derived from runtime filters.
- `FileSourceScanLike.metadata`, which is only for display, maps a scalar
subquery to the string literal `ScalarSubquery#<id>`, because the subquery has
not run yet. It now maps a field read out of a merged subquery to that string
as a whole, so `PushedFilters` shows the filter as it does for an unmerged
subquery. `MergeSubplans` keeps the `exprId` of each original subquery, so the
id is the same too. Folding the field read over the string would fail.
### Why are the changes needed?
`MergeSubplans` runs whenever `spark.sql.execution.reuseSubquery` is true,
which is the default. With Parquet tables `t1(c1 INT)` holding 1, 2, 3 and
`t2(c2 INT)` holding 4, 5, 6, the two subqueries of
```sql
SELECT * FROM t1 WHERE c1 > (SELECT min(c2) FROM t2) AND c1 < (SELECT
max(c2) FROM t2)
```
are merged, and both data filters are silently left out of the pushed
filters. The scan of `t1` returns all 3 rows:
```
PushedFilters: [IsNotNull(c1)]
```
With this PR, both are pushed, `c1 > 4` skips the row group, and the scan
returns no row:
```
PushedFilters: [IsNotNull(c1), GreaterThan(c1,ScalarSubquery#116),
LessThan(c1,ScalarSubquery#117)]
```
Each filter in a query of its own was already pushed, and so were both with
`spark.sql.execution.reuseSubquery=false`, which disables the merge.
- V1 file sources could not skip data with these filters (for example,
Parquet row groups and pages) since
[v4.0.0](https://github.com/apache/spark/releases/tag/v4.0.0) (2025-05-19),
which added the scalar subquery pushdown (SPARK-43402).
- A `SupportsRuntimeV2Filtering` scan did not receive the predicate, so it
could not prune partitions with it, since
[v4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0) (2026-07-11),
which routed scalar subquery filters to it (SPARK-56467). `V2ExpressionBuilder`
folds only `contextIndependentFoldable` expressions, and `GetStructField` is
not one.
- A `SupportsRuntimeCatalystFiltering` scan, which is not released yet,
received a filter like `part = [0,3,1].max(val)`, which carries the whole
struct, instead of `part = 3`.
Query results were correct, because the `Filter` above the scan keeps these
filters.
### Does this PR introduce _any_ user-facing change?
No change to query results. Data sources now receive filters on scalar
subqueries that `MergeSubplans` merged, as they already did for an unmerged
subquery, and can skip data with them. EXPLAIN lists them in the
`PushedFilters` of a V1 file scan.
### How was this patch tested?
Added three tests. Each fails without the main change:
- `SubquerySuite`, "SPARK-60569: FileSourceScanExec pushes down data filters
with merged scalar subqueries": the SPARK-43402 test with two subqueries that
`MergeSubplans` merges. It checks `PushedFilters`, and through the scan's
`numOutputRows` that the pushed `c1 > 4` and `c1 < 1` each skip the row group
while `c1 > -6 AND c1 < 6` keeps it. Without the change, `PushedFilters` is
`[IsNotNull(c1)]`, and `numOutputRows` is 3 instead of 0.
- `DataSourceV2SQLSuiteV2Filter`, "SPARK-60569: scalar subquery filters
merged by MergeSubplans are pushed to the scan": a `SupportsRuntimeV2Filtering`
scan keeps 1 of 10 partitions. Without the change, it keeps all 10.
- `DataSourceV2CatalystRuntimeFilterSuite`, "SPARK-60569: field read out of
a merged scalar subquery -> literalized as a whole": a
`SupportsRuntimeCatalystFiltering` scan receives `part = 3`. Without the
change, it receives `part = [0,3,1].max(val)`.
These also pass locally: `SubquerySuite`, `DataSourceV2SQLSuiteV2Filter`,
`DataSourceV2CatalystRuntimeFilterSuite`,
`DataSourceV2EnhancedRuntimePartitionFilterSuite`,
`DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite`,
`GroupBasedRowLevelOperationCatalystRuntimeFilterSuite`, the suites under
`execution/planmerging`, `ExplainSuite`, `ExplainSuiteAE`, the TPC-DS and TPC-H
plan stability suites (no approved plan changes), and the `explain*.sql` cases
of `SQLQueryTestSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
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]
