cloud-fan commented on code in PR #59524:
URL: https://github.com/apache/spark/pull/59524#discussion_r4290939095
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -9534,6 +9738,27 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
}
}
+ private def assertRepeatedPipelinedConsumptionRejected(
+ result: RDD[_], shuffleIds: Seq[Int]): Unit = {
+ for (_ <- 0 until 2) {
+ clearInvocations(scheduler, mapOutputTracker)
+ val failure = submitAndCaptureFailure(result,
result.partitions.indices.toArray)
+ assertPipelinedUnsupported(failure, "more than one consumer")
Review Comment:
**Nit (P3):** This only checks that the reason contains `more than one
consumer`. The new repeated-consumption reason (`DAGScheduler.scala:1569`) and
the existing fan-out reason (`DAGScheduler.scala:1464`) both contain that
phrase, so either guard satisfies this assertion.
The existing fan-out tests (lines 7725, 7754 and 9430) assert the same
substring, and they lose their signal because of it. Each one narrow-joins two
or three consumers of one pipelined shuffle at the same partition index, and
none of those consumers declares a `ShuffleReducePartitionMapping`. If the
fan-out throw were removed, `checkPipelinedShuffleRepeatedReads` would record
`(shuffleId, None, 0)` twice and reject those graphs with the
repeated-consumption reason, and all three tests would stay green. The shape
that only the fan-out guard rejects, consumers reading disjoint partitions such
as `PartitionPruningRDD.create(a, _ == 0).union(PartitionPruningRDD.create(b, _
== 1))`, would then be admitted with nothing failing. The new check's
reader-index key relies on that guard (the comment at
`DAGScheduler.scala:1537-1538`).
The new scheduler tests also never check the shuffle and partition that the
message reports. Only `PipelinedChannelShuffleSuite` pins the exact reason, and
only for a shape where the reduce and reader indexes are both 0.
Suggested fix:
- Have this helper require `(repeated consumption)` and the expected shuffle
ID, plus the reduce partition where it is deterministic (for example 0 in the
aliasing-reducer case).
- Have the fan-out tests require `(fan-out / branching)`.
- Add one fan-out case whose consumers read disjoint partitions, so the
fan-out guard has a failure signal the repeated-read guard cannot satisfy.
**Verification:**
- **Regression:** Disabling the fan-out rejection makes the fan-out tests
fail, including the case where two consumer RDDs read disjoint partitions of
one pipelined shuffle, which the repeated-read check alone admits.
- **Regression:** Changing the shuffle or partition reported by the
repeated-consumption rejection, for example reporting the reader partition
instead of the declared reduce partition for aliased readers or a downstream
shuffle instead of the upstream one, makes the corresponding scheduler
rejection tests fail.
- **Behavior:** All pipelined scheduler and channel-shuffle tests, including
the supported-shape and slot-admission tests, still pass with the tightened
assertions and the added disjoint-partition fan-out case.
--
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]
