[SPARK-58547][CONNECT] Expose operation IDs for end-to-end request attribution

[cloud-fan]

What changes were proposed in this pull request?

Spark Connect's protocol includes an operation ID for identifying an ExecutePlan request, but
clients currently leave it unset unless callers explicitly provide one. This PR makes operation
IDs available throughout the request lifecycle:

• Python and Scala clients generate an operation ID before sending every ExecutePlan request.
• Python exposes it through ExecutionInfo.operation_id after successful execution and
  SparkConnectException.operation_id after failure.
• Scala clients continue exposing it in ExecutePlanResponse and can retrieve it from failures
  through SparkConnectClient.getOperationId.
• Python session hooks preserve the client-generated operation ID.
• The server publishes the ID through SparkContext.SPARK_CONNECT_OPERATION_ID_PROPERTY.

No protocol change is required, and callers that explicitly provide an operation ID retain the
existing behavior.

Why are the changes needed?

Successful requests can sometimes be correlated using the operation ID returned by the server.
That is insufficient for failures before the first response, because the caller never learns a
server-generated identifier.

Generating the identifier client-side provides a stable correlation key before the RPC begins.
The server propagates the same ID using Spark's existing local-property mechanism, making it
available to driver-side listeners and executor tasks. Applications and observability integrations
can therefore correlate client failures, server logs, Spark jobs, and tasks without depending on a
specific event system.

Does this PR introduce any user-facing change?

Yes. Spark Connect clients now generate operation IDs by default. Python users can access the ID
through execution information and exceptions; Scala users can access it through responses and
exceptions. Server-side integrations can access it through the public Spark local-property key.

How was this patch tested?

Added tests for generated operation IDs, Python session-hook preservation, successful execution
metadata, exceptions, and server-side local-property propagation.

Ran:

• build/sbt 'connect-client-jvm/testOnly org.apache.spark.sql.connect.client.SparkConnectClientSuite'
• build/sbt 'connect/testOnly org.apache.spark.sql.connect.service.SparkConnectServiceE2ESuite'

The focused Python test command was also invoked, but the configured local interpreter skipped the
Connect tests because pandas 2.2 or newer was unavailable.

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

Generated-by: OpenAI Codex (GPT-5)

[zhengruifeng]

private[spark] ?

[uros-b]

Thank you @cloud-fan and @zhengruifeng!