[SPARK-52807][SDP] Proto changes to support analysis inside Declarative Pipelines query functions

[sryza]

What changes were proposed in this pull request?

Introduces a mechanism for lazy execution of Declarative Pipelines query functions. A query function is something like the mv1 in this example:

@materialized_view
def mv1():
    return spark.table("upstream_table").filter(some_condition)

Currently, query functions are always executed eagerly. I.e. the implementation of the materialized_view decorator immediately invokes the function that it decorates and then registers the resulting DataFrame with the server.

This PR introduces Spark Connect proto changes that enable executing query functions later on, initiated by the server during graph resolution. After all datasets and flows have been registered with the server, the server can tell the client to execute the query functions for flows that haven't yet successfully been executed. The way this works is that the client initiates an RPC with the server, and then the server streams back responses that indicate to the client when it's time to execute a query function for one of its flows. Relevant changes:

• New QueryFunctionFailure message
• New QueryFunctionResult message
• Replace relation field in DefineFlow with query_function_result field
• New DefineFlowQueryFunctionResult message
• New GetQueryFunctionExecutionSignalStream message
• New PipelineQueryFunctionExecutionSignal message

Why are the changes needed?

There are some situations where we can't resolve the relation immediately at the time we're registering a flow.

E.g. consider this situation:
file 1:

@materialized_view
def mv1():
    data = [("Alice", 10), ("Bob", 15), ("Alice", 5)]
    return spark.createDataFrame(data, ["name", "amount"])

file 2:

@materialized_view
def mv2():
    return spark.table("mv1").groupBy("name").agg(sum("amount").alias("total_amount"))

Unlike some other transformations, which get analyzed lazily, groupBy can trigger an AnalyzePlan Spark Connect request immediately. If the query function for mv2 gets executed before mv1, then it will hit an error, because mv1 doesn't exist yet. groupBy isn't the only example here.

Other examples of these kinds of situations:

• The set of columns for a downstream table is determined from the set of columns in an upstream table.
• When spark.sql is used.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

[sryza]

I don't have strong opinions here, but it feels a little weird to me to put this inside UserContext, because it's not related to the user. Another option could be to put it directly on messages the the user context is on.

[sryza]

Having investigated further: I think we could avoid the need for this if Spark Connect provided the ability to clone sessions. Each time we're about to execute a query function, we could clone the session, add some confs that specify the dataflow graph ID and flow name, and execute it in there.

[sryza]

Closing because superseded by #52154