Rename all arrow flight endpoint references to "Worker" (#274)

Author: gabotechs

README.md

@@ -35,10 +35,10 @@ https://datafusion-contrib.github.io/datafusion-distributed
 There are some runnable examples showcasing how to provide a localhost implementation for Distributed DataFusion in
 [examples/](examples):
 
-- [localhost_worker.rs](examples/localhost_worker.rs): code that spawns an Arrow Flight Endpoint listening for physical
+- [localhost_worker.rs](examples/localhost_worker.rs): code that spawns a Worker listening for physical
   plans over the network.
-- [localhost_run.rs](examples/localhost_run.rs): code that distributes a query across the spawned Arrow Flight Endpoints
-  and executes it.
+- [localhost_run.rs](examples/localhost_run.rs): code that distributes a query across the spawned Workers and executes
+  it.
 
 The integration tests also provide an idea about how to use the library and what can be achieved with it:
 

benchmarks/cdk/bin/worker.rs

@@ -9,8 +9,8 @@ use datafusion::execution::SessionStateBuilder;
 use datafusion::physical_plan::execute_stream;
 use datafusion::prelude::SessionContext;
 use datafusion_distributed::{
-    ArrowFlightEndpoint, DistributedExt, DistributedPhysicalOptimizerRule,
-    DistributedSessionBuilderContext, WorkerResolver, display_plan_ascii,
+    DistributedExt, DistributedPhysicalOptimizerRule, Worker, WorkerQueryContext, WorkerResolver,
+    display_plan_ascii,
 };
 use futures::{StreamExt, TryFutureExt};
 use log::{error, info, warn};
@@ -72,12 +72,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .build();
     let ctx = SessionContext::from(state);
 
-    let arrow_flight_endpoint =
-        ArrowFlightEndpoint::from_session_builder(move |ctx: DistributedSessionBuilderContext| {
-            let s3 = s3.clone();
-            let s3_url = s3_url.clone();
-            async move { Ok(ctx.builder.with_object_store(&s3_url, s3).build()) }
-        });
+    let arrow_flight_endpoint = Worker::from_session_builder(move |ctx: WorkerQueryContext| {
+        let s3 = s3.clone();
+        let s3_url = s3_url.clone();
+        async move { Ok(ctx.builder.with_object_store(&s3_url, s3).build()) }
+    });
     let http_server = axum::serve(
         listener,
         Router::new().route(

benchmarks/src/run.rs

@@ -31,7 +31,7 @@ use datafusion::prelude::*;
 use datafusion_distributed::test_utils::localhost::LocalHostWorkerResolver;
 use datafusion_distributed::test_utils::{clickbench, tpcds, tpch};
 use datafusion_distributed::{
-    ArrowFlightEndpoint, DistributedExt, DistributedPhysicalOptimizerRule, NetworkBoundaryExt,
+    DistributedExt, DistributedPhysicalOptimizerRule, NetworkBoundaryExt, Worker,
 };
 use log::info;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
@@ -185,7 +185,7 @@ impl RunOpt {
                 let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);
                 Ok::<_, Box<dyn Error + Send + Sync>>(
                     Server::builder()
-                        .add_service(ArrowFlightEndpoint::default().into_flight_server())
+                        .add_service(Worker::default().into_flight_server())
                         .serve_with_incoming(incoming)
                         .await?,
                 )

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

@@ -13,7 +13,7 @@ points:
 
 - You will need to provide your own implementation in two places:
     - in the `SessionContext` that handles your queries.
-    - while instantiating the `ArrowFlightEndpoint` with the `from_session_builder()` constructor.
+    - while instantiating the `Worker` with the `from_session_builder()` constructor.
 - If you decide to build it from scratch, make sure that Arrow Flight clients are reused across
   request rather than always building new ones.
 - You can use this library's `DefaultChannelResolver` as a backbone for your own implementation.
@@ -44,10 +44,10 @@ async fn main() {
         .with_distributed_channel_resolver(channel_resolver.clone())
         .build();
 
-    let endpoint = ArrowFlightEndpoint::from_session_builder(|ctx: DistributedSessionBuilderContext| {
+    // ... and here for each query the Worker handles.
+    let endpoint = Worker::from_session_builder(move |ctx: WorkerQueryContext| {
         let channel_resolver = channel_resolver.clone();
         async move {
-            // ... and here for each query that ArrowFlightEndpoint handles.
             Ok(ctx.builder.with_distributed_channel_resolver(channel_resolver).build())
         }
     });

docs/source/user-guide/getting-started.md

@@ -40,15 +40,16 @@ impl ChannelResolver for LocalhostChannelResolver {
 
 > NOTE: This example is not production-ready and is meant to showcase the basic concepts of the library.
 
-This `WorkerResolver` implementation should resolve URLs of Distributed DataFusion Arrow Flight servers, and it's
-also the user of this library's responsibility to spawn a Tonic server that exposes the Arrow Flight service.
+This `WorkerResolver` implementation should resolve URLs of Distributed DataFusion workers, and it's
+also the user of this library's responsibility to spawn a Tonic server that exposes the worker as an Arrow Flight
+service using Tonic.
 
 A basic example of such a server is:
 
 ```rust
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
-    let endpoint = ArrowFlightEndpoint::default();
+    let endpoint = Worker::default();
 
     Server::builder()
         .add_service(endpoint.into_flight_server())
@@ -66,7 +67,7 @@ The next two sections of this guide will walk you through tailoring the library'
 - [Build your own WorkerResolver](worker-resolver.md)
 - [Build your own ChannelResolver](channel-resolver.md)
 - [Build your own TaskEstimator](task-estimator.md)
-- [Build your own distributed DataFusion Arrow Flight endpoint](arrow-flight-endpoint.md)
+- [Build your own distributed DataFusion Worker](worker.md)
 
 Here are some other resources in the codebase:
 

docs/source/user-guide/index.md

@@ -10,7 +10,7 @@ your own distributed DataFusion cluster.
 - [Concepts](concepts.md)
 - [Getting Started](getting-started.md)
 - [Building a WorkerResolver](worker-resolver.md)
-- [Building an Arrow Flight endpoint](arrow-flight-endpoint.md)
+- [Spawning a Worker](worker.md)
 - [Building a ChannelResolver](channel-resolver.md)
 - [Building a TaskEstimator](task-estimator.md)
 - [How a distributed plan is built](how-a-distributed-plan-is-built.md)

docs/source/user-guide/worker-resolver.md

@@ -30,13 +30,13 @@ async fn main() {
 }
 ```
 
-> NOTE: It's not necessary to pass this to the Arrow Flight endpoint session builder.
+> NOTE: It's not necessary to pass this to the Worker session builder.
 
 ## Static WorkerResolver
 
 This is the simplest thing you can do, although it will not fit some common use cases. An example of this can be
-seen
-in [localhost_worker.rs](https://github.com/datafusion-contrib/datafusion-distributed/blob/fad9fa222d65b7d2ddae92fbc20082b5c434e4ff/examples/localhost_run.rs)
+seen in the
+[localhost_worker.rs](https://github.com/datafusion-contrib/datafusion-distributed/blob/fad9fa222d65b7d2ddae92fbc20082b5c434e4ff/examples/localhost_run.rs)
 example:
 
 ```rust
@@ -65,8 +65,8 @@ provider for hosting your Distributed DataFusion workers.
 It's up to you to decide how the URLs should be resolved. One important implementation note is:
 
 - Retrieving the worker URLs needs to be synchronous (as planning is synchronous), and therefore, you'll likely
-  need to spawn a background tasks that periodically refreshes the list of available URLs.
+  need to spawn background tasks that periodically refresh the list of available URLs.
 
 A good example can be found
 in https://github.com/datafusion-contrib/datafusion-distributed/blob/main/benchmarks/cdk/bin/worker.rs,
-where a cluster of AWS EC2 machines are discovered identified by tags with the AWS Rust SDK.
+where a cluster of AWS EC2 machines is discovered identified by tags with the AWS Rust SDK.

…urce/user-guide/arrow-flight-endpoint.md docs/source/user-guide/worker.mddocs/source/user-guide/arrow-flight-endpoint.md renamed to docs/source/user-guide/worker.md docs/source/user-guide/arrow-flight-endpoint.md renamed to docs/source/user-guide/worker.md

@@ -1,26 +1,26 @@
-# Building an Arrow Flight Endpoint
+# Spawn a Worker
 
-An `ArrowFlightEndpoint` is a gRPC server that implements the Arrow Flight protocol for distributed query execution.
+A `Worker` is a gRPC server that implements the Arrow Flight protocol for distributed query execution.
 Worker nodes run these endpoints to receive execution plans, execute them, and stream results back.
 
 ## Overview
 
-The `ArrowFlightEndpoint` is the core worker component in Distributed DataFusion. It:
+The `Worker` is the core worker component in Distributed DataFusion. It:
 
 - Receives serialized execution plans via Arrow Flight's `do_get` method
 - Deserializes plans using protobuf and user-provided codecs
 - Executes plans using the local DataFusion runtime
 - Streams results back as Arrow record batches through the gRPC Arrow Flight interface
 
-## Creating an ArrowFlightEndpoint
+## Launching the Arrow Flight server
 
-The `Default` implementation of the `ArrowFlightEndpoint` should satisfy most basic use cases
+The `Default` implementation of the `Worker` should satisfy most basic use cases
 
 ```rust
-use datafusion_distributed::{ArrowFlightEndpoint};
+use datafusion_distributed::Worker;
 
 async fn main() {
-    let endpoint = ArrowFlightEndpoint::default();
+    let endpoint = Worker::default();
 
     Server::builder()
         .add_service(endpoint.into_flight_server())
@@ -31,21 +31,21 @@ async fn main() {
 }
 ```
 
-If you are using DataFusion though, it's very likely that you have your own custom UDFs, execution nodes, config options
-etc...
+If you are using DataFusion, though, it's very likely that you have your own custom UDFs, execution nodes, config
+options, etc...
 
-You'll need to tell the `ArrowFlightEndpoint` how to build your DataFusion sessions:
+You'll need to tell the `Worker` how to build your DataFusion sessions:
 
 ```rust
-async fn build_state(ctx: DistributedSessionBuilderContext) -> Result<SessionState, DataFusionError> {
+async fn build_state(ctx: WorkerQueryContext) -> Result<SessionState, DataFusionError> {
     Ok(ctx
         .builder
         .with_scalar_functions(vec![your_custom_udf()])
         .build())
 }
 
 async fn main() {
-    let endpoint = ArrowFlightEndpoint::from_session_builder(build_sate);
+    let endpoint = Worker::from_session_builder(build_sate);
 
     Server::builder()
         .add_service(endpoint.into_flight_server())
@@ -56,21 +56,21 @@ async fn main() {
 }
 ```
 
-### DistributedSessionBuilder
+### WorkerSessionBuilder
 
-The `DistributedSessionBuilder` is a closure or type that implements:
+The `WorkerSessionBuilder` is a closure or type that implements:
 
 ```rust
 #[async_trait]
-pub trait DistributedSessionBuilder {
+pub trait WorkerSessionBuilder {
     async fn build_session_state(
         &self,
-        ctx: DistributedSessionBuilderContext,
+        ctx: WorkerQueryContext,
     ) -> Result<SessionState, DataFusionError>;
 }
 ```
 
-It receives a `DistributedSessionBuilderContext` containing:
+It receives a `WorkerQueryContext` containing:
 
 - `SessionStateBuilder`: A pre-populated session state builder in which you can inject your custom stuff
 - `headers`: HTTP headers from the incoming request (useful for passing metadata)
@@ -81,11 +81,11 @@ Convert the endpoint to a gRPC service and serve it:
 
 ```rust
 use tonic::transport::Server;
-use datafusion_distributed::ArrowFlightEndpoint;
+use datafusion_distributed::Worker;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 
 async fn main() {
-    let endpoint = ArrowFlightEndpoint::default();
+    let endpoint = Worker::default();
 
     Server::builder()
         .add_service(endpoint.into_flight_server())

examples/in_memory_cluster.rs

@@ -5,9 +5,8 @@ use datafusion::common::DataFusionError;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::prelude::{ParquetReadOptions, SessionContext};
 use datafusion_distributed::{
-    ArrowFlightEndpoint, BoxCloneSyncChannel, ChannelResolver, DistributedExt,
-    DistributedPhysicalOptimizerRule, DistributedSessionBuilderContext, WorkerResolver,
-    create_flight_client, display_plan_ascii,
+    BoxCloneSyncChannel, ChannelResolver, DistributedExt, DistributedPhysicalOptimizerRule, Worker,
+    WorkerQueryContext, WorkerResolver, create_flight_client, display_plan_ascii,
 };
 use futures::TryStreamExt;
 use hyper_util::rt::TokioIo;
@@ -89,12 +88,10 @@ impl InMemoryChannelResolver {
         };
         let this_clone = this.clone();
 
-        let endpoint = ArrowFlightEndpoint::from_session_builder(
-            move |ctx: DistributedSessionBuilderContext| {
-                let this = this.clone();
-                async move { Ok(ctx.builder.with_distributed_channel_resolver(this).build()) }
-            },
-        );
+        let endpoint = Worker::from_session_builder(move |ctx: WorkerQueryContext| {
+            let this = this.clone();
+            async move { Ok(ctx.builder.with_distributed_channel_resolver(this).build()) }
+        });
 
         tokio::spawn(async move {
             Server::builder()

examples/localhost.md

@@ -16,7 +16,7 @@ git lfs checkout
 
 ### Spawning the workers
 
-In two different terminals spawn two ArrowFlightEndpoints
+In two different terminals spawn two workers
 
 ```shell
 cargo run --example localhost_worker -- 8080
@@ -26,7 +26,7 @@ cargo run --example localhost_worker -- 8080
 cargo run --example localhost_worker -- 8081
 ```
 
-The positional numeric argument is the port in which each Arrow Flight endpoint will listen to.
+The positional numeric argument is the port in which each worker will listen to.
 
 ### Issuing a distributed SQL query