[SPARK-43879][CONNECT] Decouple handle command and send response on server side

[beliefer]

What changes were proposed in this pull request?

SparkConnectStreamHandler treats the proto requests from connect client and send the responses back to connect client. SparkConnectStreamHandler holds a component StreamObserver to send responses. Currently, SparkConnectPlanner also knows StreamObserver passed by method parameters. So the behavior introduces some issues.

• Visibility

SparkConnectStreamHandler holds the StreamObserver and shouldn't expose it. Now, we expose StreamObserver to SparkConnectPlanner and the latter is public, so every developers could use it.

• Safety

Based on the visibility issue, this is usually not a secure strategy.

• Code Design

Because SparkConnectStreamHandler is fully responsible for responding to RPC, which is a division of responsibilities in good programming. The better code design, the easier to extend function.

So I think we should keep the StreamObserver could be accessed only with SparkConnectStreamHandler.
This PR wraps the detail of StreamObserver into SparkConnectStreamHandler, SparkConnectPlanner only need call sendResponse if the response is ready.

This PR want decouple the process handle commands and the other process send responses on server side.

Note: As we disscussed on #41379, this PR doesn't delay to send any response "right now" and expects that it will be returned.

Why are the changes needed?

Decouple handle command and send response on server side.

Does this PR introduce any user-facing change?

'No'.
Just update the inner implementation.

How was this patch tested?

Exists test cases.

[beliefer]

ping @grundprinzip @hvanhovell cc @HyukjinKwon @zhengruifeng

[beliefer]

ping @hvanhovell @grundprinzip

[beliefer]

@grundprinzip Could you take a look? I have rebased many times.

[grundprinzip]

I think the overall approach is fine. I don't think this is the most burning refactoring that needs to be done, but it's ok.

However, there are a couple of things that we need to make sure. The APIs of the planner change their visibility and we need to make sure that this is really necessary.

In addition there is the question if the planner can really exist without response handler.

[grundprinzip]

should we make this private[connect]?

[beliefer]

It's OK for me.

[grundprinzip]

Is there a valid use-case where this can be empty? Personally, I don't think that this should ever be none. I understand for testing purposes one might inject a mock, but otherwise the planner cannot exist without the responsehandler

[beliefer]

There exists some test case only need transform proto to LogicalPlan and doesn't depend on responsehandler.

Please refer:

spark/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/ProtoToParsedPlanTestSuite.scala

Line 181 in 1a5b9f2

val planner = new SparkConnectPlanner(SessionHolder.forTesting(spark))

[grundprinzip]

why remove the private? This should at least be package private.

[beliefer]

@grundprinzip Thank you for review.

[beliefer]

The APIs of the planner change their visibility and we need to make sure that this is really necessary.

Yes. Because connect exposes the StreamObserver to users is dangerous, the refactor hide it into SparkConnectResponseHandler.
Although it seems a little break change. I think it's worth to protect it.

[beliefer]

In addition there is the question if the planner can really exist without response handler.

Spark Connect exists test case uses the planner without response handler. Could we add response handler for it ?

[beliefer]

ping @grundprinzip cc @hvanhovell

[grundprinzip]

Hey @beliefer there are some inflight changes that will conflict heavily with this work - #41315 The changes are needed for the asynchronous query execution. Can we pause this PR for a moment until the other changes are in so that we have a good understanding of how they can work together?

cc @juliuszsompolski

[juliuszsompolski]

@beliefer In what I've been working on:

Code Design

I do plan to have SparkConnectStreamHandler be responsible for sending the RPCs. That's not yet ready in my PR, in the current iteration I had the execution thread send the RPCs, but I need to hand it over back to the SparkConnectStreamHandler thread. I agree with you that it's good division of responsibilities, especially since my followup work is that I want to be able to "reconnect" to a query after the initial ExecutePlan disconnects, so I need to have another RPC call to reconnect to the stream.

Visibility

I avoid the need for the refactoring by instead passing my own class implementing the StreamObserver interface that is just going to notify the "real" StreamObserver that there is something to be sent. This way I avoid the need for refactoring, and the StreamObserver is fine as an interface to be passed around.

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!