This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 6a1b28bcb8f8 [SPARK-60569][SQL] Push down filters on scalar subqueries
merged by `MergeSubplans`
6a1b28bcb8f8 is described below
commit 6a1b28bcb8f862515e2c4044e824ecfb98632f14
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Oct 27 10:57:50 2026 -0700
[SPARK-60569][SQL] Push down filters on scalar subqueries merged by
`MergeSubplans`
### 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. A filter
with an `IN` over scalar [...]
### 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. An `IN`
over two such subqueries, like `c1 IN ((SELECT min(c2) FROM t2), (SELECT
max(c2) FROM t2))`, was not pushed either.
- 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, except a filter with an `IN`, which is left
out for an unmerged subquery too (SPARK-60577).
### How was this patch tested?
Added three tests and extended one. 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.
- `SubquerySuite`, "SPARK-60577: FileSourceScanExec supports push down IN
data filter with scalar subquery": it now also runs the `IN` over two
subqueries under the default configuration, where `MergeSubplans` merges them,
on `DATE`, `TIMESTAMP` and `DECIMAL` columns. Without the change, the `IN` is
not pushed, 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`,
`StorageFilterPushDownSuite`, `ParquetStorageFilterSuite`, the suites under
`execution/planmerging`, `ExplainSuite`, `ExplainSuiteAE`, the TPC-DS and TPC-H
plan stability suites (no approved plan changes), an [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59810 from dongjoon-hyun/SPARK-60569.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 882f8637adeee005dd12a4680d3c6bd0acab5d35)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/execution/DataSourceScanExec.scala | 23 +++++----
.../datasources/v2/DataSourceV2Strategy.scala | 5 +-
.../execution/datasources/v2/PushDownUtils.scala | 2 +-
.../org/apache/spark/sql/execution/subquery.scala | 16 +++++-
.../scala/org/apache/spark/sql/SubquerySuite.scala | 58 +++++++++++++++++++---
.../DataSourceV2CatalystRuntimeFilterSuite.scala | 26 ++++++++++
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 31 ++++++++++++
7 files changed, 140 insertions(+), 21 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
index 77f058f6269b..d26f8190f438 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
@@ -515,12 +515,10 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
private def translateToV1Filters(
dataFilters: Seq[Expression],
- scalarSubqueryToLiteral: execution.ScalarSubquery => Literal):
Seq[Filter] = {
- val scalarSubqueryReplaced = dataFilters.map(_.transform {
- // Replace scalar subquery to literal so that
`DataSourceStrategy.translateFilter` can
- // support translating it.
- case scalarSubquery: execution.ScalarSubquery =>
scalarSubqueryToLiteral(scalarSubquery)
- })
+ replaceScalarSubqueries: Expression => Expression): Seq[Filter] = {
+ // Replace scalar subquery to literal so that
`DataSourceStrategy.translateFilter` can
+ // support translating it.
+ val scalarSubqueryReplaced = dataFilters.map(replaceScalarSubqueries)
val supportNestedPredicatePushdown =
DataSourceUtils.supportNestedPredicatePushdown(relation)
// `dataFilters` should not include any constant metadata col filters
@@ -535,7 +533,8 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
// This field may execute subquery expressions and should not be accessed
during planning.
@transient
- protected lazy val pushedDownFilters: Seq[Filter] =
translateToV1Filters(dataFilters, _.toLiteral)
+ protected lazy val pushedDownFilters: Seq[Filter] =
+ translateToV1Filters(dataFilters,
execution.ScalarSubquery.replaceWithLiterals)
override lazy val metadata: Map[String, String] = {
def seqToString(seq: Seq[Any]) = seq.mkString("[", ", ", "]")
@@ -546,13 +545,19 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
// `metadata` is accessed during planning and the scalar subquery is not
executed yet. Here
// we get the pretty string of the scalar subquery, for display purpose
only. A filter with an
// `In` that has a scalar subquery in its list is not displayed, since the
values of an `In`
- // are converted to the type of its column, which the string does not have.
+ // are converted to the type of its column, which the string does not
have. A field read out
+ // of a subquery that `MergeSubplans` merged with others gets the string
as a whole, since the
+ // string has no field to read.
val pushedFiltersForDisplay = translateToV1Filters(
dataFilters.filterNot(_.exists {
case in: In =>
in.list.exists(_.exists(_.isInstanceOf[execution.ScalarSubquery]))
case _ => false
}),
- s => Literal("ScalarSubquery#" + s.exprId.id))
+ _.transform {
+ case GetStructField(s: execution.ScalarSubquery, _, _) =>
+ Literal("ScalarSubquery#" + s.exprId.id)
+ case s: execution.ScalarSubquery => Literal("ScalarSubquery#" +
s.exprId.id)
+ })
val metadata =
Map(
"Format" -> relation.fileFormat.toString,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index 32232532bbd6..d0c331282473 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -1019,10 +1019,7 @@ private[sql] object DataSourceV2Strategy extends Logging
{
*/
protected[sql] def translateScalarSubqueryFilterV2(
expr: Expression): Option[Predicate] = {
- val literalized = expr.transform {
- case s: ExecScalarSubquery => s.toLiteral
- }
- translateFilterV2(literalized)
+ translateFilterV2(ExecScalarSubquery.replaceWithLiterals(expr))
}
/**
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
index ea4cac0d221e..6d2417c3e77e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
@@ -506,7 +506,7 @@ object PushDownUtils extends Logging {
None
case DynamicPruningExpression(e) => Some(e)
case _: DynamicPruning => None
- case f => Some(f.transform { case s: ExecScalarSubquery => s.toLiteral })
+ case f => Some(ExecScalarSubquery.replaceWithLiterals(f))
}
/**
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala
index ea7f5126a78c..121e9ca14fdc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala
@@ -22,7 +22,7 @@ import scala.collection.immutable.TreeSet
import org.apache.spark.QueryContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.catalyst.{expressions, InternalRow}
-import org.apache.spark.sql.catalyst.expressions.{CreateNamedStruct,
Expression, ExprId, InSet, InterpretedOrdering, ListQuery, Literal,
PlanExpression, Predicate, SupportQueryContext}
+import org.apache.spark.sql.catalyst.expressions.{CreateNamedStruct,
Expression, ExprId, GetStructField, InSet, InterpretedOrdering, ListQuery,
Literal, PlanExpression, Predicate, SupportQueryContext}
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.catalyst.rules.Rule
@@ -115,6 +115,20 @@ case class ScalarSubquery(
}
}
+object ScalarSubquery {
+ /**
+ * `expr` with each scalar subquery replaced by a literal of its result, so
that a filter
+ * translation, which needs a bare literal, can match it. `MergeSubplans`
returns the results of
+ * the subqueries it merges in one struct, and reads each result out of it
with a
+ * [[GetStructField]], so such a field read becomes a literal of its own.
The field is read
+ * straight from the result: a literal of the whole struct would validate
every field of it.
+ */
+ def replaceWithLiterals(expr: Expression): Expression = expr.transform {
+ case g @ GetStructField(_: ScalarSubquery, _, _) => Literal(g.eval(),
g.dataType)
+ case s: ScalarSubquery => s.toLiteral
+ }
+}
+
/**
* The physical node of in-subquery. When this is used for Dynamic Partition
Pruning, as the pruning
* happens at the driver side, we don't broadcast subquery result.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
index bd2bf1caa9c2..a34862b3052b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
@@ -23,8 +23,8 @@ import org.apache.spark.{SparkArithmeticException, SparkConf,
SparkRuntimeExcept
import org.apache.spark.serializer.JavaSerializer
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{CreateNamedStruct, EqualTo,
Expression,
- ExprId, GreaterThanOrEqual, Literal, NamedExpression, OuterReference,
PlanExpression, Rand,
- SubqueryExpression}
+ ExprId, GetStructField, GreaterThanOrEqual, Literal, NamedExpression,
OuterReference,
+ PlanExpression, Rand, SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
GeneratePredicate}
import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join,
LogicalPlan, Project, Sort, Union}
@@ -2549,11 +2549,12 @@ class SubquerySuite extends SharedSparkSession
"dec" -> "CAST(5 AS DECIMAL(10, 2))").foreach { case (column,
literal) =>
checkFileSourceScan(
s"SELECT * FROM t1 WHERE $column IN ($literal, (SELECT
max($column) FROM t2))", 1)
- // Otherwise, the two scalar subqueries are merged into one.
+ val twoSubqueries = s"SELECT * FROM t1 WHERE $column IN " +
+ s"((SELECT min($column) FROM t2), (SELECT max($column) FROM t2))"
+ // `MergeSubplans` merges the two subqueries into one, unless
subquery reuse is off.
+ checkFileSourceScan(twoSubqueries, 2)
withSQLConf(SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false") {
- checkFileSourceScan(
- s"SELECT * FROM t1 WHERE $column IN " +
- s"((SELECT min($column) FROM t2), (SELECT max($column) FROM
t2))", 2)
+ checkFileSourceScan(twoSubqueries, 2)
}
}
@@ -2570,6 +2571,51 @@ class SubquerySuite extends SharedSparkSession
}
}
+ test("SPARK-60569: FileSourceScanExec pushes down data filters with merged
scalar subqueries") {
+ def checkFileSourceScan(query: String, answer: Seq[Row]): Unit = {
+ val df = sql(query)
+ checkAnswer(df, answer)
+ val fileSourceScanExec = collect(df.queryExecution.executedPlan) {
+ case f: FileSourceScanExec => f
+ }
+ sparkContext.listenerBus.waitUntilEmpty()
+ assert(fileSourceScanExec.size === 1)
+ // `MergeSubplans` merged the two subqueries into one, and each filter
reads its result out
+ // of the struct the merged one returns.
+ val mergedSubqueryResults =
fileSourceScanExec.head.dataFilters.flatMap(_.collect {
+ case g @ GetStructField(_: ScalarSubquery, _, _) => g
+ })
+ assert(mergedSubqueryResults.length === 2)
+ val pushedFilters = fileSourceScanExec.head.metadata("PushedFilters")
+ assert(pushedFilters.contains("GreaterThan(c1,ScalarSubquery#"))
+ assert(pushedFilters.contains("LessThan(c1,ScalarSubquery#"))
+ assert(fileSourceScanExec.head.metrics("numFiles").value === 1)
+ assert(fileSourceScanExec.head.metrics("numOutputRows").value ===
answer.size)
+ }
+
+ withTable("t1", "t2") {
+ withSQLConf(SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1") {
+ Seq(1, 2, 3).toDF("c1").write.format("parquet").saveAsTable("t1")
+ Seq(4, 5, 6).toDF("c2").write.format("parquet").saveAsTable("t2")
+
+ // The row group is skipped by `c1 > 4`.
+ checkFileSourceScan(
+ "SELECT * FROM t1 WHERE c1 > (SELECT min(c2) FROM t2) AND c1 <
(SELECT max(c2) FROM t2)",
+ Seq.empty)
+ // The row group is skipped by `c1 < 1`.
+ checkFileSourceScan(
+ "SELECT * FROM t1 " +
+ "WHERE c1 > (SELECT min(c2) - 10 FROM t2) AND c1 < (SELECT max(c2)
- 5 FROM t2)",
+ Seq.empty)
+ // The row group is kept by `c1 > -6` and `c1 < 6`, but not if the
fields were swapped.
+ checkFileSourceScan(
+ "SELECT * FROM t1 " +
+ "WHERE c1 > (SELECT min(c2) - 10 FROM t2) AND c1 < (SELECT max(c2)
FROM t2)",
+ Row(1) :: Row(2) :: Row(3) :: Nil)
+ }
+ }
+ }
+
test("SPARK-44562: Add OptimizeOneRowRelationSubquery in batch of Subquery")
{
withTempView("v1", "v2") {
sql(
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index 520d05e5c48b..863ce6e1af26 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -349,6 +349,32 @@ class DataSourceV2CatalystRuntimeFilterSuite extends
SharedSparkSession {
}
}
+ test("SPARK-60569: field read out of a merged scalar subquery -> literalized
as a whole") {
+ val tbl = s"$catalogName.tbl_merged"
+ val dim = s"$catalogName.dim_merged"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (1), (3)")
+
+ // `MergeSubplans` merges the two subqueries into one, and each filter
reads its result out
+ // of the struct the merged one returns. The scan gets that result, not
the whole struct.
+ val df = sql(s"SELECT * FROM $tbl " +
+ s"WHERE part = (SELECT max(val) FROM $dim) AND id > (SELECT min(val)
FROM $dim)")
+ checkAnswer(df, Row(3, 3))
+
+ assertScalarSubqueryRuntimeFilters(df)
+ val runtimeFilter = collectBatchScan(df).runtimeFilters.head
+ assert(runtimeFilter.exists(_.isInstanceOf[GetStructField]),
+ s"Expected a runtime filter on a merged scalar subquery, got
$runtimeFilter")
+ val part = AttributeReference("part", IntegerType, nullable = false)()
+ assertPushedCatalystPredicatesEqual(df, EqualTo(part, Literal(3)))
+ }
+ }
+
test("filter with no V2 translation -> pushed instead of dropped") {
val tbl = s"$catalogName.tbl_untranslatable"
val dim = s"$catalogName.dim_untranslatable"
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index e46c68d19d0b..24c4efbfb41d 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -5624,6 +5624,37 @@ class DataSourceV2SQLSuiteV2Filter extends
DataSourceV2SQLSuite {
s"Expected the non-deterministic filter above the scan, got
$postScanConditions")
}
}
+
+ test("SPARK-60569: scalar subquery filters merged by MergeSubplans are
pushed to the scan") {
+ val tbl = s"${catalogAndNamespace}tbl"
+ val dim = s"${catalogAndNamespace}dim"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Format PARTITIONED
BY (part)")
+ for (i <- 0 until 10) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i)")
+ }
+
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Format")
+ sql(s"INSERT INTO $dim VALUES (1), (3)")
+
+ // `MergeSubplans` merges the two subqueries into one, and each filter
reads its result out
+ // of the struct the merged one returns.
+ val df = sql(s"SELECT * FROM $tbl " +
+ s"WHERE part = (SELECT max(val) FROM $dim) AND id > (SELECT min(val)
FROM $dim)")
+ checkAnswer(df, Row(3, 3))
+
+ val batchScan = collect(df.queryExecution.executedPlan) {
+ case b: BatchScanExec => b
+ }.head
+
assert(batchScan.runtimeFilters.exists(_.exists(_.isInstanceOf[GetStructField])),
+ s"Expected a runtime filter on a merged scalar subquery, got
${batchScan.runtimeFilters}")
+
+ // Verify partition pruning: only 1 of 10 partitions should remain
+ val numPartitions = batchScan.filteredPartitions.count(_.isDefined)
+ assert(numPartitions == 1,
+ s"Expected 1 partition after scalar subquery pruning, got
$numPartitions")
+ }
+ }
}
class ReserveSchemaNullabilityCatalog extends InMemoryCatalog {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
