ulysses-you commented on code in PR #58591:
URL: https://github.com/apache/spark/pull/58591#discussion_r3956398899
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -274,6 +307,34 @@ case class GroupPartitionsExec(
@transient private lazy val hasCoalescing: Boolean =
groupedPartitions.exists(_._2.size > 1)
+ // All values are computed on the driver by `grouping`, so they are reported
through
+ // `sendDriverMetrics` rather than task-side accumulators. Registration
reads constructor
+ // parameters only: pruning and replication are facts of the alignment path
and are registered
+ // only in the modes that produce them, while coalescing can happen in any
mode and reports 0
+ // when nothing merged.
+ @transient override lazy val metrics: Map[String, SQLMetric] = Map(
+ "numInputPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
input partitions"),
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
partitions"),
+ "numEmptyPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
empty partitions"),
+ "numCoalescedPartitions" ->
+ SQLMetrics.createMetric(sparkContext, "number of coalesced partitions"),
+ "maxPartitionsPerGroup" ->
+ SQLMetrics.createMetric(sparkContext, "max partitions per group")) ++ {
+ if (expectedPartitionKeys.isDefined && !distributePartitions) {
+ Map("numReplicatedPartitions" -> SQLMetrics.createMetric(sparkContext,
+ "number of replicated input partition reads"))
Review Comment:
Fixed in a9a920e489a: narrowed the registration to
`expectedPartitionKeys.exists(_.exists(_._2 > 1)) && !distributePartitions` and
renamed the key to `numReplicatedPartitionReads`.
@peter-toth your single-source-of-truth follow-up applied as well: `set` is
now `metrics.get(name).foreach` and both `if`s in `sendDriverMetrics` are gone,
so the predicates cannot drift apart. A mistyped name fails the tests, which
assert every metric by value or absence.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -274,6 +307,34 @@ case class GroupPartitionsExec(
@transient private lazy val hasCoalescing: Boolean =
groupedPartitions.exists(_._2.size > 1)
+ // All values are computed on the driver by `grouping`, so they are reported
through
+ // `sendDriverMetrics` rather than task-side accumulators. Registration
reads constructor
+ // parameters only: pruning and replication are facts of the alignment path
and are registered
+ // only in the modes that produce them, while coalescing can happen in any
mode and reports 0
+ // when nothing merged.
+ @transient override lazy val metrics: Map[String, SQLMetric] = Map(
+ "numInputPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
input partitions"),
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
partitions"),
+ "numEmptyPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
empty partitions"),
+ "numCoalescedPartitions" ->
+ SQLMetrics.createMetric(sparkContext, "number of coalesced partitions"),
+ "maxPartitionsPerGroup" ->
+ SQLMetrics.createMetric(sparkContext, "max partitions per group")) ++ {
+ if (expectedPartitionKeys.isDefined && !distributePartitions) {
+ Map("numReplicatedPartitions" -> SQLMetrics.createMetric(sparkContext,
+ "number of replicated input partition reads"))
+ } else {
+ Map.empty[String, SQLMetric]
+ }
+ } ++ {
+ if (expectedPartitionKeys.isDefined) {
Review Comment:
Softened the comment in a9a920e489a: pruning is registered over the whole
alignment path; the ordering producer expects the child's own keys and so never
prunes, but no constructor parameter separates it from the join producers that
can.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -274,6 +307,34 @@ case class GroupPartitionsExec(
@transient private lazy val hasCoalescing: Boolean =
groupedPartitions.exists(_._2.size > 1)
+ // All values are computed on the driver by `grouping`, so they are reported
through
+ // `sendDriverMetrics` rather than task-side accumulators. Registration
reads constructor
+ // parameters only: pruning and replication are facts of the alignment path
and are registered
+ // only in the modes that produce them, while coalescing can happen in any
mode and reports 0
+ // when nothing merged.
+ @transient override lazy val metrics: Map[String, SQLMetric] = Map(
+ "numInputPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
input partitions"),
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
partitions"),
+ "numEmptyPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
empty partitions"),
+ "numCoalescedPartitions" ->
+ SQLMetrics.createMetric(sparkContext, "number of coalesced partitions"),
+ "maxPartitionsPerGroup" ->
+ SQLMetrics.createMetric(sparkContext, "max partitions per group")) ++ {
Review Comment:
This only matters for a repeated post, which cannot happen as peter-toth
measured: `SparkPlan.execute()` memoizes `doExecute` through a `LazyTry`
(SparkPlan.scala:185-201), so each value is posted exactly once and the SUM
aggregation sees a single update per metric. a9a920e489a notes the memoization
in a comment.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -359,7 +420,47 @@ case class GroupPartitionsExec(
private[v2] def kWayMergeOrdering: Seq[SortOrder] =
child.outputOrdering.map(_.copy(sameOrderExpressions = Seq.empty))
+ private def sendDriverMetrics(): Unit = {
+ val executionId =
sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ val driverAccumUpdates = ArrayBuffer.empty[(Long, Long)]
+ def set(name: String, value: Long): Unit = {
+ val metric = metrics(name)
+ metric.set(value)
+ driverAccumUpdates += (metric.id -> value)
+ }
+ // A single pass for the three per-group counts; an empty group and a
coalesced one are
+ // mutually exclusive.
+ var numEmptyPartitions = 0
+ var numCoalescedPartitions = 0
+ var maxPartitionsPerGroup = 0
+ groupedPartitions.foreach { case (_, group) =>
+ val size = group.size
+ if (size == 0) {
+ numEmptyPartitions += 1
+ } else if (size > 1) {
+ numCoalescedPartitions += 1
+ }
+ if (size > maxPartitionsPerGroup) {
+ maxPartitionsPerGroup = size
+ }
+ }
+ set("numInputPartitions", grouping.numInputPartitions)
+ set("numPartitions", groupedPartitions.size)
+ set("numEmptyPartitions", numEmptyPartitions)
+ set("numCoalescedPartitions", numCoalescedPartitions)
+ set("maxPartitionsPerGroup", maxPartitionsPerGroup)
+ if (expectedPartitionKeys.isDefined && !distributePartitions) {
+ set("numReplicatedPartitions", grouping.numReplicatedPartitions)
+ }
+ if (expectedPartitionKeys.isDefined) {
+ set("numPrunedPartitions", grouping.numPrunedPartitions)
+ }
+ SQLMetrics.postDriverMetricsUpdatedByValue(
+ sparkContext, executionId, driverAccumUpdates.toSeq)
+ }
+
override protected def doExecute(): RDD[InternalRow] = {
+ sendDriverMetrics()
Review Comment:
Confirmed against SparkPlan.scala:185-201: `execute()` is `executeRDD.get`
over `LazyTry { doExecute() }`, so `doExecute` runs at most once per instance
and an extra guard would memoize something already memoized. Thanks @peter-toth
for the mechanism and the measurement. a9a920e489a keeps the direct call with a
comment noting the memoization.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -380,6 +481,7 @@ case class GroupPartitionsExec(
child.supportsColumnar && !(hasCoalescing && enableSortedMerge &&
canUseSortedMerge)
override protected def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ sendDriverMetrics()
Review Comment:
Same resolution as the thread above: `executeColumnar()` memoizes
`doExecuteColumnar` through `executeColumnarRDD` (SparkPlan.scala:221-236), so
both entry points post at most once without an extra guard (a9a920e489a).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -450,12 +552,20 @@ case class GroupPartitionsExec(
}
}
-/** What a [[GroupPartitionsExec]] computes once and reports from several
members. */
+/**
+ * What a [[GroupPartitionsExec]] computes once and reports from several
members.
+ * `numInputPartitions` is the child's split count; the last two fields count
the alignment's
+ * effect on the reads of those splits (see `alignToExpectedKeys`), and are 0
outside the
+ * alignment path.
+ */
private case class PartitionGrouping(
partitions: Seq[(InternalRowComparableWrapper, Seq[Int])],
isGrouped: Boolean,
isCollapsed: Boolean,
- keysRewritten: Boolean)
+ keysRewritten: Boolean,
+ numInputPartitions: Int,
Review Comment:
Removed in a9a920e489a: `sendDriverMetrics` reads
`child.outputPartitioning.numPartitions`; `alignToExpectedKeys` keeps the
parameter.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -149,22 +150,52 @@ case class GroupPartitionsExec(
})
}
- /** Aligns partitions based on `expectedPartitionKeys` and clustering mode.
*/
- private def alignToExpectedKeys(keyMap: Map[InternalRowComparableWrapper,
Seq[Int]]) = {
+ /**
+ * Aligns partitions based on `expectedPartitionKeys` and clustering mode.
+ *
+ * Returns the aligned groups, whether their keys ended up unique, and two
counts, both in
+ * input partition reads relative to the baseline of every input being read
exactly once.
+ * The pruned count is the reads that never happen: inputs of a key the
alignment never
+ * references. The join proved such keys cannot produce output, e.g. an
inner join keeps
+ * only the intersection. Which keys are expected is orthogonal to the mode:
an
+ * `OrderedDistribution` producer expects the child's own keys and so never
prunes, while a
+ * join producer can prune in either mode. The replicated count is the reads
that happen
+ * again: in the replicate mode of partial clustering every expected
partition of a held key
+ * re-reads all its splits, counted beyond the first pass (3 splits over 2
slots count 3).
+ * Total input reads = `numInputPartitions` - pruned + replicated.
+ */
+ private def alignToExpectedKeys(
+ keyMap: Map[InternalRowComparableWrapper, Seq[Int]],
+ numInputPartitions: Int) = {
var isGrouped = true
+ var numReplicatedPartitions = 0
+ // Splits the alignment references, accumulated over the expected keys.
Every producer of
+ // `expectedPartitionKeys` deduplicates the keys, so no split is counted
twice here.
+ var numMatchedPartitions = 0
val alignedPartitions = expectedPartitionKeys.get.flatMap { case (key,
numSplits) =>
+ // Every producer derives the split counts from a `groupBy` size or the
literal 1; a
+ // non-positive count would emit a different partition count for the key
than the other
+ // side expects, breaking the pairing this alignment exists for.
+ assert(numSplits > 0, s"expected partition key split count must be
positive: $numSplits")
Review Comment:
This assert is sort of required by this change itself: `numSplits <= 0`
would make the new replicated-reads product `(numSplits - 1) * splits.size`
negative, so it guards the derivation added here, not only the pre-existing
planner contract. If you don't mind, I'd prefer to keep it in this PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala:
##########
@@ -18,17 +18,22 @@
package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute,
AttributeReference, SortOrder, TransformExpression}
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution,
KeyedPartitioning, KeyedShuffleSpec, KeyReducer, Partitioning,
PartitioningCollection, UnknownPartitioning}
import org.apache.spark.sql.catalyst.util.InternalRowComparableWrapper
+import org.apache.spark.sql.connector.KeyGroupedPartitioningSuiteBase
import org.apache.spark.sql.connector.catalog.functions.{BucketFunction,
BucketReducer, Reducer}
+import org.apache.spark.sql.connector.expressions.Expressions.identity
import org.apache.spark.sql.execution.{DummySparkPlan, LeafExecNode,
SafeForKWayMerge}
+import org.apache.spark.sql.execution.metric.SQLMetricsTestUtils
+import org.apache.spark.sql.execution.ui.SparkPlanGraphNode
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{DataType, IntegerType}
-class GroupPartitionsExecSuite extends SharedSparkSession {
+class GroupPartitionsExecSuite
+ extends KeyGroupedPartitioningSuiteBase with SQLMetricsTestUtils {
Review Comment:
Moved the three e2e tests to `KeyGroupedPartitioningSuite` (carries
`@ExtendedSQLTest`); `GroupPartitionsExecSuite` is back to `SharedSparkSession`
with unit tests only (a9a920e489a).
--
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]
