sunchao commented on code in PR #57355:
URL: https://github.com/apache/spark/pull/57355#discussion_r3610810644
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala:
##########
@@ -198,10 +199,15 @@ case class ShuffleExchangeExec(
SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
private[sql] lazy val readMetrics =
SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+
+ private lazy val customWriteMetrics: Map[String, SQLMetric] =
+ CustomShuffleMetrics.createMetrics(
+ sparkContext,
sparkContext.shuffleDriverComponents.supportedCustomMetrics())
+
override lazy val metrics = Map(
"dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size"),
"numPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
partitions")
- ) ++ readMetrics ++ writeMetrics
+ ) ++ readMetrics ++ writeMetrics ++ customWriteMetrics
Review Comment:
[P1] Keep custom metrics from replacing Spark-owned entries
Because this map is appended last, a plugin can declare a metric such as
`shuffleRecordsWritten` and silently replace the Spark accumulator exposed
through `metrics`. The real reporter still updates the separate `writeMetrics`
instance, while `runtimeStatistics` reads `metrics("shuffleRecordsWritten")`; a
custom value (or default) of zero therefore makes a materialized non-empty
shuffle report `rowCount = 0`. `AQEPropagateEmptyRelation` treats that as
definitively empty and can rewrite a non-empty branch to an empty relation,
changing query results. Please reject/reserve collisions, or make all
Spark-owned keys take precedence as `SupportsCustomDriverMetrics` does, and add
a collision regression test.
##########
core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java:
##########
@@ -288,8 +290,11 @@ private long[] mergeSpills(SpillInfo[] spills) throws
IOException {
if (spills.length == 0) {
final ShuffleMapOutputWriter mapWriter = shuffleExecutorComponents
.createMapOutputWriter(shuffleId, mapId,
partitioner.numPartitions());
- return mapWriter.commitAllPartitions(
- ShuffleChecksumHelper.EMPTY_CHECKSUM_VALUE).getPartitionLengths();
+ long[] emptyPartitionLengths =
+
mapWriter.commitAllPartitions(ShuffleChecksumHelper.EMPTY_CHECKSUM_VALUE)
+ .getPartitionLengths();
+ customMetricsValues = mapWriter.currentMetricsValues();
Review Comment:
[P2] Capture metrics from the single-spill fast path
This handles the zero-spill case, but the following one-spill branch uses
`SingleSpillShuffleMapOutputWriter` and never assigns `customMetricsValues`. A
non-empty serialized shuffle that fits in memory still produces one final
spill, so whenever a plugin exposes the optional single-file writer this common
path completes successfully with the custom array left empty. The added Unsafe
test uses an empty iterator and cannot catch this. Please expose values through
the single-spill API (or a common task-level hook), capture them after
transfer, and cover the non-empty one-spill path.
--
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]
