ulysses-you commented on code in PR #59950:
URL: https://github.com/apache/spark/pull/59950#discussion_r4450010019
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala:
##########
@@ -125,7 +125,12 @@ case class PlanDynamicPruningFilters(sparkSession:
SparkSession) extends Rule[Sp
case None =>
val aliases = broadcastKeyIndices.map(idx =>
Alias(buildKeys(idx), buildKeys(idx).toString)())
- val aggregate = Aggregate(aliases, aliases, buildPlan)
+ // Alias the keys in a Project instead of making them the grouping
expressions.
Review Comment:
Added to the user-facing section: these plan changes apply only when the
internal `spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly` is
false. With the default, DPP plans no subquery, so default plans do not change.
07c9fb47549.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala:
##########
@@ -125,7 +125,12 @@ case class PlanDynamicPruningFilters(sparkSession:
SparkSession) extends Rule[Sp
case None =>
val aliases = broadcastKeyIndices.map(idx =>
Alias(buildKeys(idx), buildKeys(idx).toString)())
- val aggregate = Aggregate(aliases, aliases, buildPlan)
+ // Alias the keys in a Project instead of making them the grouping
expressions.
+ // Grouping on plain attributes lets the aggregate keep the build
side's partitioning
+ // (e.g. KeyedPartitioning) instead of requiring a shuffle. See
+ // AliasAwareOutputExpression.
+ val groupingAttrs = aliases.map(_.toAttribute)
+ val aggregate = Aggregate(groupingAttrs, groupingAttrs,
Project(aliases, buildPlan))
Review Comment:
Reproduced. With this PR and #60030's nested-DPP test but without its fix,
both runtime-filter suites fail with `[INTERNAL_ERROR] GroupPartitionsExec's
child no longer reports the partitioning it was planned over`. With #60030's
`BatchScanExec` and `GroupPartitionsExec` changes, the test passes. So #60030
is the fix, and its nested-DPP test covers this subquery shape once this PR is
rebased on it. 07c9fb47549.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala:
##########
@@ -120,7 +120,12 @@ case class PlanAdaptiveDynamicPruningFilters(
case None =>
val aliases = indices.map(idx =>
Alias(buildKeys(idx), buildKeys(idx).toString)())
- val aggregate = Aggregate(aliases, aliases, buildPlan)
+ // Alias the keys in a Project instead of making them the grouping
expressions.
+ // Grouping on plain attributes lets the aggregate keep the build
side's partitioning
+ // (e.g. KeyedPartitioning) instead of requiring a shuffle. See
+ // AliasAwareOutputExpression.
+ val groupingAttrs = aliases.map(_.toAttribute)
+ val aggregate = Aggregate(groupingAttrs, groupingAttrs,
Project(aliases, buildPlan))
Review Comment:
Added to the user-facing section: with AQE, a build side that is itself a
shuffled join on the key loses skew-join splitting inside the subquery. I did
not keep the old shape in that case: both rules see the build side's logical
plan, and the join strategy and skew are decided during physical and AQE
planning. 07c9fb47549.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -226,6 +228,70 @@ trait KeyGroupedPartitioningRuntimeFilterTests extends
KeyGroupedPartitioningSui
}
}
+ test("SPARK-60741: dynamic partition filtering keeps the build side's keyed
partitioning") {
+ 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") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ s"(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ s"(2, 'bb', 10.5, cast('2020-01-01' as timestamp)), " +
+ s"(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ s"(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 44.0, cast('2020-01-15' as timestamp)), " +
+ s"(1, 45.0, cast('2020-01-15' as timestamp)), " +
+ s"(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ s"(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ // The false arm exercises PlanDynamicPruningFilters and the true arm
+ // PlanAdaptiveDynamicPruningFilters.
+ Seq(false, true).foreach { aqeEnabled =>
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->
aqeEnabled.toString) {
+ val df = sql(s"SELECT sum(p.price) FROM testcat.ns.$items i, " +
+ s"testcat.ns.$purchases p WHERE i.id = p.item_id AND i.price >
40.0")
+ checkAnswer(df, Seq(Row(131)))
+
+ // The subquery groups only on the build side's key column, which is
a partition key of
+ // `items`, so its aggregate must keep the scan's keyed
partitioning. These keys repeat
+ // across splits, so keeping the partitioning means a
`GroupPartitionsExec` coalesces
+ // them, and no shuffle is needed.
+ val plan = df.queryExecution.executedPlan
+ val subqueries = collectScans(plan)
+ .flatMap(_.runtimeFilters)
+ .flatMap(_.collect { case d: DynamicPruningExpression => d })
+ .flatMap(_.collect { case i: InSubqueryExec => i.plan })
+ assert(subqueries.nonEmpty, s"expected a DPP subquery, got:\n$plan")
+ subqueries.foreach {
+ case subquery: SubqueryExec =>
+ val subPlan = subquery.child
Review Comment:
Added: the AQE arm asserts the subquery is an `AdaptiveSparkPlanExec`, and
the other arm that it is not. 07c9fb47549.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala:
##########
@@ -125,7 +125,12 @@ case class PlanDynamicPruningFilters(sparkSession:
SparkSession) extends Rule[Sp
case None =>
val aliases = broadcastKeyIndices.map(idx =>
Alias(buildKeys(idx), buildKeys(idx).toString)())
- val aggregate = Aggregate(aliases, aliases, buildPlan)
+ // Alias the keys in a Project instead of making them the grouping
expressions.
+ // Grouping on plain attributes lets the aggregate keep the build
side's partitioning
+ // (e.g. KeyedPartitioning) instead of requiring a shuffle. See
+ // AliasAwareOutputExpression.
Review Comment:
The comment now lives in `DynamicPruningSubquery.planDistinctBuildKeys` and
names `PartitioningPreservingUnaryExecNode` and the partial aggregate's
grouping-only output. 07c9fb47549.
--
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]
