datafusion-distributed 2.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")]
    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),
    }
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct WorkerToCoordinatorMsg {
    #[prost(oneof = "worker_to_coordinator_msg::Inner", tags = "1")]
    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),
    }
}
#[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>,
}
#[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;
    }
}