You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2024/07/03 22:02:38 UTC
Re: [PR] [SPARK-48613][SQL] SPJ: Support auto-shuffle one side + less join keys than partition keys [spark]
sunchao commented on code in PR #47064:
URL: https://github.com/apache/spark/pull/47064#discussion_r1664859395
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -130,6 +130,15 @@ case class BatchScanExec(
}
k.copy(expressions = expressions, numPartitions = newPartValues.length,
partitionValues = newPartValues)
+ case k: KeyGroupedPartitioning if spjParams.joinKeyPositions.isDefined =>
+ val expressions = spjParams.joinKeyPositions.get.map(i => k.expressions(i))
+ val newPartValues = k.partitionValues.map{r =>
Review Comment:
Just curious why this is necessary. I see that in BatchScanExec we are also doing something similar, by checking the `joinKeyPositions` and call `uniquePartitionValues` at the end to de-dup.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -578,6 +582,20 @@ case class EnsureRequirements(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+
+ private def populateJoinKeyPositions(plan: SparkPlan,
+ joinKeyPositions: Option[Seq[Int]]): SparkPlan = plan match {
Review Comment:
nit: style
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -175,7 +175,11 @@ case class EnsureRequirements(
child
case ((child, dist), idx) =>
if (bestSpecOpt.isDefined && bestSpecOpt.get.isCompatibleWith(specs(idx))) {
- child
+ bestSpecOpt match {
Review Comment:
Can we add some comments here? to explain this is the case when `isKeyGroupCompatible` is false, but we still want to avoid shuffling on one side?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -130,6 +130,15 @@ case class BatchScanExec(
}
k.copy(expressions = expressions, numPartitions = newPartValues.length,
partitionValues = newPartValues)
+ case k: KeyGroupedPartitioning if spjParams.joinKeyPositions.isDefined =>
Review Comment:
I wonder if it is better to combine this clause with the clause of
```
case k: KeyGroupedPartitioning if spjParams.commonPartitionValues.isDefined
```
above, since there are some common logic. It's also worthwhile to add some comments here too, indicating that this case only applies when only one side of the join is `KeyGroupedPartitioning` and thus `commonPartitionValues` is not defined?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org