This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 029731cc0fb9 [SPARK-57349][CONNECT] Split udf protocol into message
and grpc service.
029731cc0fb9 is described below
commit 029731cc0fb92123cc51c233f3112a32f4c05af8
Author: Haiyang Sun <[email protected]>
AuthorDate: Tue Jun 9 12:00:27 2026 -0700
[SPARK-57349][CONNECT] Split udf protocol into message and grpc service.
### What changes were proposed in this pull request?
This PR splits the UDF worker protocol definition into two `.proto` files
in the `udf-worker-proto` module:
- `udf_protocol.proto` is renamed to `udf_message.proto` and now contains
**only** the protocol's message types (the `Execute`-stream envelopes,
`Init`/`Finish`/`Cancel`/`Data` request and response messages, `ErrorResponse`,
etc.).
- A new `udf_service.proto` holds the gRPC **service** definition --
`service UdfWorker` with its `Execute(stream UdfRequest) returns (stream
UdfResponse)` and `Manage(WorkerRequest) returns (WorkerResponse)` RPCs -- and
imports `udf_message.proto`.
This is a pure reorganization of the protocol definitions. There is no
change to message types, field numbers, or generated message classes, and no
Scala/behavior change.
### Why are the changes needed?
Clarity between message definition and grpc service.
### Does this PR introduce _any_ user-facing change?
No. The protocol is part of the experimental, currently-unused UDF worker
interface, and this PR only reorganizes the `.proto` source layout without
altering the generated message classes.
### How was this patch tested?
Existing build/codegen coverage. Verified `build/sbt
udf-worker-proto/compile` succeeds -- the proto module compiles all four
`.proto` files (including the relocated service definition) into
`protobuf-java` message classes with no gRPC dependency, confirming the split
is well-formed.
### Was this patch authored or co-authored using generative AI tooling?
Yes
Closes #56391 from haiyangsun-db/pr1-udf-proto-split.
Authored-by: Haiyang Sun <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
udf/worker/README.md | 12 ++--
.../{udf_protocol.proto => udf_message.proto} | 46 +------------
.../proto/src/main/protobuf/udf_service.proto | 75 ++++++++++++++++++++++
3 files changed, 85 insertions(+), 48 deletions(-)
diff --git a/udf/worker/README.md b/udf/worker/README.md
index cf3253fdaa89..2250c8df1053 100644
--- a/udf/worker/README.md
+++ b/udf/worker/README.md
@@ -35,7 +35,8 @@ provisioning service or daemon).
udf/worker/
├── proto/ -- protobuf message classes only
(protobuf-java)
│ worker_spec.proto -- UDFWorkerSpecification protobuf
-│ udf_protocol.proto -- UDF execution protocol (Init, UdfPayload,
...)
+│ udf_message.proto -- UDF execution protocol messages (Init,
UdfPayload, ...)
+│ udf_service.proto -- UdfWorker gRPC service (Execute, Manage)
│ common.proto -- shared enums (UDFWorkerDataFormat, etc.)
│
├── core/ -- abstract interfaces
@@ -50,7 +51,7 @@ udf/worker/
│ DirectWorkerSession.scala -- session backed by a direct process
│
└── grpc/ -- gRPC transport (gRPC runtime confined
here)
- (generated) -- UdfWorkerGrpc service stubs from
proto/udf_protocol.proto
+ (generated) -- UdfWorkerGrpc service stubs from
proto/udf_service.proto
```
The `core/` package defines abstract interfaces that are independent of how
@@ -60,7 +61,7 @@ worker creation where Spark spawns local OS processes. Future
packages
obtaining workers from a provisioning service or daemon.
The `grpc/` module owns the gRPC service-stub generation (from
-`proto/`'s `udf_protocol.proto`) and the gRPC runtime dependencies. Keeping
+`proto/`'s `udf_service.proto`) and the gRPC runtime dependencies. Keeping
gRPC here means `proto/`, `core/`, and their consumers (`core`, `catalyst`,
`sql/core`) carry no gRPC dependency on their classpath.
@@ -74,8 +75,9 @@ Engine -> Worker: Init -> PayloadChunk* -> (DataRequest)* ->
Finish (Cancel)?
Worker -> Engine: InitResponse -> (DataResponse)* ->
(ErrorResponse)? -> (FinishResponse | CancelResponse)
```
-See `udf/worker/proto/src/main/protobuf/udf_protocol.proto` for the complete
-protocol definition, ordering invariants, and error contract.
+See `udf/worker/proto/src/main/protobuf/udf_message.proto` for the complete
+message definitions, ordering invariants, and error contract, and
+`udf_service.proto` for the gRPC service.
### Direct worker creation
diff --git a/udf/worker/proto/src/main/protobuf/udf_protocol.proto
b/udf/worker/proto/src/main/protobuf/udf_message.proto
similarity index 93%
rename from udf/worker/proto/src/main/protobuf/udf_protocol.proto
rename to udf/worker/proto/src/main/protobuf/udf_message.proto
index a66e09af09fe..dd7c9ceb963b 100644
--- a/udf/worker/proto/src/main/protobuf/udf_protocol.proto
+++ b/udf/worker/proto/src/main/protobuf/udf_message.proto
@@ -29,50 +29,10 @@ option java_multiple_files = true;
//
// The Spark engine acts as the gRPC client; a UDF worker (in any
// language) acts as the gRPC server.
-// =====================================================================
-
-// The default UDF gRPC service. A worker that exposes this service
-// MUST do so over the default connection of the worker specification.
//
-// Future revisions of the worker specification may introduce additional
-// dedicated connections for specific purposes (e.g. a separate channel
-// for streaming state store access in stateful UDFs).
-service UdfWorker {
- // Per-execution stream. See [[UdfControlRequest]] for the complete
- // wire protocol and ordering invariants.
- //
- // Error contract: a gRPC error on this stream signals a transport or
- // connection failure. Application-level errors (user code exceptions,
- // worker errors, protocol violations) are reported via [[ErrorResponse]]
- // or [[InitResponse.error]] so the stream lifecycle remains intact.
- //
- // Hanging UDFs: this protocol does not define what constitutes a
- // hanging UDF -- the protocol cannot distinguish user code that is
- // genuinely stuck from code that is merely slow or intentionally
- // long-running. In-band [[Cancel]] is also not a reliable remedy:
- // a Cancel queued behind a blocked batch (or behind a full sender
- // buffer) is never delivered to the worker. Detection and recovery
- // are layered on top of this protocol -- typically via worker-side
- // watchdog timers configurable at the session or UDF level.
- //
- // Stream lifecycle: the engine MUST half-close the request stream
- // (call onCompleted() on the gRPC request side) only after the session
- // outcome is known: on receiving [[FinishResponse]] or [[CancelResponse]]
- // (clean termination), or on receiving a gRPC error. Deferring the
- // half-close keeps the stream open long enough for [[Cancel]] to follow
- // [[Finish]] when needed (see [[Finish]] for the full contract).
- //
- // For stateful execution, the state is maintained per bi-directional
- // stream, mapping to a `WorkerSession` on the engine side.
- rpc Execute(stream UdfRequest) returns (stream UdfResponse);
-
- // Worker-scoped management RPC for heartbeat, capability query, and
- // graceful shutdown. Workers MUST ensure this RPC remains serviceable
- // regardless of how many [[Execute]] streams are in flight; failing to
- // do so can prevent the engine from detecting a hung worker or
- // initiating a clean shutdown.
- rpc Manage(WorkerRequest) returns (WorkerResponse);
-}
+// This file defines only the protocol's message types. The gRPC service
+// that carries them ([[UdfWorker]]) lives in udf_service.proto.
+// =====================================================================
// =====================================================================
// Execute stream -- envelope
diff --git a/udf/worker/proto/src/main/protobuf/udf_service.proto
b/udf/worker/proto/src/main/protobuf/udf_service.proto
new file mode 100644
index 000000000000..2140d4f48bca
--- /dev/null
+++ b/udf/worker/proto/src/main/protobuf/udf_service.proto
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = "proto3";
+
+import "udf_message.proto";
+
+package org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+option java_multiple_files = true;
+
+// =====================================================================
+// gRPC service definition for the language-agnostic UDF protocol.
+//
+// The service is kept in its own file, separate from the protocol's
+// message definitions (udf_message.proto).
+// =====================================================================
+
+// The default UDF gRPC service. A worker that exposes this service
+// MUST do so over the default connection of the worker specification.
+//
+// Future revisions of the worker specification may introduce additional
+// dedicated connections for specific purposes (e.g. a separate channel
+// for streaming state store access in stateful UDFs).
+service UdfWorker {
+ // Per-execution stream. See [[UdfControlRequest]] for the complete
+ // wire protocol and ordering invariants.
+ //
+ // Error contract: a gRPC error on this stream signals a transport or
+ // connection failure. Application-level errors (user code exceptions,
+ // worker errors, protocol violations) are reported via [[ErrorResponse]]
+ // or [[InitResponse.error]] so the stream lifecycle remains intact.
+ //
+ // Hanging UDFs: this protocol does not define what constitutes a
+ // hanging UDF -- the protocol cannot distinguish user code that is
+ // genuinely stuck from code that is merely slow or intentionally
+ // long-running. In-band [[Cancel]] is also not a reliable remedy:
+ // a Cancel queued behind a blocked batch (or behind a full sender
+ // buffer) is never delivered to the worker. Detection and recovery
+ // are layered on top of this protocol -- typically via worker-side
+ // watchdog timers configurable at the session or UDF level.
+ //
+ // Stream lifecycle: the engine MUST half-close the request stream
+ // (call onCompleted() on the gRPC request side) only after the session
+ // outcome is known: on receiving [[FinishResponse]] or [[CancelResponse]]
+ // (clean termination), or on receiving a gRPC error. Deferring the
+ // half-close keeps the stream open long enough for [[Cancel]] to follow
+ // [[Finish]] when needed (see [[Finish]] for the full contract).
+ //
+ // For stateful execution, the state is maintained per bi-directional
+ // stream, mapping to a `WorkerSession` on the engine side.
+ rpc Execute(stream UdfRequest) returns (stream UdfResponse);
+
+ // Worker-scoped management RPC for heartbeat, capability query, and
+ // graceful shutdown. Workers MUST ensure this RPC remains serviceable
+ // regardless of how many [[Execute]] streams are in flight; failing to
+ // do so can prevent the engine from detecting a hung worker or
+ // initiating a clean shutdown.
+ rpc Manage(WorkerRequest) returns (WorkerResponse);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
