risingwave_pb/
stream_service.rs

1// This file is @generated by prost-build.
2#[derive(prost_helpers::AnyPB)]
3#[derive(Clone, PartialEq, ::prost::Message)]
4pub struct InjectBarrierRequest {
5    #[prost(string, tag = "1")]
6    pub request_id: ::prost::alloc::string::String,
7    #[prost(message, optional, tag = "2")]
8    pub barrier: ::core::option::Option<super::stream_plan::Barrier>,
9    #[prost(uint32, repeated, tag = "4", wrapper = "crate::id::ActorId")]
10    pub actor_ids_to_collect: ::prost::alloc::vec::Vec<crate::id::ActorId>,
11    #[prost(uint32, repeated, tag = "5", wrapper = "crate::id::TableId")]
12    pub table_ids_to_sync: ::prost::alloc::vec::Vec<crate::id::TableId>,
13    #[prost(uint64, tag = "6", wrapper = "crate::id::PartialGraphId")]
14    pub partial_graph_id: crate::id::PartialGraphId,
15    #[prost(message, repeated, tag = "9")]
16    pub actors_to_build: ::prost::alloc::vec::Vec<
17        inject_barrier_request::FragmentBuildActorInfo,
18    >,
19}
20/// Nested message and enum types in `InjectBarrierRequest`.
21pub mod inject_barrier_request {
22    #[derive(prost_helpers::AnyPB)]
23    #[derive(Clone, PartialEq, ::prost::Message)]
24    pub struct FragmentBuildActorInfo {
25        #[prost(uint32, tag = "1", wrapper = "crate::id::FragmentId")]
26        pub fragment_id: crate::id::FragmentId,
27        #[prost(message, optional, tag = "2")]
28        pub node: ::core::option::Option<super::super::stream_plan::StreamNode>,
29        #[prost(message, repeated, tag = "3")]
30        pub actors: ::prost::alloc::vec::Vec<BuildActorInfo>,
31    }
32    #[derive(prost_helpers::AnyPB)]
33    #[derive(Clone, PartialEq, ::prost::Message)]
34    pub struct BuildActorInfo {
35        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
36        pub actor_id: crate::id::ActorId,
37        #[prost(map = "uint32, message", tag = "2", wrapper = "crate::id::FragmentId")]
38        pub fragment_upstreams: ::std::collections::HashMap<
39            crate::id::FragmentId,
40            build_actor_info::UpstreamActors,
41        >,
42        #[prost(message, repeated, tag = "3")]
43        pub dispatchers: ::prost::alloc::vec::Vec<super::super::stream_plan::Dispatcher>,
44        #[prost(message, optional, tag = "4")]
45        pub vnode_bitmap: ::core::option::Option<super::super::common::Buffer>,
46        #[prost(string, tag = "5")]
47        pub mview_definition: ::prost::alloc::string::String,
48        #[prost(message, optional, tag = "6")]
49        pub expr_context: ::core::option::Option<super::super::plan_common::ExprContext>,
50        #[prost(string, tag = "9")]
51        pub config_override: ::prost::alloc::string::String,
52        #[prost(uint32, repeated, tag = "7", wrapper = "crate::id::SubscriberId")]
53        pub initial_subscriber_ids: ::prost::alloc::vec::Vec<crate::id::SubscriberId>,
54    }
55    /// Nested message and enum types in `BuildActorInfo`.
56    pub mod build_actor_info {
57        #[derive(prost_helpers::AnyPB)]
58        #[derive(Clone, PartialEq, ::prost::Message)]
59        pub struct UpstreamActors {
60            #[prost(message, repeated, tag = "1")]
61            pub actors: ::prost::alloc::vec::Vec<super::super::super::common::ActorInfo>,
62        }
63    }
64}
65#[derive(prost_helpers::AnyPB)]
66#[derive(Clone, PartialEq, ::prost::Message)]
67pub struct BarrierCompleteResponse {
68    #[prost(string, tag = "1")]
69    pub request_id: ::prost::alloc::string::String,
70    #[prost(message, optional, tag = "2")]
71    pub status: ::core::option::Option<super::common::Status>,
72    #[prost(message, repeated, tag = "3")]
73    pub create_mview_progress: ::prost::alloc::vec::Vec<
74        barrier_complete_response::CreateMviewProgress,
75    >,
76    #[prost(message, repeated, tag = "4")]
77    pub synced_sstables: ::prost::alloc::vec::Vec<
78        barrier_complete_response::LocalSstableInfo,
79    >,
80    #[prost(uint32, tag = "5", wrapper = "crate::id::WorkerId")]
81    pub worker_id: crate::id::WorkerId,
82    #[prost(map = "uint32, message", tag = "6", wrapper = "crate::id::TableId")]
83    pub table_watermarks: ::std::collections::HashMap<
84        crate::id::TableId,
85        super::hummock::TableWatermarks,
86    >,
87    #[prost(message, repeated, tag = "7")]
88    pub old_value_sstables: ::prost::alloc::vec::Vec<super::hummock::SstableInfo>,
89    #[prost(uint64, tag = "8", wrapper = "crate::id::PartialGraphId")]
90    pub partial_graph_id: crate::id::PartialGraphId,
91    /// prev_epoch of barrier
92    #[prost(uint64, tag = "9")]
93    pub epoch: u64,
94    #[prost(message, repeated, tag = "11")]
95    pub load_finished_sources: ::prost::alloc::vec::Vec<
96        barrier_complete_response::LoadFinishedSource,
97    >,
98    #[prost(map = "uint32, message", tag = "12", wrapper = "crate::id::TableId")]
99    pub vector_index_adds: ::std::collections::HashMap<
100        crate::id::TableId,
101        super::hummock::vector_index_delta::VectorIndexAdds,
102    >,
103    #[prost(message, repeated, tag = "13")]
104    pub cdc_table_backfill_progress: ::prost::alloc::vec::Vec<
105        barrier_complete_response::CdcTableBackfillProgress,
106    >,
107    #[prost(message, repeated, tag = "16")]
108    pub list_finished_sources: ::prost::alloc::vec::Vec<
109        barrier_complete_response::ListFinishedSource,
110    >,
111    #[prost(message, repeated, tag = "17")]
112    pub cdc_source_offset_updated: ::prost::alloc::vec::Vec<
113        barrier_complete_response::CdcSourceOffsetUpdated,
114    >,
115    #[prost(message, repeated, tag = "18")]
116    pub iceberg_pk_index_sink_metadata: ::prost::alloc::vec::Vec<
117        barrier_complete_response::IcebergPkIndexSinkMetadata,
118    >,
119    #[prost(message, repeated, tag = "19")]
120    pub refresh_finished_actors: ::prost::alloc::vec::Vec<
121        barrier_complete_response::RefreshFinishedActor,
122    >,
123}
124/// Nested message and enum types in `BarrierCompleteResponse`.
125pub mod barrier_complete_response {
126    #[derive(prost_helpers::AnyPB)]
127    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
128    pub struct CreateMviewProgress {
129        /// Note: ideally we should use `executor_id`, but `actor_id` is ok-ish.
130        /// See <<https://github.com/risingwavelabs/risingwave/issues/6236>.>
131        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
132        pub backfill_actor_id: crate::id::ActorId,
133        #[prost(bool, tag = "2")]
134        pub done: bool,
135        /// MV backfill snapshot read epoch (0 for Done / Source backfill)
136        #[prost(uint64, tag = "3")]
137        pub consumed_epoch: u64,
138        /// MV backfill snapshot read rows / Source backfilled rows
139        #[prost(uint64, tag = "4")]
140        pub consumed_rows: u64,
141        /// Buffered rows that are yet to be consumed (used by locality backfill to report precise progress)
142        #[prost(uint64, tag = "6")]
143        pub buffered_rows: u64,
144        #[prost(uint32, tag = "7", wrapper = "crate::id::FragmentId")]
145        pub fragment_id: crate::id::FragmentId,
146    }
147    #[derive(prost_helpers::AnyPB)]
148    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
149    pub struct CdcTableBackfillProgress {
150        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
151        pub actor_id: crate::id::ActorId,
152        #[prost(uint64, tag = "2")]
153        pub epoch: u64,
154        #[prost(bool, tag = "3")]
155        pub done: bool,
156        #[prost(int64, tag = "4")]
157        pub split_id_start_inclusive: i64,
158        #[prost(int64, tag = "5")]
159        pub split_id_end_inclusive: i64,
160        #[prost(uint64, tag = "6")]
161        pub generation: u64,
162        #[prost(uint32, tag = "7", wrapper = "crate::id::FragmentId")]
163        pub fragment_id: crate::id::FragmentId,
164    }
165    #[derive(prost_helpers::AnyPB)]
166    #[derive(Clone, PartialEq, ::prost::Message)]
167    pub struct LocalSstableInfo {
168        #[prost(message, optional, tag = "2")]
169        pub sst: ::core::option::Option<super::super::hummock::SstableInfo>,
170        #[prost(map = "uint32, message", tag = "3", wrapper = "crate::id::TableId")]
171        pub table_stats_map: ::std::collections::HashMap<
172            crate::id::TableId,
173            super::super::hummock::TableStats,
174        >,
175        #[prost(uint64, tag = "4")]
176        pub created_at: u64,
177    }
178    /// Used for refreshable batch source.
179    /// SourceExecutor reports the source load is finished, and then
180    /// meta will issue a LoadFinish barrier to notify MaterializeExecutor to start diff calculation.
181    #[derive(prost_helpers::AnyPB)]
182    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
183    pub struct LoadFinishedSource {
184        /// The actor that reported the completion event.
185        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
186        pub reporter_actor_id: crate::id::ActorId,
187        /// The table ID for the refreshable batch source.
188        #[prost(uint32, tag = "2", wrapper = "crate::id::TableId")]
189        pub table_id: crate::id::TableId,
190        /// The source identifier associated with the finished load.
191        #[prost(uint32, tag = "3", wrapper = "crate::id::SourceId")]
192        pub associated_source_id: crate::id::SourceId,
193    }
194    /// SourceExecutor reports the source list is finished, and then
195    /// meta will issue a ListFinish barrier to notify the source to start loading.
196    #[derive(prost_helpers::AnyPB)]
197    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
198    pub struct ListFinishedSource {
199        /// The actor that reported the completion event.
200        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
201        pub reporter_actor_id: crate::id::ActorId,
202        /// The table ID for the refreshable batch source.
203        #[prost(uint32, tag = "2", wrapper = "crate::id::TableId")]
204        pub table_id: crate::id::TableId,
205        /// The source identifier associated with the completed listing.
206        #[prost(uint32, tag = "3", wrapper = "crate::id::SourceId")]
207        pub associated_source_id: crate::id::SourceId,
208    }
209    /// SourceExecutor reports that CDC source has updated offset at least once.
210    /// Meta will mark the CDC source as created only after receiving this notification.
211    #[derive(prost_helpers::AnyPB)]
212    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
213    pub struct CdcSourceOffsetUpdated {
214        /// The actor that reported the offset update.
215        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
216        pub reporter_actor_id: crate::id::ActorId,
217        /// The source identifier for the CDC source.
218        #[prost(uint32, tag = "2", wrapper = "crate::id::SourceId")]
219        pub source_id: crate::id::SourceId,
220    }
221    #[derive(prost_helpers::AnyPB)]
222    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
223    pub struct IcebergPkIndexSinkMetadata {
224        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
225        pub reporter_actor_id: crate::id::ActorId,
226        #[prost(uint32, tag = "2", wrapper = "crate::id::SinkId")]
227        pub sink_id: crate::id::SinkId,
228        #[prost(uint64, tag = "3")]
229        pub prev_epoch: u64,
230        #[prost(enumeration = "super::PbIcebergPkIndexSinkRole", tag = "4")]
231        pub role: i32,
232        #[prost(message, optional, tag = "5")]
233        pub metadata: ::core::option::Option<
234            super::super::connector_service::SinkMetadata,
235        >,
236    }
237    /// A materialize actor has finished its part of a table refresh. Once all materialize actors
238    /// of the table have reported, meta finishes the refresh cycle.
239    #[derive(prost_helpers::AnyPB)]
240    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
241    pub struct RefreshFinishedActor {
242        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
243        pub reporter_actor_id: crate::id::ActorId,
244        #[prost(uint32, tag = "2", wrapper = "crate::id::TableId")]
245        pub table_id: crate::id::TableId,
246    }
247}
248#[derive(prost_helpers::AnyPB)]
249#[derive(Clone, PartialEq, ::prost::Message)]
250pub struct StreamingControlStreamRequest {
251    #[prost(oneof = "streaming_control_stream_request::Request", tags = "1, 2, 3, 4, 5")]
252    pub request: ::core::option::Option<streaming_control_stream_request::Request>,
253}
254/// Nested message and enum types in `StreamingControlStreamRequest`.
255pub mod streaming_control_stream_request {
256    #[derive(prost_helpers::AnyPB)]
257    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
258    pub struct InitRequest {}
259    #[derive(prost_helpers::AnyPB)]
260    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
261    pub struct CreatePartialGraphRequest {
262        #[prost(uint64, tag = "1", wrapper = "crate::id::PartialGraphId")]
263        pub partial_graph_id: crate::id::PartialGraphId,
264        /// Identifies the database recovery incarnation of this partial graph.
265        #[prost(string, tag = "3")]
266        pub term_id: ::prost::alloc::string::String,
267    }
268    #[derive(prost_helpers::AnyPB)]
269    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
270    pub struct RemovePartialGraphRequest {
271        #[prost(uint64, repeated, tag = "1", wrapper = "crate::id::PartialGraphId")]
272        pub partial_graph_ids: ::prost::alloc::vec::Vec<crate::id::PartialGraphId>,
273    }
274    #[derive(prost_helpers::AnyPB)]
275    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
276    pub struct ResetPartialGraphsRequest {
277        #[prost(uint64, repeated, tag = "1", wrapper = "crate::id::PartialGraphId")]
278        pub partial_graph_ids: ::prost::alloc::vec::Vec<crate::id::PartialGraphId>,
279    }
280    #[derive(prost_helpers::AnyPB)]
281    #[derive(Clone, PartialEq, ::prost::Oneof)]
282    pub enum Request {
283        #[prost(message, tag = "1")]
284        Init(InitRequest),
285        #[prost(message, tag = "2")]
286        InjectBarrier(super::InjectBarrierRequest),
287        #[prost(message, tag = "3")]
288        RemovePartialGraph(RemovePartialGraphRequest),
289        #[prost(message, tag = "4")]
290        CreatePartialGraph(CreatePartialGraphRequest),
291        #[prost(message, tag = "5")]
292        ResetPartialGraphs(ResetPartialGraphsRequest),
293    }
294}
295#[derive(prost_helpers::AnyPB)]
296#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
297pub struct ScoredError {
298    #[prost(string, tag = "1")]
299    pub err_msg: ::prost::alloc::string::String,
300    #[prost(int32, tag = "2")]
301    pub score: i32,
302}
303#[derive(prost_helpers::AnyPB)]
304#[derive(Clone, PartialEq, ::prost::Message)]
305pub struct StreamingControlStreamResponse {
306    #[prost(
307        oneof = "streaming_control_stream_response::Response",
308        tags = "1, 2, 3, 4, 5"
309    )]
310    pub response: ::core::option::Option<streaming_control_stream_response::Response>,
311}
312/// Nested message and enum types in `StreamingControlStreamResponse`.
313pub mod streaming_control_stream_response {
314    #[derive(prost_helpers::AnyPB)]
315    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
316    pub struct InitResponse {}
317    #[derive(prost_helpers::AnyPB)]
318    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
319    pub struct ShutdownResponse {}
320    #[derive(prost_helpers::AnyPB)]
321    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
322    pub struct ReportPartialGraphFailureResponse {
323        #[prost(uint64, tag = "1", wrapper = "crate::id::PartialGraphId")]
324        pub partial_graph_id: crate::id::PartialGraphId,
325    }
326    #[derive(prost_helpers::AnyPB)]
327    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
328    pub struct ResetPartialGraphResponse {
329        #[prost(uint64, tag = "1", wrapper = "crate::id::PartialGraphId")]
330        pub partial_graph_id: crate::id::PartialGraphId,
331        #[prost(message, optional, tag = "2")]
332        pub root_err: ::core::option::Option<super::ScoredError>,
333    }
334    #[derive(prost_helpers::AnyPB)]
335    #[derive(Clone, PartialEq, ::prost::Oneof)]
336    pub enum Response {
337        #[prost(message, tag = "1")]
338        Init(InitResponse),
339        #[prost(message, tag = "2")]
340        CompleteBarrier(super::BarrierCompleteResponse),
341        #[prost(message, tag = "3")]
342        Shutdown(ShutdownResponse),
343        #[prost(message, tag = "4")]
344        ReportPartialGraphFailure(ReportPartialGraphFailureResponse),
345        #[prost(message, tag = "5")]
346        ResetPartialGraph(ResetPartialGraphResponse),
347    }
348}
349#[derive(prost_helpers::AnyPB)]
350#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
351pub struct GetMinUncommittedObjectIdRequest {}
352#[derive(prost_helpers::AnyPB)]
353#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
354pub struct GetMinUncommittedObjectIdResponse {
355    #[prost(uint64, tag = "1", wrapper = "crate::id::HummockRawObjectId")]
356    pub min_uncommitted_object_id: crate::id::HummockRawObjectId,
357}
358#[derive(prost_helpers::AnyPB)]
359#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
360pub struct WarmUpTableCacheRequest {
361    #[prost(uint32, tag = "1", wrapper = "crate::id::TableId")]
362    pub table_id: crate::id::TableId,
363    #[prost(message, optional, tag = "2")]
364    pub vnode_bitmap: ::core::option::Option<super::common::Buffer>,
365    #[prost(uint64, tag = "3")]
366    pub committed_epoch: u64,
367    /// Maximum number of vnodes scanned concurrently on this compute node.
368    #[prost(uint32, tag = "4")]
369    pub concurrency: u32,
370}
371#[derive(prost_helpers::AnyPB)]
372#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
373pub struct WarmUpTableCacheResponse {
374    #[prost(uint64, tag = "1")]
375    pub key_count: u64,
376}
377#[derive(prost_helpers::AnyPB)]
378#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
379#[repr(i32)]
380pub enum PbIcebergPkIndexSinkRole {
381    Unspecified = 0,
382    Writer = 1,
383    PositionDeleteMerger = 2,
384    CompactionResolver = 3,
385}
386impl PbIcebergPkIndexSinkRole {
387    /// String value of the enum field names used in the ProtoBuf definition.
388    ///
389    /// The values are not transformed in any way and thus are considered stable
390    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
391    pub fn as_str_name(&self) -> &'static str {
392        match self {
393            Self::Unspecified => "PB_ICEBERG_PK_INDEX_SINK_ROLE_UNSPECIFIED",
394            Self::Writer => "PB_ICEBERG_PK_INDEX_SINK_ROLE_WRITER",
395            Self::PositionDeleteMerger => {
396                "PB_ICEBERG_PK_INDEX_SINK_ROLE_POSITION_DELETE_MERGER"
397            }
398            Self::CompactionResolver => {
399                "PB_ICEBERG_PK_INDEX_SINK_ROLE_COMPACTION_RESOLVER"
400            }
401        }
402    }
403    /// Creates an enum from field names used in the ProtoBuf definition.
404    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
405        match value {
406            "PB_ICEBERG_PK_INDEX_SINK_ROLE_UNSPECIFIED" => Some(Self::Unspecified),
407            "PB_ICEBERG_PK_INDEX_SINK_ROLE_WRITER" => Some(Self::Writer),
408            "PB_ICEBERG_PK_INDEX_SINK_ROLE_POSITION_DELETE_MERGER" => {
409                Some(Self::PositionDeleteMerger)
410            }
411            "PB_ICEBERG_PK_INDEX_SINK_ROLE_COMPACTION_RESOLVER" => {
412                Some(Self::CompactionResolver)
413            }
414            _ => None,
415        }
416    }
417}
418/// Generated client implementations.
419pub mod stream_service_client {
420    #![allow(
421        unused_variables,
422        dead_code,
423        missing_docs,
424        clippy::wildcard_imports,
425        clippy::let_unit_value,
426    )]
427    use tonic::codegen::*;
428    use tonic::codegen::http::Uri;
429    #[derive(Debug, Clone)]
430    pub struct StreamServiceClient<T> {
431        inner: tonic::client::Grpc<T>,
432    }
433    impl StreamServiceClient<tonic::transport::Channel> {
434        /// Attempt to create a new client by connecting to a given endpoint.
435        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
436        where
437            D: TryInto<tonic::transport::Endpoint>,
438            D::Error: Into<StdError>,
439        {
440            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
441            Ok(Self::new(conn))
442        }
443    }
444    impl<T> StreamServiceClient<T>
445    where
446        T: tonic::client::GrpcService<tonic::body::Body>,
447        T::Error: Into<StdError>,
448        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
449        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
450    {
451        pub fn new(inner: T) -> Self {
452            let inner = tonic::client::Grpc::new(inner);
453            Self { inner }
454        }
455        pub fn with_origin(inner: T, origin: Uri) -> Self {
456            let inner = tonic::client::Grpc::with_origin(inner, origin);
457            Self { inner }
458        }
459        pub fn with_interceptor<F>(
460            inner: T,
461            interceptor: F,
462        ) -> StreamServiceClient<InterceptedService<T, F>>
463        where
464            F: tonic::service::Interceptor,
465            T::ResponseBody: Default,
466            T: tonic::codegen::Service<
467                http::Request<tonic::body::Body>,
468                Response = http::Response<
469                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
470                >,
471            >,
472            <T as tonic::codegen::Service<
473                http::Request<tonic::body::Body>,
474            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
475        {
476            StreamServiceClient::new(InterceptedService::new(inner, interceptor))
477        }
478        /// Compress requests with the given encoding.
479        ///
480        /// This requires the server to support it otherwise it might respond with an
481        /// error.
482        #[must_use]
483        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
484            self.inner = self.inner.send_compressed(encoding);
485            self
486        }
487        /// Enable decompressing responses.
488        #[must_use]
489        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
490            self.inner = self.inner.accept_compressed(encoding);
491            self
492        }
493        /// Limits the maximum size of a decoded message.
494        ///
495        /// Default: `4MB`
496        #[must_use]
497        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
498            self.inner = self.inner.max_decoding_message_size(limit);
499            self
500        }
501        /// Limits the maximum size of an encoded message.
502        ///
503        /// Default: `usize::MAX`
504        #[must_use]
505        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
506            self.inner = self.inner.max_encoding_message_size(limit);
507            self
508        }
509        pub async fn streaming_control_stream(
510            &mut self,
511            request: impl tonic::IntoStreamingRequest<
512                Message = super::StreamingControlStreamRequest,
513            >,
514        ) -> std::result::Result<
515            tonic::Response<
516                tonic::codec::Streaming<super::StreamingControlStreamResponse>,
517            >,
518            tonic::Status,
519        > {
520            self.inner
521                .ready()
522                .await
523                .map_err(|e| {
524                    tonic::Status::unknown(
525                        format!("Service was not ready: {}", e.into()),
526                    )
527                })?;
528            let codec = tonic_prost::ProstCodec::default();
529            let path = http::uri::PathAndQuery::from_static(
530                "/stream_service.StreamService/StreamingControlStream",
531            );
532            let mut req = request.into_streaming_request();
533            req.extensions_mut()
534                .insert(
535                    GrpcMethod::new(
536                        "stream_service.StreamService",
537                        "StreamingControlStream",
538                    ),
539                );
540            self.inner.streaming(req, path, codec).await
541        }
542        pub async fn get_min_uncommitted_object_id(
543            &mut self,
544            request: impl tonic::IntoRequest<super::GetMinUncommittedObjectIdRequest>,
545        ) -> std::result::Result<
546            tonic::Response<super::GetMinUncommittedObjectIdResponse>,
547            tonic::Status,
548        > {
549            self.inner
550                .ready()
551                .await
552                .map_err(|e| {
553                    tonic::Status::unknown(
554                        format!("Service was not ready: {}", e.into()),
555                    )
556                })?;
557            let codec = tonic_prost::ProstCodec::default();
558            let path = http::uri::PathAndQuery::from_static(
559                "/stream_service.StreamService/GetMinUncommittedObjectId",
560            );
561            let mut req = request.into_request();
562            req.extensions_mut()
563                .insert(
564                    GrpcMethod::new(
565                        "stream_service.StreamService",
566                        "GetMinUncommittedObjectId",
567                    ),
568                );
569            self.inner.unary(req, path, codec).await
570        }
571        pub async fn warm_up_table_cache(
572            &mut self,
573            request: impl tonic::IntoRequest<super::WarmUpTableCacheRequest>,
574        ) -> std::result::Result<
575            tonic::Response<super::WarmUpTableCacheResponse>,
576            tonic::Status,
577        > {
578            self.inner
579                .ready()
580                .await
581                .map_err(|e| {
582                    tonic::Status::unknown(
583                        format!("Service was not ready: {}", e.into()),
584                    )
585                })?;
586            let codec = tonic_prost::ProstCodec::default();
587            let path = http::uri::PathAndQuery::from_static(
588                "/stream_service.StreamService/WarmUpTableCache",
589            );
590            let mut req = request.into_request();
591            req.extensions_mut()
592                .insert(
593                    GrpcMethod::new("stream_service.StreamService", "WarmUpTableCache"),
594                );
595            self.inner.unary(req, path, codec).await
596        }
597    }
598}
599/// Generated server implementations.
600pub mod stream_service_server {
601    #![allow(
602        unused_variables,
603        dead_code,
604        missing_docs,
605        clippy::wildcard_imports,
606        clippy::let_unit_value,
607    )]
608    use tonic::codegen::*;
609    /// Generated trait containing gRPC methods that should be implemented for use with StreamServiceServer.
610    #[async_trait]
611    pub trait StreamService: std::marker::Send + std::marker::Sync + 'static {
612        /// Server streaming response type for the StreamingControlStream method.
613        type StreamingControlStreamStream: tonic::codegen::tokio_stream::Stream<
614                Item = std::result::Result<
615                    super::StreamingControlStreamResponse,
616                    tonic::Status,
617                >,
618            >
619            + std::marker::Send
620            + 'static;
621        async fn streaming_control_stream(
622            &self,
623            request: tonic::Request<
624                tonic::Streaming<super::StreamingControlStreamRequest>,
625            >,
626        ) -> std::result::Result<
627            tonic::Response<Self::StreamingControlStreamStream>,
628            tonic::Status,
629        >;
630        async fn get_min_uncommitted_object_id(
631            &self,
632            request: tonic::Request<super::GetMinUncommittedObjectIdRequest>,
633        ) -> std::result::Result<
634            tonic::Response<super::GetMinUncommittedObjectIdResponse>,
635            tonic::Status,
636        >;
637        async fn warm_up_table_cache(
638            &self,
639            request: tonic::Request<super::WarmUpTableCacheRequest>,
640        ) -> std::result::Result<
641            tonic::Response<super::WarmUpTableCacheResponse>,
642            tonic::Status,
643        >;
644    }
645    #[derive(Debug)]
646    pub struct StreamServiceServer<T> {
647        inner: Arc<T>,
648        accept_compression_encodings: EnabledCompressionEncodings,
649        send_compression_encodings: EnabledCompressionEncodings,
650        max_decoding_message_size: Option<usize>,
651        max_encoding_message_size: Option<usize>,
652    }
653    impl<T> StreamServiceServer<T> {
654        pub fn new(inner: T) -> Self {
655            Self::from_arc(Arc::new(inner))
656        }
657        pub fn from_arc(inner: Arc<T>) -> Self {
658            Self {
659                inner,
660                accept_compression_encodings: Default::default(),
661                send_compression_encodings: Default::default(),
662                max_decoding_message_size: None,
663                max_encoding_message_size: None,
664            }
665        }
666        pub fn with_interceptor<F>(
667            inner: T,
668            interceptor: F,
669        ) -> InterceptedService<Self, F>
670        where
671            F: tonic::service::Interceptor,
672        {
673            InterceptedService::new(Self::new(inner), interceptor)
674        }
675        /// Enable decompressing requests with the given encoding.
676        #[must_use]
677        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
678            self.accept_compression_encodings.enable(encoding);
679            self
680        }
681        /// Compress responses with the given encoding, if the client supports it.
682        #[must_use]
683        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
684            self.send_compression_encodings.enable(encoding);
685            self
686        }
687        /// Limits the maximum size of a decoded message.
688        ///
689        /// Default: `4MB`
690        #[must_use]
691        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
692            self.max_decoding_message_size = Some(limit);
693            self
694        }
695        /// Limits the maximum size of an encoded message.
696        ///
697        /// Default: `usize::MAX`
698        #[must_use]
699        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
700            self.max_encoding_message_size = Some(limit);
701            self
702        }
703    }
704    impl<T, B> tonic::codegen::Service<http::Request<B>> for StreamServiceServer<T>
705    where
706        T: StreamService,
707        B: Body + std::marker::Send + 'static,
708        B::Error: Into<StdError> + std::marker::Send + 'static,
709    {
710        type Response = http::Response<tonic::body::Body>;
711        type Error = std::convert::Infallible;
712        type Future = BoxFuture<Self::Response, Self::Error>;
713        fn poll_ready(
714            &mut self,
715            _cx: &mut Context<'_>,
716        ) -> Poll<std::result::Result<(), Self::Error>> {
717            Poll::Ready(Ok(()))
718        }
719        fn call(&mut self, req: http::Request<B>) -> Self::Future {
720            match req.uri().path() {
721                "/stream_service.StreamService/StreamingControlStream" => {
722                    #[allow(non_camel_case_types)]
723                    struct StreamingControlStreamSvc<T: StreamService>(pub Arc<T>);
724                    impl<
725                        T: StreamService,
726                    > tonic::server::StreamingService<
727                        super::StreamingControlStreamRequest,
728                    > for StreamingControlStreamSvc<T> {
729                        type Response = super::StreamingControlStreamResponse;
730                        type ResponseStream = T::StreamingControlStreamStream;
731                        type Future = BoxFuture<
732                            tonic::Response<Self::ResponseStream>,
733                            tonic::Status,
734                        >;
735                        fn call(
736                            &mut self,
737                            request: tonic::Request<
738                                tonic::Streaming<super::StreamingControlStreamRequest>,
739                            >,
740                        ) -> Self::Future {
741                            let inner = Arc::clone(&self.0);
742                            let fut = async move {
743                                <T as StreamService>::streaming_control_stream(
744                                        &inner,
745                                        request,
746                                    )
747                                    .await
748                            };
749                            Box::pin(fut)
750                        }
751                    }
752                    let accept_compression_encodings = self.accept_compression_encodings;
753                    let send_compression_encodings = self.send_compression_encodings;
754                    let max_decoding_message_size = self.max_decoding_message_size;
755                    let max_encoding_message_size = self.max_encoding_message_size;
756                    let inner = self.inner.clone();
757                    let fut = async move {
758                        let method = StreamingControlStreamSvc(inner);
759                        let codec = tonic_prost::ProstCodec::default();
760                        let mut grpc = tonic::server::Grpc::new(codec)
761                            .apply_compression_config(
762                                accept_compression_encodings,
763                                send_compression_encodings,
764                            )
765                            .apply_max_message_size_config(
766                                max_decoding_message_size,
767                                max_encoding_message_size,
768                            );
769                        let res = grpc.streaming(method, req).await;
770                        Ok(res)
771                    };
772                    Box::pin(fut)
773                }
774                "/stream_service.StreamService/GetMinUncommittedObjectId" => {
775                    #[allow(non_camel_case_types)]
776                    struct GetMinUncommittedObjectIdSvc<T: StreamService>(pub Arc<T>);
777                    impl<
778                        T: StreamService,
779                    > tonic::server::UnaryService<
780                        super::GetMinUncommittedObjectIdRequest,
781                    > for GetMinUncommittedObjectIdSvc<T> {
782                        type Response = super::GetMinUncommittedObjectIdResponse;
783                        type Future = BoxFuture<
784                            tonic::Response<Self::Response>,
785                            tonic::Status,
786                        >;
787                        fn call(
788                            &mut self,
789                            request: tonic::Request<
790                                super::GetMinUncommittedObjectIdRequest,
791                            >,
792                        ) -> Self::Future {
793                            let inner = Arc::clone(&self.0);
794                            let fut = async move {
795                                <T as StreamService>::get_min_uncommitted_object_id(
796                                        &inner,
797                                        request,
798                                    )
799                                    .await
800                            };
801                            Box::pin(fut)
802                        }
803                    }
804                    let accept_compression_encodings = self.accept_compression_encodings;
805                    let send_compression_encodings = self.send_compression_encodings;
806                    let max_decoding_message_size = self.max_decoding_message_size;
807                    let max_encoding_message_size = self.max_encoding_message_size;
808                    let inner = self.inner.clone();
809                    let fut = async move {
810                        let method = GetMinUncommittedObjectIdSvc(inner);
811                        let codec = tonic_prost::ProstCodec::default();
812                        let mut grpc = tonic::server::Grpc::new(codec)
813                            .apply_compression_config(
814                                accept_compression_encodings,
815                                send_compression_encodings,
816                            )
817                            .apply_max_message_size_config(
818                                max_decoding_message_size,
819                                max_encoding_message_size,
820                            );
821                        let res = grpc.unary(method, req).await;
822                        Ok(res)
823                    };
824                    Box::pin(fut)
825                }
826                "/stream_service.StreamService/WarmUpTableCache" => {
827                    #[allow(non_camel_case_types)]
828                    struct WarmUpTableCacheSvc<T: StreamService>(pub Arc<T>);
829                    impl<
830                        T: StreamService,
831                    > tonic::server::UnaryService<super::WarmUpTableCacheRequest>
832                    for WarmUpTableCacheSvc<T> {
833                        type Response = super::WarmUpTableCacheResponse;
834                        type Future = BoxFuture<
835                            tonic::Response<Self::Response>,
836                            tonic::Status,
837                        >;
838                        fn call(
839                            &mut self,
840                            request: tonic::Request<super::WarmUpTableCacheRequest>,
841                        ) -> Self::Future {
842                            let inner = Arc::clone(&self.0);
843                            let fut = async move {
844                                <T as StreamService>::warm_up_table_cache(&inner, request)
845                                    .await
846                            };
847                            Box::pin(fut)
848                        }
849                    }
850                    let accept_compression_encodings = self.accept_compression_encodings;
851                    let send_compression_encodings = self.send_compression_encodings;
852                    let max_decoding_message_size = self.max_decoding_message_size;
853                    let max_encoding_message_size = self.max_encoding_message_size;
854                    let inner = self.inner.clone();
855                    let fut = async move {
856                        let method = WarmUpTableCacheSvc(inner);
857                        let codec = tonic_prost::ProstCodec::default();
858                        let mut grpc = tonic::server::Grpc::new(codec)
859                            .apply_compression_config(
860                                accept_compression_encodings,
861                                send_compression_encodings,
862                            )
863                            .apply_max_message_size_config(
864                                max_decoding_message_size,
865                                max_encoding_message_size,
866                            );
867                        let res = grpc.unary(method, req).await;
868                        Ok(res)
869                    };
870                    Box::pin(fut)
871                }
872                _ => {
873                    Box::pin(async move {
874                        let mut response = http::Response::new(
875                            tonic::body::Body::default(),
876                        );
877                        let headers = response.headers_mut();
878                        headers
879                            .insert(
880                                tonic::Status::GRPC_STATUS,
881                                (tonic::Code::Unimplemented as i32).into(),
882                            );
883                        headers
884                            .insert(
885                                http::header::CONTENT_TYPE,
886                                tonic::metadata::GRPC_CONTENT_TYPE,
887                            );
888                        Ok(response)
889                    })
890                }
891            }
892        }
893    }
894    impl<T> Clone for StreamServiceServer<T> {
895        fn clone(&self) -> Self {
896            let inner = self.inner.clone();
897            Self {
898                inner,
899                accept_compression_encodings: self.accept_compression_encodings,
900                send_compression_encodings: self.send_compression_encodings,
901                max_decoding_message_size: self.max_decoding_message_size,
902                max_encoding_message_size: self.max_encoding_message_size,
903            }
904        }
905    }
906    /// Generated gRPC service name
907    pub const SERVICE_NAME: &str = "stream_service.StreamService";
908    impl<T> tonic::server::NamedService for StreamServiceServer<T> {
909        const NAME: &'static str = SERVICE_NAME;
910    }
911}
risingwave_pb/stream_service.rs

risingwave_pb/
stream_service.rs
