wangyum commented on a change in pull request #32781:
URL: https://github.com/apache/spark/pull/32781#discussion_r646673351
##########
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
##########
@@ -1351,6 +1351,45 @@ object RepartitionByExpression {
}
}
+/**
+ * This method repartitions data using [[Expression]]s into
`numShufflePartitions` defined in
+ * `SQLConf`, and could be coalesced by AQE. Usually used to merge small files.
+ */
+case class CoalescePartitions(
+ partitionExpressions: Seq[Expression],
+ child: LogicalPlan) extends RepartitionOperation {
+
+ override val numPartitions = conf.numShufflePartitions
+
+ override val partitioning: Partitioning = {
+ val (sortOrder, nonSortOrder) =
partitionExpressions.partition(_.isInstanceOf[SortOrder])
+
+ require(sortOrder.isEmpty || nonSortOrder.isEmpty,
+ s"${getClass.getSimpleName} expects that either all its
`partitionExpressions` are of type " +
+ "`SortOrder`, which means `RangePartitioning`, or none of them are
`SortOrder`, which " +
+ "means `HashPartitioning`. In this case we have:" +
+ s"""
+ |SortOrder: $sortOrder
+ |NonSortOrder: $nonSortOrder
+ """.stripMargin)
+
+ if (numPartitions == 1) {
+ SinglePartition
+ } else if (sortOrder.nonEmpty) {
+ RangePartitioning(sortOrder.map(_.asInstanceOf[SortOrder]),
numPartitions)
Review comment:
It needs to set its own `ShuffleOrigin`:
https://github.com/apache/spark/blob/ba75b09281f04e4376b6da0b402828c33075f8d1/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala#L721-L722
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
