Skip to content

[SPARK-43755][CONNECT] Move execution out of SparkExecutePlanStreamHandler and to a different thread #41315

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
juliuszsompolski wants to merge 19 commits into from
Closed

[SPARK-43755][CONNECT] Move execution out of SparkExecutePlanStreamHandler and to a different thread #41315

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
3d5ee30
move execution
juliuszsompolski May 23, 2023
2ba6d26
fix compile
juliuszsompolski Jul 6, 2023
bbf8761
renames
juliuszsompolski Jul 6, 2023
1a8db60
dump
juliuszsompolski Jul 7, 2023
bbba4ca
todo - deadlocks and needs debug
juliuszsompolski Jul 10, 2023
e4e3615
passes tests, needs cleanup, needs a bit of thought about error handling
juliuszsompolski Jul 11, 2023
6f39e7b
separate ExecuteHolder / ExecuteRunner; todo comments and docs
juliuszsompolski Jul 11, 2023
20cb3dd
fix hang and cancellation
juliuszsompolski Jul 12, 2023
bcea9d4
docs and comments
juliuszsompolski Jul 12, 2023
0f88963
nits
juliuszsompolski Jul 12, 2023
2e6effa
scalastyle
juliuszsompolski Jul 13, 2023
e98db60
change job tag pattern
juliuszsompolski Jul 13, 2023
fee8838
self review nits and cleanup
juliuszsompolski Jul 13, 2023
0757c1c
add error class to documentation
juliuszsompolski Jul 13, 2023
a0d0792
typo
juliuszsompolski Jul 13, 2023
428dd77
Merge branch 'master' into sc-execute-thread
juliuszsompolski Jul 13, 2023
9104557
fix OPERATION_CANCELED error class
juliuszsompolski Jul 13, 2023
aa4093c
Merge branch 'master' into sc-execute-thread
juliuszsompolski Jul 14, 2023
c5008b5
Update connector/connect/server/src/main/scala/org/apache/spark/sql/c…
HyukjinKwon Jul 14, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
fix compile
  • Loading branch information
@juliuszsompolski
juliuszsompolski committed Jul 6, 2023
commit 2ba6d26ddfe07effecf0eb6c34da9d4733c138b1
12 changes: 0 additions & 12 deletions ...r/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseObserver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,18 +33,6 @@ import org.apache.spark.connect.proto.ExecutePlanResponse
class ExecutePlanResponseObserver(responseObserver: StreamObserver[ExecutePlanResponse])
extends StreamObserver[ExecutePlanResponse] {

val jobTag =
"SparkConnect_" +
s"User_${sessionHolder.userId}_Session_${sessionHolder.sessionId}_Request_${operationId}"

def interrupt(): Unit = {
// TODO/WIP: This only interrupts active Spark jobs that are actively running.
// This would then throw the error from ExecutePlan and terminate it.
// But if the query is not running a Spark job, but executing code on Spark driver, this
// would be a noop and the execution will keep running.
sessionHolder.session.sparkContext.cancelJobsWithTag(jobTag)
}

def onNext(r: ExecutePlanResponse): Unit = {
responseObserver.onNext(r)
}
Expand Down
4 changes: 2 additions & 2 deletions ...ver/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,10 +63,10 @@ class SparkConnectPlanExecution(executionHolder: ExecutionHolder) {
Dataset.ofRows(sessionHolder.session, planner.transformRelation(request.getPlan.getRoot))
responseObserver.onNext(createSchemaResponse(request.getSessionId, dataframe.schema))
processAsArrowBatches(request.getSessionId, dataframe, responseObserver)
responseObserver.onNext(createMetricsResponse(request.getSessionId, dataframe))
responseObserver.onNext(MetricGenerator.createMetricsResponse(request.getSessionId, dataframe))
if (dataframe.queryExecution.observedMetrics.nonEmpty) {
responseObserver.onNext(
SparkConnectStreamHandler.sendObservedMetricsToResponse(request.getSessionId, dataframe))
createObservedMetricsResponse(request.getSessionId, dataframe))
}
responseObserver.onCompleted()
}
Expand Down
9 changes: 4 additions & 5 deletions .../connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ import org.apache.spark.SparkSQLException
import org.apache.spark.connect.proto
import org.apache.spark.connect.proto.{ExecutePlanRequest, ExecutePlanResponse}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.connect.artifact.SparkConnectArtifactManager
import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.execution.{ExecutePlanResponseObserver, SparkConnectPlanExecution}
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
Expand Down Expand Up @@ -96,7 +95,7 @@ case class ExecutionHolder(operationId: String, sessionHolder: SessionHolder) ex
}
}

protected def execute() = SparkConnectArtifactManager.withArtifactClassLoader {
protected def execute() = {
try {
// synchronized - check if already got interrupted while starting.
synchronized {
Expand All @@ -111,8 +110,8 @@ case class ExecutionHolder(operationId: String, sessionHolder: SessionHolder) ex
val debugString = requestString(executePlanRequest.get)

// Set tag for query cancellation
session.sparkContext.addJobTag(executeHolder.jobTag)
session.sparkContext.setDescription(
session.sparkContext.addJobTag(jobTag)
session.sparkContext.setJobDescription(
s"Spark Connect - ${StringUtils.abbreviate(debugString, 128)}")
session.sparkContext.setInterruptOnCancel(true)

Expand Down Expand Up @@ -141,7 +140,7 @@ case class ExecutionHolder(operationId: String, sessionHolder: SessionHolder) ex
// scalastyle:off
logDebug(s"Exception in execute: $e")
// Always cancel all remaining execution after error.
sessionHolder.session.sparkContext.cancelJobGroup(jobGroupId)
sessionHolder.session.sparkContext.cancelJobsWithTag(jobTag)
executionError = if (interrupted) {
// Turn the interrupt into OPERATION_CANCELLED error.
Some(new SparkSQLException("OPERATION_CANCELLED", Map.empty))
Expand Down
1 change: 0 additions & 1 deletion ...or/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import scala.util.control.NonFatal

import org.apache.spark.JobArtifactSet
import org.apache.spark.SparkException
import org.apache.spark.connect.proto
import org.apache.spark.internal.Logging
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.SparkSession
Expand Down