[SPARK-42942][SQL] Support coalesce table cache stage partitions #40574
Conversation
|
cc @dongjoon-hyun @cloud-fan @viirya @yaooqinn thank you |
| /** | ||
| * The [[Partition]] used by [[CachedRDD]]. | ||
| */ | ||
| case class CachedRDDPartition( |
There was a problem hiding this comment.
can we add some PR comments to indicate which code are copied from somewhere and which code are new?
There was a problem hiding this comment.
Most are from CoalescedRDD,
spark/core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala
Lines 74 to 78 in 5a56c17
that has similar use case to coalesce RDD partition. The main difference is
CachedRDD know its target partitions which make things simple. All core methods in CachedRDD can reuse the origin of previous RDD.
There was a problem hiding this comment.
So if that's the case, is there a reason why we can not implement this with CoalescedRDD and supplying the correct partitionCoalescer: Option[PartitionCoalescer]? It feels a little strange having an RDD classes code copied from core to SQL.
ulysses-you
force-pushed
the
coalesce-cache-partition
branch
from
75409df
to
229a57c
Compare
| // `RoundRobinPartitioning` but we don't need to retain the number of partitions. | ||
| case r: RoundRobinPartitioning => | ||
| r.copy(numPartitions = numPartitions) | ||
| case other@SinglePartition => |
There was a problem hiding this comment.
nit: spaces around @?
| r.copy(numPartitions = numPartitions) | ||
| case other@SinglePartition => | ||
| throw new IllegalStateException( | ||
| "Unexpected partitioning for coalesced shuffle read: " + other) |
There was a problem hiding this comment.
nit: s/Unexpected/Illegal ?
| case _ => | ||
| // Spark plugins may have custom partitioning and may replace this operator | ||
| // during the postStageOptimization phase, so return UnknownPartitioning here | ||
| // rather than throw an exception |
There was a problem hiding this comment.
Can we make this comment a DEBUG message?
| abstract class AQERead extends UnaryExecNode { | ||
| def child: SparkPlan | ||
| def partitionSpecs: Seq[ShufflePartitionSpec] |
There was a problem hiding this comment.
Can we have a comment about what AQERead is/does?
| if (specsMap.nonEmpty) { | ||
| updateCacheReads(plan, specsMap) | ||
| } else { | ||
| plan | ||
| } |
There was a problem hiding this comment.
So this only applies to reads?
|
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable. |
What changes were proposed in this pull request?
Add a new rule
CoalesceCachePartitionsto support coalesce partitions withTableCacheQueryStageExec. In order to reuse the code path withCoalesceShufflePartitions, this pr also does a small refactor about how we coalesce partitions.RDD cache use the RDD id and partition id as the block id, so it seems not possible to split skewd partitions like shuffle. To reduce complexity, this pr does not allow coalesce partitions with both shuffle and cache stage since shuffle read may contain skewed partition spec.
For example, the follow case can not be coalesced by both
CoalesceCachePartitionsandCoalesceShufflePartitions.Why are the changes needed?
Make AQE support coalesce table cache stage partitions.
Does this PR introduce any user-facing change?
yes, add a new config to control if coalesce partitions for table cache stage.
How was this patch tested?
add tests