risingwave_pb/
lib.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![allow(unfulfilled_lint_expectations)]
16#![allow(clippy::doc_overindented_list_items)]
17// for derived code of `Message`
18#![expect(clippy::doc_markdown)]
19#![expect(clippy::upper_case_acronyms)]
20#![expect(clippy::needless_lifetimes)]
21// For tonic::transport::Endpoint::connect
22#![expect(clippy::disallowed_methods)]
23#![expect(clippy::enum_variant_names)]
24#![expect(clippy::module_inception)]
25// FIXME: This should be fixed!!! https://github.com/risingwavelabs/risingwave/issues/19906
26#![expect(clippy::large_enum_variant)]
27
28use std::str::FromStr;
29
30use event_recovery::RecoveryEvent;
31use plan_common::AdditionalColumn;
32pub use prost::Message;
33use risingwave_error::tonic::ToTonicStatus;
34use thiserror::Error;
35
36use crate::common::WorkerType;
37use crate::meta::event_log::event_recovery;
38use crate::stream_plan::PbStreamScanType;
39
40#[rustfmt::skip]
41#[cfg_attr(madsim, path = "sim/catalog.rs")]
42pub mod catalog;
43#[rustfmt::skip]
44#[cfg_attr(madsim, path = "sim/common.rs")]
45pub mod common;
46#[rustfmt::skip]
47#[cfg_attr(madsim, path = "sim/compute.rs")]
48pub mod compute;
49#[rustfmt::skip]
50#[cfg_attr(madsim, path = "sim/cloud_service.rs")]
51pub mod cloud_service;
52#[rustfmt::skip]
53#[cfg_attr(madsim, path = "sim/data.rs")]
54pub mod data;
55#[rustfmt::skip]
56#[cfg_attr(madsim, path = "sim/ddl_service.rs")]
57pub mod ddl_service;
58#[rustfmt::skip]
59#[cfg_attr(madsim, path = "sim/expr.rs")]
60pub mod expr;
61#[rustfmt::skip]
62#[cfg_attr(madsim, path = "sim/meta.rs")]
63pub mod meta;
64#[rustfmt::skip]
65#[cfg_attr(madsim, path = "sim/plan_common.rs")]
66pub mod plan_common;
67#[rustfmt::skip]
68#[cfg_attr(madsim, path = "sim/batch_plan.rs")]
69pub mod batch_plan;
70#[rustfmt::skip]
71#[cfg_attr(madsim, path = "sim/task_service.rs")]
72pub mod task_service;
73#[rustfmt::skip]
74#[cfg_attr(madsim, path = "sim/connector_service.rs")]
75pub mod connector_service;
76#[rustfmt::skip]
77#[cfg_attr(madsim, path = "sim/stream_plan.rs")]
78pub mod stream_plan;
79#[rustfmt::skip]
80#[cfg_attr(madsim, path = "sim/stream_service.rs")]
81pub mod stream_service;
82#[rustfmt::skip]
83#[cfg_attr(madsim, path = "sim/hummock.rs")]
84pub mod hummock;
85#[rustfmt::skip]
86#[cfg_attr(madsim, path = "sim/compactor.rs")]
87pub mod compactor;
88#[rustfmt::skip]
89#[cfg_attr(madsim, path = "sim/user.rs")]
90pub mod user;
91#[rustfmt::skip]
92#[cfg_attr(madsim, path = "sim/source.rs")]
93pub mod source;
94#[rustfmt::skip]
95#[cfg_attr(madsim, path = "sim/monitor_service.rs")]
96pub mod monitor_service;
97#[rustfmt::skip]
98#[cfg_attr(madsim, path = "sim/backup_service.rs")]
99pub mod backup_service;
100#[rustfmt::skip]
101#[cfg_attr(madsim, path = "sim/serverless_backfill_controller.rs")]
102pub mod serverless_backfill_controller;
103#[rustfmt::skip]
104#[cfg_attr(madsim, path = "sim/frontend_service.rs")]
105pub mod frontend_service;
106#[rustfmt::skip]
107#[cfg_attr(madsim, path = "sim/java_binding.rs")]
108pub mod java_binding;
109#[rustfmt::skip]
110#[cfg_attr(madsim, path = "sim/health.rs")]
111pub mod health;
112#[rustfmt::skip]
113#[path = "sim/telemetry.rs"]
114pub mod telemetry;
115#[rustfmt::skip]
116#[cfg_attr(madsim, path = "sim/iceberg_compaction.rs")]
117pub mod iceberg_compaction;
118
119#[rustfmt::skip]
120#[path = "sim/secret.rs"]
121pub mod secret;
122#[rustfmt::skip]
123#[path = "connector_service.serde.rs"]
124pub mod connector_service_serde;
125#[rustfmt::skip]
126#[path = "catalog.serde.rs"]
127pub mod catalog_serde;
128#[rustfmt::skip]
129#[path = "common.serde.rs"]
130pub mod common_serde;
131#[rustfmt::skip]
132#[path = "compute.serde.rs"]
133pub mod compute_serde;
134#[rustfmt::skip]
135#[path = "cloud_service.serde.rs"]
136pub mod cloud_service_serde;
137#[rustfmt::skip]
138#[path = "data.serde.rs"]
139pub mod data_serde;
140#[rustfmt::skip]
141#[path = "ddl_service.serde.rs"]
142pub mod ddl_service_serde;
143#[rustfmt::skip]
144#[path = "expr.serde.rs"]
145pub mod expr_serde;
146#[rustfmt::skip]
147#[path = "meta.serde.rs"]
148pub mod meta_serde;
149#[rustfmt::skip]
150#[path = "plan_common.serde.rs"]
151pub mod plan_common_serde;
152#[rustfmt::skip]
153#[path = "batch_plan.serde.rs"]
154pub mod batch_plan_serde;
155#[rustfmt::skip]
156#[path = "task_service.serde.rs"]
157pub mod task_service_serde;
158#[rustfmt::skip]
159#[path = "stream_plan.serde.rs"]
160pub mod stream_plan_serde;
161#[rustfmt::skip]
162#[path = "stream_service.serde.rs"]
163pub mod stream_service_serde;
164#[rustfmt::skip]
165#[path = "hummock.serde.rs"]
166pub mod hummock_serde;
167#[rustfmt::skip]
168#[path = "compactor.serde.rs"]
169pub mod compactor_serde;
170#[rustfmt::skip]
171#[path = "user.serde.rs"]
172pub mod user_serde;
173#[rustfmt::skip]
174#[path = "source.serde.rs"]
175pub mod source_serde;
176#[rustfmt::skip]
177#[path = "monitor_service.serde.rs"]
178pub mod monitor_service_serde;
179#[rustfmt::skip]
180#[path = "backup_service.serde.rs"]
181pub mod backup_service_serde;
182#[rustfmt::skip]
183#[path = "java_binding.serde.rs"]
184pub mod java_binding_serde;
185#[rustfmt::skip]
186#[path = "telemetry.serde.rs"]
187pub mod telemetry_serde;
188
189#[rustfmt::skip]
190#[path = "secret.serde.rs"]
191pub mod secret_serde;
192#[rustfmt::skip]
193#[path = "serverless_backfill_controller.serde.rs"]
194pub mod serverless_backfill_controller_serde;
195
196#[derive(Clone, PartialEq, Eq, Debug, Error)]
197#[error("field `{0}` not found")]
198pub struct PbFieldNotFound(pub &'static str);
199
200impl From<PbFieldNotFound> for tonic::Status {
201    fn from(e: PbFieldNotFound) -> Self {
202        e.to_status_unnamed(tonic::Code::Internal)
203    }
204}
205
206impl FromStr for crate::expr::table_function::PbType {
207    type Err = ();
208
209    fn from_str(s: &str) -> Result<Self, Self::Err> {
210        Self::from_str_name(&s.to_uppercase()).ok_or(())
211    }
212}
213
214impl FromStr for crate::expr::agg_call::PbKind {
215    type Err = ();
216
217    fn from_str(s: &str) -> Result<Self, Self::Err> {
218        Self::from_str_name(&s.to_uppercase()).ok_or(())
219    }
220}
221
222impl stream_plan::MaterializeNode {
223    pub fn dist_key_indices(&self) -> Vec<u32> {
224        self.get_table()
225            .unwrap()
226            .distribution_key
227            .iter()
228            .map(|i| *i as u32)
229            .collect()
230    }
231
232    pub fn column_descs(&self) -> Vec<plan_common::PbColumnDesc> {
233        self.get_table()
234            .unwrap()
235            .columns
236            .iter()
237            .map(|c| c.get_column_desc().unwrap().clone())
238            .collect()
239    }
240}
241
242impl stream_plan::StreamScanNode {
243    /// See [`Self::upstream_column_ids`].
244    pub fn upstream_columns(&self) -> Vec<plan_common::PbColumnDesc> {
245        self.upstream_column_ids
246            .iter()
247            .map(|id| {
248                (self.table_desc.as_ref().unwrap().columns.iter())
249                    .find(|c| c.column_id == *id)
250                    .unwrap()
251                    .clone()
252            })
253            .collect()
254    }
255}
256
257impl stream_plan::SourceBackfillNode {
258    pub fn column_descs(&self) -> Vec<plan_common::PbColumnDesc> {
259        self.columns
260            .iter()
261            .map(|c| c.column_desc.as_ref().unwrap().clone())
262            .collect()
263    }
264}
265
266// Encapsulating the use of parallelism.
267impl common::WorkerNode {
268    pub fn compute_node_parallelism(&self) -> usize {
269        assert_eq!(self.r#type(), WorkerType::ComputeNode);
270        self.property
271            .as_ref()
272            .expect("property should be exist")
273            .parallelism as usize
274    }
275
276    pub fn parallelism(&self) -> Option<usize> {
277        if WorkerType::ComputeNode == self.r#type() {
278            Some(self.compute_node_parallelism())
279        } else {
280            None
281        }
282    }
283
284    pub fn resource_group(&self) -> Option<String> {
285        self.property
286            .as_ref()
287            .and_then(|p| p.resource_group.clone())
288    }
289}
290
291impl stream_plan::SourceNode {
292    pub fn column_descs(&self) -> Option<Vec<plan_common::PbColumnDesc>> {
293        Some(
294            self.source_inner
295                .as_ref()?
296                .columns
297                .iter()
298                .map(|c| c.get_column_desc().unwrap().clone())
299                .collect(),
300        )
301    }
302}
303
304impl meta::table_fragments::ActorStatus {
305    pub fn worker_id(&self) -> u32 {
306        self.location
307            .as_ref()
308            .expect("actor location should be exist")
309            .worker_node_id
310    }
311}
312
313impl common::WorkerNode {
314    pub fn is_streaming_schedulable(&self) -> bool {
315        let property = self.property.as_ref();
316        property.is_some_and(|p| p.is_streaming) && !property.is_some_and(|p| p.is_unschedulable)
317    }
318}
319
320impl common::ActorLocation {
321    pub fn from_worker(worker_node_id: u32) -> Option<Self> {
322        Some(Self { worker_node_id })
323    }
324}
325
326impl meta::event_log::EventRecovery {
327    pub fn event_type(&self) -> &str {
328        match self.recovery_event.as_ref() {
329            Some(RecoveryEvent::DatabaseStart(_)) => "DATABASE_RECOVERY_START",
330            Some(RecoveryEvent::DatabaseSuccess(_)) => "DATABASE_RECOVERY_SUCCESS",
331            Some(RecoveryEvent::DatabaseFailure(_)) => "DATABASE_RECOVERY_FAILURE",
332            Some(RecoveryEvent::GlobalStart(_)) => "GLOBAL_RECOVERY_START",
333            Some(RecoveryEvent::GlobalSuccess(_)) => "GLOBAL_RECOVERY_SUCCESS",
334            Some(RecoveryEvent::GlobalFailure(_)) => "GLOBAL_RECOVERY_FAILURE",
335            None => "UNKNOWN_RECOVERY_EVENT",
336        }
337    }
338
339    pub fn database_recovery_start(database_id: u32) -> Self {
340        Self {
341            recovery_event: Some(RecoveryEvent::DatabaseStart(
342                event_recovery::DatabaseRecoveryStart { database_id },
343            )),
344        }
345    }
346
347    pub fn database_recovery_failure(database_id: u32) -> Self {
348        Self {
349            recovery_event: Some(RecoveryEvent::DatabaseFailure(
350                event_recovery::DatabaseRecoveryFailure { database_id },
351            )),
352        }
353    }
354
355    pub fn database_recovery_success(database_id: u32) -> Self {
356        Self {
357            recovery_event: Some(RecoveryEvent::DatabaseSuccess(
358                event_recovery::DatabaseRecoverySuccess { database_id },
359            )),
360        }
361    }
362
363    pub fn global_recovery_start(reason: String) -> Self {
364        Self {
365            recovery_event: Some(RecoveryEvent::GlobalStart(
366                event_recovery::GlobalRecoveryStart { reason },
367            )),
368        }
369    }
370
371    pub fn global_recovery_success(
372        reason: String,
373        duration_secs: f32,
374        running_database_ids: Vec<u32>,
375        recovering_database_ids: Vec<u32>,
376    ) -> Self {
377        Self {
378            recovery_event: Some(RecoveryEvent::GlobalSuccess(
379                event_recovery::GlobalRecoverySuccess {
380                    reason,
381                    duration_secs,
382                    running_database_ids,
383                    recovering_database_ids,
384                },
385            )),
386        }
387    }
388
389    pub fn global_recovery_failure(reason: String, error: String) -> Self {
390        Self {
391            recovery_event: Some(RecoveryEvent::GlobalFailure(
392                event_recovery::GlobalRecoveryFailure { reason, error },
393            )),
394        }
395    }
396}
397
398impl stream_plan::StreamNode {
399    /// Find the external stream source info inside the stream node, if any.
400    ///
401    /// Returns `source_id`.
402    pub fn find_stream_source(&self) -> Option<u32> {
403        if let Some(crate::stream_plan::stream_node::NodeBody::Source(source)) =
404            self.node_body.as_ref()
405            && let Some(inner) = &source.source_inner
406        {
407            return Some(inner.source_id);
408        }
409
410        for child in &self.input {
411            if let Some(source) = child.find_stream_source() {
412                return Some(source);
413            }
414        }
415
416        None
417    }
418
419    /// Find the external stream source info inside the stream node, if any.
420    ///
421    /// Returns (`source_id`, `upstream_source_fragment_id`).
422    ///
423    /// Note: we must get upstream fragment id from the merge node, not from the fragment's
424    /// `upstream_fragment_ids`. e.g., DynamicFilter may have 2 upstream fragments, but only
425    /// one is the upstream source fragment.
426    pub fn find_source_backfill(&self) -> Option<(u32, u32)> {
427        if let Some(crate::stream_plan::stream_node::NodeBody::SourceBackfill(source)) =
428            self.node_body.as_ref()
429        {
430            if let crate::stream_plan::stream_node::NodeBody::Merge(merge) =
431                self.input[0].node_body.as_ref().unwrap()
432            {
433                // Note: avoid using `merge.upstream_actor_id` to prevent misuse.
434                // See comments there for details.
435                return Some((source.upstream_source_id, merge.upstream_fragment_id));
436            } else {
437                unreachable!(
438                    "source backfill must have a merge node as its input: {:?}",
439                    self
440                );
441            }
442        }
443
444        for child in &self.input {
445            if let Some(source) = child.find_source_backfill() {
446                return Some(source);
447            }
448        }
449
450        None
451    }
452}
453impl stream_plan::Dispatcher {
454    pub fn as_strategy(&self) -> stream_plan::DispatchStrategy {
455        stream_plan::DispatchStrategy {
456            r#type: self.r#type,
457            dist_key_indices: self.dist_key_indices.clone(),
458            output_mapping: self.output_mapping.clone(),
459        }
460    }
461}
462
463impl stream_plan::DispatchOutputMapping {
464    /// Create a mapping that forwards all columns.
465    pub fn identical(len: usize) -> Self {
466        Self {
467            indices: (0..len as u32).collect(),
468            types: Vec::new(),
469        }
470    }
471
472    /// Create a mapping that forwards columns with given indices, without type conversion.
473    pub fn simple(indices: Vec<u32>) -> Self {
474        Self {
475            indices,
476            types: Vec::new(),
477        }
478    }
479
480    /// Assert that this mapping does not involve type conversion and return the indices.
481    pub fn into_simple_indices(self) -> Vec<u32> {
482        assert!(
483            self.types.is_empty(),
484            "types must be empty for simple mapping"
485        );
486        self.indices
487    }
488}
489
490impl catalog::StreamSourceInfo {
491    /// Refer to [`Self::cdc_source_job`] for details.
492    pub fn is_shared(&self) -> bool {
493        self.cdc_source_job
494    }
495}
496
497impl stream_plan::PbStreamScanType {
498    pub fn is_reschedulable(&self) -> bool {
499        match self {
500            // todo: should this be true?
501            PbStreamScanType::UpstreamOnly => false,
502            PbStreamScanType::ArrangementBackfill => true,
503            PbStreamScanType::CrossDbSnapshotBackfill => true,
504            PbStreamScanType::SnapshotBackfill => true,
505            _ => false,
506        }
507    }
508}
509
510impl catalog::Sink {
511    // TODO: remove this placeholder
512    // creating table sink does not have an id, so we need a placeholder
513    pub const UNIQUE_IDENTITY_FOR_CREATING_TABLE_SINK: &'static str = "PLACE_HOLDER";
514
515    pub fn unique_identity(&self) -> String {
516        // TODO: use a more unique name
517        format!("{}", self.id)
518    }
519}
520
521impl std::fmt::Debug for meta::SystemParams {
522    /// Directly formatting `SystemParams` can be inaccurate or leak sensitive information.
523    ///
524    /// Use `SystemParamsReader` instead.
525    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
526        f.debug_struct("SystemParams").finish_non_exhaustive()
527    }
528}
529
530// More compact formats for debugging
531
532impl std::fmt::Debug for data::DataType {
533    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
534        let data::DataType {
535            precision,
536            scale,
537            interval_type,
538            field_type,
539            field_names,
540            field_ids,
541            type_name,
542            // currently all data types are nullable
543            is_nullable: _,
544        } = self;
545
546        let type_name = data::data_type::TypeName::try_from(*type_name)
547            .map(|t| t.as_str_name())
548            .unwrap_or("Unknown");
549
550        let mut s = f.debug_struct(type_name);
551        if self.precision != 0 {
552            s.field("precision", precision);
553        }
554        if self.scale != 0 {
555            s.field("scale", scale);
556        }
557        if self.interval_type != 0 {
558            s.field("interval_type", interval_type);
559        }
560        if !self.field_type.is_empty() {
561            s.field("field_type", field_type);
562        }
563        if !self.field_names.is_empty() {
564            s.field("field_names", field_names);
565        }
566        if !self.field_ids.is_empty() {
567            s.field("field_ids", field_ids);
568        }
569        s.finish()
570    }
571}
572
573impl std::fmt::Debug for plan_common::column_desc::GeneratedOrDefaultColumn {
574    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
575        match self {
576            Self::GeneratedColumn(arg0) => f.debug_tuple("GeneratedColumn").field(arg0).finish(),
577            Self::DefaultColumn(arg0) => f.debug_tuple("DefaultColumn").field(arg0).finish(),
578        }
579    }
580}
581
582impl std::fmt::Debug for plan_common::ColumnDesc {
583    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
584        // destruct here to avoid missing new fields in the future.
585        let plan_common::ColumnDesc {
586            column_type,
587            column_id,
588            name,
589            description,
590            additional_column_type,
591            additional_column,
592            generated_or_default_column,
593            version,
594            nullable,
595        } = self;
596
597        let mut s = f.debug_struct("ColumnDesc");
598        if let Some(column_type) = column_type {
599            s.field("column_type", column_type);
600        } else {
601            s.field("column_type", &"Unknown");
602        }
603        s.field("column_id", column_id).field("name", name);
604        if let Some(description) = description {
605            s.field("description", description);
606        }
607        if self.additional_column_type != 0 {
608            s.field("additional_column_type", additional_column_type);
609        }
610        s.field("version", version);
611        if let Some(AdditionalColumn {
612            column_type: Some(column_type),
613        }) = additional_column
614        {
615            // AdditionalColumn { None } means a normal column
616            s.field("additional_column", &column_type);
617        }
618        if let Some(generated_or_default_column) = generated_or_default_column {
619            s.field("generated_or_default_column", &generated_or_default_column);
620        }
621        s.field("nullable", nullable);
622        s.finish()
623    }
624}
625
626impl expr::UserDefinedFunction {
627    pub fn name_in_runtime(&self) -> Option<&str> {
628        if self.version() < expr::UdfExprVersion::NameInRuntime {
629            if self.language == "rust" || self.language == "wasm" {
630                // The `identifier` value of Rust and WASM UDF before `NameInRuntime`
631                // is not used any more. The real bound function name should be the same
632                // as `name`.
633                Some(&self.name)
634            } else {
635                // `identifier`s of other UDFs already mean `name_in_runtime` before `NameInRuntime`.
636                self.identifier.as_deref()
637            }
638        } else {
639            // after `PbUdfExprVersion::NameInRuntime`, `identifier` means `name_in_runtime`
640            self.identifier.as_deref()
641        }
642    }
643}
644
645impl expr::UserDefinedFunctionMetadata {
646    pub fn name_in_runtime(&self) -> Option<&str> {
647        if self.version() < expr::UdfExprVersion::NameInRuntime {
648            if self.language == "rust" || self.language == "wasm" {
649                // The `identifier` value of Rust and WASM UDF before `NameInRuntime`
650                // is not used any more. And unfortunately, we don't have the original name
651                // in `PbUserDefinedFunctionMetadata`, so we need to extract the name from
652                // the old `identifier` value (e.g. `foo()->int32`).
653                let old_identifier = self
654                    .identifier
655                    .as_ref()
656                    .expect("Rust/WASM UDF must have identifier");
657                Some(
658                    old_identifier
659                        .split_once("(")
660                        .expect("the old identifier must contain `(`")
661                        .0,
662                )
663            } else {
664                // `identifier`s of other UDFs already mean `name_in_runtime` before `NameInRuntime`.
665                self.identifier.as_deref()
666            }
667        } else {
668            // after `PbUdfExprVersion::NameInRuntime`, `identifier` means `name_in_runtime`
669            self.identifier.as_deref()
670        }
671    }
672}
673
674#[cfg(test)]
675mod tests {
676    use crate::data::{DataType, data_type};
677    use crate::plan_common::Field;
678    use crate::stream_plan::stream_node::NodeBody;
679
680    #[test]
681    fn test_getter() {
682        let data_type: DataType = DataType {
683            is_nullable: true,
684            ..Default::default()
685        };
686        let field = Field {
687            data_type: Some(data_type),
688            name: "".to_owned(),
689        };
690        assert!(field.get_data_type().unwrap().is_nullable);
691    }
692
693    #[test]
694    fn test_enum_getter() {
695        let mut data_type: DataType = DataType::default();
696        data_type.type_name = data_type::TypeName::Double as i32;
697        assert_eq!(
698            data_type::TypeName::Double,
699            data_type.get_type_name().unwrap()
700        );
701    }
702
703    #[test]
704    fn test_enum_unspecified() {
705        let mut data_type: DataType = DataType::default();
706        data_type.type_name = data_type::TypeName::TypeUnspecified as i32;
707        assert!(data_type.get_type_name().is_err());
708    }
709
710    #[test]
711    fn test_primitive_getter() {
712        let data_type: DataType = DataType::default();
713        let new_data_type = DataType {
714            is_nullable: data_type.get_is_nullable(),
715            ..Default::default()
716        };
717        assert!(!new_data_type.is_nullable);
718    }
719
720    #[test]
721    fn test_size() {
722        use static_assertions::const_assert_eq;
723        // box all fields in NodeBody to avoid large_enum_variant
724        // see https://github.com/risingwavelabs/risingwave/issues/19910
725        const_assert_eq!(std::mem::size_of::<NodeBody>(), 16);
726    }
727}
risingwave_pb/lib.rs

risingwave_pb/
lib.rs
