dongjoon-hyun commented on code in PR #60030:
URL: https://github.com/apache/spark/pull/60030#discussion_r4448632048
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -46,6 +46,12 @@ import org.apache.spark.util.ArrayImplicits._
* post-scan `Filter` too.
* @param appendsCheckedColumn Whether the readers mark the rows they checked
in a boolean column,
* the last of `output`, see
`StorageFilterPlanning.checkedColumn`.
+ * @param plannedInputPartitions Keeps the input partitions once the node
plans them. The copies
+ * later rules make of the node share it, so
they report the layout
Review Comment:
**1. [Medium] A copy still re-derives the keyed layout from the live
bucketing conf**
The holder shares the input partitions, but each copy still builds its own
`reportedKeyedPartitioning` on first use, and that is gated on the live
`conf.v2BucketingEnabled` (`DataSourceV2ScanExecBase.scala` L103-L106;
`SparkPlan.conf` is the session conf, `SparkPlan.scala` L70-L72). The copies
made after `EnsureRequirements` read it for the first time at execution, so
they can still report a layout other than the one the plan was built for.
With the tables and confs of the second new test, AQE off:
```scala
val df = sql(query) // the nested-DPP query of the second new
test
df.queryExecution.executedPlan // planned with bucketing on: no shuffle
withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
df.collect()
}
```
- `ReuseExchangeAndSubquery` runs at the end of the preparations
(`QueryExecution.scala` L886). The `purchases` scan it rebuilds, and the
`GroupPartitionsExec` above it, have not computed their partitioning yet.
- At execution the copy sees bucketing off, gets `None` for
`reportedKeyedPartitioning`, and reports `UnknownPartitioning(0)`.
- `checkChildStillMatches` then throws `[INTERNAL_ERROR]
GroupPartitionsExec's child no longer reports the partitioning it was planned
over` (`GroupPartitionsExec.scala` L354-L357).
- With AQE on, `ReuseAdaptiveSubquery` makes the copies inside `collect()`,
after the conf change, and the same happens.
- Without a `GroupPartitionsExec`, `filteredPartitions` skips the
None-padding and the key sort (`PushDownUtils.scala` L339-L341 and L354). The
join then fails with "Can't zip RDDs", or can zip partitions of different keys
when the source prunes nothing.
This is the recipe of the SPARK-59252 test
(`KeyGroupedPartitioningSuite.scala` L2183-L2198), which has no subquery and so
never builds a copy. The mechanism predates this PR, but this `@param` and the
new sentence at `GroupPartitionsExec.scala` L130-L132 now say that the copies
report the planned layout.
Keeping the first node's bucketing decision in the holder, next to the
partitions, would make that true, the same way `StampUnionDecisions` freezes a
`UnionExec` decision under the conf the exchanges were planned against
(`QueryExecution.scala` L849-L851). Storing the `KeyedPartitioning` itself
would be harder, because `doCanonicalize` shares the holder through `copy`, so
the canonicalized node would pick up expressions that were not normalized. I
traced this from the code and did not run it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -109,7 +116,7 @@ case class BatchScanExec(
keyGroupedPartitioning.map(_.filter(_.references.subsetOf(outputSet)))
@transient override lazy val inputPartitions: Seq[InputPartition] =
- batch.planInputPartitions().toImmutableArraySeq
+
plannedInputPartitions.getOrPlan(scan)(batch.planInputPartitions().toImmutableArraySeq)
Review Comment:
**2. [Low] The executed node reads partitions that another `Batch` planned**
The holder is also shared with the node that runs, because `executedPlan` is
built from `sparkPlan.clone()` (`QueryExecution.scala` L393). For the default
`ColumnarSupportMode.PARTITION_DEFINED` (`Scan.java` L157-L159),
`DataSourceV2Strategy` already asks the planner's node for `supportsColumnar`
(`DataSourceV2Strategy.scala` L242), which plans the partitions with that
node's `Batch` (`DataSourceV2ScanExecBase.scala` L195-L200). Every node after
it, including the executed one, takes those partitions from the holder, while
its `readerFactory` comes from its own `batch` (L132), whose
`planInputPartitions()` is never called.
Before this PR, a scan without a pushed runtime filter planned and read
through the same `Batch`. Spark mixed `Batch` instances only on the
runtime-filter path (`PushDownUtils.scala` L299). Now it does so for every scan
in the default mode, not only for the copies that later rules make.
- `Scan.toBatch` may be called more than once (`Scan.java` L73-L75), and
nothing in the DSv2 API pairs the partitions with the reader factory of the
same `Batch`, so this breaks no written contract.
- A connector whose `toBatch` returns a new `Batch`, which opens a read
session in `planInputPartitions()` and passes it to `createReaderFactory()`,
would now get a factory without a session, or with another one.
- No connector in this repo depends on that: the file scans, the in-memory
test tables and `StateScan` return `this`, `ColumnarScan` caches its `Batch`,
and the Kafka and Python sources are `UNSUPPORTED`.
Keeping the first node's `Batch` in the holder too, and building
`readerFactory` from it when the `Scan` matches, would keep the partitions and
the factory together. Otherwise a sentence in the description or in the
`Scan.toBatch` Javadoc would tell connector authors.
##########
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
+ * that instance. Only the nodes over the `Scan` that planned the partitions
use them. A copy given
+ * another `Scan` plans its own.
+ */
+class PlannedInputPartitions {
+ private var planned: Option[(Scan, Seq[InputPartition])] = None
+
+ def getOrPlan(scan: Scan)(plan: => Seq[InputPartition]): Seq[InputPartition]
= synchronized {
+ planned match {
+ case Some((plannedScan, partitions)) if plannedScan eq scan => partitions
+ case Some(_) => plan
Review Comment:
**3. [Low] A copy given another `Scan` drops the sharing for its own copies**
The holder keeps one `(Scan, partitions)` entry, and this branch plans
without storing the result. So:
- after a rule makes `b.copy(scan = s2)`, every later copy of that node,
with the same holder and the same `s2`, plans again from the live `Scan`;
- if the `s2` node asks first, it takes the slot, and the copies of the
original node plan again.
Either way the protection of this PR is off for that chain. For example, an
extension rule that wraps the scan after `EnsureRequirements`, such as
`b.copy(scan = DelegatingScan(b.scan))` from an injected columnar rule,
followed by `ReuseExchangeAndSubquery`, gives a copy that plans after the DPP
build side pruned the shared `Scan`. The join then fails with `Can't zip RDDs
... List(7, 1)` as before.
Nothing in this repo replaces `scan` on a `BatchScanExec` (only the new test
at `KeyGroupedPartitioningSuite.scala` L479 does), so only extensions are
affected. The class comment covers the first copy ("A copy given another `Scan`
plans its own"), but the `@param` at L49-L51 says "The copies later rules make
of the node share it" without that limit.
Keeping one entry per `Scan` instance, for example in a
`java.util.IdentityHashMap`, would make such a chain behave like the original
one. At least the `@param` could say "the copies over the same `Scan`". This
branch also runs the connector's planning under the holder's monitor, which is
not needed, since the result is not kept.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -55,7 +61,8 @@ case class BatchScanExec(
@transient table: Table,
keyGroupedPartitioning: Option[Seq[Expression]] = None,
storageFilters: Seq[Expression] = Nil,
- appendsCheckedColumn: Boolean = false
+ appendsCheckedColumn: Boolean = false,
+ @transient plannedInputPartitions: PlannedInputPartitions = new
PlannedInputPartitions
Review Comment:
**5. [Low] A 9th case-class field changes the signatures of `BatchScanExec`**
Adding the holder to the first parameter list changes the constructor,
`apply`, `copy` and `unapply`. On master that costs the 13 positional patterns
in 8 test files that this PR widens, a week after SPARK-60465 widened the same
13 from 6 to 8 fields. SPARK-42454 bundled the SPJ parameters to avoid exactly
this.
It matters more for the backports:
- branch-4.2, branch-4.3 and branch-4.x have a 6-field `BatchScanExec` and
no SPARK-60465, so this diff does not apply there.
- A 7th field in 4.2.x or 4.3.x would break plugins compiled against the
released class. Their constructor, `apply` and `copy` calls fail with
`NoSuchMethodError`, and their positional patterns stop compiling.
- MiMa does not flag it, because `org.apache.spark.sql.execution.*` is
excluded (`MimaExcludes.scala` L187). The field count of `BatchScanExec` has
not changed within a maintenance line from 3.3.0 to 4.3.0.
A second parameter list with `otherCopyArgs`, as SPARK-58265 did for
`SubqueryAdaptiveBroadcastExec.broadcastValueProjection`
(`SubqueryAdaptiveBroadcastExec.scala` L41-L45), keeps the extractor and
`productArity`, but not the JVM signatures, since the lists are flattened in
bytecode. Keeping every signature needs a `@transient private var` that
`makeCopy` and `clone` overrides hand over, as `InMemoryRelation` does for
`statsOfPlanToCache` (`InMemoryRelation.scala` L671 and L734-L745). A plain
case-class `copy` would then start a new holder, so the direct test would copy
through a transform instead. That shape may be worth it, at least for the
maintenance branches.
##########
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 " +
Review Comment:
**4. [Medium] branch-4.1 and branch-4.0 fail too, and the affected releases
are not listed**
The description says that branch-4.1 and branch-4.0 are not affected,
because their `BatchScanExec` pads to the partition values in `spjParams`. That
holds only when `EnsureRequirements` pushed those values and the copy still has
partitions. With AQE off, this test's query fails on branch-4.1 in two cases
(line numbers at d4521911899):
- With `spark.sql.sources.v2.bucketing.pushPartValues.enabled=false`, a
public conf that the 4.1 DPP tests also run with, `EnsureRequirements` only
checks `isCompatibleWith` and never calls `populateCommonPartitionInfo`
(`EnsureRequirements.scala` L509-L512 and L662-L666). The copy then reports its
own re-planned keys (`KeyGroupedPartitionedScan.scala` L57), and the join fails
with `Can't zip RDDs with unequal numbers of partitions: List(7, 3)`.
- With the default confs but no `dim` row with `flag = 1`, the build side
prunes every split, and the copy plans no partitions. `groupPartitions` returns
`None` (`DataSourceV2ScanExecBase.scala` L140), so `outputPartitioning` is not
a `KeyGroupedPartitioning` and `inputRDD` skips the padding
(`BatchScanExec.scala` L125-L130 and L139-L142). The query fails with `List(7,
0)` instead of returning no rows.
branch-4.0 behaves the same for row-based scans once
`spark.sql.sources.v2.bucketing.enabled` is on. In the other direction, 4.2.0,
4.2.1 and 4.3.0 (branch-4.2, branch-4.3 and branch-4.x) look affected as
described, since SPARK-55535 removed `spjParams` before 4.2.0 and their copies
plan `inputPartitions` again, but the description does not list them. The
`GroupPartitionsExec` INTERNAL_ERROR form needs SPARK-59289, which only master
and branch-4.x have.
Could the description list the affected versions, so that the backports can
be planned? A third case in this test, with no `dim` row matching, would also
pin the fully pruned build side on master, which this PR already handles. I
traced the branch-4.1 and branch-4.0 paths from the code and did not run them.
##########
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(
+ 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).flatMap(i =>
+ Seq.fill(2)(s"($i, $i.0, cast('2020-01-01' as
timestamp))")).mkString(", "))
+ createTable(dim, Array(Column.create("k", LongType),
Column.create("flag", IntegerType)),
+ Array.empty)
+ sql(s"INSERT INTO testcat.ns.$dim VALUES " +
+ (1 to 7).map(i => s"($i, ${if (i >= 5) 1 else 0})").mkString(", "))
+
+ Seq(true, false).foreach { aqe =>
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ val df = sql(s"SELECT i.id, p.price FROM testcat.ns.$items i JOIN " +
+ s"(SELECT /*+ BROADCAST(d) */ p.item_id, p.price FROM
testcat.ns.$purchases p " +
+ s"JOIN testcat.ns.$dim d ON p.item_id = d.k WHERE d.flag = 1) p ON
i.id = p.item_id")
+ checkAnswer(df, (5 to 7).flatMap(i => Seq.fill(2)(Row(i.toLong,
i.toFloat))))
+ val plan = df.queryExecution.executedPlan
+ assert(collectShuffles(plan).isEmpty,
+ s"aqe=$aqe: the join should stay storage-partitioned:\n$plan")
+ }
+ }
+ }
+ }
+
+ test("SPARK-60752: copies of a BatchScanExec share the input partitions it
planned") {
Review Comment:
**8. [Nit] The copy test runs in both runtime-filter suites**
This test has no join and no runtime filter. The trait is scoped to "runtime
filtering under a storage-partitioned join, whose outcome depends on how the
scan takes runtime filters" (L125-L128), and both
`KeyGroupedPartitioningRuntimeFilterSuite` and
`KeyGroupedPartitioningCatalystRuntimeFilterSuite` mix it in (L11164 and
L11178), so the test runs twice with the same outcome.
`KeyGroupedPartitioningSuite` (L484) has the same helpers and the same
`clearTables` cleanup, and would run it once.
##########
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(
+ 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).flatMap(i =>
+ Seq.fill(2)(s"($i, $i.0, cast('2020-01-01' as
timestamp))")).mkString(", "))
+ createTable(dim, Array(Column.create("k", LongType),
Column.create("flag", IntegerType)),
+ Array.empty)
+ sql(s"INSERT INTO testcat.ns.$dim VALUES " +
+ (1 to 7).map(i => s"($i, ${if (i >= 5) 1 else 0})").mkString(", "))
+
+ Seq(true, false).foreach { aqe =>
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ val df = sql(s"SELECT i.id, p.price FROM testcat.ns.$items i JOIN " +
+ s"(SELECT /*+ BROADCAST(d) */ p.item_id, p.price FROM
testcat.ns.$purchases p " +
+ s"JOIN testcat.ns.$dim d ON p.item_id = d.k WHERE d.flag = 1) p ON
i.id = p.item_id")
+ checkAnswer(df, (5 to 7).flatMap(i => Seq.fill(2)(Row(i.toLong,
i.toFloat))))
+ val plan = df.queryExecution.executedPlan
+ assert(collectShuffles(plan).isEmpty,
+ s"aqe=$aqe: the join should stay storage-partitioned:\n$plan")
+ }
+ }
+ }
+ }
+
+ test("SPARK-60752: copies of a BatchScanExec share the input partitions it
planned") {
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp))")
+ def plannedScan(): BatchScanExec =
+ collectScans(sql(s"SELECT * FROM
testcat.ns.$items").queryExecution.executedPlan).head
+ val scan = plannedScan()
+ // A copy, like the ones runtime filter rewrites make, uses the partitions
the node planned.
+ assert(scan.copy(runtimeFilters = Nil).inputPartitions eq
scan.inputPartitions)
+ // A copy over another `Scan` plans its own.
+ assert(!(scan.copy(scan = plannedScan().scan).inputPartitions eq
scan.inputPartitions))
Review Comment:
**6. [Low] This assertion cannot catch an other-`Scan` copy that takes over
the holder**
`toImmutableArraySeq` wraps the array in a new `ArraySeq` on every call, so
this assertion holds with or without the holder. It would also hold if `case
Some(_) => plan` stored its result (`planned = Some((scan, p))`), which would
turn the sharing off for every later copy over the original `Scan`. None of the
three new tests would fail then, since in the first two every copy keeps the
same `Scan`. Repeating the assertion of L477 after this line would catch it:
```scala
assert(scan.copy(runtimeFilters = Nil).inputPartitions eq
scan.inputPartitions)
```
##########
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:
**9. [Nit] The two end-to-end tests repeat their setup**
This test and the previous one repeat the four-conf `withSQLConf`, the
`items` table, the `purchases` table with one or two rows per key, the AQE loop
and the no-shuffle check, about 16 lines. A private helper that takes the rows
per key would remove about 10 lines net. The older tests in this trait inline
the same DPP confs too, so this is only a suggestion.
##########
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:
**12. [Nit] Newly planned nodes over the same `Scan` still plan it again**
As this comment says, each newly built node gets its own holder, so the
nodes planned again over the same `Scan` still call `planInputPartitions()`
once each. Those are the DPP build side (`PlanDynamicPruningFilters.scala`
L60-L61 and L128-L129, where the first plan only checks broadcast reuse and is
dropped), its AQE counterparts (`PlanAdaptiveDynamicPruningFilters.scala` L124
and `InsertAdaptiveSparkPlan.compileSubquery`), and `reOptimize`.
For the first new test, this PR brings the calls on the shared `purchases`
scan from 8 to 5 without AQE and from 10 to 5 with it, and 2 of the remaining 5
come from the rebuilt build side. So "runs less often" in the description is
right, and this is only a note. A holder per `Scan` on the logical relation
would cover these nodes, but it needs `otherCopyArgs`-style plumbing on
`DataSourceV2ScanRelation`, which does not look worth 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
}
}
+
+/**
Review Comment:
**10. [Nit] The contract of the holder is described twice**
The `@param` at L49-L54 and this comment both say that the copies later
rules make of the node share the holder. Keeping the contract here, and only
the reason (a shared `Scan` that a DPP build side prunes) in the `@param` with
a link, would leave one place to update when the contract changes, for example
for item 3.
##########
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 {
Review Comment:
**7. [Low] The `>=` iteration repeats the path of the `=` one**
Both predicates become runtime filters through the same `SCALAR_SUBQUERY`
condition (`DataSourceV2Strategy.scala` L207-L213), the Catalyst fixture prunes
on both through the same interpreted predicate, and the fix does not depend on
the operator. Without the fix, the `=` iteration already fails first. The three
keys that `>=` keeps exercise the padding over several keys, which the existing
tests of this trait already cover.
The test runs in two suites, in two AQE modes and with two predicates, and
`checkAnswer` executes each query twice, so it makes 16 executions. The 8 in
`KeyGroupedPartitioningRuntimeFilterSuite` pass with or without the fix
(L400-L401). Dropping `>=` halves the count.
From reading the code, the AQE-on iteration may also pass without the fix.
There the copy plans its partitions when the final stage is created (through
its `supportsColumnar`), before the subquery runs, so it is the AQE-off
iteration that catches the regression. I did not run it.
##########
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:
**11. [Nit] Each copy still sorts the partition keys again**
The rebuilt node shares the partitions, but it still builds its own
`reportedKeyedPartitioning` from them: one key extraction per split, a sort
with the interpreted grouped-key ordering, a wrapper per key and a `distinct`
(`DataSourceV2ScanExecBase.scala` L111-L113). This node then compares the
result with `childPartitioning` by value (L139, and again in
`checkChildStillMatches`). For N splits that is about N log N comparisons and N
wrappers on the driver per copy. This predates the PR and is small next to the
planning itself. If the holder keeps the bucketing decision for item 1, it
could keep the derived partitioning too, keyed by the node's
`keyGroupedPartitioning` and `output` so that the canonicalized node does not
reuse it.
--
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]
