dongjoon-hyun commented on code in PR #60030:
URL: https://github.com/apache/spark/pull/60030#discussion_r4470331673
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala:
##########
@@ -321,6 +322,29 @@ class SparkPlanSuite extends SharedSparkSession {
assert(canonicalizedTypes == Seq(IntegerType, StringType),
s"Expected [IntegerType, StringType] but got $canonicalizedTypes")
}
+
+ test("SPARK-60752: copies of a BatchScanExec share its Batch and the
partitions it planned") {
+ // A `Scan` that returns a new `Batch` from each `toBatch`, which a
connector may do.
+ class NewBatchScan extends Scan {
+ override def readSchema(): StructType = new StructType()
+ override def toBatch: Batch = new Batch {
+ override def planInputPartitions(): Array[InputPartition] = Array(new
InputPartition {})
+ override def createReaderFactory(): PartitionReaderFactory =
+ throw new UnsupportedOperationException()
+ }
+ }
+ val scan = BatchScanExec(Seq.empty, new PlannedScan(new NewBatchScan),
Seq.empty, table = null)
+ def sharesPlanning(copy: SparkPlan): Boolean = copy match {
+ case c: BatchScanExec =>
+ (c.batch eq scan.batch) && (c.inputPartitions eq scan.inputPartitions)
Review Comment:
**24. [Nit] No test checks that the copies share the reader factory**
Thanks for moving the reader factory into the holder (item 17). Here
`createReaderFactory()` throws (L332-L333) and `sharesPlanning` compares the
`Batch` and the partitions only, so moving `readerFactory` back to a lazy val
of each node would still pass every test. If `createReaderFactory()` returned a
new factory on each call, one whose `createReader` throws, this could also
compare the factories:
```scala
(c.batch eq scan.batch) && (c.inputPartitions eq
scan.inputPartitions) &&
(c.readerFactory eq scan.readerFactory)
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -239,8 +239,8 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
val storageFilters = storageFiltersFor(relation, filtersAbove)
val checked = StorageFilterPlanning.checkedColumn(
session.sessionState.conf, storageFilters, relation.output)
- val batchExec = BatchScanExec(relation.output ++ checked, relation.scan,
runtimeFilters,
- relation.ordering, relation.relation.table, keyGroupedPartitioning,
+ val batchExec = BatchScanExec(relation.output ++ checked, new
PlannedScan(relation.scan),
+ runtimeFilters, relation.ordering, relation.relation.table,
keyGroupedPartitioning,
Review Comment:
**22. [Nit] The description still says the copies read the live bucketing
conf**
After the rebase onto SPARK-60260, this line passes the
`keyGroupedPartitioning` decided once at planning (L228-L232), and
`reportedKeyedPartitioning` no longer reads the conf. The first section of the
description still ends with "The copies still decide their keyed layout from
the live `spark.sql.sources.v2.bucketing.enabled`. SPARK-60260 (#59789) moves
that decision to `DataSourceV2Strategy`." Since the merge script puts the
description into the commit message, that would record a limitation that master
no longer has. Could the paragraph go, or say that SPARK-60260 already decides
it here?
Also, branch-4.x does not have SPARK-60260, so for "branch-4.x will take the
same change" this argument needs SPARK-60260 there first, or
`relation.keyGroupedPartitioning` as before.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -511,6 +511,82 @@ trait KeyGroupedPartitioningRuntimeFilterTests extends
KeyGroupedPartitioningSui
}
}
}
+
+ /**
+ * Runs `f` over `items` and `purchases`, both partitioned by the ids 1 to
7, with `rowsPerKey`
+ * rows in each `purchases` key, under the confs of a DPP subquery that
reuses no broadcast.
+ */
+ private def withDppSubqueryTables(rowsPerKey: Int)(f: => Unit): Unit = {
+ 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(rowsPerKey)(s"($i, $i.0, cast('2020-01-01' as
timestamp))")).mkString(", "))
+ f
+ }
+ }
+
+ /**
+ * Checks the rows of `query`, and that its join plans no shuffle. With AQE
off only, since with
Review Comment:
**20. [Low] With AQE on, the base commit also fails for a scan that declares
its columnar support mode**
This corrects my item 15, which I based on the in-memory fixtures. The
reason given here, that with AQE on the copies plan their partitions when the
result stage is created, holds only for a scan in the default
`PARTITION_DEFINED` mode. Only that mode reads the partitions in
`supportsColumnar` (`DataSourceV2ScanExecBase.scala` L200-L211). `SUPPORTED`
and `UNSUPPORTED` return without planning anything, which is what SPARK-44505
added them for (`DataSourceV2Suite.scala` L477-L492).
Reading the AQE path on the base commit, for a keyed scan with runtime
filtering whose `columnarSupportMode()` returns `UNSUPPORTED`, with the confs
of `withDppSubqueryTables` and AQE on, the default:
1. At result-stage creation, `ReuseAdaptiveSubquery` (first test) or
`PlanAdaptiveDynamicPruningFilters` (nested test, `flag = 1`) copies the
`purchases` scan. `ApplyColumnarRulesAndInsertTransitions` asks the copy
`supportsColumnar`, which plans nothing in this mode, and with no shuffle stage
nothing validates the plan.
2. The sort-merge join executes `items` first, which waits for the DPP
subquery in `waitForSubqueries`. The subquery's build side calls `filter()` on
the shared `Scan`.
3. Only then is the copy asked, by `checkChildStillMatches`
(`GroupPartitionsExec.scala` L355-L360), which `doExecute` calls first, or by
its own `filteredPartitions`. On the base commit the copy plans from the pruned
`Scan`, so the query fails with the INTERNAL_ERROR or `Can't zip RDDs ...
List(7, 1)`, the same as with AQE off.
This PR fixes that case too, since the copy shares the `PlannedScan` that
`EnsureRequirements` planned. But the description scopes the bug to AQE off
("With AQE off:" and the AQE-on note in the testing section), and no test runs
this path. No in-repo source reaches it: the keyed in-memory fixtures keep the
default mode, and the sources that declare a mode report no partitioning. So
the exposure is external connectors, with the default AQE.
Could the doc here and the description say that this holds for
`PARTITION_DEFINED` scans, and that a scan that declares `SUPPORTED` or
`UNSUPPORTED` fails the same way with AQE on? An AQE-on run of these checks
over such a scan would also cover the fix, for example with a catalog like
`CustomReportingCatalog` whose scan builder returns an `InMemoryBatchScan` that
overrides `columnarSupportMode()`. It would need `flag = 1` or the first test's
query, since with `flag = 2` the empty broadcast lets AQE drop the join (item
14). I traced this from the code and did not run 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]
