ammarchalifah commented on code in PR #59599:
URL: https://github.com/apache/spark/pull/59599#discussion_r4377517437
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -144,7 +145,9 @@ case class EnsureRequirements(
// To choose a KeyGroupedShuffleSpec, we must be able to push down
SPJ parameters into
// the scan (for join key positions). If these parameters can't be
pushed down, this
// spec can't be used to shuffle other children.
- case (idx, _: KeyGroupedShuffleSpec) =>
canPushDownSPJParamsToScan(children(idx))
+ case (idx, s: KeyGroupedShuffleSpec) =>
+ canPushDownSPJParamsToScan(children(idx)) &&
+ !isBuiltFromReducedScan(children(idx), s)
Review Comment:
Addressed
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -698,7 +709,121 @@ case class EnsureRequirements(
joinType == LeftAnti || joinType == LeftOuter
}
- // Populate the common partition information down to the scan nodes
+ /**
+ * Compose join-key positions from an outer SPJ with positions already
pushed by an inner SPJ.
+ * Outer positions index the layout the outer join sees; inner positions
index the scan's raw
+ * partition expressions, so they are composed rather than replaced.
+ */
+ private def composeJoinKeyPositions(
+ existing: Option[Seq[Int]],
+ newPositions: Option[Seq[Int]]): Option[Seq[Int]] = {
+ (existing, newPositions) match {
+ case (Some(old), Some(newPos)) => Some(newPos.map(old))
+ case (_, None) => existing
+ case (None, Some(newPos)) => Some(newPos)
+ }
+ }
+
+ /**
+ * Project common partition values onto `projectionPositions` of the layout
`expressions`
+ * describes and deduplicate. Values that collapse to the same key are read
as one group, so
+ * they become one partition, unless the scan is partially clustered, where
each split of the
+ * group still gets its own partition and the split counts add up.
+ */
+ private def projectAndDedupeCommonPartitionValues(
+ values: Seq[(InternalRow, Int)],
+ expressions: Seq[Expression],
+ projectionPositions: Seq[Int],
+ partiallyClustered: Boolean): Seq[(InternalRow, Int)] = {
+ val projectedExpressions = projectionPositions.map(expressions(_))
+ val partitionOrdering = RowOrdering.createNaturalAscendingOrdering(
+ projectedExpressions.map(_.dataType))
+ values.map { case (partValue, numSplits) =>
+ val projected = KeyGroupedPartitioning.project(
+ expressions, projectionPositions, partValue)
+ (InternalRowComparableWrapper(projected, projectedExpressions),
numSplits)
+ }.groupBy(_._1).map { case (wrapper, entries) =>
+ (wrapper.row, if (partiallyClustered) entries.map(_._2).sum else 1)
+ }.toSeq.sortBy(_._1)(partitionOrdering)
Review Comment:
Addressed
--
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]
