Update docs (#549)

Update docs to adhere to the `WorkerChannel` abstaction

Author: gabotechs

docs/source/advanced/07-worker-versioning.md

@@ -27,15 +27,15 @@ let worker = Worker::default()
 
 ## Querying a worker's version
 
-From the coordinating context, use `DefaultChannelResolver` to get a cached
-channel and `create_worker_client` to build a client, then call `get_worker_info`:
+From the coordinating context, use `grpc::DefaultChannelResolver` to get a cached
+channel and `grpc::create_worker_client` to build a client, then call `get_worker_info`:
 
 ```rust
-use datafusion_distributed::{DefaultChannelResolver, GetWorkerInfoRequest, create_worker_client};
+use datafusion_distributed::{grpc, GetWorkerInfoRequest};
 
-let channel_resolver = DefaultChannelResolver::default();
+let channel_resolver = grpc::DefaultChannelResolver::default();
 let channel = channel_resolver.get_channel(&worker_url).await?;
-let mut client = create_worker_client(channel);
+let mut client = grpc::create_worker_client(channel);
 
 let response = client.get_worker_info(GetWorkerInfoRequest {}).await?;
 println!("version: {}", response.into_inner().version);
@@ -66,9 +66,7 @@ use std::sync::{Arc, RwLock};
 use std::time::Duration;
 use url::Url;
 use datafusion::common::{HashMap, DataFusionError};
-use datafusion_distributed::{
-    DefaultChannelResolver, GetWorkerInfoRequest, WorkerResolver, create_worker_client,
-};
+use datafusion_distributed::{grpc, GetWorkerInfoRequest, WorkerResolver};
 
 struct VersionAwareWorkerResolver {
     compatible_urls: Arc<RwLock<Vec<Url>>>,
@@ -80,7 +78,7 @@ async fn background_version_resolver(
     all_worker_urls: Vec<Url>,
     local_version: String,
     compatible_urls: Arc<RwLock<Vec<Url>>>,
-    channel_resolver: Arc<DefaultChannelResolver>,
+    channel_resolver: Arc<grpc::DefaultChannelResolver>,
 ) {
     let mut version_cache: HashMap<Url, String> = HashMap::new();
 
@@ -94,7 +92,7 @@ async fn background_version_resolver(
             let cr = Arc::clone(&channel_resolver);
             async move {
                 let channel = cr.get_channel(url).await.ok()?;
-                let mut client = create_worker_client(channel);
+                let mut client = grpc::create_worker_client(channel);
                 let resp = client.get_worker_info(GetWorkerInfoRequest {}).await.ok()?;
                 Some(resp.into_inner().version)
             }
@@ -122,7 +120,7 @@ impl VersionAwareWorkerResolver {
     fn start_version_filtering(
         all_worker_urls: Vec<Url>,
         expected_version: String,
-        channel_resolver: Arc<DefaultChannelResolver>,
+        channel_resolver: Arc<grpc::DefaultChannelResolver>,
     ) -> Self {
         let compatible_urls = Arc::new(RwLock::new(vec![]));
 

docs/source/learn/01-concepts.md

@@ -52,7 +52,7 @@ to serialized plans that get executed over the wire.
 Users are expected to build these and spawn them in ports so that the network
 boundary nodes can reach them.
 
-## [WorkerResolver](https://github.com/datafusion-contrib/datafusion-distributed/blob/main/src/networking/worker_resolver.rs)
+## [WorkerResolver](https://github.com/datafusion-contrib/datafusion-distributed/blob/main/src/worker_resolver.rs)
 
 Determines the available workers in the Distributed DataFusion cluster by
 returning their URLs.
@@ -87,7 +87,7 @@ logic can select the appropriate data subset. For example, task 0 of 3 might
 return the first third of rows, task 2 the last third, and so on. See the
 `TaskEstimator` documentation for guidance on which approach to use.
 
-## [ChannelResolver](https://github.com/datafusion-contrib/datafusion-distributed/blob/main/src/networking/channel_resolver.rs)
+## [ChannelResolver](https://github.com/datafusion-contrib/datafusion-distributed/blob/main/src/protocol/channel_resolver.rs)
 
 Optional extension trait that allows to customize how connections are
 established to workers. Given one of the URLs returned by the `WorkerResolver`,

docs/source/user-guide/06-channel-resolver.md

@@ -20,34 +20,34 @@ pub trait ChannelResolver {
     async fn get_worker_client_for_url(
         &self,
         url: &Url,
-    ) -> Result<WorkerServiceClient<BoxCloneSyncChannel>, DataFusionError>;
+    ) -> Result<Box<dyn WorkerChannel>, DataFusionError>;
 }
 ```
 
 ```{note}
 `get_worker_client_for_url` is called on **every** gRPC request. Reuse clients
 rather than reconnecting each time, or you'll open a fresh connection per request.
-The easiest way is to build on `DefaultChannelResolver` (which caches channels),
-or to use the `create_worker_client` helper.
+The easiest way is to build on `grpc::DefaultChannelResolver` (which caches channels),
+or to use the `grpc::create_worker_client` helper.
 ```
 
 ## Providing your own
 
-The simplest custom resolver wraps `DefaultChannelResolver`, delegating to it for
+The simplest custom resolver wraps `grpc::DefaultChannelResolver`, delegating to it for
 channel caching and only customizing what you need:
 
 ```rust
 #[derive(Clone)]
 struct CustomChannelResolver {
-    inner: DefaultChannelResolver,
+    inner: grpc::DefaultChannelResolver,
 }
 
 #[async_trait]
 impl ChannelResolver for CustomChannelResolver {
     async fn get_worker_client_for_url(
         &self,
         url: &Url,
-    ) -> Result<WorkerServiceClient<BoxCloneSyncChannel>, DataFusionError> {
+    ) -> Result<Box<dyn WorkerChannel>, DataFusionError> {
         // Delegate to the default (cached channels), or build your own client
         // here — e.g. wrapped in tower layers, or on a custom runtime.
         self.inner.get_worker_client_for_url(url).await
@@ -62,7 +62,7 @@ too:
 
 ```rust
 let channel_resolver = CustomChannelResolver {
-    inner: DefaultChannelResolver::default(),
+    inner: grpc::DefaultChannelResolver::default(),
 };
 
 // On the coordinating context:

examples/custom_distributed_partial_reduction_tree.md

@@ -3,7 +3,7 @@
 Demonstrates how to **build your own distributed plan by injecting network boundaries directly**,
 instead of relying on the automatic distributed planner to decide where the stages go.
 
-See the [Custom Distributed Plans user guide](../docs/source/user-guide/custom-distributed-plans.md)
+See the [Custom Distributed Plans guide](../docs/source/advanced/05-custom-distributed-plans.md)
 for the concepts.
 
 ## What it shows

examples/custom_worker_url_routing.md

@@ -3,7 +3,7 @@
 Demonstrates **custom task routing** for **cache affinity**: consistently routing each parquet file
 to the *same* worker so that worker can serve it from an in-memory cache on repeat queries. This is
 the
-[`TaskEstimator::route_tasks`](../docs/source/user-guide/task-estimator.md#routing-tasks-to-specific-workers)
+[`TaskEstimator::route_tasks`](../docs/source/advanced/06-worker-routing.md)
 API.
 
 ## Scenario

examples/work_unit_feed.md

@@ -4,7 +4,7 @@ Demonstrates a **work unit feed**: a distributed leaf node whose work is discove
 coordinator *at runtime* and streamed to the workers as the query executes, instead of being fully
 known at planning time.
 
-See the [Work Unit Feeds user guide](../docs/source/user-guide/work-unit-feeds.md) for the concepts.
+See the [Work Unit Feeds guide](../docs/source/advanced/04-work-unit-feeds.md) for the concepts.
 
 ## Scenario