You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "jerrypeng (via GitHub)" <gi...@apache.org> on 2026/08/01 00:50:20 UTC
[PR] [WIP] Stack/pipelined shuffle pr8 rtm [spark]
jerrypeng opened a new pull request, #57692:
URL: https://github.com/apache/spark/pull/57692
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3706745105
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:
##########
@@ -663,7 +665,96 @@ class IncrementalExecution(
}
}
- override def preparations: Seq[Rule[SparkPlan]] = state +: super.preparations
+ /**
+ * For a Real-Time Mode batch, mark the shuffle exchanges as pipelined so the DAGScheduler
+ * co-schedules a stateful query's producer (source scan) and consumer (stateful operator) stages
+ * as one pipelined group -- records stream through a transient shuffle instead of the consumer
+ * waiting for the producer to fully materialize. The exchange carries the decision as a field
+ * (see ShuffleExchangeExec.pipelined); the PipelinedShuffleDependency it then builds is the whole
+ * opt-in -- routing to the streaming shuffle manager and pipelined-group scheduling both follow
+ * from that dependency type.
+ *
+ * Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf (there is no
+ * RTM-specific plan flag). Inert for a non-RTM batch, so the ordinary microbatch path is
+ * unchanged.
+ *
+ * Marks EVERY shuffle exchange, so a plan with several pipelined shuffles in a chain (e.g. two
+ * repartitions, or a repartition feeding a keyed stateful operator) is handled: each becomes a
+ * PipelinedShuffleDependency and the whole all-pipelined job is co-scheduled as one pipelined
+ * group (the DAGScheduler treats an all-pipelined job's stage graph as a single group). No
+ * shuffle-count restriction is needed here.
+ *
+ * transformUp does not descend into a ReusedExchangeExec (a leaf whose wrapped exchange is a
+ * field, not a tree child), so a REUSED shuffle exchange would keep pipelined=false while its
+ * standalone twin flips to true. That divergence is not reachable: a reused shuffle requires
+ * referencing the same streaming source more than once (self-join / self-union / CTE read twice),
+ * which Real-Time Mode rejects when the query starts (MicroBatchExecution,
+ * IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED) before this rule runs. The only ReusedExchangeExec
+ * that reaches an RTM plan wraps a BROADCAST exchange (multiple broadcast joins on the same
+ * static table, SC-209926), which this rule does not match.
+ *
+ * A pipelined shuffle read by more than one consumer (fan-out) is rejected by the DAGScheduler
+ * (checkPipelinedGroupsSupportedInRDDGraph). Note that check runs in handleJobSubmitted, so it
+ * rejects the batch's job rather than the query: such a query fails the same way on every batch
+ * instead of failing once when it is planned. Marking is not what makes the shape unsupported, so
+ * a plan-time guard here would only improve the failure mode, not the outcome.
+ */
+ object MarkPipelinedShuffleForRealTimeMode extends Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = {
+ val isRealTimeMode = plan.exists(_.isInstanceOf[RealTimeStreamScanExec])
+ if (!isRealTimeMode) {
+ plan
+ } else {
+ markStreamingPath(plan)._1
+ }
+ }
+
+ /**
+ * Marks the shuffles that are on the streaming path -- those whose subtree reaches a
+ * [[RealTimeStreamScanExec]] -- and returns the rewritten plan along with whether this
+ * subtree reaches one.
+ *
+ * A plan can hold a static subtree alongside the streaming one: the static side of a
+ * broadcast stream-static join is planned in the same physical plan and may contain its own
+ * shuffle. That shuffle materializes normally and is not part of the pipelined group -- and
+ * cannot be, since a static side runs to completion rather than streaming. Marking it
+ * pipelined would pull it into the group and demand slots for stages that must instead
+ * finish, which fails admission (CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT). This mirrors the
+ * streaming-path detection the operator allowlist uses (RealTimeModeAllowlist), which only
+ * inspects nodes whose subtree reaches the real-time scan; marking a wider set than the
+ * allowlist checks would flip shuffles it never validated.
+ */
+ private def markStreamingPath(plan: SparkPlan): (SparkPlan, Boolean) = plan match {
+ case rts: RealTimeStreamScanExec => (rts, true)
+ case p if p.children.isEmpty => (p, false)
+ case p =>
+ val results = p.children.map(markStreamingPath)
+ val onStreamingPath = results.exists(_._2)
+ val newPlan = p.withNewChildren(results.map(_._1))
+ newPlan match {
+ case s: ShuffleExchangeExec
+ if onStreamingPath && !s.pipelined && canBePipelined(s.outputPartitioning) =>
+ (s.copy(pipelined = true), true)
Review Comment:
will fix
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3717221997
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala:
##########
@@ -393,4 +466,330 @@ 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 a
+ // stage from any other streaming query would otherwise satisfy the co-scheduling assertion
+ // below even if this query's producer and consumer actually ran one after the other. The id is
+ // captured from the query once it is running, and every job is matched against it.
+ val queryId = new AtomicReference[String](null)
+ val listener = new SparkListener {
+ override def onJobStart(e: SparkListenerJobStart): Unit = {
+ // StreamExecution tags every streaming job with its query id.
+ val id = queryId.get()
+ if (id != null && e.properties.getProperty(StreamExecution.QUERY_ID_KEY) == id) {
+ e.stageIds.foreach(queryStageIds.add(_))
+ }
+ }
+ override def onStageSubmitted(e: SparkListenerStageSubmitted): Unit = {
+ if (queryStageIds.contains(e.stageInfo.stageId)) {
+ runningStages.add(e.stageInfo.stageId)
+ maxConcurrentStages.accumulateAndGet(runningStages.size(), Math.max)
+ }
+ }
+ override def onStageCompleted(e: SparkListenerStageCompleted): Unit = {
+ runningStages.remove(e.stageInfo.stageId)
+ }
+ }
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ val inputData = LowLatencyMemoryStream[(String, Int)]
+ // scan --shuffle(repartition by key)--> streaming dropDuplicates --> sink.
+ val result = inputData.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+ testStream(result, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ AddData(inputData, ("a", 1), ("b", 1), ("c", 1), ("a", 2), ("b", 2), ("c", 2)),
+ StartStream(),
+ // Record the id before any batch is awaited, so the listener attributes this query's jobs
+ // from the first one.
+ Execute(q => queryId.set(q.id.toString)),
Review Comment:
will fix
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3702359158
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:
##########
@@ -663,7 +665,96 @@ class IncrementalExecution(
}
}
- override def preparations: Seq[Rule[SparkPlan]] = state +: super.preparations
+ /**
+ * For a Real-Time Mode batch, mark the shuffle exchanges as pipelined so the DAGScheduler
+ * co-schedules a stateful query's producer (source scan) and consumer (stateful operator) stages
+ * as one pipelined group -- records stream through a transient shuffle instead of the consumer
+ * waiting for the producer to fully materialize. The exchange carries the decision as a field
+ * (see ShuffleExchangeExec.pipelined); the PipelinedShuffleDependency it then builds is the whole
+ * opt-in -- routing to the streaming shuffle manager and pipelined-group scheduling both follow
+ * from that dependency type.
+ *
+ * Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf (there is no
+ * RTM-specific plan flag). Inert for a non-RTM batch, so the ordinary microbatch path is
+ * unchanged.
+ *
+ * Marks EVERY shuffle exchange, so a plan with several pipelined shuffles in a chain (e.g. two
+ * repartitions, or a repartition feeding a keyed stateful operator) is handled: each becomes a
+ * PipelinedShuffleDependency and the whole all-pipelined job is co-scheduled as one pipelined
+ * group (the DAGScheduler treats an all-pipelined job's stage graph as a single group). No
+ * shuffle-count restriction is needed here.
+ *
+ * transformUp does not descend into a ReusedExchangeExec (a leaf whose wrapped exchange is a
+ * field, not a tree child), so a REUSED shuffle exchange would keep pipelined=false while its
+ * standalone twin flips to true. That divergence is not reachable: a reused shuffle requires
+ * referencing the same streaming source more than once (self-join / self-union / CTE read twice),
+ * which Real-Time Mode rejects when the query starts (MicroBatchExecution,
+ * IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED) before this rule runs. The only ReusedExchangeExec
+ * that reaches an RTM plan wraps a BROADCAST exchange (multiple broadcast joins on the same
+ * static table, SC-209926), which this rule does not match.
+ *
+ * A pipelined shuffle read by more than one consumer (fan-out) is rejected by the DAGScheduler
+ * (checkPipelinedGroupsSupportedInRDDGraph). Note that check runs in handleJobSubmitted, so it
+ * rejects the batch's job rather than the query: such a query fails the same way on every batch
+ * instead of failing once when it is planned. Marking is not what makes the shape unsupported, so
+ * a plan-time guard here would only improve the failure mode, not the outcome.
+ */
+ object MarkPipelinedShuffleForRealTimeMode extends Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = {
+ val isRealTimeMode = plan.exists(_.isInstanceOf[RealTimeStreamScanExec])
+ if (!isRealTimeMode) {
+ plan
+ } else {
+ markStreamingPath(plan)._1
+ }
+ }
+
+ /**
+ * Marks the shuffles that are on the streaming path -- those whose subtree reaches a
+ * [[RealTimeStreamScanExec]] -- and returns the rewritten plan along with whether this
+ * subtree reaches one.
+ *
+ * A plan can hold a static subtree alongside the streaming one: the static side of a
+ * broadcast stream-static join is planned in the same physical plan and may contain its own
+ * shuffle. That shuffle materializes normally and is not part of the pipelined group -- and
+ * cannot be, since a static side runs to completion rather than streaming. Marking it
+ * pipelined would pull it into the group and demand slots for stages that must instead
+ * finish, which fails admission (CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT). This mirrors the
+ * streaming-path detection the operator allowlist uses (RealTimeModeAllowlist), which only
+ * inspects nodes whose subtree reaches the real-time scan; marking a wider set than the
+ * allowlist checks would flip shuffles it never validated.
+ */
+ private def markStreamingPath(plan: SparkPlan): (SparkPlan, Boolean) = plan match {
+ case rts: RealTimeStreamScanExec => (rts, true)
+ case p if p.children.isEmpty => (p, false)
+ case p =>
+ val results = p.children.map(markStreamingPath)
+ val onStreamingPath = results.exists(_._2)
+ val newPlan = p.withNewChildren(results.map(_._1))
+ newPlan match {
+ case s: ShuffleExchangeExec
+ if onStreamingPath && !s.pipelined && canBePipelined(s.outputPartitioning) =>
+ (s.copy(pipelined = true), true)
Review Comment:
Preserve the original exchange's tree-node tags when setting `pipelined=true`. This case-class copy bypasses `TreeNode.copyTagsFrom`, so the marked exchange loses metadata such as its logical link; use a tag-preserving tree operation or explicitly copy the tags onto the replacement.
##########
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:
Filter these events by this query's actual ID or run ID. Checking only that `QUERY_ID_KEY` is present admits stages from any concurrent streaming query, so `maxConcurrentStages` can reach two even when this query's stages run sequentially. Please apply the same fix to the listener near the two-shuffle test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:
##########
@@ -663,7 +665,96 @@ class IncrementalExecution(
}
}
- override def preparations: Seq[Rule[SparkPlan]] = state +: super.preparations
+ /**
+ * For a Real-Time Mode batch, mark the shuffle exchanges as pipelined so the DAGScheduler
+ * co-schedules a stateful query's producer (source scan) and consumer (stateful operator) stages
+ * as one pipelined group -- records stream through a transient shuffle instead of the consumer
+ * waiting for the producer to fully materialize. The exchange carries the decision as a field
+ * (see ShuffleExchangeExec.pipelined); the PipelinedShuffleDependency it then builds is the whole
+ * opt-in -- routing to the streaming shuffle manager and pipelined-group scheduling both follow
+ * from that dependency type.
+ *
+ * Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf (there is no
+ * RTM-specific plan flag). Inert for a non-RTM batch, so the ordinary microbatch path is
+ * unchanged.
+ *
+ * Marks EVERY shuffle exchange, so a plan with several pipelined shuffles in a chain (e.g. two
Review Comment:
Please narrow this to every eligible shuffle on the streaming path. The rule intentionally skips static-side and range-partitioned exchanges, so the universal wording contradicts the implementation.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on PR #57692:
URL: https://github.com/apache/spark/pull/57692#issuecomment-5188319006
@cloud-fan I believe that was just a unrelated flaky test failure. On re-run the test passed. CI is green now.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on PR #57692:
URL: https://github.com/apache/spark/pull/57692#issuecomment-5174585857
@cloud-fan thank you for the review! I have addressed your comments. PTAL. I have also rebased because there were some unrelated tests failing.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57692:
URL: https://github.com/apache/spark/pull/57692#issuecomment-5188342707
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/8debe32663ad988684a2ecfb11351a0c9c6641e0
- merged into branch-4.x https://github.com/apache/spark/commit/aa68679f1bb4abb16c52b59779b40dd1ab37f721
- merged into branch-4.3 https://github.com/apache/spark/commit/3afa984ace964569c7e921e81cede5144b84f50f
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3717345462
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala:
##########
@@ -393,4 +466,330 @@ 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 a
+ // stage from any other streaming query would otherwise satisfy the co-scheduling assertion
+ // below even if this query's producer and consumer actually ran one after the other. The id is
+ // captured from the query once it is running, and every job is matched against it.
+ val queryId = new AtomicReference[String](null)
+ val listener = new SparkListener {
+ override def onJobStart(e: SparkListenerJobStart): Unit = {
+ // StreamExecution tags every streaming job with its query id.
+ val id = queryId.get()
+ if (id != null && e.properties.getProperty(StreamExecution.QUERY_ID_KEY) == id) {
+ e.stageIds.foreach(queryStageIds.add(_))
+ }
+ }
+ override def onStageSubmitted(e: SparkListenerStageSubmitted): Unit = {
+ if (queryStageIds.contains(e.stageInfo.stageId)) {
+ runningStages.add(e.stageInfo.stageId)
+ maxConcurrentStages.accumulateAndGet(runningStages.size(), Math.max)
+ }
+ }
+ override def onStageCompleted(e: SparkListenerStageCompleted): Unit = {
+ runningStages.remove(e.stageInfo.stageId)
+ }
+ }
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ val inputData = LowLatencyMemoryStream[(String, Int)]
+ // scan --shuffle(repartition by key)--> streaming dropDuplicates --> sink.
+ val result = inputData.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+ testStream(result, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ AddData(inputData, ("a", 1), ("b", 1), ("c", 1), ("a", 2), ("b", 2), ("c", 2)),
+ StartStream(),
+ // Record the id before any batch is awaited, so the listener attributes this query's jobs
+ // from the first one.
+ Execute(q => queryId.set(q.id.toString)),
Review Comment:
I took the second half of your suggestion (a dedicated observed batch) but not the first, because the id can't be set before StartStream.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3709557379
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:
##########
@@ -663,7 +664,91 @@ class IncrementalExecution(
}
}
- override def preparations: Seq[Rule[SparkPlan]] = state +: super.preparations
+ /**
+ * For a Real-Time Mode batch, mark the shuffle exchanges as pipelined so the DAGScheduler
+ * co-schedules a stateful query's producer (source scan) and consumer (stateful operator) stages
+ * as one pipelined group -- records stream through a transient shuffle instead of the consumer
+ * waiting for the producer to fully materialize. The exchange carries the decision as a field
+ * (see ShuffleExchangeExec.pipelined); the PipelinedShuffleDependency it then builds is the whole
+ * opt-in -- routing to the streaming shuffle manager and pipelined-group scheduling both follow
+ * from that dependency type.
+ *
+ * Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf (there is no
+ * RTM-specific plan flag). Inert for a non-RTM batch, so the ordinary microbatch path is
+ * unchanged.
+ *
+ * Marks every eligible shuffle exchange on the streaming path, so a plan with several pipelined
+ * shuffles in a chain (e.g. two repartitions, or a repartition feeding a keyed stateful operator)
+ * is handled: each becomes a PipelinedShuffleDependency and the whole all-pipelined job is
+ * co-scheduled as one pipelined group (the DAGScheduler treats an all-pipelined job's stage graph
+ * as a single group). There is no shuffle-count restriction. An exchange whose subtree does not
+ * reach the real-time scan is skipped -- the static side of a broadcast stream-static join must
+ * materialize, because it runs to completion rather than streaming. A partitioning the pipelined
+ * path cannot serve, such as range partitioning, is rejected up front by RealTimeModeAllowlist
+ * rather than being handled here.
+ *
+ * The walk does not descend into a ReusedExchangeExec (a leaf whose wrapped exchange is a field,
+ * not a tree child), so a REUSED shuffle exchange would keep pipelined=false while its standalone
+ * twin flips to true. That divergence is not reachable: a reused shuffle requires
+ * referencing the same streaming source more than once (self-join / self-union / CTE read twice),
+ * which Real-Time Mode rejects when the query starts (MicroBatchExecution,
+ * IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED) before this rule runs. The only ReusedExchangeExec
+ * that reaches an RTM plan wraps a BROADCAST exchange (multiple broadcast joins on the same
+ * static table, SC-209926), which this rule does not match.
+ *
+ * A pipelined shuffle read by more than one consumer (fan-out) is rejected by the DAGScheduler
+ * (checkPipelinedGroupsSupportedInRDDGraph). Note that check runs in handleJobSubmitted, so it
+ * rejects the batch's job rather than the query: such a query fails the same way on every batch
+ * instead of failing once when it is planned. Marking is not what makes the shape unsupported, so
Review Comment:
This reverses the cause of the rejection: the scheduler rejects fan-out specifically for a `PipelinedShuffleDependency`, while a regular materialized shuffle supports it. Please describe fan-out as a limitation activated by the pipelined path; a plan-time guard would improve only when the error is reported.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala:
##########
@@ -393,4 +466,330 @@ 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 a
+ // stage from any other streaming query would otherwise satisfy the co-scheduling assertion
+ // below even if this query's producer and consumer actually ran one after the other. The id is
+ // captured from the query once it is running, and every job is matched against it.
+ val queryId = new AtomicReference[String](null)
+ val listener = new SparkListener {
+ override def onJobStart(e: SparkListenerJobStart): Unit = {
+ // StreamExecution tags every streaming job with its query id.
+ val id = queryId.get()
+ if (id != null && e.properties.getProperty(StreamExecution.QUERY_ID_KEY) == id) {
+ e.stageIds.foreach(queryStageIds.add(_))
+ }
+ }
+ override def onStageSubmitted(e: SparkListenerStageSubmitted): Unit = {
+ if (queryStageIds.contains(e.stageInfo.stageId)) {
+ runningStages.add(e.stageInfo.stageId)
+ maxConcurrentStages.accumulateAndGet(runningStages.size(), Math.max)
+ }
+ }
+ override def onStageCompleted(e: SparkListenerStageCompleted): Unit = {
+ runningStages.remove(e.stageInfo.stageId)
+ }
+ }
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ val inputData = LowLatencyMemoryStream[(String, Int)]
+ // scan --shuffle(repartition by key)--> streaming dropDuplicates --> sink.
+ val result = inputData.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+ testStream(result, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ AddData(inputData, ("a", 1), ("b", 1), ("c", 1), ("a", 2), ("b", 2), ("c", 2)),
+ StartStream(),
+ // Record the id before any batch is awaited, so the listener attributes this query's jobs
+ // from the first one.
+ Execute(q => queryId.set(q.id.toString)),
Review Comment:
`StartStream` can launch the first job before this action sets `queryId`, so that job never enters `queryStageIds`. This is especially flaky in the two-shuffle test, which asserts before running another batch. Set the ID, then trigger a dedicated observed batch in both listener-based tests.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3706748824
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:
##########
@@ -663,7 +665,96 @@ class IncrementalExecution(
}
}
- override def preparations: Seq[Rule[SparkPlan]] = state +: super.preparations
+ /**
+ * For a Real-Time Mode batch, mark the shuffle exchanges as pipelined so the DAGScheduler
+ * co-schedules a stateful query's producer (source scan) and consumer (stateful operator) stages
+ * as one pipelined group -- records stream through a transient shuffle instead of the consumer
+ * waiting for the producer to fully materialize. The exchange carries the decision as a field
+ * (see ShuffleExchangeExec.pipelined); the PipelinedShuffleDependency it then builds is the whole
+ * opt-in -- routing to the streaming shuffle manager and pipelined-group scheduling both follow
+ * from that dependency type.
+ *
+ * Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf (there is no
+ * RTM-specific plan flag). Inert for a non-RTM batch, so the ordinary microbatch path is
+ * unchanged.
+ *
+ * Marks EVERY shuffle exchange, so a plan with several pipelined shuffles in a chain (e.g. two
Review Comment:
will reword
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on PR #57692:
URL: https://github.com/apache/spark/pull/57692#issuecomment-5186341604
@cloud-fan thank you for your review! I have addressed your comments. PTAL.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57692:
URL: https://github.com/apache/spark/pull/57692#issuecomment-5187516880
@jerrypeng can you look into `ClientStreamingQuerySuite.listener events` test failure? maybe related to the scheduler change
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #57692: [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries
URL: https://github.com/apache/spark/pull/57692
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58508][CORE][SS] Activate pipelined shuffles for Real-Time Mode streaming queries to support Stateful Queries [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #57692:
URL: https://github.com/apache/spark/pull/57692#discussion_r3716980138
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:
##########
@@ -663,7 +664,91 @@ class IncrementalExecution(
}
}
- override def preparations: Seq[Rule[SparkPlan]] = state +: super.preparations
+ /**
+ * For a Real-Time Mode batch, mark the shuffle exchanges as pipelined so the DAGScheduler
+ * co-schedules a stateful query's producer (source scan) and consumer (stateful operator) stages
+ * as one pipelined group -- records stream through a transient shuffle instead of the consumer
+ * waiting for the producer to fully materialize. The exchange carries the decision as a field
+ * (see ShuffleExchangeExec.pipelined); the PipelinedShuffleDependency it then builds is the whole
+ * opt-in -- routing to the streaming shuffle manager and pipelined-group scheduling both follow
+ * from that dependency type.
+ *
+ * Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf (there is no
+ * RTM-specific plan flag). Inert for a non-RTM batch, so the ordinary microbatch path is
+ * unchanged.
+ *
+ * Marks every eligible shuffle exchange on the streaming path, so a plan with several pipelined
+ * shuffles in a chain (e.g. two repartitions, or a repartition feeding a keyed stateful operator)
+ * is handled: each becomes a PipelinedShuffleDependency and the whole all-pipelined job is
+ * co-scheduled as one pipelined group (the DAGScheduler treats an all-pipelined job's stage graph
+ * as a single group). There is no shuffle-count restriction. An exchange whose subtree does not
+ * reach the real-time scan is skipped -- the static side of a broadcast stream-static join must
+ * materialize, because it runs to completion rather than streaming. A partitioning the pipelined
+ * path cannot serve, such as range partitioning, is rejected up front by RealTimeModeAllowlist
+ * rather than being handled here.
+ *
+ * The walk does not descend into a ReusedExchangeExec (a leaf whose wrapped exchange is a field,
+ * not a tree child), so a REUSED shuffle exchange would keep pipelined=false while its standalone
+ * twin flips to true. That divergence is not reachable: a reused shuffle requires
+ * referencing the same streaming source more than once (self-join / self-union / CTE read twice),
+ * which Real-Time Mode rejects when the query starts (MicroBatchExecution,
+ * IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED) before this rule runs. The only ReusedExchangeExec
+ * that reaches an RTM plan wraps a BROADCAST exchange (multiple broadcast joins on the same
+ * static table, SC-209926), which this rule does not match.
+ *
+ * A pipelined shuffle read by more than one consumer (fan-out) is rejected by the DAGScheduler
+ * (checkPipelinedGroupsSupportedInRDDGraph). Note that check runs in handleJobSubmitted, so it
+ * rejects the batch's job rather than the query: such a query fails the same way on every batch
+ * instead of failing once when it is planned. Marking is not what makes the shape unsupported, so
Review Comment:
fair will reword
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org