sunchao commented on code in PR #57355:
URL: https://github.com/apache/spark/pull/57355#discussion_r3706596055
##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleWriteProcessor.scala:
##########
@@ -57,6 +63,7 @@ private[spark] class ShuffleWriteProcessor extends
Serializable with Logging {
createMetricsReporter(context))
writer.write(inputs.asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
val mapStatus = writer.stop(success = true)
+ reportCustomMetrics(writer.currentMetricsValues())
Review Comment:
[P2] Read custom shuffle metrics before closing the writer
The preceding `writer.stop(success = true)` closes the writer, while
`ShuffleWriter.currentMetricsValues()` only promises values after a successful
`write`. A valid custom writer can release its task/client state in `stop`;
reading its metrics afterward therefore loses the values or throws and fails an
already-written map task. A writer compiled against this exact head
successfully returned metrics before `stop` and threw `IllegalStateException`
afterward. Please capture/report the metrics after `writer.write(...)` but
before `writer.stop(true)`, and add coverage for a writer that rejects metric
access after close.
##########
core/src/main/java/org/apache/spark/shuffle/api/SingleSpillShuffleMapOutputWriter.java:
##########
@@ -36,4 +37,11 @@ void transferMapSpillFile(
File mapOutputFile,
long[] partitionLengths,
long[] checksums) throws IOException;
+
+ /**
+ * The values of the custom shuffle metrics for this map task.
+ */
+ default CustomShuffleTaskMetric[] currentMetricsValues() {
Review Comment:
[P1] Avoid conflicting defaults on the two writer interfaces
`ShuffleMapOutputWriter` introduces the same `currentMetricsValues()`
default. A previously valid shuffle plugin can implement both interfaces with
one writer. Recompiling that implementation now fails with `inherits unrelated
defaults for currentMetricsValues()`, while running an already-compiled
implementation against these exact PR interfaces throws
`IncompatibleClassChangeError: Conflicting default methods` when Spark invokes
the hook. I reproduced both failures using the actual head interfaces. Please
place the default on one shared parent interface, or otherwise remove the
conflicting independent defaults, and add source/binary coverage for a writer
implementing both.
##########
core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java:
##########
@@ -348,6 +355,7 @@ private long[] mergeSpillsUsingStandardWriter(SpillInfo[]
spills) throws IOExcep
mergeSpillsWithFileStream(spills, mapWriter, compressionCodec);
}
partitionLengths =
mapWriter.commitAllPartitions(sorter.getChecksums()).getPartitionLengths();
+ customMetricsValues = mapWriter.currentMetricsValues();
Review Comment:
[P2] Do not abort committed output when optional metric collection fails
`commitAllPartitions()` has already succeeded immediately above, and its SPI
explicitly permits closing resources during commit. If `currentMetricsValues()`
then reads released state and throws, the surrounding handler invokes
`mapWriter.abort(e)`, whose contract invalidates the written output. A focused
reproduction against the actual SPI produced `aborted=true` and
`committed_output_retained=false`; `BypassMergeSortShuffleWriter` has the same
post-commit/abort path. This also affects RDD shuffles even though their custom
metrics are discarded. Please keep best-effort metric collection outside
destructive commit-failure handling and add a throwing-getter regression test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -38,14 +38,34 @@ trait LimitExec extends UnaryExecNode {
def limit: Int
}
+/**
+ * Shuffle read/write metric plumbing for operators that shuffle via
+ * [[ShuffleExchangeExec.prepareShuffleDependency]]. Publishes the built-in
read/write metrics plus
+ * collision-filtered plugin metrics (`customWriteMetrics`) through `metrics`.
Operators with
+ * additional Spark-owned metrics reserve those names against plugin
collisions via
+ * [[extraReservedMetrics]].
+ */
+trait ShuffleMetricsSupport { self: SparkPlan =>
+ protected lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ protected lazy val readMetrics =
Review Comment:
[P2] Preserve package-level access to existing shuffle read metrics
`ShuffleExchangeExec` previously declared `private[sql] lazy val
readMetrics`; after mixing in this trait it instead inherits this `protected`
accessor. Existing callers elsewhere under `org.apache.spark.sql.*` that access
`exchange.readMetrics` and do not subclass the exchange can no longer compile.
I reproduced successful Scala compilation with the base definition and `Access
to protected lazy value readMetrics not permitted` with this head. The JVM
accessor remains public, so this is specifically a source-compatibility
regression distinct from the already-restored helper overloads. Please preserve
`private[sql]` visibility or expose an equivalent package-visible accessor on
`ShuffleExchangeExec`.
--
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]
