peter-toth commented on code in PR #60030:
URL: https://github.com/apache/spark/pull/60030#discussion_r4471438347
##########
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:
Thanks, confirmed by a run. On the base commit, with AQE on, the nested
query fails with the internal error over a scan that declares `UNSUPPORTED`,
and it passes with this PR. The default mode passes there, as before.
[ce08e19](https://github.com/apache/spark/pull/60030/commits/ce08e1973f21b9e15d8e7c2daafe0e2ca59c7de8)
adds a test over such a scan with AQE on, and rewords this doc and the
description.
##########
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:
Thanks, removed from the description. On branch-4.x this line will pass
`relation.keyGroupedPartitioning`, unless SPARK-60260 is there first.
##########
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
+ * AQE on the copies of the join's scan plan their partitions when the
result stage is created,
+ * before the DPP subquery prunes the shared `Scan`.
+ */
+ private def checkStoragePartitionedJoin(query: String, expected: Seq[Row]):
Unit = {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = sql(query)
+ checkAnswer(df, expected)
+ val plan = df.queryExecution.executedPlan
+ assert(collectShuffles(plan).isEmpty,
Review Comment:
Thanks, I'd keep the checks as they are. Both tests fail without the fix,
which is what they are for.
##########
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:
Thanks, I'd keep it. The `Batch` check shows the copies share the
`PlannedScan`, and the factory is one of its lazy vals.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -181,3 +184,25 @@ case class BatchScanExec(
s"BatchScan ${table.name()}".trim
}
}
+
+/**
+ * A `Scan`, with the `Batch` and the input partitions a [[BatchScanExec]]
plans from it once, when
Review Comment:
The `PlannedScan` doc says the copies read the partitions with a reader
factory from the `Batch` that planned them, and the `@param` points there. I'd
keep 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]
