You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "LuciferYang (via GitHub)" <gi...@apache.org> on 2026/06/23 11:50:46 UTC
[PR] [SPARK-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
LuciferYang opened a new pull request, #56700:
URL: https://github.com/apache/spark/pull/56700
### What changes were proposed in this pull request?
`DataflowGraphTransformer.transformDownNodes` resolves flows on a bounded thread pool and drives them from a `while` loop that, each pass, partitioned the in-flight futures with the non-blocking `future.isDone`, reaped the completed ones, and scheduled a new flow if a slot was free. When all slots were in flight (or the queue was drained and only the last futures remained) and none had completed, the pass reaped nothing and scheduled nothing, then looped again immediately - busy-spinning on `isDone` and pinning a core for the duration of resolution.
This drives the loop with an `ExecutorCompletionService` instead: completed tasks are drained with the non-blocking `poll()`, and when nothing can be scheduled but tasks are still running, the loop blocks on `take()` until the next one finishes rather than spinning. Behavior is otherwise unchanged - the same flows are scheduled in the same order, exceptions are still propagated via `Future.get()`, and an `outstanding` counter replaces the `ArrayBuffer[Future]` for slot bookkeeping.
### Why are the changes needed?
Resolving a graph with more flows than the parallelism (10) kept one CPU core busy at 100% doing no useful work for the whole resolution, which is wasteful and shows up as unexplained driver CPU.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing graph-resolution suites (`ConnectValidPipelineSuite`, `ConnectInvalidPipelineSuite`, `SqlPipelineSuite`, `TriggeredGraphExecutionSuite`, `MaterializeTablesSuite`) still pass; the change only affects how the loop waits, not what it resolves. A dedicated test is not included because asserting the absence of a busy-wait reliably requires CPU-time or timing measurements that are flaky in CI.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
--
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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #56700:
URL: https://github.com/apache/spark/pull/56700#discussion_r3461701004
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraphTransformer.scala:
##########
@@ -134,159 +134,165 @@ class DataflowGraphTransformer(graph: DataflowGraph) extends AutoCloseable {
val failedFlowsQueue = new ConcurrentLinkedQueue[ResolutionFailedFlow]()
val failedDependentFlows = new ConcurrentHashMap[TableIdentifier, Seq[ResolutionFailedFlow]]()
- var futures = ArrayBuffer[Future[Unit]]()
+ val completionService = new ExecutorCompletionService[Unit](executor)
+ var outstanding = 0
val toBeResolvedFlows = new ConcurrentLinkedDeque[Flow]()
toBeResolvedFlows.addAll(flows.asJava)
- while (futures.nonEmpty || toBeResolvedFlows.peekFirst() != null) {
- val (done, notDone) = futures.partition(_.isDone)
- // Explicitly call future.get() to propagate exceptions one by one if any
+ // Waits on a finished resolution task and propagates its exception, if any.
+ def reap(finished: Future[Unit]): Unit = {
try {
- done.foreach(_.get())
+ finished.get()
} catch {
case exn: ExecutionException =>
// Computation threw the exception that is the cause of exn
throw exn.getCause
}
- futures = notDone
- val flowOpt = {
- // We only schedule [[batchSize]] number of flows in parallel.
- if (futures.size < batchSize) {
- Option(toBeResolvedFlows.pollFirst())
- } else {
- None
- }
+ outstanding -= 1
+ }
+
+ while (outstanding > 0 || toBeResolvedFlows.peekFirst() != null) {
+ // Reap every resolution task that has already finished, without blocking.
+ var finished = completionService.poll()
+ while (finished != null) {
+ reap(finished)
+ finished = completionService.poll()
}
- flowOpt.foreach { flow =>
- futures.append(
- executor.submit(
- () =>
+ // We only schedule [[batchSize]] number of flows in parallel.
+ if (outstanding < batchSize && toBeResolvedFlows.peekFirst() != null) {
+ val flow = toBeResolvedFlows.pollFirst()
+ outstanding += 1
+ completionService.submit(
+ () =>
+ try {
try {
- try {
- // Note: Flow don't need their inputs passed, so for now we send empty Seq.
- val result = transformer(flow, Seq.empty)
- require(
- result.forall(_.isInstanceOf[ResolvedFlow]),
- "transformer must return a Seq[Flow]"
- )
+ // Note: Flow don't need their inputs passed, so for now we send empty Seq.
+ val result = transformer(flow, Seq.empty)
+ require(
+ result.forall(_.isInstanceOf[ResolvedFlow]),
+ "transformer must return a Seq[Flow]"
+ )
- val transformedFlows = result.map(_.asInstanceOf[ResolvedFlow])
- resolvedFlowsMap.put(flow.identifier, transformedFlows)
- resolvedFlows.addAll(transformedFlows.asJava)
- } catch {
- case e: TransformNodeRetryableException =>
- val datasetIdentifier = e.datasetIdentifier
- failedDependentFlows.compute(
- datasetIdentifier,
- (_, flows) => {
- // Don't add the input flow back but the failed flow object
- // back which has relevant failure information.
- val failedFlow = e.failedNode
- if (flows == null) {
- Seq(failedFlow)
- } else {
- flows :+ failedFlow
- }
+ val transformedFlows = result.map(_.asInstanceOf[ResolvedFlow])
+ resolvedFlowsMap.put(flow.identifier, transformedFlows)
+ resolvedFlows.addAll(transformedFlows.asJava)
+ } catch {
+ case e: TransformNodeRetryableException =>
+ val datasetIdentifier = e.datasetIdentifier
+ failedDependentFlows.compute(
+ datasetIdentifier,
+ (_, flows) => {
+ // Don't add the input flow back but the failed flow object
+ // back which has relevant failure information.
+ val failedFlow = e.failedNode
+ if (flows == null) {
+ Seq(failedFlow)
+ } else {
+ flows :+ failedFlow
}
- )
- // Between the time the flow started and finished resolving, perhaps the
- // dependent dataset was resolved
- resolvedFlowDestinationsMap.computeIfPresent(
- datasetIdentifier,
- (_, resolved) => {
- if (resolved) {
- // Check if the dataset that the flow is dependent on has been resolved
- // and if so, remove all dependent flows from the failedDependentFlows and
- // add them to the toBeResolvedFlows queue for retry.
- failedDependentFlows.computeIfPresent(
- datasetIdentifier,
- (_, toRetryFlows) => {
- toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
- null
- }
- )
- }
- resolved
+ }
+ )
+ // Between the time the flow started and finished resolving, perhaps the
+ // dependent dataset was resolved
+ resolvedFlowDestinationsMap.computeIfPresent(
+ datasetIdentifier,
+ (_, resolved) => {
+ if (resolved) {
+ // Check if the dataset that the flow is dependent on has been resolved
+ // and if so, remove all dependent flows from the failedDependentFlows and
+ // add them to the toBeResolvedFlows queue for retry.
+ failedDependentFlows.computeIfPresent(
+ datasetIdentifier,
+ (_, toRetryFlows) => {
+ toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
+ null
+ }
+ )
}
+ resolved
+ }
+ )
+ case other: Throwable => throw other
+ }
+ // If all flows to this particular destination are resolved, move to the destination
+ // node transformer
+ if (flowsTo(flow.destinationIdentifier).forall({ flowToDestination =>
+ resolvedFlowsMap.containsKey(flowToDestination.identifier)
+ })) {
+ // If multiple flows completed in parallel, ensure we resolve the destination only
+ // once by electing a leader via computeIfAbsent
+ var isCurrentThreadLeader = false
+ resolvedFlowDestinationsMap.computeIfAbsent(flow.destinationIdentifier, _ => {
+ isCurrentThreadLeader = true
+ // Set initial value as false as flow destination is not resolved yet.
+ false
+ })
+ if (isCurrentThreadLeader) {
+ if (tableMap.contains(flow.destinationIdentifier)) {
+ val transformed =
+ transformer(
+ tableMap(flow.destinationIdentifier),
+ flowsTo(flow.destinationIdentifier)
+ )
+ resolvedTables.addAll(
+ transformed.collect { case t: Table => t }.asJava
)
- case other: Throwable => throw other
- }
- // If all flows to this particular destination are resolved, move to the destination
- // node transformer
- if (flowsTo(flow.destinationIdentifier).forall({ flowToDestination =>
- resolvedFlowsMap.containsKey(flowToDestination.identifier)
- })) {
- // If multiple flows completed in parallel, ensure we resolve the destination only
- // once by electing a leader via computeIfAbsent
- var isCurrentThreadLeader = false
- resolvedFlowDestinationsMap.computeIfAbsent(flow.destinationIdentifier, _ => {
- isCurrentThreadLeader = true
- // Set initial value as false as flow destination is not resolved yet.
- false
- })
- if (isCurrentThreadLeader) {
- if (tableMap.contains(flow.destinationIdentifier)) {
+ resolvedFlows.addAll(
+ transformed.collect { case f: ResolvedFlow => f }.asJava
+ )
+ } else if (viewMap.contains(flow.destinationIdentifier)) {
+ resolvedViews.addAll {
val transformed =
transformer(
- tableMap(flow.destinationIdentifier),
+ viewMap(flow.destinationIdentifier),
flowsTo(flow.destinationIdentifier)
)
- resolvedTables.addAll(
- transformed.collect { case t: Table => t }.asJava
- )
- resolvedFlows.addAll(
- transformed.collect { case f: ResolvedFlow => f }.asJava
- )
- } else if (viewMap.contains(flow.destinationIdentifier)) {
- resolvedViews.addAll {
- val transformed =
- transformer(
- viewMap(flow.destinationIdentifier),
- flowsTo(flow.destinationIdentifier)
- )
- transformed.map(_.asInstanceOf[View]).asJava
- }
- } else if (sinkMap.contains(flow.destinationIdentifier)) {
- resolvedSinks.addAll {
- val transformed =
- transformer(
- sinkMap(flow.destinationIdentifier), flowsTo(flow.destinationIdentifier)
- )
- require(
- transformed.forall(_.isInstanceOf[Sink]),
- "transformer must return a Seq[Sink]"
+ transformed.map(_.asInstanceOf[View]).asJava
+ }
+ } else if (sinkMap.contains(flow.destinationIdentifier)) {
+ resolvedSinks.addAll {
+ val transformed =
+ transformer(
+ sinkMap(flow.destinationIdentifier), flowsTo(flow.destinationIdentifier)
)
- transformed.map(_.asInstanceOf[Sink]).asJava
- }
- } else {
- throw new IllegalArgumentException(
- s"Unsupported destination ${flow.destinationIdentifier.unquotedString}" +
- s" in flow: ${flow.displayName} at transformDownNodes"
+ require(
+ transformed.forall(_.isInstanceOf[Sink]),
+ "transformer must return a Seq[Sink]"
)
+ transformed.map(_.asInstanceOf[Sink]).asJava
}
- // Set flow destination as resolved now.
- resolvedFlowDestinationsMap.computeIfPresent(
- flow.destinationIdentifier,
- (_, _) => {
- // If there are any other node failures dependent on this destination, retry
- // them
- failedDependentFlows.computeIfPresent(
- flow.destinationIdentifier,
- (_, toRetryFlows) => {
- toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
- null
- }
- )
- true
- }
+ } else {
+ throw new IllegalArgumentException(
+ s"Unsupported destination ${flow.destinationIdentifier.unquotedString}" +
+ s" in flow: ${flow.displayName} at transformDownNodes"
)
}
+ // Set flow destination as resolved now.
+ resolvedFlowDestinationsMap.computeIfPresent(
+ flow.destinationIdentifier,
+ (_, _) => {
+ // If there are any other node failures dependent on this destination, retry
+ // them
+ failedDependentFlows.computeIfPresent(
+ flow.destinationIdentifier,
+ (_, toRetryFlows) => {
+ toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
+ null
+ }
+ )
+ true
+ }
+ )
}
- } catch {
- case ex: TransformNodeFailedException => failedFlowsQueue.add(ex.failedNode)
}
- )
+ } catch {
+ case ex: TransformNodeFailedException => failedFlowsQueue.add(ex.failedNode)
+ }
)
+ } else if (outstanding > 0) {
+ // Nothing finished and nothing could be scheduled, but tasks are still running:
+ // block until the next one finishes instead of busy-spinning on Future.isDone.
+ reap(completionService.take())
Review Comment:
Nit: the `else if (outstanding > 0)` guard is effectively always true when reached. Control only reaches it when the if is false, i.e. `outstanding >= batchSize || queue.isEmpty`. Since `batchSize >= 1`, the first disjunct implies outstanding > 0; and if the queue is empty the loop invariant guarantees outstanding > 0. So it could just be a plain `else`?
--
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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #56700:
URL: https://github.com/apache/spark/pull/56700#issuecomment-5093345722
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/0747e287579272958600e3db3ccf03044378576f
- merged into branch-4.x https://github.com/apache/spark/commit/3401f3d91124e689922768b7b08d186520196006
- merged into branch-4.2 https://github.com/apache/spark/commit/b7c92259a0ef366f5d88dc7e6a41b18445113c23
*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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #56700:
URL: https://github.com/apache/spark/pull/56700#issuecomment-5052765205
@anew @jose-torres do you want to take a look as well?
--
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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #56700:
URL: https://github.com/apache/spark/pull/56700#issuecomment-5093318890
> +1, LGTM. BTW, shall we remove the following sentence, `A dedicated test is not included ...`, because this PR added two new test cases, @LuciferYang ?
>
> > ### How was this patch tested?
> > A dedicated test is not included because asserting the absence of a busy-wait reliably requires CPU-time or timing measurements that are flaky in CI.
Thank you @dongjoon-hyun . I have update the `How was this patch tested?` section to describe the new 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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #56700:
URL: https://github.com/apache/spark/pull/56700#discussion_r3464790941
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraphTransformer.scala:
##########
@@ -134,159 +134,165 @@ class DataflowGraphTransformer(graph: DataflowGraph) extends AutoCloseable {
val failedFlowsQueue = new ConcurrentLinkedQueue[ResolutionFailedFlow]()
val failedDependentFlows = new ConcurrentHashMap[TableIdentifier, Seq[ResolutionFailedFlow]]()
- var futures = ArrayBuffer[Future[Unit]]()
+ val completionService = new ExecutorCompletionService[Unit](executor)
+ var outstanding = 0
val toBeResolvedFlows = new ConcurrentLinkedDeque[Flow]()
toBeResolvedFlows.addAll(flows.asJava)
- while (futures.nonEmpty || toBeResolvedFlows.peekFirst() != null) {
- val (done, notDone) = futures.partition(_.isDone)
- // Explicitly call future.get() to propagate exceptions one by one if any
+ // Waits on a finished resolution task and propagates its exception, if any.
+ def reap(finished: Future[Unit]): Unit = {
try {
- done.foreach(_.get())
+ finished.get()
} catch {
case exn: ExecutionException =>
// Computation threw the exception that is the cause of exn
throw exn.getCause
}
- futures = notDone
- val flowOpt = {
- // We only schedule [[batchSize]] number of flows in parallel.
- if (futures.size < batchSize) {
- Option(toBeResolvedFlows.pollFirst())
- } else {
- None
- }
+ outstanding -= 1
+ }
+
+ while (outstanding > 0 || toBeResolvedFlows.peekFirst() != null) {
+ // Reap every resolution task that has already finished, without blocking.
+ var finished = completionService.poll()
+ while (finished != null) {
+ reap(finished)
+ finished = completionService.poll()
}
- flowOpt.foreach { flow =>
- futures.append(
- executor.submit(
- () =>
+ // We only schedule [[batchSize]] number of flows in parallel.
+ if (outstanding < batchSize && toBeResolvedFlows.peekFirst() != null) {
+ val flow = toBeResolvedFlows.pollFirst()
+ outstanding += 1
+ completionService.submit(
+ () =>
+ try {
try {
- try {
- // Note: Flow don't need their inputs passed, so for now we send empty Seq.
- val result = transformer(flow, Seq.empty)
- require(
- result.forall(_.isInstanceOf[ResolvedFlow]),
- "transformer must return a Seq[Flow]"
- )
+ // Note: Flow don't need their inputs passed, so for now we send empty Seq.
+ val result = transformer(flow, Seq.empty)
+ require(
+ result.forall(_.isInstanceOf[ResolvedFlow]),
+ "transformer must return a Seq[Flow]"
+ )
- val transformedFlows = result.map(_.asInstanceOf[ResolvedFlow])
- resolvedFlowsMap.put(flow.identifier, transformedFlows)
- resolvedFlows.addAll(transformedFlows.asJava)
- } catch {
- case e: TransformNodeRetryableException =>
- val datasetIdentifier = e.datasetIdentifier
- failedDependentFlows.compute(
- datasetIdentifier,
- (_, flows) => {
- // Don't add the input flow back but the failed flow object
- // back which has relevant failure information.
- val failedFlow = e.failedNode
- if (flows == null) {
- Seq(failedFlow)
- } else {
- flows :+ failedFlow
- }
+ val transformedFlows = result.map(_.asInstanceOf[ResolvedFlow])
+ resolvedFlowsMap.put(flow.identifier, transformedFlows)
+ resolvedFlows.addAll(transformedFlows.asJava)
+ } catch {
+ case e: TransformNodeRetryableException =>
+ val datasetIdentifier = e.datasetIdentifier
+ failedDependentFlows.compute(
+ datasetIdentifier,
+ (_, flows) => {
+ // Don't add the input flow back but the failed flow object
+ // back which has relevant failure information.
+ val failedFlow = e.failedNode
+ if (flows == null) {
+ Seq(failedFlow)
+ } else {
+ flows :+ failedFlow
}
- )
- // Between the time the flow started and finished resolving, perhaps the
- // dependent dataset was resolved
- resolvedFlowDestinationsMap.computeIfPresent(
- datasetIdentifier,
- (_, resolved) => {
- if (resolved) {
- // Check if the dataset that the flow is dependent on has been resolved
- // and if so, remove all dependent flows from the failedDependentFlows and
- // add them to the toBeResolvedFlows queue for retry.
- failedDependentFlows.computeIfPresent(
- datasetIdentifier,
- (_, toRetryFlows) => {
- toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
- null
- }
- )
- }
- resolved
+ }
+ )
+ // Between the time the flow started and finished resolving, perhaps the
+ // dependent dataset was resolved
+ resolvedFlowDestinationsMap.computeIfPresent(
+ datasetIdentifier,
+ (_, resolved) => {
+ if (resolved) {
+ // Check if the dataset that the flow is dependent on has been resolved
+ // and if so, remove all dependent flows from the failedDependentFlows and
+ // add them to the toBeResolvedFlows queue for retry.
+ failedDependentFlows.computeIfPresent(
+ datasetIdentifier,
+ (_, toRetryFlows) => {
+ toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
+ null
+ }
+ )
}
+ resolved
+ }
+ )
+ case other: Throwable => throw other
+ }
+ // If all flows to this particular destination are resolved, move to the destination
+ // node transformer
+ if (flowsTo(flow.destinationIdentifier).forall({ flowToDestination =>
+ resolvedFlowsMap.containsKey(flowToDestination.identifier)
+ })) {
+ // If multiple flows completed in parallel, ensure we resolve the destination only
+ // once by electing a leader via computeIfAbsent
+ var isCurrentThreadLeader = false
+ resolvedFlowDestinationsMap.computeIfAbsent(flow.destinationIdentifier, _ => {
+ isCurrentThreadLeader = true
+ // Set initial value as false as flow destination is not resolved yet.
+ false
+ })
+ if (isCurrentThreadLeader) {
+ if (tableMap.contains(flow.destinationIdentifier)) {
+ val transformed =
+ transformer(
+ tableMap(flow.destinationIdentifier),
+ flowsTo(flow.destinationIdentifier)
+ )
+ resolvedTables.addAll(
+ transformed.collect { case t: Table => t }.asJava
)
- case other: Throwable => throw other
- }
- // If all flows to this particular destination are resolved, move to the destination
- // node transformer
- if (flowsTo(flow.destinationIdentifier).forall({ flowToDestination =>
- resolvedFlowsMap.containsKey(flowToDestination.identifier)
- })) {
- // If multiple flows completed in parallel, ensure we resolve the destination only
- // once by electing a leader via computeIfAbsent
- var isCurrentThreadLeader = false
- resolvedFlowDestinationsMap.computeIfAbsent(flow.destinationIdentifier, _ => {
- isCurrentThreadLeader = true
- // Set initial value as false as flow destination is not resolved yet.
- false
- })
- if (isCurrentThreadLeader) {
- if (tableMap.contains(flow.destinationIdentifier)) {
+ resolvedFlows.addAll(
+ transformed.collect { case f: ResolvedFlow => f }.asJava
+ )
+ } else if (viewMap.contains(flow.destinationIdentifier)) {
+ resolvedViews.addAll {
val transformed =
transformer(
- tableMap(flow.destinationIdentifier),
+ viewMap(flow.destinationIdentifier),
flowsTo(flow.destinationIdentifier)
)
- resolvedTables.addAll(
- transformed.collect { case t: Table => t }.asJava
- )
- resolvedFlows.addAll(
- transformed.collect { case f: ResolvedFlow => f }.asJava
- )
- } else if (viewMap.contains(flow.destinationIdentifier)) {
- resolvedViews.addAll {
- val transformed =
- transformer(
- viewMap(flow.destinationIdentifier),
- flowsTo(flow.destinationIdentifier)
- )
- transformed.map(_.asInstanceOf[View]).asJava
- }
- } else if (sinkMap.contains(flow.destinationIdentifier)) {
- resolvedSinks.addAll {
- val transformed =
- transformer(
- sinkMap(flow.destinationIdentifier), flowsTo(flow.destinationIdentifier)
- )
- require(
- transformed.forall(_.isInstanceOf[Sink]),
- "transformer must return a Seq[Sink]"
+ transformed.map(_.asInstanceOf[View]).asJava
+ }
+ } else if (sinkMap.contains(flow.destinationIdentifier)) {
+ resolvedSinks.addAll {
+ val transformed =
+ transformer(
+ sinkMap(flow.destinationIdentifier), flowsTo(flow.destinationIdentifier)
)
- transformed.map(_.asInstanceOf[Sink]).asJava
- }
- } else {
- throw new IllegalArgumentException(
- s"Unsupported destination ${flow.destinationIdentifier.unquotedString}" +
- s" in flow: ${flow.displayName} at transformDownNodes"
+ require(
+ transformed.forall(_.isInstanceOf[Sink]),
+ "transformer must return a Seq[Sink]"
)
+ transformed.map(_.asInstanceOf[Sink]).asJava
}
- // Set flow destination as resolved now.
- resolvedFlowDestinationsMap.computeIfPresent(
- flow.destinationIdentifier,
- (_, _) => {
- // If there are any other node failures dependent on this destination, retry
- // them
- failedDependentFlows.computeIfPresent(
- flow.destinationIdentifier,
- (_, toRetryFlows) => {
- toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
- null
- }
- )
- true
- }
+ } else {
+ throw new IllegalArgumentException(
+ s"Unsupported destination ${flow.destinationIdentifier.unquotedString}" +
+ s" in flow: ${flow.displayName} at transformDownNodes"
)
}
+ // Set flow destination as resolved now.
+ resolvedFlowDestinationsMap.computeIfPresent(
+ flow.destinationIdentifier,
+ (_, _) => {
+ // If there are any other node failures dependent on this destination, retry
+ // them
+ failedDependentFlows.computeIfPresent(
+ flow.destinationIdentifier,
+ (_, toRetryFlows) => {
+ toRetryFlows.foreach(toBeResolvedFlows.addFirst(_))
+ null
+ }
+ )
+ true
+ }
+ )
}
- } catch {
- case ex: TransformNodeFailedException => failedFlowsQueue.add(ex.failedNode)
}
- )
+ } catch {
+ case ex: TransformNodeFailedException => failedFlowsQueue.add(ex.failedNode)
+ }
)
+ } else if (outstanding > 0) {
+ // Nothing finished and nothing could be scheduled, but tasks are still running:
+ // block until the next one finishes instead of busy-spinning on Future.isDone.
+ reap(completionService.take())
Review Comment:
Good question, but the guard is needed - a plain `else` would deadlock the last iteration. The reasoning misses that the `poll()` drain loop runs *after* the loop-condition check and mutates `outstanding`. The invariant `outstanding > 0 || queue.nonEmpty` only holds at the top of the loop; by the time we reach this branch, the drain may have reaped the last in-flight tasks and taken `outstanding` to 0 while the queue is already empty. In that case the `if` is false (empty queue) and there is nothing left to wait for - `take()` would block forever. The `outstanding > 0` guard lets the loop fall through so the next condition check (`0 > 0 || empty`) exits cleanly. I added a comment spelling this out.
--
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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang closed pull request #56700: [SPARK-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution
URL: https://github.com/apache/spark/pull/56700
--
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-57638][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #56700:
URL: https://github.com/apache/spark/pull/56700#issuecomment-5093367290
Thanks @dongjoon-hyun @uros-b @yadavay-amzn
--
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