peter-toth commented on code in PR #60030:
URL: https://github.com/apache/spark/pull/60030#discussion_r4456685990
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -389,6 +389,95 @@ trait KeyGroupedPartitioningRuntimeFilterTests extends
KeyGroupedPartitioningSui
checkAnswer(df, Seq(Row(1L, 42.0f), Row(1L, 44.0f)))
}
}
+
+ test("SPARK-60752: a DPP subquery's build side does not change the layout of
the scan it " +
+ "shares with the join") {
+ // A DPP subquery that does not reuse a broadcast plans its build side
again, over the same
+ // `Scan` as the join's `purchases` scan. A scalar subquery filter on
`item_id` is a runtime
+ // filter of both, and the subquery runs it first, which prunes the shared
`Scan`. The join's
+ // scan is a copy made after planning, when the scalar subquery is
replaced by a reused one. Had
+ // it planned its partitions again, it would report fewer of them than
`items`, and the join
+ // would fail with "Can't zip RDDs". Of the two suites that run this, only
the Catalyst one
+ // prunes on these predicates.
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " + (1 to 7).map(i =>
+ s"($i, 'x', 10.0, cast('2020-01-01' as timestamp))").mkString(", "))
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " + (1 to 7).map(i =>
+ s"($i, $i.0, cast('2020-01-01' as timestamp))").mkString(", "))
+
+ def bound(agg: String): String =
+ s"(SELECT $agg(item_id) FROM testcat.ns.$purchases WHERE price > 4.0)"
+ Seq(true, false).foreach { aqe =>
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ Seq(s"i.id = ${bound("max")}" -> Seq(7), s"i.id >= ${bound("min")}"
-> (5 to 7)).foreach {
+ case (filter, ids) =>
+ val df = sql(s"SELECT i.id, p.price FROM testcat.ns.$items i " +
+ s"JOIN testcat.ns.$purchases p ON i.id = p.item_id WHERE
$filter")
+ checkAnswer(df, ids.map(i => Row(i.toLong, i.toFloat)))
+ val plan = df.queryExecution.executedPlan
+ assert(collectShuffles(plan).isEmpty,
+ s"aqe=$aqe, $filter: the join should stay
storage-partitioned:\n$plan")
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60752: a nested DPP filter on a DPP subquery's build side does
not change the " +
+ "layout of the scan it shares with the join") {
+ // As above, with a DPP filter from a broadcast dimension as the build
side's own runtime
+ // filter, which every fixture prunes on. Two rows per key give
`purchases` two splits per key,
+ // so a GroupPartitionsExec groups that side and checks the layout of the
copied scan.
+ val dim = "dim"
+ withSQLConf(
Review Comment:
Thanks, done in
[c016753](https://github.com/apache/spark/pull/60030/commits/c016753d73373f5bdd2a3a0cc65eb9f97f730179)
with `withDppSubqueryTables` and `checkStoragePartitionedJoin`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -181,3 +188,24 @@ case class BatchScanExec(
s"BatchScan ${table.name()}".trim
}
}
+
+/**
Review Comment:
Thanks, done in
[474ef79](https://github.com/apache/spark/pull/60030/commits/474ef79bb89401903919eefd1d8495443a7b6843).
The contract is in the `PlannedScan` doc, and the `@param` points there.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -126,9 +126,10 @@ case class GroupPartitionsExec(
* The comparison is by reference for a child that reports the same object,
which the columnar and
* codegen wrappers do. A rule that rewrites a subquery in the runtime or
storage filters of a
* `BatchScanExec` after `EnsureRequirements` rebuilds it, so there it
compares the partition keys
- * by value, and reads equal only while the connector returns the same input
partitions for the
- * same scan. Such rules are `PlanAdaptiveDynamicPruningFilters` and
`ReuseAdaptiveSubquery` with
- * AQE, and `ReuseExchangeAndSubquery` without it.
+ * by value. Such rules are `PlanAdaptiveDynamicPruningFilters` and
`ReuseAdaptiveSubquery` with
+ * AQE, and `ReuseExchangeAndSubquery` without it. The rebuilt node shares
the input partitions
+ * the first one planned (`BatchScanExec.plannedInputPartitions`), so its
keys come from the same
Review Comment:
Agreed, it predates this PR and is small next to the planning, so I left it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -181,3 +188,24 @@ case class BatchScanExec(
s"BatchScan ${table.name()}".trim
}
}
+
+/**
+ * The input partitions of a [[BatchScanExec]], planned once, by the first
node that asks for them.
+ * Each newly built node gets its own instance, and the copies later rules
make of the node share
Review Comment:
Agreed, thanks for counting.
--
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]
