jerrypeng commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3709431559
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala:
##########
@@ -393,4 +466,322 @@ class StreamRealTimeModeWithManualClockSuite extends
StreamRealTimeModeManualClo
StopStream
)
}
+
+ //
========================================================================================
+ // Pipelined (streaming) shuffle: a stateful/repartition Real-Time Mode
query whose shuffle is a
+ // PipelinedShuffleDependency, so the producer (source scan) and consumer
stages are co-scheduled
+ // and stream records through a transient shuffle instead of the consumer
waiting for the producer
+ // to fully materialize.
+ //
========================================================================================
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ StreamRealTimeModeSuite.failTasks = false
+ }
+
+ /** Assert every shuffle exchange in the query's last executed plan is
pipelined. */
+ private def assertAllExchangesPipelined(q: StreamExecution): Unit = {
+ val exchanges = q.lastExecution.executedPlan.collect { case s:
ShuffleExchangeExec => s }
+ assert(exchanges.nonEmpty, "expected at least one shuffle exchange in the
plan")
+ assert(exchanges.forall(_.pipelined),
+ "expected all Real-Time Mode shuffle exchanges to be pipelined, got: " +
+ exchanges.map(e => s"pipelined=${e.pipelined}").mkString(", "))
+ }
+
+ test("pipelined shuffle: stateful dedup runs in Real-Time Mode and
co-schedules its stages") {
+ // Track, from the driver, whether the producer (source scan) and consumer
(dedup) stages of the
+ // pipelined group were ever RUNNING simultaneously. A sequential
producer-then-consumer
+ // schedule never exceeds one running stage at a time; >= 2 proves genuine
co-scheduling.
+ val runningStages = ConcurrentHashMap.newKeySet[Int]()
+ val maxConcurrentStages = new AtomicInteger(0)
+ val queryStageIds = ConcurrentHashMap.newKeySet[Int]()
+ // Count only stages belonging to the query under test. The suite shares
one SparkContext, so
Review Comment:
will add.
--
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]
