diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3c5da3fd..5748469b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -41,6 +41,17 @@ jobs:
       - uses: ./.github/actions/setup
       - run: cargo test --features integration
 
+  # Runs the whole lib unit suite with `grpc` off, the proof that the abstraction builds and runs
+  # without gRPC. `lfs: true` because some planner unit tests load parquet testdata.
+  unit-test-no-grpc:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          lfs: true
+      - uses: ./.github/actions/setup
+      - run: cargo test --no-default-features --lib
+
   tpch-correctness-test:
     runs-on: ubuntu-latest
     strategy:
@@ -50,7 +61,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/setup
-      - run: cargo test --features tpch --test tpch_correctness_test
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpch --test tpch_correctness_test
         env:
           ADAPTIVE: ${{ matrix.planning_mode == 'adaptive' }}
 
@@ -59,7 +70,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/setup
-      - run: cargo test --features tpch --test tpch_plans_test
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpch --test tpch_plans_test
 
   tpcds-correctness-test:
     runs-on: ubuntu-latest
@@ -73,9 +84,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/tpcds/main.zip
+          path: benchmarks/testdata/tpcds/main.zip
           key: "main.zip"
-      - run: cargo test --features tpcds --test tpcds_correctness_test shard${{ matrix.shard }}
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpcds --test tpcds_correctness_test shard${{ matrix.shard }}
         env:
           ADAPTIVE: ${{ matrix.planning_mode == 'adaptive' }}
 
@@ -86,9 +97,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/tpcds/main.zip
+          path: benchmarks/testdata/tpcds/main.zip
           key: "main.zip"
-      - run: cargo test --features tpcds --test tpcds_plans_test
+      - run: cargo test -p datafusion-distributed-benchmarks --features tpcds --test tpcds_plans_test
 
   clickbench-correctness-test:
     runs-on: ubuntu-latest
@@ -101,9 +112,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/clickbench/
+          path: benchmarks/testdata/clickbench/
           key: "data"
-      - run: cargo test --features clickbench --test clickbench_correctness_test
+      - run: cargo test -p datafusion-distributed-benchmarks --features clickbench --test clickbench_correctness_test
         env:
           ADAPTIVE: ${{ matrix.planning_mode == 'adaptive' }}
 
@@ -114,9 +125,9 @@ jobs:
       - uses: ./.github/actions/setup
       - uses: actions/cache@v4
         with:
-          path: testdata/clickbench/
+          path: benchmarks/testdata/clickbench/
           key: "data"
-      - run: cargo test --features clickbench --test clickbench_plans_test
+      - run: cargo test -p datafusion-distributed-benchmarks --features clickbench --test clickbench_plans_test
 
   format-check:
     runs-on: ubuntu-latest
diff --git a/Cargo.lock b/Cargo.lock
index dbe6096c..12a4e62b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2199,7 +2199,6 @@ dependencies = [
  "crossbeam-queue",
  "dashmap",
  "datafusion",
- "datafusion-distributed-benchmarks",
  "datafusion-proto",
  "delegate",
  "futures",
@@ -2249,12 +2248,14 @@ dependencies = [
  "object_store",
  "openssl",
  "parquet",
+ "pretty_assertions",
  "reqwest",
  "serde",
  "serde_json",
  "sketches-ddsketch",
  "structopt",
  "sysinfo",
+ "test-case",
  "tokio",
  "tonic",
  "tpchgen",
diff --git a/Cargo.toml b/Cargo.toml
index 79366ed3..1bb4874d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -45,7 +45,7 @@ sysinfo = { version = "0.30", optional = true }
 sketches-ddsketch = { version = "0.3", features = ["use_serde"] }
 num-traits = "0.2"
 bincode = "1"
-tonic-prost = "0.14.2"
+tonic-prost = { version = "0.14.2", optional = true }
 
 # grpc-specific features
 arrow-flight = { version = "58", optional = true }
@@ -68,6 +68,7 @@ grpc = [
     "arrow-select",
     "arrow-ipc",
     "tonic",
+    "tonic-prost",
     "tower"
 ]
 
@@ -75,14 +76,9 @@ integration = ["insta", "parquet", "arrow", "hyper-util", "grpc"]
 
 system-metrics = ["sysinfo"]
 
-tpch = ["integration"]
-tpcds = ["integration"]
-clickbench = ["integration"]
-slow-tests = []
 sysinfo = ["dep:sysinfo"]
 
 [dev-dependencies]
-datafusion-distributed-benchmarks = { path = "benchmarks" }
 structopt = "0.3"
 insta = { version = "1.46.0", features = ["filters"] }
 parquet = "58"
@@ -92,6 +88,43 @@ hyper-util = "0.1.16"
 pretty_assertions = "1.4"
 test-case = "3.3.1"
 
+# Every example drives the gRPC transport, so a no-grpc build skips them.
+[[example]]
+name = "custom_distributed_partial_reduction_tree"
+required-features = ["integration"]
+
+[[example]]
+name = "custom_execution_plan"
+required-features = ["integration"]
+
+[[example]]
+name = "custom_worker_url_routing"
+required-features = ["integration"]
+
+[[example]]
+name = "work_unit_feed"
+required-features = ["integration"]
+
+[[example]]
+name = "in_memory_cluster"
+required-features = ["grpc"]
+
+[[example]]
+name = "localhost_run"
+required-features = ["grpc"]
+
+[[example]]
+name = "localhost_versioned_run"
+required-features = ["grpc"]
+
+[[example]]
+name = "localhost_worker"
+required-features = ["grpc"]
+
+[[example]]
+name = "localhost_versioned_worker"
+required-features = ["grpc"]
+
 [workspace.lints.clippy]
 disallowed_types = "deny"
 disallowed-methods = "deny"
diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml
index 531a8aba..58736c66 100644
--- a/benchmarks/Cargo.toml
+++ b/benchmarks/Cargo.toml
@@ -34,9 +34,20 @@ openssl = { version = "0.10", features = ["vendored"] } # Keep this. Necessary f
 mimalloc = "0.1"
 sketches-ddsketch = "0.3"
 
+[features]
+# Gates for the dataset test suites under `tests/`. They live here rather than in the library
+# because, as a dev-dependency of the library, this crate re-enables `grpc` on every test build
+# through feature unification, which made the no-grpc config untestable.
+tpch = []
+tpcds = []
+clickbench = []
+slow-tests = []
+
 [dev-dependencies]
 criterion = "0.5"
 sysinfo = "0.30"
+pretty_assertions = "1.4"
+test-case = "3.3.1"
 
 [build-dependencies]
 built = { version = "0.8", features = ["git2", "chrono"] }
diff --git a/benchmarks/cdk/bin/worker.rs b/benchmarks/cdk/bin/worker.rs
index 96a399c8..cdeb64a3 100644
--- a/benchmarks/cdk/bin/worker.rs
+++ b/benchmarks/cdk/bin/worker.rs
@@ -1,3 +1,4 @@
+#![allow(clippy::disallowed_types)]
 use async_trait::async_trait;
 use aws_config::BehaviorVersion;
 use aws_sdk_ec2::Client as Ec2Client;
diff --git a/tests/clickbench_correctness_test.rs b/benchmarks/tests/clickbench_correctness_test.rs
similarity index 99%
rename from tests/clickbench_correctness_test.rs
rename to benchmarks/tests/clickbench_correctness_test.rs
index acf3daa8..25fb38f8 100644
--- a/tests/clickbench_correctness_test.rs
+++ b/benchmarks/tests/clickbench_correctness_test.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "clickbench", test))]
+#[cfg(all(feature = "clickbench", test))]
 mod tests {
     use datafusion::arrow::array::RecordBatch;
     use datafusion::common::plan_err;
diff --git a/tests/clickbench_plans_test.rs b/benchmarks/tests/clickbench_plans_test.rs
similarity index 99%
rename from tests/clickbench_plans_test.rs
rename to benchmarks/tests/clickbench_plans_test.rs
index b9c32bb1..7291d4f7 100644
--- a/tests/clickbench_plans_test.rs
+++ b/benchmarks/tests/clickbench_plans_test.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "clickbench", test))]
+#[cfg(all(feature = "clickbench", test))]
 mod tests {
     use datafusion::error::Result;
     use datafusion_distributed::test_utils::in_memory_channel_resolver::start_in_memory_context;
diff --git a/tests/stateful_data_cleanup.rs b/benchmarks/tests/stateful_data_cleanup.rs
similarity index 98%
rename from tests/stateful_data_cleanup.rs
rename to benchmarks/tests/stateful_data_cleanup.rs
index fae892d2..f977c61a 100644
--- a/tests/stateful_data_cleanup.rs
+++ b/benchmarks/tests/stateful_data_cleanup.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpch", test))]
+#[cfg(all(feature = "tpch", test))]
 mod tests {
     use datafusion::common::instant::Instant;
     use datafusion::error::Result;
diff --git a/tests/tpcds_correctness_test.rs b/benchmarks/tests/tpcds_correctness_test.rs
similarity index 99%
rename from tests/tpcds_correctness_test.rs
rename to benchmarks/tests/tpcds_correctness_test.rs
index e4baeba9..8d5d8722 100644
--- a/tests/tpcds_correctness_test.rs
+++ b/benchmarks/tests/tpcds_correctness_test.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpcds", test))]
+#[cfg(all(feature = "tpcds", test))]
 mod tests {
     use datafusion::arrow::array::RecordBatch;
     use datafusion::common::plan_err;
diff --git a/tests/tpcds_plans_test.rs b/benchmarks/tests/tpcds_plans_test.rs
similarity index 99%
rename from tests/tpcds_plans_test.rs
rename to benchmarks/tests/tpcds_plans_test.rs
index 412c50a2..b6363ad4 100644
--- a/tests/tpcds_plans_test.rs
+++ b/benchmarks/tests/tpcds_plans_test.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpcds", test))]
+#[cfg(all(feature = "tpcds", test))]
 mod tests {
     use datafusion::error::Result;
     use datafusion_distributed::test_utils::in_memory_channel_resolver::start_in_memory_context;
diff --git a/tests/tpch_correctness_test.rs b/benchmarks/tests/tpch_correctness_test.rs
similarity index 99%
rename from tests/tpch_correctness_test.rs
rename to benchmarks/tests/tpch_correctness_test.rs
index 3ae089dc..7d6ab2ce 100644
--- a/tests/tpch_correctness_test.rs
+++ b/benchmarks/tests/tpch_correctness_test.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpch", test))]
+#[cfg(all(feature = "tpch", test))]
 mod tests {
     use datafusion::physical_plan::execute_stream;
     use datafusion::prelude::SessionContext;
diff --git a/tests/tpch_plans_test.rs b/benchmarks/tests/tpch_plans_test.rs
similarity index 99%
rename from tests/tpch_plans_test.rs
rename to benchmarks/tests/tpch_plans_test.rs
index 016e21ea..f625cfec 100644
--- a/tests/tpch_plans_test.rs
+++ b/benchmarks/tests/tpch_plans_test.rs
@@ -1,4 +1,4 @@
-#[cfg(all(feature = "integration", feature = "tpch", test))]
+#[cfg(all(feature = "tpch", test))]
 mod tests {
     use datafusion_distributed::test_utils::in_memory_channel_resolver::start_in_memory_context;
     use datafusion_distributed::{
diff --git a/console/src/main.rs b/console/src/main.rs
index 731b504b..ab1171d7 100644
--- a/console/src/main.rs
+++ b/console/src/main.rs
@@ -63,10 +63,10 @@ async fn run_app(
         terminal.draw(|frame| ui::render(frame, app))?;
 
         // Check for keyboard input (16ms timeout ~ 60fps responsiveness)
-        if event::poll(Duration::from_millis(16))? {
-            if let Event::Key(key) = event::read()? {
-                input::handle_key_event(app, key);
-            }
+        if event::poll(Duration::from_millis(16))?
+            && let Event::Key(key) = event::read()?
+        {
+            input::handle_key_event(app, key);
         }
 
         if app.should_quit {
diff --git a/console/src/worker.rs b/console/src/worker.rs
index 4a23c5b9..44c06260 100644
--- a/console/src/worker.rs
+++ b/console/src/worker.rs
@@ -176,32 +176,32 @@ impl WorkerConn {
 
                 // Detect completed tasks: tasks that were running but disappeared
                 for old_task in &self.tasks {
-                    if old_task.status == TaskStatus::Running as i32 {
-                        if let Some(sk) = &old_task.task_key {
-                            let key = (sk.query_id.clone(), sk.stage_id, sk.task_number);
-                            if !new_task_keys.contains(&key) {
-                                // Task disappeared — assume completed
-                                let observed_duration = self
-                                    .task_first_seen
-                                    .get(&key)
-                                    .map(|first| first.elapsed())
-                                    .unwrap_or_default();
-
-                                self.completed_tasks.push_front(CompletedTaskRecord {
-                                    query_id: sk.query_id.clone(),
-                                    stage_id: sk.stage_id,
-                                    task_number: sk.task_number,
-                                    observed_duration,
-                                });
-
-                                // Maintain bounded size
-                                while self.completed_tasks.len() > MAX_COMPLETED_TASKS {
-                                    self.completed_tasks.pop_back();
-                                }
-
-                                // Remove from first_seen tracking
-                                self.task_first_seen.remove(&key);
+                    if old_task.status == TaskStatus::Running as i32
+                        && let Some(sk) = &old_task.task_key
+                    {
+                        let key = (sk.query_id.clone(), sk.stage_id, sk.task_number);
+                        if !new_task_keys.contains(&key) {
+                            // Task disappeared — assume completed
+                            let observed_duration = self
+                                .task_first_seen
+                                .get(&key)
+                                .map(|first| first.elapsed())
+                                .unwrap_or_default();
+
+                            self.completed_tasks.push_front(CompletedTaskRecord {
+                                query_id: sk.query_id.clone(),
+                                stage_id: sk.stage_id,
+                                task_number: sk.task_number,
+                                observed_duration,
+                            });
+
+                            // Maintain bounded size
+                            while self.completed_tasks.len() > MAX_COMPLETED_TASKS {
+                                self.completed_tasks.pop_back();
                             }
+
+                            // Remove from first_seen tracking
+                            self.task_first_seen.remove(&key);
                         }
                     }
                 }
diff --git a/src/coordinator/metrics_store.rs b/src/coordinator/metrics_store.rs
index 8ccc1f7c..ed55db2c 100644
--- a/src/coordinator/metrics_store.rs
+++ b/src/coordinator/metrics_store.rs
@@ -27,7 +27,7 @@ impl MetricsStore {
         self.rx.borrow().get(key).cloned()
     }
 
-    #[cfg(test)]
+    #[cfg(all(test, feature = "grpc"))]
     pub(crate) fn from_entries(entries: impl IntoIterator<Item = (TaskKey, TaskMetrics)>) -> Self {
         let map: HashMap<_, _> = entries.into_iter().collect();
         let (tx, rx) = watch::channel(map);
diff --git a/src/execution_plans/metrics.rs b/src/execution_plans/metrics.rs
index 0fefdf3e..de97ed04 100644
--- a/src/execution_plans/metrics.rs
+++ b/src/execution_plans/metrics.rs
@@ -21,7 +21,7 @@ impl MetricsWrapperExec {
         Self { inner, metrics }
     }
 
-    #[cfg(test)]
+    #[cfg(all(test, feature = "grpc"))]
     pub(crate) fn inner(&self) -> &Arc<dyn ExecutionPlan> {
         &self.inner
     }
diff --git a/src/execution_plans/mod.rs b/src/execution_plans/mod.rs
index ecdcbc35..97904ef1 100644
--- a/src/execution_plans/mod.rs
+++ b/src/execution_plans/mod.rs
@@ -8,7 +8,9 @@ mod network_coalesce;
 mod network_shuffle;
 mod sampler;
 
-#[cfg(any(test, feature = "integration"))]
+// The benchmark fixtures spin up real gRPC workers (`tonic` servers, Flight channels), so they
+// only exist when that transport is compiled in.
+#[cfg(all(feature = "grpc", any(test, feature = "integration")))]
 pub mod benchmarks;
 
 pub use broadcast::BroadcastExec;
diff --git a/src/lib.rs b/src/lib.rs
index b132caf5..f7becd63 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -46,14 +46,19 @@ pub mod test_utils;
 #[cfg(feature = "grpc")]
 pub use protocol::grpc;
 
+/// The worker-protocol prost message types, independent of any transport. A non-gRPC transport
+/// reaches for these to speak the same wire shape the gRPC path serializes. Unstable: this
+/// mirrors `worker.proto`, which is regenerated freely.
+pub use protocol::generated::worker as proto;
+
 pub use codec::DistributedCodec;
 pub use worker_resolver::{WorkerResolver, get_distributed_worker_resolver};
 
 pub use protocol::{
     ChannelResolver, CoordinatorToWorkerMsg, ExecuteTaskRequest, GetWorkerInfoRequest,
-    GetWorkerInfoResponse, LoadInfo, ProducerHeadSpec, SetPlanRequest, TaskKey, TaskMetrics,
-    WorkUnitBatch, WorkUnitFeedDeclaration, WorkUnitMsg, WorkerChannel, WorkerToCoordinatorMsg,
-    get_distributed_channel_resolver,
+    GetWorkerInfoResponse, InProcessChannelResolver, LoadInfo, ProducerHeadSpec, SetPlanRequest,
+    TaskKey, TaskMetrics, WorkUnitBatch, WorkUnitFeedDeclaration, WorkUnitMsg, WorkerChannel,
+    WorkerToCoordinatorMsg, get_distributed_channel_resolver,
 };
 pub use stage::{
     DistributedTaskContext, Stage, display_plan_ascii, display_plan_graphviz, explain_analyze,
@@ -66,7 +71,7 @@ pub use worker::{
     Worker, WorkerQueryContext, WorkerSessionBuilder,
 };
 
-#[cfg(any(feature = "integration", test))]
+#[cfg(all(feature = "grpc", any(feature = "integration", test)))]
 pub use execution_plans::benchmarks::{
     LocalRepartitionBench, LocalRepartitionFixture, LocalRepartitionMode, ShuffleBench,
     ShuffleFixture, TransportBench, TransportBenchMode, TransportFixture,
diff --git a/src/metrics/task_metrics_collector.rs b/src/metrics/task_metrics_collector.rs
index ba6e6855..f0647b95 100644
--- a/src/metrics/task_metrics_collector.rs
+++ b/src/metrics/task_metrics_collector.rs
@@ -19,7 +19,8 @@ pub fn collect_plan_metrics(plan: &Arc<dyn ExecutionPlan>) -> Result<Vec<Metrics
     Ok(metrics)
 }
 
-#[cfg(test)]
+// These tests execute over the in-memory gRPC transport, so they need that transport compiled in.
+#[cfg(all(test, feature = "grpc"))]
 mod tests {
 
     use super::*;
diff --git a/src/metrics/task_metrics_rewriter.rs b/src/metrics/task_metrics_rewriter.rs
index 4e62cfd8..a683bfa3 100644
--- a/src/metrics/task_metrics_rewriter.rs
+++ b/src/metrics/task_metrics_rewriter.rs
@@ -280,7 +280,8 @@ pub fn stage_metrics_rewriter(
         .map(|v| v.data)
 }
 
-#[cfg(test)]
+// These tests execute over the in-memory gRPC transport, so they need that transport compiled in.
+#[cfg(all(test, feature = "grpc"))]
 mod tests {
     use crate::DistributedExt;
     use crate::coordinator::MetricsStore;
diff --git a/src/protocol/channel_resolver.rs b/src/protocol/channel_resolver.rs
index e781aa90..05da3069 100644
--- a/src/protocol/channel_resolver.rs
+++ b/src/protocol/channel_resolver.rs
@@ -70,6 +70,9 @@ pub fn get_distributed_channel_resolver(
 
     #[cfg(not(feature = "grpc"))]
     {
+        // With `grpc` off there is no built-in default. A co-located deployment can register
+        // [`crate::InProcessChannelResolver`] (or another transport) via
+        // `with_distributed_channel_resolver`.
         panic!(
             "gRPC feature is not enabled, and no channel resolver was provided, so no default ChannelResolver can be provided"
         );
diff --git a/src/protocol/generated/mod.rs b/src/protocol/generated/mod.rs
new file mode 100644
index 00000000..a787a5fc
--- /dev/null
+++ b/src/protocol/generated/mod.rs
@@ -0,0 +1,5 @@
+pub mod worker;
+
+#[cfg(feature = "grpc")]
+#[allow(clippy::all)]
+pub mod worker_grpc;
diff --git a/src/protocol/generated/worker.rs b/src/protocol/generated/worker.rs
new file mode 100644
index 00000000..7e70450c
--- /dev/null
+++ b/src/protocol/generated/worker.rs
@@ -0,0 +1,468 @@
+// This file is @generated by prost-build.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct CoordinatorToWorkerMsg {
+    #[prost(oneof = "coordinator_to_worker_msg::Inner", tags = "1, 2, 3")]
+    pub inner: ::core::option::Option<coordinator_to_worker_msg::Inner>,
+}
+/// Nested message and enum types in `CoordinatorToWorkerMsg`.
+pub mod coordinator_to_worker_msg {
+    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum Inner {
+        /// Sends a subplan to a worker so that a future ExecuteTask call can actually execute it.
+        /// The plan is identified by a TaskKey.
+        #[prost(message, tag = "1")]
+        SetPlanRequest(super::SetPlanRequest),
+        /// A batch of messages from a work unit feed belonging to different partitions from one node from the plan set in
+        /// set_plan_request. A work unit feed is a per-partition stream of information that tells the node what should
+        /// be executed within a partition, for example, a stream of file addresses that should be read.
+        #[prost(message, tag = "2")]
+        WorkUnitBatch(super::WorkUnitBatch),
+        /// Signals an EOS for WorkUnits. After this message is received, no more WorkUnits will be sent.
+        #[prost(bool, tag = "3")]
+        WorkUnitEos(bool),
+    }
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WorkerToCoordinatorMsg {
+    #[prost(oneof = "worker_to_coordinator_msg::Inner", tags = "1, 2, 3")]
+    pub inner: ::core::option::Option<worker_to_coordinator_msg::Inner>,
+}
+/// Nested message and enum types in `WorkerToCoordinatorMsg`.
+pub mod worker_to_coordinator_msg {
+    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum Inner {
+        /// Sends the metrics collected during task execution back to the coordinator.
+        /// This is sent after all partitions of a task have finished (or been dropped),
+        /// ensuring metrics are never lost due to early stream termination.
+        /// metrics\[i\] is the set of metrics for plan node i in pre-order traversal order.
+        #[prost(message, tag = "1")]
+        TaskMetrics(super::TaskMetrics),
+        /// Load information reported by a task. This information is used for dynamically
+        /// sizing the number of workers involved in a query.
+        #[prost(message, tag = "2")]
+        LoadInfo(super::LoadInfo),
+        #[prost(bool, tag = "3")]
+        LoadInfoEos(bool),
+    }
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TaskMetrics {
+    /// Metrics for a single task's plan nodes in pre-order traversal order.
+    /// The TaskKey is implicit — it is determined by the SetPlanRequest that
+    /// opened this coordinator channel connection.
+    #[prost(message, repeated, tag = "1")]
+    pub pre_order_plan_metrics: ::prost::alloc::vec::Vec<MetricsSet>,
+    /// Metrics related to the execution of a task within a stage. This metrics, instead of being
+    /// associated to a specific node, they are global to the task, like the time at which the plan
+    /// was fed by the coordinator to the worker.
+    #[prost(message, optional, tag = "2")]
+    pub task_metrics: ::core::option::Option<MetricsSet>,
+}
+/// Load information reported for a specific partition with information about this
+/// amount of data flowing through the plan.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct LoadInfo {
+    /// The partition index to which this message belongs to.
+    #[prost(uint64, tag = "1")]
+    pub partition: u64,
+    /// The amount of rows ready to be returned.
+    #[prost(uint64, tag = "2")]
+    pub rows_ready: u64,
+    /// The amount of bytes ready to be returned per column.
+    #[prost(uint64, repeated, tag = "4")]
+    pub per_column_bytes_ready: ::prost::alloc::vec::Vec<u64>,
+    /// Approximate ratio of NDV for each column.
+    #[prost(float, repeated, tag = "6")]
+    pub per_column_ndv_percentage: ::prost::alloc::vec::Vec<f32>,
+    /// Approximate ratio of null count for each column.
+    #[prost(float, repeated, tag = "7")]
+    pub per_column_null_percentage: ::prost::alloc::vec::Vec<f32>,
+    /// The amount of rows that were pulled from leaf nodes while this partition was sampling data.
+    #[prost(uint64, tag = "8")]
+    pub rows_pulled_from_leaf: u64,
+    /// Whether the sampled partition stream reached end-of-stream (i.e. the partition finished
+    /// producing all of its output) by the time this LoadInfo was captured.
+    #[prost(bool, tag = "9")]
+    pub reached_eos: bool,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetWorkerInfoRequest {}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetWorkerInfoResponse {
+    #[prost(string, tag = "1")]
+    pub version: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SetPlanRequest {
+    /// The unique identifier of the task to which the subplan belongs to.
+    #[prost(message, optional, tag = "1")]
+    pub task_key: ::core::option::Option<TaskKey>,
+    /// The amount of tasks that share the same subplan. Necessary for building the DistributedTaskContext during execution.
+    #[prost(uint64, tag = "2")]
+    pub task_count: u64,
+    /// The serialized subplan the worker is expected to execute on an ExecuteTask gRPC call.
+    #[prost(bytes = "vec", tag = "3")]
+    pub plan_proto: ::prost::alloc::vec::Vec<u8>,
+    /// Information about all the work unit feeds that will be streamed from coordinator to worker.
+    /// This information is needed here because at the moment of setting the plan, all the appropriate
+    /// channels for the incoming work unit feeds need to be constructed.
+    ///
+    /// If no WorkUnitFeedExec nodes are present in the plan, this should be empty.
+    #[prost(message, repeated, tag = "4")]
+    pub work_unit_feed_declarations:
+        ::prost::alloc::vec::Vec<set_plan_request::WorkUnitFeedDeclaration>,
+    /// The worker URL to which this message will go. The receiving worker will use this information to identify
+    /// itself, and avoid further gRPC calls in case it needs to call itself for executing remote tasks.
+    #[prost(string, tag = "5")]
+    pub target_worker_url: ::prost::alloc::string::String,
+    /// Unix nanos when the query started as reported by the coordinator. Used for collecting temporal metrics
+    /// relative to when the query was fired in the coordinator.
+    #[prost(uint64, tag = "6")]
+    pub query_start_time_ns: u64,
+}
+/// Nested message and enum types in `SetPlanRequest`.
+pub mod set_plan_request {
+    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+    pub struct WorkUnitFeedDeclaration {
+        /// Unique identifier of the node to which work unit feeds are expected to be streamed.
+        #[prost(bytes = "vec", tag = "1")]
+        pub id: ::prost::alloc::vec::Vec<u8>,
+        /// The amount of partitions expected to be streamed.
+        #[prost(uint64, tag = "2")]
+        pub partitions: u64,
+    }
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WorkUnitBatch {
+    /// A batch of WorkUnits.
+    #[prost(message, repeated, tag = "1")]
+    pub batch: ::prost::alloc::vec::Vec<WorkUnit>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct WorkUnit {
+    /// Identifier of the node to which this work unit feed belongs to.
+    #[prost(bytes = "vec", tag = "1")]
+    pub id: ::prost::alloc::vec::Vec<u8>,
+    /// The partition index within the node to which the work unit feed belongs to.
+    #[prost(uint64, tag = "2")]
+    pub partition: u64,
+    /// Arbitrary user-defined data (e.g., a file address) necessary during execution.
+    #[prost(bytes = "vec", tag = "3")]
+    pub body: ::prost::alloc::vec::Vec<u8>,
+    /// Unix timestamp in nanoseconds at which this message was created in the coordinator.
+    #[prost(uint64, tag = "4")]
+    pub created_timestamp_unix_nanos: u64,
+    /// Unix timestamp in nanoseconds at which this message was sent by the coordinator.
+    #[prost(uint64, tag = "5")]
+    pub sent_timestamp_unix_nanos: u64,
+    /// Unix timestamp in nanoseconds at which this message was received by a worker.
+    #[prost(uint64, tag = "6")]
+    pub received_timestamp_unix_nanos: u64,
+    /// Unix timestamp in nanoseconds at which this message started being processed.
+    #[prost(uint64, tag = "7")]
+    pub processed_timestamp_unix_nanos: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ExecuteTaskRequest {
+    /// The unique identifier of the task that is going to get executed.
+    #[prost(message, optional, tag = "1")]
+    pub task_key: ::core::option::Option<TaskKey>,
+    /// The start of the partition range of the specified task that is going to be executed.
+    #[prost(uint64, tag = "2")]
+    pub target_partition_start: u64,
+    /// The end of the partition range of the specified task that is going to be executed.
+    #[prost(uint64, tag = "3")]
+    pub target_partition_end: u64,
+    /// The head node the requested task should have. Depending on the network boundary executing
+    /// the task, the head node should be prepared differently, for example:
+    ///
+    /// * A RepartitionExecHead implies a RepartitionExec at the head of the task.
+    /// * A BroadcastExecHead implies a BroadcastExec at the head of the task.
+    /// * A NoneHead does not need any specific head.
+    #[prost(oneof = "execute_task_request::ProducerHead", tags = "6, 7, 8")]
+    pub producer_head: ::core::option::Option<execute_task_request::ProducerHead>,
+}
+/// Nested message and enum types in `ExecuteTaskRequest`.
+pub mod execute_task_request {
+    /// The head node the requested task should have. Depending on the network boundary executing
+    /// the task, the head node should be prepared differently, for example:
+    ///
+    /// * A RepartitionExecHead implies a RepartitionExec at the head of the task.
+    /// * A BroadcastExecHead implies a BroadcastExec at the head of the task.
+    /// * A NoneHead does not need any specific head.
+    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
+    pub enum ProducerHead {
+        /// The boundary executing the task is a NetworkCoalesceExec.
+        #[prost(message, tag = "6")]
+        None(super::NoneHead),
+        /// The boundary executing the task is a NetworkBroadcastExec.
+        #[prost(message, tag = "7")]
+        Broadcast(super::BroadcastExecHead),
+        /// The boundary executing the task is a NetworkShuffleExec.
+        #[prost(message, tag = "8")]
+        Repartition(super::RepartitionExecHead),
+    }
+}
+/// Head needed by a NetworkCoalesceExec.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NoneHead {}
+/// Head needed by a NetworkBroadcastExec.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct BroadcastExecHead {
+    /// The amount of output partitions
+    #[prost(uint64, tag = "1")]
+    pub output_partitions: u64,
+}
+/// Head needed by a NetworkShuffleExec.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct RepartitionExecHead {
+    /// `Partitioning` message from datafusion.proto
+    #[prost(bytes = "vec", tag = "1")]
+    pub partitioning: ::prost::alloc::vec::Vec<u8>,
+}
+/// A key that uniquely identifies a task in a query.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct TaskKey {
+    /// Our query id.
+    #[prost(bytes = "vec", tag = "1")]
+    pub query_id: ::prost::alloc::vec::Vec<u8>,
+    /// Our stage id.
+    #[prost(uint64, tag = "2")]
+    pub stage_id: u64,
+    /// The task number within the stage.
+    #[prost(uint64, tag = "3")]
+    pub task_number: u64,
+}
+/// FlightAppMetadata represents all types of app_metadata which we use in the distributed execution.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct FlightAppMetadata {
+    #[prost(uint64, tag = "1")]
+    pub partition: u64,
+    /// Unix timestamp in nanoseconds at which this message was created.
+    #[prost(uint64, tag = "2")]
+    pub created_timestamp_unix_nanos: u64,
+}
+/// A Label mirrors datafusion::physical_plan::metrics::Label.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct Label {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub value: ::prost::alloc::string::String,
+}
+/// A Metric is a protobuf mirror of datafusion::physical_plan::metrics::Metric.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Metric {
+    #[prost(message, repeated, tag = "1")]
+    pub labels: ::prost::alloc::vec::Vec<Label>,
+    #[prost(uint64, optional, tag = "2")]
+    pub partition: ::core::option::Option<u64>,
+    #[prost(
+        oneof = "metric::Value",
+        tags = "10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34"
+    )]
+    pub value: ::core::option::Option<metric::Value>,
+}
+/// Nested message and enum types in `Metric`.
+pub mod metric {
+    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
+    pub enum Value {
+        #[prost(message, tag = "10")]
+        OutputRows(super::OutputRows),
+        #[prost(message, tag = "11")]
+        ElapsedCompute(super::ElapsedCompute),
+        #[prost(message, tag = "12")]
+        SpillCount(super::SpillCount),
+        #[prost(message, tag = "13")]
+        SpilledBytes(super::SpilledBytes),
+        #[prost(message, tag = "14")]
+        SpilledRows(super::SpilledRows),
+        #[prost(message, tag = "15")]
+        CurrentMemoryUsage(super::CurrentMemoryUsage),
+        #[prost(message, tag = "16")]
+        Count(super::NamedCount),
+        #[prost(message, tag = "17")]
+        Gauge(super::NamedGauge),
+        #[prost(message, tag = "18")]
+        Time(super::NamedTime),
+        #[prost(message, tag = "19")]
+        StartTimestamp(super::StartTimestamp),
+        #[prost(message, tag = "20")]
+        EndTimestamp(super::EndTimestamp),
+        #[prost(message, tag = "21")]
+        OutputBytes(super::OutputBytes),
+        #[prost(message, tag = "22")]
+        OutputBatches(super::OutputBatches),
+        #[prost(message, tag = "23")]
+        PruningMetrics(super::NamedPruningMetrics),
+        #[prost(message, tag = "24")]
+        Ratio(super::NamedRatio),
+        #[prost(message, tag = "25")]
+        CustomMinLatency(super::MinLatency),
+        #[prost(message, tag = "26")]
+        CustomMaxLatency(super::MaxLatency),
+        #[prost(message, tag = "27")]
+        CustomAvgLatency(super::AvgLatency),
+        #[prost(message, tag = "28")]
+        CustomFirstLatency(super::FirstLatency),
+        #[prost(message, tag = "29")]
+        CustomBytesCount(super::BytesCount),
+        #[prost(message, tag = "30")]
+        CustomP50Latency(super::PercentileLatency),
+        #[prost(message, tag = "31")]
+        CustomP75Latency(super::PercentileLatency),
+        #[prost(message, tag = "32")]
+        CustomP95Latency(super::PercentileLatency),
+        #[prost(message, tag = "33")]
+        CustomP99Latency(super::PercentileLatency),
+        #[prost(message, tag = "34")]
+        CustomMaxGauge(super::MaxGauge),
+    }
+}
+/// A MetricsSet is a protobuf mirror of datafusion::physical_plan::metrics::MetricsSet. It represents
+/// a collection of metrics for one ExecutionPlan node.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MetricsSet {
+    #[prost(message, repeated, tag = "1")]
+    pub metrics: ::prost::alloc::vec::Vec<Metric>,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct OutputRows {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ElapsedCompute {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SpillCount {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SpilledBytes {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SpilledRows {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct CurrentMemoryUsage {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NamedCount {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NamedGauge {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NamedTime {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct StartTimestamp {
+    #[prost(int64, optional, tag = "1")]
+    pub value: ::core::option::Option<i64>,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct EndTimestamp {
+    #[prost(int64, optional, tag = "1")]
+    pub value: ::core::option::Option<i64>,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct OutputBytes {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct OutputBatches {
+    #[prost(uint64, tag = "1")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NamedPruningMetrics {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub pruned: u64,
+    #[prost(uint64, tag = "3")]
+    pub matched: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NamedRatio {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub part: u64,
+    #[prost(uint64, tag = "3")]
+    pub total: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct BytesCount {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct MinLatency {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct MaxLatency {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct AvgLatency {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub nanos_sum: u64,
+    #[prost(uint64, tag = "3")]
+    pub count: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct FirstLatency {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct PercentileLatency {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "4")]
+    pub sketch_bytes: ::prost::alloc::vec::Vec<u8>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct MaxGauge {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "2")]
+    pub value: u64,
+}
diff --git a/src/protocol/generated/worker_grpc.rs b/src/protocol/generated/worker_grpc.rs
new file mode 100644
index 00000000..ba955c27
--- /dev/null
+++ b/src/protocol/generated/worker_grpc.rs
@@ -0,0 +1,447 @@
+// This file is @generated by prost-build.
+/// Generated client implementations.
+pub mod worker_service_client {
+    #![allow(
+        unused_variables,
+        dead_code,
+        missing_docs,
+        clippy::wildcard_imports,
+        clippy::let_unit_value
+    )]
+    use tonic::codegen::http::Uri;
+    use tonic::codegen::*;
+    #[derive(Debug, Clone)]
+    pub struct WorkerServiceClient<T> {
+        inner: tonic::client::Grpc<T>,
+    }
+    impl WorkerServiceClient<tonic::transport::Channel> {
+        /// Attempt to create a new client by connecting to a given endpoint.
+        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+        where
+            D: TryInto<tonic::transport::Endpoint>,
+            D::Error: Into<StdError>,
+        {
+            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+            Ok(Self::new(conn))
+        }
+    }
+    impl<T> WorkerServiceClient<T>
+    where
+        T: tonic::client::GrpcService<tonic::body::Body>,
+        T::Error: Into<StdError>,
+        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
+        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
+    {
+        pub fn new(inner: T) -> Self {
+            let inner = tonic::client::Grpc::new(inner);
+            Self { inner }
+        }
+        pub fn with_origin(inner: T, origin: Uri) -> Self {
+            let inner = tonic::client::Grpc::with_origin(inner, origin);
+            Self { inner }
+        }
+        pub fn with_interceptor<F>(
+            inner: T,
+            interceptor: F,
+        ) -> WorkerServiceClient<InterceptedService<T, F>>
+        where
+            F: tonic::service::Interceptor,
+            T::ResponseBody: Default,
+            T: tonic::codegen::Service<
+                    http::Request<tonic::body::Body>,
+                    Response = http::Response<
+                        <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
+                    >,
+                >,
+            <T as tonic::codegen::Service<http::Request<tonic::body::Body>>>::Error:
+                Into<StdError> + std::marker::Send + std::marker::Sync,
+        {
+            WorkerServiceClient::new(InterceptedService::new(inner, interceptor))
+        }
+        /// Compress requests with the given encoding.
+        ///
+        /// This requires the server to support it otherwise it might respond with an
+        /// error.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.send_compressed(encoding);
+            self
+        }
+        /// Enable decompressing responses.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.accept_compressed(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_decoding_message_size(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_encoding_message_size(limit);
+            self
+        }
+        /// Establishes a bidirectional message stream between a coordinator and a worker, over which messages
+        /// will be exchanged at any time during a query's lifetime. It's expected to be one coordinator channel
+        /// per task.
+        pub async fn coordinator_channel(
+            &mut self,
+            request: impl tonic::IntoStreamingRequest<
+                Message = super::super::worker::CoordinatorToWorkerMsg,
+            >,
+        ) -> std::result::Result<
+            tonic::Response<tonic::codec::Streaming<super::super::worker::WorkerToCoordinatorMsg>>,
+            tonic::Status,
+        > {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/worker_grpc.WorkerService/CoordinatorChannel",
+            );
+            let mut req = request.into_streaming_request();
+            req.extensions_mut().insert(GrpcMethod::new(
+                "worker_grpc.WorkerService",
+                "CoordinatorChannel",
+            ));
+            self.inner.streaming(req, path, codec).await
+        }
+        /// Executes the requested partition range of a subplan previously sent by the coordinator channel.
+        pub async fn execute_task(
+            &mut self,
+            request: impl tonic::IntoRequest<super::super::worker::ExecuteTaskRequest>,
+        ) -> std::result::Result<
+            tonic::Response<tonic::codec::Streaming<::arrow_flight::FlightData>>,
+            tonic::Status,
+        > {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path =
+                http::uri::PathAndQuery::from_static("/worker_grpc.WorkerService/ExecuteTask");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("worker_grpc.WorkerService", "ExecuteTask"));
+            self.inner.server_streaming(req, path, codec).await
+        }
+        /// Returns metadata about a worker. Currently only used for worker versioning.
+        pub async fn get_worker_info(
+            &mut self,
+            request: impl tonic::IntoRequest<super::super::worker::GetWorkerInfoRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::super::worker::GetWorkerInfoResponse>,
+            tonic::Status,
+        > {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path =
+                http::uri::PathAndQuery::from_static("/worker_grpc.WorkerService/GetWorkerInfo");
+            let mut req = request.into_request();
+            req.extensions_mut().insert(GrpcMethod::new(
+                "worker_grpc.WorkerService",
+                "GetWorkerInfo",
+            ));
+            self.inner.unary(req, path, codec).await
+        }
+    }
+}
+/// Generated server implementations.
+pub mod worker_service_server {
+    #![allow(
+        unused_variables,
+        dead_code,
+        missing_docs,
+        clippy::wildcard_imports,
+        clippy::let_unit_value
+    )]
+    use tonic::codegen::*;
+    /// Generated trait containing gRPC methods that should be implemented for use with WorkerServiceServer.
+    #[async_trait]
+    pub trait WorkerService: std::marker::Send + std::marker::Sync + 'static {
+        /// Server streaming response type for the CoordinatorChannel method.
+        type CoordinatorChannelStream: tonic::codegen::tokio_stream::Stream<
+                Item = std::result::Result<
+                    super::super::worker::WorkerToCoordinatorMsg,
+                    tonic::Status,
+                >,
+            > + std::marker::Send
+            + 'static;
+        /// Establishes a bidirectional message stream between a coordinator and a worker, over which messages
+        /// will be exchanged at any time during a query's lifetime. It's expected to be one coordinator channel
+        /// per task.
+        async fn coordinator_channel(
+            &self,
+            request: tonic::Request<tonic::Streaming<super::super::worker::CoordinatorToWorkerMsg>>,
+        ) -> std::result::Result<tonic::Response<Self::CoordinatorChannelStream>, tonic::Status>;
+        /// Server streaming response type for the ExecuteTask method.
+        type ExecuteTaskStream: tonic::codegen::tokio_stream::Stream<
+                Item = std::result::Result<::arrow_flight::FlightData, tonic::Status>,
+            > + std::marker::Send
+            + 'static;
+        /// Executes the requested partition range of a subplan previously sent by the coordinator channel.
+        async fn execute_task(
+            &self,
+            request: tonic::Request<super::super::worker::ExecuteTaskRequest>,
+        ) -> std::result::Result<tonic::Response<Self::ExecuteTaskStream>, tonic::Status>;
+        /// Returns metadata about a worker. Currently only used for worker versioning.
+        async fn get_worker_info(
+            &self,
+            request: tonic::Request<super::super::worker::GetWorkerInfoRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::super::worker::GetWorkerInfoResponse>,
+            tonic::Status,
+        >;
+    }
+    #[derive(Debug)]
+    pub struct WorkerServiceServer<T> {
+        inner: Arc<T>,
+        accept_compression_encodings: EnabledCompressionEncodings,
+        send_compression_encodings: EnabledCompressionEncodings,
+        max_decoding_message_size: Option<usize>,
+        max_encoding_message_size: Option<usize>,
+    }
+    impl<T> WorkerServiceServer<T> {
+        pub fn new(inner: T) -> Self {
+            Self::from_arc(Arc::new(inner))
+        }
+        pub fn from_arc(inner: Arc<T>) -> Self {
+            Self {
+                inner,
+                accept_compression_encodings: Default::default(),
+                send_compression_encodings: Default::default(),
+                max_decoding_message_size: None,
+                max_encoding_message_size: None,
+            }
+        }
+        pub fn with_interceptor<F>(inner: T, interceptor: F) -> InterceptedService<Self, F>
+        where
+            F: tonic::service::Interceptor,
+        {
+            InterceptedService::new(Self::new(inner), interceptor)
+        }
+        /// Enable decompressing requests with the given encoding.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.accept_compression_encodings.enable(encoding);
+            self
+        }
+        /// Compress responses with the given encoding, if the client supports it.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.send_compression_encodings.enable(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.max_decoding_message_size = Some(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.max_encoding_message_size = Some(limit);
+            self
+        }
+    }
+    impl<T, B> tonic::codegen::Service<http::Request<B>> for WorkerServiceServer<T>
+    where
+        T: WorkerService,
+        B: Body + std::marker::Send + 'static,
+        B::Error: Into<StdError> + std::marker::Send + 'static,
+    {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = std::convert::Infallible;
+        type Future = BoxFuture<Self::Response, Self::Error>;
+        fn poll_ready(
+            &mut self,
+            _cx: &mut Context<'_>,
+        ) -> Poll<std::result::Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+        fn call(&mut self, req: http::Request<B>) -> Self::Future {
+            match req.uri().path() {
+                "/worker_grpc.WorkerService/CoordinatorChannel" => {
+                    #[allow(non_camel_case_types)]
+                    struct CoordinatorChannelSvc<T: WorkerService>(pub Arc<T>);
+                    impl<T: WorkerService>
+                        tonic::server::StreamingService<
+                            super::super::worker::CoordinatorToWorkerMsg,
+                        > for CoordinatorChannelSvc<T>
+                    {
+                        type Response = super::super::worker::WorkerToCoordinatorMsg;
+                        type ResponseStream = T::CoordinatorChannelStream;
+                        type Future =
+                            BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<
+                                tonic::Streaming<super::super::worker::CoordinatorToWorkerMsg>,
+                            >,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as WorkerService>::coordinator_channel(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = CoordinatorChannelSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.streaming(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/worker_grpc.WorkerService/ExecuteTask" => {
+                    #[allow(non_camel_case_types)]
+                    struct ExecuteTaskSvc<T: WorkerService>(pub Arc<T>);
+                    impl<T: WorkerService>
+                        tonic::server::ServerStreamingService<
+                            super::super::worker::ExecuteTaskRequest,
+                        > for ExecuteTaskSvc<T>
+                    {
+                        type Response = ::arrow_flight::FlightData;
+                        type ResponseStream = T::ExecuteTaskStream;
+                        type Future =
+                            BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::super::worker::ExecuteTaskRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as WorkerService>::execute_task(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ExecuteTaskSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.server_streaming(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/worker_grpc.WorkerService/GetWorkerInfo" => {
+                    #[allow(non_camel_case_types)]
+                    struct GetWorkerInfoSvc<T: WorkerService>(pub Arc<T>);
+                    impl<T: WorkerService>
+                        tonic::server::UnaryService<super::super::worker::GetWorkerInfoRequest>
+                        for GetWorkerInfoSvc<T>
+                    {
+                        type Response = super::super::worker::GetWorkerInfoResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::super::worker::GetWorkerInfoRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as WorkerService>::get_worker_info(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = GetWorkerInfoSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                _ => Box::pin(async move {
+                    let mut response = http::Response::new(tonic::body::Body::default());
+                    let headers = response.headers_mut();
+                    headers.insert(
+                        tonic::Status::GRPC_STATUS,
+                        (tonic::Code::Unimplemented as i32).into(),
+                    );
+                    headers.insert(
+                        http::header::CONTENT_TYPE,
+                        tonic::metadata::GRPC_CONTENT_TYPE,
+                    );
+                    Ok(response)
+                }),
+            }
+        }
+    }
+    impl<T> Clone for WorkerServiceServer<T> {
+        fn clone(&self) -> Self {
+            let inner = self.inner.clone();
+            Self {
+                inner,
+                accept_compression_encodings: self.accept_compression_encodings,
+                send_compression_encodings: self.send_compression_encodings,
+                max_decoding_message_size: self.max_decoding_message_size,
+                max_encoding_message_size: self.max_encoding_message_size,
+            }
+        }
+    }
+    /// Generated gRPC service name
+    pub const SERVICE_NAME: &str = "worker_grpc.WorkerService";
+    impl<T> tonic::server::NamedService for WorkerServiceServer<T> {
+        const NAME: &'static str = SERVICE_NAME;
+    }
+}
diff --git a/src/protocol/grpc/gen/src/main.rs b/src/protocol/grpc/gen/src/main.rs
index 7505aae5..9c6addbe 100644
--- a/src/protocol/grpc/gen/src/main.rs
+++ b/src/protocol/grpc/gen/src/main.rs
@@ -4,15 +4,17 @@ use std::fs;
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let repo_root = env::current_dir()?;
 
-    let proto_dir = repo_root.join("src/protocol/grpc");
-    let proto_file = proto_dir.join("worker.proto");
-    let out_dir = repo_root.join("src/protocol/grpc/generated");
+    let protocol_dir = repo_root.join("src/protocol");
+    let messages_proto = protocol_dir.join("messages.proto");
+    let worker_service_proto = protocol_dir.join("grpc/worker_service.proto");
+    let out_dir = repo_root.join("src/protocol/generated");
 
     fs::create_dir_all(&out_dir)?;
 
     println!("Generating protobuf code...");
-    println!("Proto dir: {proto_dir:?}");
-    println!("Proto file: {proto_file:?}");
+    println!("Protocol dir: {protocol_dir:?}");
+    println!("Messages proto: {messages_proto:?}");
+    println!("Worker service proto: {worker_service_proto:?}");
     println!("Output dir: {out_dir:?}");
 
     tonic_prost_build::configure()
@@ -24,7 +26,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             ".worker.FlightDescriptor",
             "::arrow_flight::FlightDescriptor",
         )
-        .compile_protos(&[proto_file], &[proto_dir])?;
+        .compile_protos(&[messages_proto, worker_service_proto], &[protocol_dir])?;
 
     println!("Successfully generated worker proto code");
 
diff --git a/src/protocol/grpc/generated/mod.rs b/src/protocol/grpc/generated/mod.rs
deleted file mode 100644
index 844c269c..00000000
--- a/src/protocol/grpc/generated/mod.rs
+++ /dev/null
@@ -1 +0,0 @@
-pub(crate) mod worker;
diff --git a/src/protocol/grpc/generated/worker.rs b/src/protocol/grpc/generated/worker.rs
deleted file mode 100644
index d1f0fb4a..00000000
--- a/src/protocol/grpc/generated/worker.rs
+++ /dev/null
@@ -1,896 +0,0 @@
-// This file is @generated by prost-build.
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct CoordinatorToWorkerMsg {
-    #[prost(oneof = "coordinator_to_worker_msg::Inner", tags = "1, 2, 3")]
-    pub inner: ::core::option::Option<coordinator_to_worker_msg::Inner>,
-}
-/// Nested message and enum types in `CoordinatorToWorkerMsg`.
-pub mod coordinator_to_worker_msg {
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
-    pub enum Inner {
-        /// Sends a subplan to a worker so that a future ExecuteTask call can actually execute it.
-        /// The plan is identified by a TaskKey.
-        #[prost(message, tag = "1")]
-        SetPlanRequest(super::SetPlanRequest),
-        /// A batch of messages from a work unit feed belonging to different partitions from one node from the plan set in
-        /// set_plan_request. A work unit feed is a per-partition stream of information that tells the node what should
-        /// be executed within a partition, for example, a stream of file addresses that should be read.
-        #[prost(message, tag = "2")]
-        WorkUnitBatch(super::WorkUnitBatch),
-        /// Signals an EOS for WorkUnits. After this message is received, no more WorkUnits will be sent.
-        #[prost(bool, tag = "3")]
-        WorkUnitEos(bool),
-    }
-}
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct WorkerToCoordinatorMsg {
-    #[prost(oneof = "worker_to_coordinator_msg::Inner", tags = "1, 2, 3")]
-    pub inner: ::core::option::Option<worker_to_coordinator_msg::Inner>,
-}
-/// Nested message and enum types in `WorkerToCoordinatorMsg`.
-pub mod worker_to_coordinator_msg {
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
-    pub enum Inner {
-        /// Sends the metrics collected during task execution back to the coordinator.
-        /// This is sent after all partitions of a task have finished (or been dropped),
-        /// ensuring metrics are never lost due to early stream termination.
-        /// metrics\[i\] is the set of metrics for plan node i in pre-order traversal order.
-        #[prost(message, tag = "1")]
-        TaskMetrics(super::TaskMetrics),
-        /// Load information reported by a task. This information is used for dynamically
-        /// sizing the number of workers involved in a query.
-        #[prost(message, tag = "2")]
-        LoadInfo(super::LoadInfo),
-        #[prost(bool, tag = "3")]
-        LoadInfoEos(bool),
-    }
-}
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct TaskMetrics {
-    /// Metrics for a single task's plan nodes in pre-order traversal order.
-    /// The TaskKey is implicit — it is determined by the SetPlanRequest that
-    /// opened this coordinator channel connection.
-    #[prost(message, repeated, tag = "1")]
-    pub pre_order_plan_metrics: ::prost::alloc::vec::Vec<MetricsSet>,
-    /// Metrics related to the execution of a task within a stage. This metrics, instead of being
-    /// associated to a specific node, they are global to the task, like the time at which the plan
-    /// was fed by the coordinator to the worker.
-    #[prost(message, optional, tag = "2")]
-    pub task_metrics: ::core::option::Option<MetricsSet>,
-}
-/// Load information reported for a specific partition with information about this
-/// amount of data flowing through the plan.
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct LoadInfo {
-    /// The partition index to which this message belongs to.
-    #[prost(uint64, tag = "1")]
-    pub partition: u64,
-    /// The amount of rows ready to be returned.
-    #[prost(uint64, tag = "2")]
-    pub rows_ready: u64,
-    /// The amount of bytes ready to be returned per column.
-    #[prost(uint64, repeated, tag = "4")]
-    pub per_column_bytes_ready: ::prost::alloc::vec::Vec<u64>,
-    /// Approximate ratio of NDV for each column.
-    #[prost(float, repeated, tag = "6")]
-    pub per_column_ndv_percentage: ::prost::alloc::vec::Vec<f32>,
-    /// Approximate ratio of null count for each column.
-    #[prost(float, repeated, tag = "7")]
-    pub per_column_null_percentage: ::prost::alloc::vec::Vec<f32>,
-    /// The amount of rows that were pulled from leaf nodes while this partition was sampling data.
-    #[prost(uint64, tag = "8")]
-    pub rows_pulled_from_leaf: u64,
-    /// Whether the sampled partition stream reached end-of-stream by the time this LoadInfo was
-    /// captured.
-    #[prost(bool, tag = "9")]
-    pub reached_eos: bool,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct GetWorkerInfoRequest {}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct GetWorkerInfoResponse {
-    #[prost(string, tag = "1")]
-    pub version: ::prost::alloc::string::String,
-}
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct SetPlanRequest {
-    /// The unique identifier of the task to which the subplan belongs to.
-    #[prost(message, optional, tag = "1")]
-    pub task_key: ::core::option::Option<TaskKey>,
-    /// The amount of tasks that share the same subplan. Necessary for building the DistributedTaskContext during execution.
-    #[prost(uint64, tag = "2")]
-    pub task_count: u64,
-    /// The serialized subplan the worker is expected to execute on an ExecuteTask gRPC call.
-    #[prost(bytes = "vec", tag = "3")]
-    pub plan_proto: ::prost::alloc::vec::Vec<u8>,
-    /// Information about all the work unit feeds that will be streamed from coordinator to worker.
-    /// This information is needed here because at the moment of setting the plan, all the appropriate
-    /// channels for the incoming work unit feeds need to be constructed.
-    ///
-    /// If no WorkUnitFeedExec nodes are present in the plan, this should be empty.
-    #[prost(message, repeated, tag = "4")]
-    pub work_unit_feed_declarations:
-        ::prost::alloc::vec::Vec<set_plan_request::WorkUnitFeedDeclaration>,
-    /// The worker URL to which this message will go. The receiving worker will use this information to identify
-    /// itself, and avoid further gRPC calls in case it needs to call itself for executing remote tasks.
-    #[prost(string, tag = "5")]
-    pub target_worker_url: ::prost::alloc::string::String,
-    /// Unix nanos when the query started as reported by the coordinator. Used for collecting temporal metrics
-    /// relative to when the query was fired in the coordinator.
-    #[prost(uint64, tag = "6")]
-    pub query_start_time_ns: u64,
-}
-/// Nested message and enum types in `SetPlanRequest`.
-pub mod set_plan_request {
-    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-    pub struct WorkUnitFeedDeclaration {
-        /// Unique identifier of the node to which work unit feeds are expected to be streamed.
-        #[prost(bytes = "vec", tag = "1")]
-        pub id: ::prost::alloc::vec::Vec<u8>,
-        /// The amount of partitions expected to be streamed.
-        #[prost(uint64, tag = "2")]
-        pub partitions: u64,
-    }
-}
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct WorkUnitBatch {
-    /// A batch of WorkUnits.
-    #[prost(message, repeated, tag = "1")]
-    pub batch: ::prost::alloc::vec::Vec<WorkUnit>,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct WorkUnit {
-    /// Identifier of the node to which this work unit feed belongs to.
-    #[prost(bytes = "vec", tag = "1")]
-    pub id: ::prost::alloc::vec::Vec<u8>,
-    /// The partition index within the node to which the work unit feed belongs to.
-    #[prost(uint64, tag = "2")]
-    pub partition: u64,
-    /// Arbitrary user-defined data (e.g., a file address) necessary during execution.
-    #[prost(bytes = "vec", tag = "3")]
-    pub body: ::prost::alloc::vec::Vec<u8>,
-    /// Unix timestamp in nanoseconds at which this message was created in the coordinator.
-    #[prost(uint64, tag = "4")]
-    pub created_timestamp_unix_nanos: u64,
-    /// Unix timestamp in nanoseconds at which this message was sent by the coordinator.
-    #[prost(uint64, tag = "5")]
-    pub sent_timestamp_unix_nanos: u64,
-    /// Unix timestamp in nanoseconds at which this message was received by a worker.
-    #[prost(uint64, tag = "6")]
-    pub received_timestamp_unix_nanos: u64,
-    /// Unix timestamp in nanoseconds at which this message started being processed.
-    #[prost(uint64, tag = "7")]
-    pub processed_timestamp_unix_nanos: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct ExecuteTaskRequest {
-    /// The unique identifier of the task that is going to get executed.
-    #[prost(message, optional, tag = "1")]
-    pub task_key: ::core::option::Option<TaskKey>,
-    /// The start of the partition range of the specified task that is going to be executed.
-    #[prost(uint64, tag = "2")]
-    pub target_partition_start: u64,
-    /// The end of the partition range of the specified task that is going to be executed.
-    #[prost(uint64, tag = "3")]
-    pub target_partition_end: u64,
-    /// The head node the requested task should have. Depending on the network boundary executing
-    /// the task, the head node should be prepared differently, for example:
-    ///
-    /// * A RepartitionExecHead implies a RepartitionExec at the head of the task.
-    /// * A BroadcastExecHead implies a BroadcastExec at the head of the task.
-    /// * A NoneHead does not need any specific head.
-    #[prost(oneof = "execute_task_request::ProducerHead", tags = "6, 7, 8")]
-    pub producer_head: ::core::option::Option<execute_task_request::ProducerHead>,
-}
-/// Nested message and enum types in `ExecuteTaskRequest`.
-pub mod execute_task_request {
-    /// The head node the requested task should have. Depending on the network boundary executing
-    /// the task, the head node should be prepared differently, for example:
-    ///
-    /// * A RepartitionExecHead implies a RepartitionExec at the head of the task.
-    /// * A BroadcastExecHead implies a BroadcastExec at the head of the task.
-    /// * A NoneHead does not need any specific head.
-    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
-    pub enum ProducerHead {
-        /// The boundary executing the task is a NetworkCoalesceExec.
-        #[prost(message, tag = "6")]
-        None(super::NoneHead),
-        /// The boundary executing the task is a NetworkBroadcastExec.
-        #[prost(message, tag = "7")]
-        Broadcast(super::BroadcastExecHead),
-        /// The boundary executing the task is a NetworkShuffleExec.
-        #[prost(message, tag = "8")]
-        Repartition(super::RepartitionExecHead),
-    }
-}
-/// Head needed by a NetworkCoalesceExec.
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct NoneHead {}
-/// Head needed by a NetworkBroadcastExec.
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct BroadcastExecHead {
-    /// The amount of output partitions
-    #[prost(uint64, tag = "1")]
-    pub output_partitions: u64,
-}
-/// Head needed by a NetworkShuffleExec.
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct RepartitionExecHead {
-    /// `Partitioning` message from datafusion.proto
-    #[prost(bytes = "vec", tag = "1")]
-    pub partitioning: ::prost::alloc::vec::Vec<u8>,
-}
-/// A key that uniquely identifies a task in a query.
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct TaskKey {
-    /// Our query id.
-    #[prost(bytes = "vec", tag = "1")]
-    pub query_id: ::prost::alloc::vec::Vec<u8>,
-    /// Our stage id.
-    #[prost(uint64, tag = "2")]
-    pub stage_id: u64,
-    /// The task number within the stage.
-    #[prost(uint64, tag = "3")]
-    pub task_number: u64,
-}
-/// FlightAppMetadata represents all types of app_metadata which we use in the distributed execution.
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct FlightAppMetadata {
-    #[prost(uint64, tag = "1")]
-    pub partition: u64,
-    /// Unix timestamp in nanoseconds at which this message was created.
-    #[prost(uint64, tag = "2")]
-    pub created_timestamp_unix_nanos: u64,
-}
-/// A Label mirrors datafusion::physical_plan::metrics::Label.
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct Label {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(string, tag = "2")]
-    pub value: ::prost::alloc::string::String,
-}
-/// A Metric is a protobuf mirror of datafusion::physical_plan::metrics::Metric.
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct Metric {
-    #[prost(message, repeated, tag = "1")]
-    pub labels: ::prost::alloc::vec::Vec<Label>,
-    #[prost(uint64, optional, tag = "2")]
-    pub partition: ::core::option::Option<u64>,
-    #[prost(
-        oneof = "metric::Value",
-        tags = "10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34"
-    )]
-    pub value: ::core::option::Option<metric::Value>,
-}
-/// Nested message and enum types in `Metric`.
-pub mod metric {
-    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
-    pub enum Value {
-        #[prost(message, tag = "10")]
-        OutputRows(super::OutputRows),
-        #[prost(message, tag = "11")]
-        ElapsedCompute(super::ElapsedCompute),
-        #[prost(message, tag = "12")]
-        SpillCount(super::SpillCount),
-        #[prost(message, tag = "13")]
-        SpilledBytes(super::SpilledBytes),
-        #[prost(message, tag = "14")]
-        SpilledRows(super::SpilledRows),
-        #[prost(message, tag = "15")]
-        CurrentMemoryUsage(super::CurrentMemoryUsage),
-        #[prost(message, tag = "16")]
-        Count(super::NamedCount),
-        #[prost(message, tag = "17")]
-        Gauge(super::NamedGauge),
-        #[prost(message, tag = "18")]
-        Time(super::NamedTime),
-        #[prost(message, tag = "19")]
-        StartTimestamp(super::StartTimestamp),
-        #[prost(message, tag = "20")]
-        EndTimestamp(super::EndTimestamp),
-        #[prost(message, tag = "21")]
-        OutputBytes(super::OutputBytes),
-        #[prost(message, tag = "22")]
-        OutputBatches(super::OutputBatches),
-        #[prost(message, tag = "23")]
-        PruningMetrics(super::NamedPruningMetrics),
-        #[prost(message, tag = "24")]
-        Ratio(super::NamedRatio),
-        #[prost(message, tag = "25")]
-        CustomMinLatency(super::MinLatency),
-        #[prost(message, tag = "26")]
-        CustomMaxLatency(super::MaxLatency),
-        #[prost(message, tag = "27")]
-        CustomAvgLatency(super::AvgLatency),
-        #[prost(message, tag = "28")]
-        CustomFirstLatency(super::FirstLatency),
-        #[prost(message, tag = "29")]
-        CustomBytesCount(super::BytesCount),
-        #[prost(message, tag = "30")]
-        CustomP50Latency(super::PercentileLatency),
-        #[prost(message, tag = "31")]
-        CustomP75Latency(super::PercentileLatency),
-        #[prost(message, tag = "32")]
-        CustomP95Latency(super::PercentileLatency),
-        #[prost(message, tag = "33")]
-        CustomP99Latency(super::PercentileLatency),
-        #[prost(message, tag = "34")]
-        CustomMaxGauge(super::MaxGauge),
-    }
-}
-/// A MetricsSet is a protobuf mirror of datafusion::physical_plan::metrics::MetricsSet. It represents
-/// a collection of metrics for one ExecutionPlan node.
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct MetricsSet {
-    #[prost(message, repeated, tag = "1")]
-    pub metrics: ::prost::alloc::vec::Vec<Metric>,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct OutputRows {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct ElapsedCompute {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct SpillCount {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct SpilledBytes {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct SpilledRows {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct CurrentMemoryUsage {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct NamedCount {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct NamedGauge {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct NamedTime {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct StartTimestamp {
-    #[prost(int64, optional, tag = "1")]
-    pub value: ::core::option::Option<i64>,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct EndTimestamp {
-    #[prost(int64, optional, tag = "1")]
-    pub value: ::core::option::Option<i64>,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct OutputBytes {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct OutputBatches {
-    #[prost(uint64, tag = "1")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct NamedPruningMetrics {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub pruned: u64,
-    #[prost(uint64, tag = "3")]
-    pub matched: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct NamedRatio {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub part: u64,
-    #[prost(uint64, tag = "3")]
-    pub total: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct BytesCount {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct MinLatency {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct MaxLatency {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct AvgLatency {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub nanos_sum: u64,
-    #[prost(uint64, tag = "3")]
-    pub count: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct FirstLatency {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct PercentileLatency {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(bytes = "vec", tag = "4")]
-    pub sketch_bytes: ::prost::alloc::vec::Vec<u8>,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct MaxGauge {
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    #[prost(uint64, tag = "2")]
-    pub value: u64,
-}
-/// Generated client implementations.
-pub mod worker_service_client {
-    #![allow(
-        unused_variables,
-        dead_code,
-        missing_docs,
-        clippy::wildcard_imports,
-        clippy::let_unit_value
-    )]
-    use tonic::codegen::http::Uri;
-    use tonic::codegen::*;
-    #[derive(Debug, Clone)]
-    pub struct WorkerServiceClient<T> {
-        inner: tonic::client::Grpc<T>,
-    }
-    impl WorkerServiceClient<tonic::transport::Channel> {
-        /// Attempt to create a new client by connecting to a given endpoint.
-        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
-        where
-            D: TryInto<tonic::transport::Endpoint>,
-            D::Error: Into<StdError>,
-        {
-            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
-            Ok(Self::new(conn))
-        }
-    }
-    impl<T> WorkerServiceClient<T>
-    where
-        T: tonic::client::GrpcService<tonic::body::Body>,
-        T::Error: Into<StdError>,
-        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
-        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
-    {
-        pub fn new(inner: T) -> Self {
-            let inner = tonic::client::Grpc::new(inner);
-            Self { inner }
-        }
-        pub fn with_origin(inner: T, origin: Uri) -> Self {
-            let inner = tonic::client::Grpc::with_origin(inner, origin);
-            Self { inner }
-        }
-        pub fn with_interceptor<F>(
-            inner: T,
-            interceptor: F,
-        ) -> WorkerServiceClient<InterceptedService<T, F>>
-        where
-            F: tonic::service::Interceptor,
-            T::ResponseBody: Default,
-            T: tonic::codegen::Service<
-                    http::Request<tonic::body::Body>,
-                    Response = http::Response<
-                        <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
-                    >,
-                >,
-            <T as tonic::codegen::Service<http::Request<tonic::body::Body>>>::Error:
-                Into<StdError> + std::marker::Send + std::marker::Sync,
-        {
-            WorkerServiceClient::new(InterceptedService::new(inner, interceptor))
-        }
-        /// Compress requests with the given encoding.
-        ///
-        /// This requires the server to support it otherwise it might respond with an
-        /// error.
-        #[must_use]
-        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
-            self.inner = self.inner.send_compressed(encoding);
-            self
-        }
-        /// Enable decompressing responses.
-        #[must_use]
-        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
-            self.inner = self.inner.accept_compressed(encoding);
-            self
-        }
-        /// Limits the maximum size of a decoded message.
-        ///
-        /// Default: `4MB`
-        #[must_use]
-        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
-            self.inner = self.inner.max_decoding_message_size(limit);
-            self
-        }
-        /// Limits the maximum size of an encoded message.
-        ///
-        /// Default: `usize::MAX`
-        #[must_use]
-        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
-            self.inner = self.inner.max_encoding_message_size(limit);
-            self
-        }
-        /// Establishes a bidirectional message stream between a coordinator and a worker, over which messages
-        /// will be exchanged at any time during a query's lifetime. It's expected to be one coordinator channel
-        /// per task.
-        pub async fn coordinator_channel(
-            &mut self,
-            request: impl tonic::IntoStreamingRequest<Message = super::CoordinatorToWorkerMsg>,
-        ) -> std::result::Result<
-            tonic::Response<tonic::codec::Streaming<super::WorkerToCoordinatorMsg>>,
-            tonic::Status,
-        > {
-            self.inner.ready().await.map_err(|e| {
-                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
-            })?;
-            let codec = tonic_prost::ProstCodec::default();
-            let path =
-                http::uri::PathAndQuery::from_static("/worker.WorkerService/CoordinatorChannel");
-            let mut req = request.into_streaming_request();
-            req.extensions_mut().insert(GrpcMethod::new(
-                "worker.WorkerService",
-                "CoordinatorChannel",
-            ));
-            self.inner.streaming(req, path, codec).await
-        }
-        /// Executes the requested partition range of a subplan previously sent by the coordinator channel.
-        pub async fn execute_task(
-            &mut self,
-            request: impl tonic::IntoRequest<super::ExecuteTaskRequest>,
-        ) -> std::result::Result<
-            tonic::Response<tonic::codec::Streaming<::arrow_flight::FlightData>>,
-            tonic::Status,
-        > {
-            self.inner.ready().await.map_err(|e| {
-                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
-            })?;
-            let codec = tonic_prost::ProstCodec::default();
-            let path = http::uri::PathAndQuery::from_static("/worker.WorkerService/ExecuteTask");
-            let mut req = request.into_request();
-            req.extensions_mut()
-                .insert(GrpcMethod::new("worker.WorkerService", "ExecuteTask"));
-            self.inner.server_streaming(req, path, codec).await
-        }
-        /// Returns metadata about a worker. Currently only used for worker versioning.
-        pub async fn get_worker_info(
-            &mut self,
-            request: impl tonic::IntoRequest<super::GetWorkerInfoRequest>,
-        ) -> std::result::Result<tonic::Response<super::GetWorkerInfoResponse>, tonic::Status>
-        {
-            self.inner.ready().await.map_err(|e| {
-                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
-            })?;
-            let codec = tonic_prost::ProstCodec::default();
-            let path = http::uri::PathAndQuery::from_static("/worker.WorkerService/GetWorkerInfo");
-            let mut req = request.into_request();
-            req.extensions_mut()
-                .insert(GrpcMethod::new("worker.WorkerService", "GetWorkerInfo"));
-            self.inner.unary(req, path, codec).await
-        }
-    }
-}
-/// Generated server implementations.
-pub mod worker_service_server {
-    #![allow(
-        unused_variables,
-        dead_code,
-        missing_docs,
-        clippy::wildcard_imports,
-        clippy::let_unit_value
-    )]
-    use tonic::codegen::*;
-    /// Generated trait containing gRPC methods that should be implemented for use with WorkerServiceServer.
-    #[async_trait]
-    pub trait WorkerService: std::marker::Send + std::marker::Sync + 'static {
-        /// Server streaming response type for the CoordinatorChannel method.
-        type CoordinatorChannelStream: tonic::codegen::tokio_stream::Stream<
-                Item = std::result::Result<super::WorkerToCoordinatorMsg, tonic::Status>,
-            > + std::marker::Send
-            + 'static;
-        /// Establishes a bidirectional message stream between a coordinator and a worker, over which messages
-        /// will be exchanged at any time during a query's lifetime. It's expected to be one coordinator channel
-        /// per task.
-        async fn coordinator_channel(
-            &self,
-            request: tonic::Request<tonic::Streaming<super::CoordinatorToWorkerMsg>>,
-        ) -> std::result::Result<tonic::Response<Self::CoordinatorChannelStream>, tonic::Status>;
-        /// Server streaming response type for the ExecuteTask method.
-        type ExecuteTaskStream: tonic::codegen::tokio_stream::Stream<
-                Item = std::result::Result<::arrow_flight::FlightData, tonic::Status>,
-            > + std::marker::Send
-            + 'static;
-        /// Executes the requested partition range of a subplan previously sent by the coordinator channel.
-        async fn execute_task(
-            &self,
-            request: tonic::Request<super::ExecuteTaskRequest>,
-        ) -> std::result::Result<tonic::Response<Self::ExecuteTaskStream>, tonic::Status>;
-        /// Returns metadata about a worker. Currently only used for worker versioning.
-        async fn get_worker_info(
-            &self,
-            request: tonic::Request<super::GetWorkerInfoRequest>,
-        ) -> std::result::Result<tonic::Response<super::GetWorkerInfoResponse>, tonic::Status>;
-    }
-    #[derive(Debug)]
-    pub struct WorkerServiceServer<T> {
-        inner: Arc<T>,
-        accept_compression_encodings: EnabledCompressionEncodings,
-        send_compression_encodings: EnabledCompressionEncodings,
-        max_decoding_message_size: Option<usize>,
-        max_encoding_message_size: Option<usize>,
-    }
-    impl<T> WorkerServiceServer<T> {
-        pub fn new(inner: T) -> Self {
-            Self::from_arc(Arc::new(inner))
-        }
-        pub fn from_arc(inner: Arc<T>) -> Self {
-            Self {
-                inner,
-                accept_compression_encodings: Default::default(),
-                send_compression_encodings: Default::default(),
-                max_decoding_message_size: None,
-                max_encoding_message_size: None,
-            }
-        }
-        pub fn with_interceptor<F>(inner: T, interceptor: F) -> InterceptedService<Self, F>
-        where
-            F: tonic::service::Interceptor,
-        {
-            InterceptedService::new(Self::new(inner), interceptor)
-        }
-        /// Enable decompressing requests with the given encoding.
-        #[must_use]
-        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
-            self.accept_compression_encodings.enable(encoding);
-            self
-        }
-        /// Compress responses with the given encoding, if the client supports it.
-        #[must_use]
-        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
-            self.send_compression_encodings.enable(encoding);
-            self
-        }
-        /// Limits the maximum size of a decoded message.
-        ///
-        /// Default: `4MB`
-        #[must_use]
-        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
-            self.max_decoding_message_size = Some(limit);
-            self
-        }
-        /// Limits the maximum size of an encoded message.
-        ///
-        /// Default: `usize::MAX`
-        #[must_use]
-        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
-            self.max_encoding_message_size = Some(limit);
-            self
-        }
-    }
-    impl<T, B> tonic::codegen::Service<http::Request<B>> for WorkerServiceServer<T>
-    where
-        T: WorkerService,
-        B: Body + std::marker::Send + 'static,
-        B::Error: Into<StdError> + std::marker::Send + 'static,
-    {
-        type Response = http::Response<tonic::body::Body>;
-        type Error = std::convert::Infallible;
-        type Future = BoxFuture<Self::Response, Self::Error>;
-        fn poll_ready(
-            &mut self,
-            _cx: &mut Context<'_>,
-        ) -> Poll<std::result::Result<(), Self::Error>> {
-            Poll::Ready(Ok(()))
-        }
-        fn call(&mut self, req: http::Request<B>) -> Self::Future {
-            match req.uri().path() {
-                "/worker.WorkerService/CoordinatorChannel" => {
-                    #[allow(non_camel_case_types)]
-                    struct CoordinatorChannelSvc<T: WorkerService>(pub Arc<T>);
-                    impl<T: WorkerService>
-                        tonic::server::StreamingService<super::CoordinatorToWorkerMsg>
-                        for CoordinatorChannelSvc<T>
-                    {
-                        type Response = super::WorkerToCoordinatorMsg;
-                        type ResponseStream = T::CoordinatorChannelStream;
-                        type Future =
-                            BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
-                        fn call(
-                            &mut self,
-                            request: tonic::Request<
-                                tonic::Streaming<super::CoordinatorToWorkerMsg>,
-                            >,
-                        ) -> Self::Future {
-                            let inner = Arc::clone(&self.0);
-                            let fut = async move {
-                                <T as WorkerService>::coordinator_channel(&inner, request).await
-                            };
-                            Box::pin(fut)
-                        }
-                    }
-                    let accept_compression_encodings = self.accept_compression_encodings;
-                    let send_compression_encodings = self.send_compression_encodings;
-                    let max_decoding_message_size = self.max_decoding_message_size;
-                    let max_encoding_message_size = self.max_encoding_message_size;
-                    let inner = self.inner.clone();
-                    let fut = async move {
-                        let method = CoordinatorChannelSvc(inner);
-                        let codec = tonic_prost::ProstCodec::default();
-                        let mut grpc = tonic::server::Grpc::new(codec)
-                            .apply_compression_config(
-                                accept_compression_encodings,
-                                send_compression_encodings,
-                            )
-                            .apply_max_message_size_config(
-                                max_decoding_message_size,
-                                max_encoding_message_size,
-                            );
-                        let res = grpc.streaming(method, req).await;
-                        Ok(res)
-                    };
-                    Box::pin(fut)
-                }
-                "/worker.WorkerService/ExecuteTask" => {
-                    #[allow(non_camel_case_types)]
-                    struct ExecuteTaskSvc<T: WorkerService>(pub Arc<T>);
-                    impl<T: WorkerService>
-                        tonic::server::ServerStreamingService<super::ExecuteTaskRequest>
-                        for ExecuteTaskSvc<T>
-                    {
-                        type Response = ::arrow_flight::FlightData;
-                        type ResponseStream = T::ExecuteTaskStream;
-                        type Future =
-                            BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
-                        fn call(
-                            &mut self,
-                            request: tonic::Request<super::ExecuteTaskRequest>,
-                        ) -> Self::Future {
-                            let inner = Arc::clone(&self.0);
-                            let fut = async move {
-                                <T as WorkerService>::execute_task(&inner, request).await
-                            };
-                            Box::pin(fut)
-                        }
-                    }
-                    let accept_compression_encodings = self.accept_compression_encodings;
-                    let send_compression_encodings = self.send_compression_encodings;
-                    let max_decoding_message_size = self.max_decoding_message_size;
-                    let max_encoding_message_size = self.max_encoding_message_size;
-                    let inner = self.inner.clone();
-                    let fut = async move {
-                        let method = ExecuteTaskSvc(inner);
-                        let codec = tonic_prost::ProstCodec::default();
-                        let mut grpc = tonic::server::Grpc::new(codec)
-                            .apply_compression_config(
-                                accept_compression_encodings,
-                                send_compression_encodings,
-                            )
-                            .apply_max_message_size_config(
-                                max_decoding_message_size,
-                                max_encoding_message_size,
-                            );
-                        let res = grpc.server_streaming(method, req).await;
-                        Ok(res)
-                    };
-                    Box::pin(fut)
-                }
-                "/worker.WorkerService/GetWorkerInfo" => {
-                    #[allow(non_camel_case_types)]
-                    struct GetWorkerInfoSvc<T: WorkerService>(pub Arc<T>);
-                    impl<T: WorkerService> tonic::server::UnaryService<super::GetWorkerInfoRequest>
-                        for GetWorkerInfoSvc<T>
-                    {
-                        type Response = super::GetWorkerInfoResponse;
-                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
-                        fn call(
-                            &mut self,
-                            request: tonic::Request<super::GetWorkerInfoRequest>,
-                        ) -> Self::Future {
-                            let inner = Arc::clone(&self.0);
-                            let fut = async move {
-                                <T as WorkerService>::get_worker_info(&inner, request).await
-                            };
-                            Box::pin(fut)
-                        }
-                    }
-                    let accept_compression_encodings = self.accept_compression_encodings;
-                    let send_compression_encodings = self.send_compression_encodings;
-                    let max_decoding_message_size = self.max_decoding_message_size;
-                    let max_encoding_message_size = self.max_encoding_message_size;
-                    let inner = self.inner.clone();
-                    let fut = async move {
-                        let method = GetWorkerInfoSvc(inner);
-                        let codec = tonic_prost::ProstCodec::default();
-                        let mut grpc = tonic::server::Grpc::new(codec)
-                            .apply_compression_config(
-                                accept_compression_encodings,
-                                send_compression_encodings,
-                            )
-                            .apply_max_message_size_config(
-                                max_decoding_message_size,
-                                max_encoding_message_size,
-                            );
-                        let res = grpc.unary(method, req).await;
-                        Ok(res)
-                    };
-                    Box::pin(fut)
-                }
-                _ => Box::pin(async move {
-                    let mut response = http::Response::new(tonic::body::Body::default());
-                    let headers = response.headers_mut();
-                    headers.insert(
-                        tonic::Status::GRPC_STATUS,
-                        (tonic::Code::Unimplemented as i32).into(),
-                    );
-                    headers.insert(
-                        http::header::CONTENT_TYPE,
-                        tonic::metadata::GRPC_CONTENT_TYPE,
-                    );
-                    Ok(response)
-                }),
-            }
-        }
-    }
-    impl<T> Clone for WorkerServiceServer<T> {
-        fn clone(&self) -> Self {
-            let inner = self.inner.clone();
-            Self {
-                inner,
-                accept_compression_encodings: self.accept_compression_encodings,
-                send_compression_encodings: self.send_compression_encodings,
-                max_decoding_message_size: self.max_decoding_message_size,
-                max_encoding_message_size: self.max_encoding_message_size,
-            }
-        }
-    }
-    /// Generated gRPC service name
-    pub const SERVICE_NAME: &str = "worker.WorkerService";
-    impl<T> tonic::server::NamedService for WorkerServiceServer<T> {
-        const NAME: &'static str = SERVICE_NAME;
-    }
-}
diff --git a/src/protocol/grpc/metrics_proto.rs b/src/protocol/grpc/metrics_proto.rs
index 6db8cda9..654b6cee 100644
--- a/src/protocol/grpc/metrics_proto.rs
+++ b/src/protocol/grpc/metrics_proto.rs
@@ -1,4 +1,4 @@
-use super::generated::worker as pb;
+use crate::protocol::generated::worker as pb;
 use chrono::DateTime;
 use datafusion::common::internal_err;
 use datafusion::error::DataFusionError;
diff --git a/src/protocol/grpc/mod.rs b/src/protocol/grpc/mod.rs
index e432b607..fc367585 100644
--- a/src/protocol/grpc/mod.rs
+++ b/src/protocol/grpc/mod.rs
@@ -1,6 +1,5 @@
 mod channel_resolver;
 mod errors;
-mod generated;
 mod metrics_proto;
 mod observability;
 mod on_drop_stream;
diff --git a/src/protocol/grpc/observability/gen/src/main.rs b/src/protocol/grpc/observability/gen/src/main.rs
index 124f0f1b..1b55fb7d 100644
--- a/src/protocol/grpc/observability/gen/src/main.rs
+++ b/src/protocol/grpc/observability/gen/src/main.rs
@@ -21,7 +21,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .out_dir(&out_dir)
         .extern_path(
             ".observability.TaskKey",
-            "crate::protocol::grpc::generated::worker::TaskKey",
+            "crate::protocol::generated::worker::TaskKey",
         )
         .compile_protos(&[proto_file], &[proto_dir])?;
 
diff --git a/src/protocol/grpc/observability/generated/observability.rs b/src/protocol/grpc/observability/generated/observability.rs
index 3b0a95cf..98aa592d 100644
--- a/src/protocol/grpc/observability/generated/observability.rs
+++ b/src/protocol/grpc/observability/generated/observability.rs
@@ -12,7 +12,7 @@ pub struct GetTaskProgressRequest {}
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct TaskProgress {
     #[prost(message, optional, tag = "1")]
-    pub task_key: ::core::option::Option<crate::protocol::grpc::generated::worker::TaskKey>,
+    pub task_key: ::core::option::Option<crate::protocol::generated::worker::TaskKey>,
     #[prost(enumeration = "TaskStatus", tag = "4")]
     pub status: i32,
     #[prost(uint64, tag = "5")]
diff --git a/src/protocol/grpc/observability/service.rs b/src/protocol/grpc/observability/service.rs
index 27edee2e..04c2aa26 100644
--- a/src/protocol/grpc/observability/service.rs
+++ b/src/protocol/grpc/observability/service.rs
@@ -4,7 +4,7 @@ use super::{
 };
 use crate::common::serialize_uuid;
 use crate::grpc::{GetClusterWorkersRequest, GetClusterWorkersResponse};
-use crate::protocol::grpc::generated::worker as worker_pb;
+use crate::protocol::generated::worker as worker_pb;
 use crate::worker::{SingleWriteMultiRead, TaskData};
 use crate::{TaskKey, WorkerResolver};
 use datafusion::error::DataFusionError;
diff --git a/src/protocol/grpc/worker_client.rs b/src/protocol/grpc/worker_client.rs
index 56612671..28ac9d38 100644
--- a/src/protocol/grpc/worker_client.rs
+++ b/src/protocol/grpc/worker_client.rs
@@ -1,10 +1,11 @@
 use super::channel_resolver::BoxCloneSyncChannel;
 use super::errors::{map_flight_to_datafusion_error, map_status_to_datafusion_error};
-use super::generated::worker as pb;
 use super::metrics_proto::metrics_set_proto_to_df;
 use crate::common::serialize_uuid;
-use crate::grpc::generated::worker::FlightAppMetadata;
 use crate::grpc::on_drop_stream::on_drop_stream;
+use crate::protocol::generated::worker as pb;
+use crate::protocol::generated::worker::FlightAppMetadata;
+use crate::protocol::generated::worker_grpc::worker_service_client::WorkerServiceClient;
 use crate::{
     BytesMetricExt, CoordinatorToWorkerMsg, DistributedConfig, ExecuteTaskRequest,
     FirstLatencyMetric, GetWorkerInfoRequest, GetWorkerInfoResponse, LatencyMetricExt, LoadInfo,
@@ -43,7 +44,7 @@ use tonic::metadata::MetadataMap;
 use tonic::{Request, Status};
 
 #[async_trait]
-impl WorkerChannel for pb::worker_service_client::WorkerServiceClient<BoxCloneSyncChannel> {
+impl WorkerChannel for WorkerServiceClient<BoxCloneSyncChannel> {
     async fn coordinator_channel(
         &mut self,
         headers: HeaderMap,
@@ -567,7 +568,7 @@ fn fanout(o_txs: &[UnboundedSender<WorkerMsg>], err: Status) {
 /// ```
 pub fn create_worker_client(channel: BoxCloneSyncChannel) -> Box<dyn WorkerChannel> {
     Box::new(
-        pb::worker_service_client::WorkerServiceClient::new(channel)
+        WorkerServiceClient::new(channel)
             .max_decoding_message_size(usize::MAX)
             .max_encoding_message_size(usize::MAX),
     )
diff --git a/src/protocol/grpc/worker_service.proto b/src/protocol/grpc/worker_service.proto
new file mode 100644
index 00000000..9ba6c327
--- /dev/null
+++ b/src/protocol/grpc/worker_service.proto
@@ -0,0 +1,15 @@
+syntax = "proto3";
+package worker_grpc;
+
+import "messages.proto";
+
+service WorkerService {
+  // Establishes a bidirectional message stream between a coordinator and a worker, over which messages
+  // will be exchanged at any time during a query's lifetime. It's expected to be one coordinator channel
+  // per task.
+  rpc CoordinatorChannel(stream worker.CoordinatorToWorkerMsg) returns (stream worker.WorkerToCoordinatorMsg);
+  // Executes the requested partition range of a subplan previously sent by the coordinator channel.
+  rpc ExecuteTask(worker.ExecuteTaskRequest) returns (stream worker.FlightData);
+  // Returns metadata about a worker. Currently only used for worker versioning.
+  rpc GetWorkerInfo(worker.GetWorkerInfoRequest) returns (worker.GetWorkerInfoResponse);
+}
diff --git a/src/protocol/grpc/worker_service.rs b/src/protocol/grpc/worker_service.rs
index 6fba14ee..c3666d5c 100644
--- a/src/protocol/grpc/worker_service.rs
+++ b/src/protocol/grpc/worker_service.rs
@@ -1,10 +1,12 @@
 use super::errors::{datafusion_error_to_tonic_status, map_status_to_datafusion_error};
-use super::generated::worker as pb;
 use super::metrics_proto::df_metrics_set_to_proto;
 use super::spawn_select_all::spawn_select_all;
-
 use crate::common::{deserialize_uuid, now_ns};
 use crate::protocol::ProducerHeadSpec;
+use crate::protocol::generated::worker as pb;
+use crate::protocol::generated::worker_grpc::worker_service_server::{
+    WorkerService, WorkerServiceServer,
+};
 use crate::protocol::grpc::{ObservabilityServiceImpl, ObservabilityServiceServer};
 use crate::{
     CoordinatorToWorkerMsg, DistributedConfig, ExecuteTaskRequest, LoadInfo, SetPlanRequest,
@@ -59,8 +61,8 @@ impl Worker {
     ///
     /// # }
     /// ```
-    pub fn into_worker_server(self) -> pb::worker_service_server::WorkerServiceServer<Self> {
-        pb::worker_service_server::WorkerServiceServer::new(self)
+    pub fn into_worker_server(self) -> WorkerServiceServer<Self> {
+        WorkerServiceServer::new(self)
             .max_decoding_message_size(usize::MAX)
             .max_encoding_message_size(usize::MAX)
     }
@@ -86,7 +88,7 @@ impl Worker {
 /// The methods are delegated to plan `impl Worker` implementations so that they can be implemented
 /// in different files.
 #[async_trait]
-impl pb::worker_service_server::WorkerService for Worker {
+impl WorkerService for Worker {
     type CoordinatorChannelStream = BoxStream<'static, Result<pb::WorkerToCoordinatorMsg, Status>>;
 
     async fn coordinator_channel(
diff --git a/src/protocol/in_process.rs b/src/protocol/in_process.rs
new file mode 100644
index 00000000..9a59085d
--- /dev/null
+++ b/src/protocol/in_process.rs
@@ -0,0 +1,230 @@
+//! The reference implementation of the worker protocol for a co-located worker.
+//!
+//! It implements [`WorkerChannel`] by calling a [`Worker`] that lives in the same process, with no
+//! gRPC, no IPC, and no networking. Its purpose is twofold: it lets the crate run distributed
+//! queries with the `grpc` feature off (the protocol abstraction is only real if something other
+//! than gRPC implements it), and it is the shape a custom co-located transport (for example a
+//! shared-memory mesh spanning sibling processes) follows: implement [`ChannelResolver`] to hand
+//! out a channel for a URL, then route the three protocol methods to wherever the worker runs.
+
+use crate::protocol::{ChannelResolver, WorkerChannel};
+use crate::{
+    CoordinatorToWorkerMsg, DefaultSessionBuilder, DistributedExt, ExecuteTaskRequest,
+    GetWorkerInfoRequest, GetWorkerInfoResponse, MappedWorkerSessionBuilderExt, Worker,
+    WorkerSessionBuilder, WorkerToCoordinatorMsg,
+};
+use async_trait::async_trait;
+use datafusion::arrow::array::RecordBatch;
+use datafusion::common::{DataFusionError, Result, internal_datafusion_err};
+use datafusion::execution::TaskContext;
+use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
+use futures::stream::{BoxStream, StreamExt};
+use http::HeaderMap;
+use std::sync::{Arc, Weak};
+use url::Url;
+
+/// A [`ChannelResolver`] backed by a single co-located [`Worker`].
+///
+/// Every URL resolves to that one worker: with no network there is nothing to dial, so the URLs a
+/// [`crate::WorkerResolver`] hands out only label the tasks the planner routes. One worker holds the
+/// state for every task, keyed by [`crate::TaskKey`], the same way the gRPC worker does when several
+/// tasks of a query land on it.
+#[derive(Clone)]
+pub struct InProcessChannelResolver {
+    worker: Arc<Worker>,
+}
+
+impl InProcessChannelResolver {
+    /// Builds the co-located worker from `session_builder`, registering this resolver into the
+    /// worker's own per-query sessions so a worker reading a downstream stage stays in process
+    /// rather than falling back to the gRPC default.
+    pub fn from_session_builder(
+        session_builder: impl WorkerSessionBuilder + Send + Sync + 'static,
+    ) -> Self {
+        // The worker and the resolver point at each other: the resolver runs tasks on the worker,
+        // and the worker resolves its own nested reads back through the resolver. A `Weak` on the
+        // worker's side breaks the cycle, so the returned `InProcessChannelResolver` owns the only
+        // strong reference and dropping it frees the worker.
+        let worker = Arc::new_cyclic(|weak: &Weak<Worker>| {
+            let weak = weak.clone();
+            Worker::from_session_builder(session_builder.map(move |builder| {
+                Ok(builder
+                    .with_distributed_channel_resolver(WeakInProcessChannelResolver(weak.clone()))
+                    .build())
+            }))
+        });
+        Self { worker }
+    }
+}
+
+impl Default for InProcessChannelResolver {
+    fn default() -> Self {
+        Self::from_session_builder(DefaultSessionBuilder)
+    }
+}
+
+#[async_trait]
+impl ChannelResolver for InProcessChannelResolver {
+    async fn get_worker_client_for_url(
+        &self,
+        _url: &Url,
+    ) -> Result<Box<dyn WorkerChannel>, DataFusionError> {
+        Ok(Box::new(InProcessWorkerChannel {
+            worker: Arc::clone(&self.worker),
+        }))
+    }
+}
+
+/// The resolver a worker installs in its own sessions. It upgrades a [`Weak`] reference to the
+/// co-located worker so a read of a downstream stage routes back in process instead of dialing out.
+struct WeakInProcessChannelResolver(Weak<Worker>);
+
+#[async_trait]
+impl ChannelResolver for WeakInProcessChannelResolver {
+    async fn get_worker_client_for_url(
+        &self,
+        _url: &Url,
+    ) -> Result<Box<dyn WorkerChannel>, DataFusionError> {
+        let worker = self
+            .0
+            .upgrade()
+            .ok_or_else(|| internal_datafusion_err!("the in-process worker has been dropped"))?;
+        Ok(Box::new(InProcessWorkerChannel { worker }))
+    }
+}
+
+/// A [`WorkerChannel`] that calls a co-located [`Worker`] directly.
+struct InProcessWorkerChannel {
+    worker: Arc<Worker>,
+}
+
+#[async_trait]
+impl WorkerChannel for InProcessWorkerChannel {
+    async fn coordinator_channel(
+        &mut self,
+        headers: HeaderMap,
+        c2w_stream: BoxStream<'static, CoordinatorToWorkerMsg>,
+    ) -> Result<BoxStream<'static, Result<WorkerToCoordinatorMsg>>> {
+        // The worker reads a fallible stream so a wire transport can surface its decode errors.
+        // Handing messages over in process has no such step, so each one is already an `Ok`.
+        self.worker
+            .coordinator_channel(headers, c2w_stream.map(Ok).boxed())
+            .await
+    }
+
+    async fn execute_task(
+        &mut self,
+        _headers: HeaderMap,
+        request: ExecuteTaskRequest,
+        _metrics: ExecutionPlanMetricsSet,
+        _task_ctx: &Arc<TaskContext>,
+    ) -> Result<Vec<BoxStream<'static, Result<RecordBatch>>>> {
+        // Reading a partition runs the producer in place: the returned streams are the worker's own
+        // task output, so there is no IPC decode pass and no network metrics to record. The
+        // consumer's `task_ctx` is the consumer side's; the producer runs under the worker's own.
+        let (streams, _task_ctx) = self.worker.execute_task(request).await?;
+        Ok(streams.into_iter().map(|stream| stream.boxed()).collect())
+    }
+
+    async fn get_worker_info(
+        &mut self,
+        _request: GetWorkerInfoRequest,
+    ) -> Result<GetWorkerInfoResponse> {
+        Ok(GetWorkerInfoResponse {
+            version: self.worker.version().to_string(),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{SessionStateBuilderExt, WorkerResolver, display_plan_ascii};
+    use datafusion::arrow::util::pretty::pretty_format_batches;
+    use datafusion::execution::SessionStateBuilder;
+    use datafusion::physical_plan::collect;
+    use datafusion::prelude::{CsvReadOptions, SessionConfig, SessionContext};
+    use std::io::Write;
+
+    /// Hands out as many placeholder URLs as workers. With one co-located worker behind the
+    /// transport, these only label the tasks the planner routes; nothing is dialed.
+    struct InProcessWorkers(usize);
+
+    impl WorkerResolver for InProcessWorkers {
+        fn get_urls(&self) -> Result<Vec<Url>, DataFusionError> {
+            (0..self.0)
+                .map(|i| Url::parse(&format!("http://worker-{i}")))
+                .collect::<Result<_, _>>()
+                .map_err(|err| DataFusionError::External(Box::new(err)))
+        }
+    }
+
+    /// Drives a real distributed query end to end through the in-process transport. With the `grpc`
+    /// feature off this is the only transport that can run it; `cargo check --no-default-features`
+    /// covers the no-gRPC compile.
+    #[tokio::test]
+    async fn in_process_transport_runs_a_distributed_query() -> Result<()> {
+        const N: usize = 4;
+
+        // A file scan round-trips through `datafusion-proto`, so a worker can rebuild it from the
+        // serialized stage plan. An in-memory table would not, hence a CSV on disk.
+        let path = std::env::temp_dir().join(format!("dfd_in_process_{}.csv", std::process::id()));
+        let mut file =
+            std::fs::File::create(&path).map_err(|e| DataFusionError::External(Box::new(e)))?;
+        writeln!(file, "k,v").unwrap();
+        for i in 0..200 {
+            writeln!(file, "{},{}", ["a", "b", "c", "d"][i % 4], i).unwrap();
+        }
+        drop(file);
+        let path = path.to_str().unwrap().to_string();
+
+        let query = "SELECT k, COUNT(*) AS c FROM t GROUP BY k ORDER BY k";
+
+        // Single-node reference result.
+        let ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(N));
+        ctx.register_csv("t", &path, CsvReadOptions::new()).await?;
+        let expected = collect(
+            ctx.sql(query).await?.create_physical_plan().await?,
+            ctx.task_ctx(),
+        )
+        .await?;
+        let expected = pretty_format_batches(&expected)?.to_string();
+
+        // Distributed over the in-process transport.
+        let state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_config(SessionConfig::new().with_target_partitions(N))
+            .with_distributed_planner()
+            .with_distributed_worker_resolver(InProcessWorkers(N))
+            .with_distributed_channel_resolver(InProcessChannelResolver::default())
+            .with_distributed_file_scan_config_bytes_per_partition(1)
+            .unwrap()
+            .build();
+        let ctx_distributed = SessionContext::from(state);
+        ctx_distributed
+            .register_csv("t", &path, CsvReadOptions::new())
+            .await?;
+
+        let physical = ctx_distributed
+            .sql(query)
+            .await?
+            .create_physical_plan()
+            .await?;
+        let rendered = display_plan_ascii(physical.as_ref(), false);
+        assert!(
+            rendered.contains("DistributedExec"),
+            "plan was not distributed:\n{rendered}"
+        );
+        assert!(
+            rendered.contains("NetworkShuffleExec"),
+            "no shuffle boundary, so the transport never carried a cross-task stream:\n{rendered}"
+        );
+
+        let actual = collect(physical, ctx_distributed.task_ctx()).await?;
+        let actual = pretty_format_batches(&actual)?.to_string();
+
+        let _ = std::fs::remove_file(&path);
+        assert_eq!(actual, expected);
+        Ok(())
+    }
+}
diff --git a/src/protocol/grpc/worker.proto b/src/protocol/messages.proto
similarity index 94%
rename from src/protocol/grpc/worker.proto
rename to src/protocol/messages.proto
index 24f5c987..f5127325 100644
--- a/src/protocol/grpc/worker.proto
+++ b/src/protocol/messages.proto
@@ -1,16 +1,6 @@
 syntax = "proto3";
 package worker;
 
-service WorkerService {
-  // Establishes a bidirectional message stream between a coordinator and a worker, over which messages
-  // will be exchanged at any time during a query's lifetime. It's expected to be one coordinator channel
-  // per task.
-  rpc CoordinatorChannel(stream CoordinatorToWorkerMsg) returns (stream WorkerToCoordinatorMsg);
-  // Executes the requested partition range of a subplan previously sent by the coordinator channel.
-  rpc ExecuteTask(ExecuteTaskRequest) returns (stream FlightData);
-  // Returns metadata about a worker. Currently only used for worker versioning.
-  rpc GetWorkerInfo(GetWorkerInfoRequest) returns (GetWorkerInfoResponse);
-}
 
 message CoordinatorToWorkerMsg {
   oneof inner {
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index 6df0e11b..3aee1273 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -2,10 +2,16 @@
 pub mod grpc;
 
 mod channel_resolver;
+// The prost message types carry no tonic dependency, so a non-gRPC transport (an in-process
+// worker, a shared-memory mesh) can speak the same wire shape without pulling in the whole gRPC
+// stack.
+pub(crate) mod generated;
+mod in_process;
 mod worker_channel;
 
 pub(crate) use channel_resolver::set_distributed_channel_resolver;
 pub use channel_resolver::{ChannelResolver, get_distributed_channel_resolver};
+pub use in_process::InProcessChannelResolver;
 
 pub use worker_channel::{
     CoordinatorToWorkerMsg, ExecuteTaskRequest, GetWorkerInfoRequest, GetWorkerInfoResponse,
diff --git a/src/test_utils/in_memory_channel_resolver.rs b/src/test_utils/in_memory_channel_resolver.rs
index 92f710c7..28105c58 100644
--- a/src/test_utils/in_memory_channel_resolver.rs
+++ b/src/test_utils/in_memory_channel_resolver.rs
@@ -1,91 +1,9 @@
-use crate::{
-    ChannelResolver, DefaultSessionBuilder, DistributedExt, MappedWorkerSessionBuilderExt,
-    SessionStateBuilderExt, Worker, WorkerChannel, WorkerResolver, WorkerSessionBuilder, grpc,
-};
-use async_trait::async_trait;
+use crate::WorkerResolver;
 use datafusion::common::DataFusionError;
-use datafusion::execution::SessionStateBuilder;
-use datafusion::prelude::{SessionConfig, SessionContext};
-use hyper_util::rt::TokioIo;
-use tonic::transport::{Endpoint, Server};
 use url::Url;
 
-const DUMMY_URL: &str = "http://localhost:50051";
 const DUMMY_URL_PREFIX: &str = "http://url-";
 
-/// [ChannelResolver] implementation that returns gRPC clients backed by an in-memory
-/// tokio duplex rather than a TCP connection.
-#[derive(Clone)]
-pub struct InMemoryChannelResolver {
-    channel: grpc::BoxCloneSyncChannel,
-}
-
-impl InMemoryChannelResolver {
-    /// Build an [InMemoryChannelResolver] with a custom [WorkerSessionBuilder].
-    /// This allows you to inject your own DataFusion extensions in the in-memory worker
-    /// spawned by this method.
-    pub fn from_session_builder(
-        builder: impl WorkerSessionBuilder + Send + Sync + 'static,
-    ) -> Self {
-        Self::from_configured_worker(builder, |worker| worker)
-    }
-
-    /// Build an [InMemoryChannelResolver] with a custom [WorkerSessionBuilder] and worker setup.
-    pub fn from_configured_worker(
-        builder: impl WorkerSessionBuilder + Send + Sync + 'static,
-        configure_worker: impl Fn(Worker) -> Worker + Send + Sync + 'static,
-    ) -> Self {
-        let (client, server) = tokio::io::duplex(1024 * 1024);
-
-        let mut client = Some(client);
-        let channel = Endpoint::try_from(DUMMY_URL)
-            .expect("Invalid dummy URL for building an endpoint. This should never happen")
-            .connect_with_connector_lazy(tower::service_fn(move |_| {
-                let client = client
-                    .take()
-                    .expect("Client taken twice. This should never happen");
-                async move { Ok::<_, std::io::Error>(TokioIo::new(client)) }
-            }));
-
-        let this = Self {
-            channel: grpc::BoxCloneSyncChannel::new(channel),
-        };
-        let this_clone = this.clone();
-
-        let endpoint = Worker::from_session_builder(builder.map(move |builder| {
-            let this = this.clone();
-            Ok(builder.with_distributed_channel_resolver(this).build())
-        }));
-        let endpoint = configure_worker(endpoint);
-
-        #[allow(clippy::disallowed_methods)]
-        tokio::spawn(async move {
-            Server::builder()
-                .add_service(endpoint.into_worker_server())
-                .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
-                .await
-        });
-
-        this_clone
-    }
-}
-
-impl Default for InMemoryChannelResolver {
-    fn default() -> Self {
-        Self::from_session_builder(DefaultSessionBuilder)
-    }
-}
-
-#[async_trait]
-impl ChannelResolver for InMemoryChannelResolver {
-    async fn get_worker_client_for_url(
-        &self,
-        _: &url::Url,
-    ) -> Result<Box<dyn WorkerChannel>, DataFusionError> {
-        Ok(grpc::create_worker_client(self.channel.clone()))
-    }
-}
-
 pub struct InMemoryWorkerResolver {
     n_workers: usize,
 }
@@ -105,41 +23,139 @@ impl WorkerResolver for InMemoryWorkerResolver {
     }
 }
 
-/// Creates a distributed session context backed by a single in-memory worker service.
-/// The set of produced worker URLs is deterministic, taking the form http://worker-<i>.
-pub async fn start_in_memory_context(
-    num_workers: usize,
-    session_builder: impl WorkerSessionBuilder + Send + Sync + 'static,
-) -> SessionContext {
-    let channel_resolver = InMemoryChannelResolver::from_session_builder(session_builder);
-    let state = SessionStateBuilder::new()
-        .with_default_features()
-        .with_distributed_planner()
-        .with_distributed_worker_resolver(InMemoryWorkerResolver::new(num_workers))
-        .with_distributed_channel_resolver(channel_resolver)
-        .build();
-    SessionContext::from(state)
-}
+// The channel resolver runs a real `tonic` worker server over a tokio duplex, so it only compiles
+// with the gRPC transport. The worker resolver above stays available without it, so the planner
+// tests that only need worker URLs keep building in a no-gRPC config.
+#[cfg(feature = "grpc")]
+mod channel {
+    use super::InMemoryWorkerResolver;
+    use crate::{
+        ChannelResolver, DefaultSessionBuilder, DistributedExt, MappedWorkerSessionBuilderExt,
+        SessionStateBuilderExt, Worker, WorkerChannel, WorkerSessionBuilder, grpc,
+    };
+    use async_trait::async_trait;
+    use datafusion::common::DataFusionError;
+    use datafusion::execution::SessionStateBuilder;
+    use datafusion::prelude::{SessionConfig, SessionContext};
+    use hyper_util::rt::TokioIo;
+    use tonic::transport::{Endpoint, Server};
+
+    const DUMMY_URL: &str = "http://localhost:50051";
+
+    /// [ChannelResolver] implementation that returns gRPC clients backed by an in-memory
+    /// tokio duplex rather than a TCP connection.
+    #[derive(Clone)]
+    pub struct InMemoryChannelResolver {
+        channel: grpc::BoxCloneSyncChannel,
+    }
+
+    impl InMemoryChannelResolver {
+        /// Build an [InMemoryChannelResolver] with a custom [WorkerSessionBuilder].
+        /// This allows you to inject your own DataFusion extensions in the in-memory worker
+        /// spawned by this method.
+        pub fn from_session_builder(
+            builder: impl WorkerSessionBuilder + Send + Sync + 'static,
+        ) -> Self {
+            Self::from_configured_worker(builder, |worker| worker)
+        }
+
+        /// Build an [InMemoryChannelResolver] with a custom [WorkerSessionBuilder] and worker setup.
+        pub fn from_configured_worker(
+            builder: impl WorkerSessionBuilder + Send + Sync + 'static,
+            configure_worker: impl Fn(Worker) -> Worker + Send + Sync + 'static,
+        ) -> Self {
+            let (client, server) = tokio::io::duplex(1024 * 1024);
+
+            let mut client = Some(client);
+            let channel = Endpoint::try_from(DUMMY_URL)
+                .expect("Invalid dummy URL for building an endpoint. This should never happen")
+                .connect_with_connector_lazy(tower::service_fn(move |_| {
+                    let client = client
+                        .take()
+                        .expect("Client taken twice. This should never happen");
+                    async move { Ok::<_, std::io::Error>(TokioIo::new(client)) }
+                }));
+
+            let this = Self {
+                channel: grpc::BoxCloneSyncChannel::new(channel),
+            };
+            let this_clone = this.clone();
+
+            let endpoint = Worker::from_session_builder(builder.map(move |builder| {
+                let this = this.clone();
+                Ok(builder.with_distributed_channel_resolver(this).build())
+            }));
+            let endpoint = configure_worker(endpoint);
+
+            #[allow(clippy::disallowed_methods)]
+            tokio::spawn(async move {
+                Server::builder()
+                    .add_service(endpoint.into_worker_server())
+                    .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
+                    .await
+            });
+
+            this_clone
+        }
+    }
+
+    impl Default for InMemoryChannelResolver {
+        fn default() -> Self {
+            Self::from_session_builder(DefaultSessionBuilder)
+        }
+    }
 
-/// Creates a distributed session context backed by a configurable in-memory worker service.
-///
-/// Like [crate::test_utils::localhost::start_localhost_context], this uses tiny file-scan
-/// partitions so small test datasets still cross worker boundaries.
-pub async fn start_configured_in_memory_context(
-    num_workers: usize,
-    session_builder: impl WorkerSessionBuilder + Send + Sync + 'static,
-    configure_worker: impl Fn(Worker) -> Worker + Send + Sync + 'static,
-) -> SessionContext {
-    let channel_resolver =
-        InMemoryChannelResolver::from_configured_worker(session_builder, configure_worker);
-    let state = SessionStateBuilder::new()
-        .with_default_features()
-        .with_config(SessionConfig::new().with_target_partitions(num_workers))
-        .with_distributed_planner()
-        .with_distributed_worker_resolver(InMemoryWorkerResolver::new(num_workers))
-        .with_distributed_channel_resolver(channel_resolver)
-        .with_distributed_file_scan_config_bytes_per_partition(1)
-        .unwrap()
-        .build();
-    SessionContext::from(state)
+    #[async_trait]
+    impl ChannelResolver for InMemoryChannelResolver {
+        async fn get_worker_client_for_url(
+            &self,
+            _: &url::Url,
+        ) -> Result<Box<dyn WorkerChannel>, DataFusionError> {
+            Ok(grpc::create_worker_client(self.channel.clone()))
+        }
+    }
+
+    /// Creates a distributed session context backed by a single in-memory worker service.
+    /// The set of produced worker URLs is deterministic, taking the form http://worker-<i>.
+    pub async fn start_in_memory_context(
+        num_workers: usize,
+        session_builder: impl WorkerSessionBuilder + Send + Sync + 'static,
+    ) -> SessionContext {
+        let channel_resolver = InMemoryChannelResolver::from_session_builder(session_builder);
+        let state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_distributed_planner()
+            .with_distributed_worker_resolver(InMemoryWorkerResolver::new(num_workers))
+            .with_distributed_channel_resolver(channel_resolver)
+            .build();
+        SessionContext::from(state)
+    }
+
+    /// Creates a distributed session context backed by a configurable in-memory worker service.
+    ///
+    /// Like [crate::test_utils::localhost::start_localhost_context], this uses tiny file-scan
+    /// partitions so small test datasets still cross worker boundaries.
+    pub async fn start_configured_in_memory_context(
+        num_workers: usize,
+        session_builder: impl WorkerSessionBuilder + Send + Sync + 'static,
+        configure_worker: impl Fn(Worker) -> Worker + Send + Sync + 'static,
+    ) -> SessionContext {
+        let channel_resolver =
+            InMemoryChannelResolver::from_configured_worker(session_builder, configure_worker);
+        let state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_config(SessionConfig::new().with_target_partitions(num_workers))
+            .with_distributed_planner()
+            .with_distributed_worker_resolver(InMemoryWorkerResolver::new(num_workers))
+            .with_distributed_channel_resolver(channel_resolver)
+            .with_distributed_file_scan_config_bytes_per_partition(1)
+            .unwrap()
+            .build();
+        SessionContext::from(state)
+    }
 }
+
+#[cfg(feature = "grpc")]
+pub use channel::{
+    InMemoryChannelResolver, start_configured_in_memory_context, start_in_memory_context,
+};
diff --git a/src/test_utils/mod.rs b/src/test_utils/mod.rs
index dc66f977..6c84a3ff 100644
--- a/src/test_utils/mod.rs
+++ b/src/test_utils/mod.rs
@@ -1,5 +1,6 @@
 pub mod in_memory_channel_resolver;
 pub mod insta;
+#[cfg(feature = "grpc")]
 pub mod localhost;
 pub mod metrics;
 pub mod mock_exec;
diff --git a/src/test_utils/routing.rs b/src/test_utils/routing.rs
index 15fdabc6..d687e424 100644
--- a/src/test_utils/routing.rs
+++ b/src/test_utils/routing.rs
@@ -2,6 +2,7 @@ use arrow::{
     array::{Int64Array, RecordBatch, StringArray},
     datatypes::{DataType, Field, Schema, SchemaRef},
 };
+use async_trait::async_trait;
 use datafusion::{
     catalog::{Session, TableFunctionImpl, TableProvider},
     common::{ScalarValue, Statistics, internal_err, plan_err},
@@ -19,7 +20,6 @@ use datafusion_proto::{physical_plan::PhysicalExtensionCodec, protobuf::proto_er
 use futures::stream;
 use prost::Message;
 use std::{fmt::Formatter, sync::Arc};
-use tonic::async_trait;
 
 use crate::execution_plans::DistributedLeafExec;
 use crate::worker::LocalWorkerContext;
diff --git a/src/worker/mod.rs b/src/worker/mod.rs
index b2e79eac..030ac357 100644
--- a/src/worker/mod.rs
+++ b/src/worker/mod.rs
@@ -3,7 +3,9 @@ mod impl_execute_task;
 mod session_builder;
 mod single_write_multi_read;
 mod task_data;
-#[cfg(any(test, feature = "integration"))]
+// `worker_handles` builds `tonic` servers and Flight channels for the benchmark fixtures, which
+// only compile with the gRPC transport.
+#[cfg(all(feature = "grpc", any(test, feature = "integration")))]
 pub(crate) mod test_utils;
 mod worker_connection_pool;
 mod worker_service;