risingwave_pb/

stream_plan.rs

// This file is @generated by prost-build.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Dispatchers {
    #[prost(message, repeated, tag = "1")]
    pub dispatchers: ::prost::alloc::vec::Vec<Dispatcher>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct AddMutation {
    /// New dispatchers for each actor.
    #[prost(map = "uint32, message", tag = "1")]
    pub actor_dispatchers: ::std::collections::HashMap<u32, Dispatchers>,
    /// All actors to be added (to the main connected component of the graph) in this update.
    #[prost(uint32, repeated, tag = "3")]
    pub added_actors: ::prost::alloc::vec::Vec<u32>,
    /// We may embed a source change split mutation here.
    /// `Source` and `SourceBackfill` are handled together here.
    /// TODO: we may allow multiple mutations in a single barrier.
    #[prost(map = "uint32, message", tag = "2")]
    pub actor_splits: ::std::collections::HashMap<u32, super::source::ConnectorSplits>,
    /// We may embed a pause mutation here.
    /// TODO: we may allow multiple mutations in a single barrier.
    #[prost(bool, tag = "4")]
    pub pause: bool,
    #[prost(message, repeated, tag = "5")]
    pub subscriptions_to_add: ::prost::alloc::vec::Vec<SubscriptionUpstreamInfo>,
    /// nodes which should be paused initially.
    #[prost(uint32, repeated, tag = "6")]
    pub backfill_nodes_to_pause: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StopMutation {
    #[prost(uint32, repeated, tag = "1")]
    pub actors: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct UpdateMutation {
    /// Dispatcher updates.
    #[prost(message, repeated, tag = "1")]
    pub dispatcher_update: ::prost::alloc::vec::Vec<update_mutation::DispatcherUpdate>,
    /// Merge updates.
    #[prost(message, repeated, tag = "2")]
    pub merge_update: ::prost::alloc::vec::Vec<update_mutation::MergeUpdate>,
    /// Vnode bitmap updates for each actor.
    #[prost(map = "uint32, message", tag = "3")]
    pub actor_vnode_bitmap_update: ::std::collections::HashMap<
        u32,
        super::common::Buffer,
    >,
    /// All actors to be dropped in this update.
    #[prost(uint32, repeated, tag = "4")]
    pub dropped_actors: ::prost::alloc::vec::Vec<u32>,
    /// Source updates.
    /// `Source` and `SourceBackfill` are handled together here.
    #[prost(map = "uint32, message", tag = "5")]
    pub actor_splits: ::std::collections::HashMap<u32, super::source::ConnectorSplits>,
    /// When modifying the Materialized View, we need to recreate the Dispatcher from the old upstream to the new TableFragment.
    /// Consistent with the semantics in AddMutation.
    #[prost(map = "uint32, message", tag = "6")]
    pub actor_new_dispatchers: ::std::collections::HashMap<u32, Dispatchers>,
}
/// Nested message and enum types in `UpdateMutation`.
pub mod update_mutation {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct DispatcherUpdate {
        /// Dispatcher can be uniquely identified by a combination of actor id and dispatcher id.
        #[prost(uint32, tag = "1")]
        pub actor_id: u32,
        #[prost(uint64, tag = "2")]
        pub dispatcher_id: u64,
        /// The hash mapping for consistent hash.
        /// For dispatcher types other than HASH, this is ignored.
        #[prost(message, optional, tag = "3")]
        pub hash_mapping: ::core::option::Option<super::ActorMapping>,
        /// Added downstream actors.
        #[prost(uint32, repeated, tag = "4")]
        pub added_downstream_actor_id: ::prost::alloc::vec::Vec<u32>,
        /// Removed downstream actors.
        #[prost(uint32, repeated, tag = "5")]
        pub removed_downstream_actor_id: ::prost::alloc::vec::Vec<u32>,
    }
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct MergeUpdate {
        /// Merge executor can be uniquely identified by a combination of actor id and upstream fragment id.
        #[prost(uint32, tag = "1")]
        pub actor_id: u32,
        #[prost(uint32, tag = "2")]
        pub upstream_fragment_id: u32,
        /// - For scaling, this is always `None`.
        /// - For plan change, the upstream fragment will be changed to a new one, and this will be `Some`.
        ///    In this case, all the upstream actors should be removed and replaced by the `new` ones.
        #[prost(uint32, optional, tag = "5")]
        pub new_upstream_fragment_id: ::core::option::Option<u32>,
        /// Added upstream actors.
        #[prost(message, repeated, tag = "3")]
        pub added_upstream_actors: ::prost::alloc::vec::Vec<
            super::super::common::ActorInfo,
        >,
        /// Removed upstream actors.
        /// Note: this is empty for replace job.
        #[prost(uint32, repeated, tag = "4")]
        pub removed_upstream_actor_id: ::prost::alloc::vec::Vec<u32>,
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SourceChangeSplitMutation {
    /// `Source` and `SourceBackfill` are handled together here.
    #[prost(map = "uint32, message", tag = "2")]
    pub actor_splits: ::std::collections::HashMap<u32, super::source::ConnectorSplits>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct PauseMutation {}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct ResumeMutation {}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ThrottleMutation {
    #[prost(map = "uint32, message", tag = "1")]
    pub actor_throttle: ::std::collections::HashMap<u32, throttle_mutation::RateLimit>,
}
/// Nested message and enum types in `ThrottleMutation`.
pub mod throttle_mutation {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
    pub struct RateLimit {
        #[prost(uint32, optional, tag = "1")]
        pub rate_limit: ::core::option::Option<u32>,
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct CombinedMutation {
    #[prost(message, repeated, tag = "1")]
    pub mutations: ::prost::alloc::vec::Vec<BarrierMutation>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct SubscriptionUpstreamInfo {
    /// can either be subscription_id or table_id of creating TableFragments
    #[prost(uint32, tag = "1")]
    pub subscriber_id: u32,
    #[prost(uint32, tag = "2")]
    pub upstream_mv_table_id: u32,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DropSubscriptionsMutation {
    #[prost(message, repeated, tag = "1")]
    pub info: ::prost::alloc::vec::Vec<SubscriptionUpstreamInfo>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ConnectorPropsChangeMutation {
    #[prost(map = "uint32, message", tag = "1")]
    pub connector_props_infos: ::std::collections::HashMap<
        u32,
        connector_props_change_mutation::ConnectorPropsInfo,
    >,
}
/// Nested message and enum types in `ConnectorPropsChangeMutation`.
pub mod connector_props_change_mutation {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct ConnectorPropsInfo {
        #[prost(map = "string, string", tag = "1")]
        pub connector_props_info: ::std::collections::HashMap<
            ::prost::alloc::string::String,
            ::prost::alloc::string::String,
        >,
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StartFragmentBackfillMutation {
    #[prost(uint32, repeated, tag = "1")]
    pub fragment_ids: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BarrierMutation {
    #[prost(
        oneof = "barrier_mutation::Mutation",
        tags = "3, 4, 5, 6, 7, 8, 10, 12, 13, 14, 100"
    )]
    pub mutation: ::core::option::Option<barrier_mutation::Mutation>,
}
/// Nested message and enum types in `BarrierMutation`.
pub mod barrier_mutation {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Oneof)]
    pub enum Mutation {
        /// Add new dispatchers to some actors, used for creating materialized views.
        #[prost(message, tag = "3")]
        Add(super::AddMutation),
        /// Stop a set of actors, used for dropping materialized views. Empty dispatchers will be
        /// automatically removed.
        #[prost(message, tag = "4")]
        Stop(super::StopMutation),
        /// Update outputs and hash mappings for some dispatchers, used for scaling and replace table.
        #[prost(message, tag = "5")]
        Update(super::UpdateMutation),
        /// Change the split of some sources.
        #[prost(message, tag = "6")]
        Splits(super::SourceChangeSplitMutation),
        /// Pause the dataflow of the whole streaming graph, only used for scaling.
        #[prost(message, tag = "7")]
        Pause(super::PauseMutation),
        /// Resume the dataflow of the whole streaming graph, only used for scaling.
        #[prost(message, tag = "8")]
        Resume(super::ResumeMutation),
        /// Throttle specific source exec or backfill exec.
        #[prost(message, tag = "10")]
        Throttle(super::ThrottleMutation),
        /// Drop subscription on mv
        #[prost(message, tag = "12")]
        DropSubscriptions(super::DropSubscriptionsMutation),
        /// Alter sink/connector/source props
        #[prost(message, tag = "13")]
        ConnectorPropsChange(super::ConnectorPropsChangeMutation),
        /// Start backfilling for specific fragments
        /// This is separated from `ThrottleMutation::NO_RATE_LIMIT`.
        /// This is because user may concurrently rate limit + use backfill order control.
        /// If we use rate limit to pause / resume backfill fragments, if user manually
        /// resumes some fragments, this will overwrite the backfill order configuration.
        #[prost(message, tag = "14")]
        StartFragmentBackfill(super::StartFragmentBackfillMutation),
        /// Combined mutation.
        /// Currently, it can only be Add & Update, which is for sink into table.
        #[prost(message, tag = "100")]
        Combined(super::CombinedMutation),
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Barrier {
    #[prost(message, optional, tag = "1")]
    pub epoch: ::core::option::Option<super::data::Epoch>,
    #[prost(message, optional, tag = "3")]
    pub mutation: ::core::option::Option<BarrierMutation>,
    /// Used for tracing.
    #[prost(map = "string, string", tag = "2")]
    pub tracing_context: ::std::collections::HashMap<
        ::prost::alloc::string::String,
        ::prost::alloc::string::String,
    >,
    /// The kind of the barrier.
    #[prost(enumeration = "barrier::BarrierKind", tag = "9")]
    pub kind: i32,
    /// Record the actors that the barrier has passed. Only used for debugging.
    #[prost(uint32, repeated, tag = "255")]
    pub passed_actors: ::prost::alloc::vec::Vec<u32>,
}
/// Nested message and enum types in `Barrier`.
pub mod barrier {
    #[derive(prost_helpers::AnyPB)]
    #[derive(::enum_as_inner::EnumAsInner)]
    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        Hash,
        PartialOrd,
        Ord,
        ::prost::Enumeration
    )]
    #[repr(i32)]
    pub enum BarrierKind {
        Unspecified = 0,
        /// The first barrier after a fresh start or recovery.
        /// There will be no data associated with the previous epoch of the barrier.
        Initial = 1,
        /// A normal barrier. Data should be flushed locally.
        Barrier = 2,
        /// A checkpoint barrier. Data should be synchorized to the shared storage.
        Checkpoint = 3,
    }
    impl BarrierKind {
        /// String value of the enum field names used in the ProtoBuf definition.
        ///
        /// The values are not transformed in any way and thus are considered stable
        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
        pub fn as_str_name(&self) -> &'static str {
            match self {
                Self::Unspecified => "BARRIER_KIND_UNSPECIFIED",
                Self::Initial => "BARRIER_KIND_INITIAL",
                Self::Barrier => "BARRIER_KIND_BARRIER",
                Self::Checkpoint => "BARRIER_KIND_CHECKPOINT",
            }
        }
        /// Creates an enum from field names used in the ProtoBuf definition.
        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
            match value {
                "BARRIER_KIND_UNSPECIFIED" => Some(Self::Unspecified),
                "BARRIER_KIND_INITIAL" => Some(Self::Initial),
                "BARRIER_KIND_BARRIER" => Some(Self::Barrier),
                "BARRIER_KIND_CHECKPOINT" => Some(Self::Checkpoint),
                _ => None,
            }
        }
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Watermark {
    /// The reference to the watermark column in the stream's schema.
    #[prost(message, optional, tag = "1")]
    pub column: ::core::option::Option<super::expr::InputRef>,
    /// The watermark value, there will be no record having a greater value in the watermark column.
    #[prost(message, optional, tag = "3")]
    pub val: ::core::option::Option<super::data::Datum>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamMessage {
    #[prost(oneof = "stream_message::StreamMessage", tags = "1, 2, 3")]
    pub stream_message: ::core::option::Option<stream_message::StreamMessage>,
}
/// Nested message and enum types in `StreamMessage`.
pub mod stream_message {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Oneof)]
    pub enum StreamMessage {
        #[prost(message, tag = "1")]
        StreamChunk(super::super::data::StreamChunk),
        #[prost(message, tag = "2")]
        Barrier(super::Barrier),
        #[prost(message, tag = "3")]
        Watermark(super::Watermark),
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamMessageBatch {
    #[prost(oneof = "stream_message_batch::StreamMessageBatch", tags = "1, 2, 3")]
    pub stream_message_batch: ::core::option::Option<
        stream_message_batch::StreamMessageBatch,
    >,
}
/// Nested message and enum types in `StreamMessageBatch`.
pub mod stream_message_batch {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct BarrierBatch {
        #[prost(message, repeated, tag = "1")]
        pub barriers: ::prost::alloc::vec::Vec<super::Barrier>,
    }
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Oneof)]
    pub enum StreamMessageBatch {
        #[prost(message, tag = "1")]
        StreamChunk(super::super::data::StreamChunk),
        #[prost(message, tag = "2")]
        BarrierBatch(BarrierBatch),
        #[prost(message, tag = "3")]
        Watermark(super::Watermark),
    }
}
/// Hash mapping for compute node. Stores mapping from virtual node to actor id.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ActorMapping {
    #[prost(uint32, repeated, tag = "1")]
    pub original_indices: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, repeated, tag = "2")]
    pub data: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamSource {
    #[prost(uint32, tag = "1")]
    pub source_id: u32,
    #[prost(message, optional, tag = "2")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(uint32, optional, tag = "3")]
    pub row_id_index: ::core::option::Option<u32>,
    #[prost(message, repeated, tag = "4")]
    pub columns: ::prost::alloc::vec::Vec<super::plan_common::ColumnCatalog>,
    #[prost(btree_map = "string, string", tag = "6")]
    pub with_properties: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        ::prost::alloc::string::String,
    >,
    #[prost(message, optional, tag = "7")]
    pub info: ::core::option::Option<super::catalog::StreamSourceInfo>,
    #[prost(string, tag = "8")]
    pub source_name: ::prost::alloc::string::String,
    /// Source rate limit
    #[prost(uint32, optional, tag = "9")]
    pub rate_limit: ::core::option::Option<u32>,
    #[prost(btree_map = "string, message", tag = "10")]
    pub secret_refs: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        super::secret::SecretRef,
    >,
}
/// copy contents from StreamSource to prevent compatibility issues in the future
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamFsFetch {
    #[prost(uint32, tag = "1")]
    pub source_id: u32,
    #[prost(message, optional, tag = "2")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(uint32, optional, tag = "3")]
    pub row_id_index: ::core::option::Option<u32>,
    #[prost(message, repeated, tag = "4")]
    pub columns: ::prost::alloc::vec::Vec<super::plan_common::ColumnCatalog>,
    #[prost(btree_map = "string, string", tag = "6")]
    pub with_properties: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        ::prost::alloc::string::String,
    >,
    #[prost(message, optional, tag = "7")]
    pub info: ::core::option::Option<super::catalog::StreamSourceInfo>,
    #[prost(string, tag = "8")]
    pub source_name: ::prost::alloc::string::String,
    /// Source rate limit
    #[prost(uint32, optional, tag = "9")]
    pub rate_limit: ::core::option::Option<u32>,
    #[prost(btree_map = "string, message", tag = "10")]
    pub secret_refs: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        super::secret::SecretRef,
    >,
}
/// The executor only for receiving barrier from the meta service. It always resides in the leaves
/// of the streaming graph.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct BarrierRecvNode {}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SourceNode {
    /// The source node can contain either a stream source or nothing. So here we extract all
    /// information about stream source to a message, and here it will be an `Option` in Rust.
    #[prost(message, optional, tag = "1")]
    pub source_inner: ::core::option::Option<StreamSource>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamFsFetchNode {
    #[prost(message, optional, tag = "1")]
    pub node_inner: ::core::option::Option<StreamFsFetch>,
}
/// / It's input must be a `MergeNode`, which connects to the upstream source job.
/// / See `StreamSourceScan::adhoc_to_stream_prost` for the plan.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SourceBackfillNode {
    #[prost(uint32, tag = "1")]
    pub upstream_source_id: u32,
    #[prost(uint32, optional, tag = "2")]
    pub row_id_index: ::core::option::Option<u32>,
    #[prost(message, repeated, tag = "3")]
    pub columns: ::prost::alloc::vec::Vec<super::plan_common::ColumnCatalog>,
    #[prost(message, optional, tag = "4")]
    pub info: ::core::option::Option<super::catalog::StreamSourceInfo>,
    #[prost(string, tag = "5")]
    pub source_name: ::prost::alloc::string::String,
    #[prost(btree_map = "string, string", tag = "6")]
    pub with_properties: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        ::prost::alloc::string::String,
    >,
    /// Backfill rate limit
    #[prost(uint32, optional, tag = "7")]
    pub rate_limit: ::core::option::Option<u32>,
    /// `| partition_id | backfill_progress |`
    #[prost(message, optional, tag = "8")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(btree_map = "string, message", tag = "9")]
    pub secret_refs: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        super::secret::SecretRef,
    >,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SinkDesc {
    #[prost(uint32, tag = "1")]
    pub id: u32,
    #[prost(string, tag = "2")]
    pub name: ::prost::alloc::string::String,
    #[prost(string, tag = "3")]
    pub definition: ::prost::alloc::string::String,
    #[prost(message, repeated, tag = "5")]
    pub plan_pk: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    #[prost(uint32, repeated, tag = "6")]
    pub downstream_pk: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, repeated, tag = "7")]
    pub distribution_key: ::prost::alloc::vec::Vec<u32>,
    #[prost(btree_map = "string, string", tag = "8")]
    pub properties: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        ::prost::alloc::string::String,
    >,
    /// to be deprecated
    #[prost(enumeration = "super::catalog::SinkType", tag = "9")]
    pub sink_type: i32,
    #[prost(message, repeated, tag = "10")]
    pub column_catalogs: ::prost::alloc::vec::Vec<super::plan_common::ColumnCatalog>,
    #[prost(string, tag = "11")]
    pub db_name: ::prost::alloc::string::String,
    /// If the sink is from table or mv, this is name of the table/mv. Otherwise
    /// it is the name of the sink itself.
    #[prost(string, tag = "12")]
    pub sink_from_name: ::prost::alloc::string::String,
    #[prost(message, optional, tag = "13")]
    pub format_desc: ::core::option::Option<super::catalog::SinkFormatDesc>,
    #[prost(uint32, optional, tag = "14")]
    pub target_table: ::core::option::Option<u32>,
    #[prost(uint64, optional, tag = "15")]
    pub extra_partition_col_idx: ::core::option::Option<u64>,
    #[prost(btree_map = "string, message", tag = "16")]
    pub secret_refs: ::prost::alloc::collections::BTreeMap<
        ::prost::alloc::string::String,
        super::secret::SecretRef,
    >,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SinkNode {
    #[prost(message, optional, tag = "1")]
    pub sink_desc: ::core::option::Option<SinkDesc>,
    /// A sink with a kv log store should have a table.
    #[prost(message, optional, tag = "2")]
    pub table: ::core::option::Option<super::catalog::Table>,
    #[prost(enumeration = "SinkLogStoreType", tag = "3")]
    pub log_store_type: i32,
    #[prost(uint32, optional, tag = "4")]
    pub rate_limit: ::core::option::Option<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ProjectNode {
    #[prost(message, repeated, tag = "1")]
    pub select_list: ::prost::alloc::vec::Vec<super::expr::ExprNode>,
    /// this two field is expressing a list of usize pair, which means when project receives a
    /// watermark with `watermark_input_cols\[i\]` column index, it should derive a new watermark
    /// with `watermark_output_cols\[i\]`th expression
    #[prost(uint32, repeated, tag = "2")]
    pub watermark_input_cols: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, repeated, tag = "3")]
    pub watermark_output_cols: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, repeated, tag = "4")]
    pub nondecreasing_exprs: ::prost::alloc::vec::Vec<u32>,
    /// Whether there are likely no-op updates in the output chunks, so that eliminating them with
    /// `StreamChunk::eliminate_adjacent_noop_update` could be beneficial.
    #[prost(bool, tag = "5")]
    pub noop_update_hint: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct FilterNode {
    #[prost(message, optional, tag = "1")]
    pub search_condition: ::core::option::Option<super::expr::ExprNode>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct ChangeLogNode {
    /// Whether or not there is an op in the final output.
    #[prost(bool, tag = "1")]
    pub need_op: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct CdcFilterNode {
    #[prost(message, optional, tag = "1")]
    pub search_condition: ::core::option::Option<super::expr::ExprNode>,
    #[prost(uint32, tag = "2")]
    pub upstream_source_id: u32,
}
/// A materialized view is regarded as a table.
/// In addition, we also specify primary key to MV for efficient point lookup during update and deletion.
///
/// The node will be used for both create mv and create index.
/// - When creating mv, `pk == distribution_key == column_orders`.
/// - When creating index, `column_orders` will contain both
///    arrange columns and pk columns, while distribution key will be arrange columns.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct MaterializeNode {
    #[prost(uint32, tag = "1")]
    pub table_id: u32,
    /// Column indexes and orders of primary key.
    #[prost(message, repeated, tag = "2")]
    pub column_orders: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "3")]
    pub table: ::core::option::Option<super::catalog::Table>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct AggCallState {
    #[prost(oneof = "agg_call_state::Inner", tags = "1, 3")]
    pub inner: ::core::option::Option<agg_call_state::Inner>,
}
/// Nested message and enum types in `AggCallState`.
pub mod agg_call_state {
    /// the state is stored in the intermediate state table. used for count/sum/append-only extreme.
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
    pub struct ValueState {}
    /// use the some column of the Upstream's materialization as the AggCall's state, used for extreme/string_agg/array_agg.
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct MaterializedInputState {
        #[prost(message, optional, tag = "1")]
        pub table: ::core::option::Option<super::super::catalog::Table>,
        /// for constructing state table column mapping
        #[prost(uint32, repeated, tag = "2")]
        pub included_upstream_indices: ::prost::alloc::vec::Vec<u32>,
        #[prost(uint32, repeated, tag = "3")]
        pub table_value_indices: ::prost::alloc::vec::Vec<u32>,
        #[prost(message, repeated, tag = "4")]
        pub order_columns: ::prost::alloc::vec::Vec<super::super::common::ColumnOrder>,
    }
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Oneof)]
    pub enum Inner {
        #[prost(message, tag = "1")]
        ValueState(ValueState),
        #[prost(message, tag = "3")]
        MaterializedInputState(MaterializedInputState),
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SimpleAggNode {
    #[prost(message, repeated, tag = "1")]
    pub agg_calls: ::prost::alloc::vec::Vec<super::expr::AggCall>,
    /// Only used for stateless simple agg.
    #[prost(uint32, repeated, tag = "2")]
    pub distribution_key: ::prost::alloc::vec::Vec<u32>,
    #[prost(message, repeated, tag = "3")]
    pub agg_call_states: ::prost::alloc::vec::Vec<AggCallState>,
    #[prost(message, optional, tag = "4")]
    pub intermediate_state_table: ::core::option::Option<super::catalog::Table>,
    /// Whether to optimize for append only stream.
    /// It is true when the input is append-only
    #[prost(bool, tag = "5")]
    pub is_append_only: bool,
    #[prost(map = "uint32, message", tag = "6")]
    pub distinct_dedup_tables: ::std::collections::HashMap<u32, super::catalog::Table>,
    #[prost(uint32, tag = "7")]
    pub row_count_index: u32,
    #[prost(enumeration = "AggNodeVersion", tag = "8")]
    pub version: i32,
    /// Required by the downstream `RowMergeNode`,
    /// currently only used by the `approx_percentile`'s two phase plan
    #[prost(bool, tag = "9")]
    pub must_output_per_barrier: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct HashAggNode {
    #[prost(uint32, repeated, tag = "1")]
    pub group_key: ::prost::alloc::vec::Vec<u32>,
    #[prost(message, repeated, tag = "2")]
    pub agg_calls: ::prost::alloc::vec::Vec<super::expr::AggCall>,
    #[prost(message, repeated, tag = "3")]
    pub agg_call_states: ::prost::alloc::vec::Vec<AggCallState>,
    #[prost(message, optional, tag = "4")]
    pub intermediate_state_table: ::core::option::Option<super::catalog::Table>,
    /// Whether to optimize for append only stream.
    /// It is true when the input is append-only
    #[prost(bool, tag = "5")]
    pub is_append_only: bool,
    #[prost(map = "uint32, message", tag = "6")]
    pub distinct_dedup_tables: ::std::collections::HashMap<u32, super::catalog::Table>,
    #[prost(uint32, tag = "7")]
    pub row_count_index: u32,
    #[prost(bool, tag = "8")]
    pub emit_on_window_close: bool,
    #[prost(enumeration = "AggNodeVersion", tag = "9")]
    pub version: i32,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct TopNNode {
    /// 0 means no limit as limit of 0 means this node should be optimized away
    #[prost(uint64, tag = "1")]
    pub limit: u64,
    #[prost(uint64, tag = "2")]
    pub offset: u64,
    #[prost(message, optional, tag = "3")]
    pub table: ::core::option::Option<super::catalog::Table>,
    #[prost(message, repeated, tag = "4")]
    pub order_by: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    #[prost(bool, tag = "5")]
    pub with_ties: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct GroupTopNNode {
    /// 0 means no limit as limit of 0 means this node should be optimized away
    #[prost(uint64, tag = "1")]
    pub limit: u64,
    #[prost(uint64, tag = "2")]
    pub offset: u64,
    #[prost(uint32, repeated, tag = "3")]
    pub group_key: ::prost::alloc::vec::Vec<u32>,
    #[prost(message, optional, tag = "4")]
    pub table: ::core::option::Option<super::catalog::Table>,
    #[prost(message, repeated, tag = "5")]
    pub order_by: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    #[prost(bool, tag = "6")]
    pub with_ties: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DeltaExpression {
    #[prost(enumeration = "super::expr::expr_node::Type", tag = "1")]
    pub delta_type: i32,
    #[prost(message, optional, tag = "2")]
    pub delta: ::core::option::Option<super::expr::ExprNode>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct InequalityPair {
    /// Input index of greater side of inequality.
    #[prost(uint32, tag = "1")]
    pub key_required_larger: u32,
    /// Input index of less side of inequality.
    #[prost(uint32, tag = "2")]
    pub key_required_smaller: u32,
    /// Whether this condition is used to clean state table of `HashJoinExecutor`.
    #[prost(bool, tag = "3")]
    pub clean_state: bool,
    /// greater >= less + delta_expression, if `None`, it represents that greater >= less
    #[prost(message, optional, tag = "4")]
    pub delta_expression: ::core::option::Option<DeltaExpression>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct HashJoinNode {
    #[prost(enumeration = "super::plan_common::JoinType", tag = "1")]
    pub join_type: i32,
    #[prost(int32, repeated, tag = "2")]
    pub left_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(int32, repeated, tag = "3")]
    pub right_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(message, optional, tag = "4")]
    pub condition: ::core::option::Option<super::expr::ExprNode>,
    #[prost(message, repeated, tag = "5")]
    pub inequality_pairs: ::prost::alloc::vec::Vec<InequalityPair>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "6")]
    pub left_table: ::core::option::Option<super::catalog::Table>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "7")]
    pub right_table: ::core::option::Option<super::catalog::Table>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "8")]
    pub left_degree_table: ::core::option::Option<super::catalog::Table>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "9")]
    pub right_degree_table: ::core::option::Option<super::catalog::Table>,
    /// The output indices of current node
    #[prost(uint32, repeated, tag = "10")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
    /// Left deduped input pk indices. The pk of the left_table and
    /// left_degree_table is  \[left_join_key | left_deduped_input_pk_indices\]
    /// and is expected to be the shortest key which starts with
    /// the join key and satisfies unique constrain.
    #[prost(uint32, repeated, tag = "11")]
    pub left_deduped_input_pk_indices: ::prost::alloc::vec::Vec<u32>,
    /// Right deduped input pk indices. The pk of the right_table and
    /// right_degree_table is  \[right_join_key | right_deduped_input_pk_indices\]
    /// and is expected to be the shortest key which starts with
    /// the join key and satisfies unique constrain.
    #[prost(uint32, repeated, tag = "12")]
    pub right_deduped_input_pk_indices: ::prost::alloc::vec::Vec<u32>,
    #[prost(bool, repeated, tag = "13")]
    pub null_safe: ::prost::alloc::vec::Vec<bool>,
    /// Whether to optimize for append only stream.
    /// It is true when the input is append-only
    #[prost(bool, tag = "14")]
    pub is_append_only: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct AsOfJoinNode {
    #[prost(enumeration = "super::plan_common::AsOfJoinType", tag = "1")]
    pub join_type: i32,
    #[prost(int32, repeated, tag = "2")]
    pub left_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(int32, repeated, tag = "3")]
    pub right_key: ::prost::alloc::vec::Vec<i32>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "4")]
    pub left_table: ::core::option::Option<super::catalog::Table>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "5")]
    pub right_table: ::core::option::Option<super::catalog::Table>,
    /// The output indices of current node
    #[prost(uint32, repeated, tag = "6")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
    /// Left deduped input pk indices. The pk of the left_table and
    /// The pk of the left_table is  \[left_join_key | left_inequality_key | left_deduped_input_pk_indices\]
    /// left_inequality_key is not used but for forward compatibility.
    #[prost(uint32, repeated, tag = "7")]
    pub left_deduped_input_pk_indices: ::prost::alloc::vec::Vec<u32>,
    /// Right deduped input pk indices.
    /// The pk of the right_table is  \[right_join_key | right_inequality_key | right_deduped_input_pk_indices\]
    /// right_inequality_key is not used but for forward compatibility.
    #[prost(uint32, repeated, tag = "8")]
    pub right_deduped_input_pk_indices: ::prost::alloc::vec::Vec<u32>,
    #[prost(bool, repeated, tag = "9")]
    pub null_safe: ::prost::alloc::vec::Vec<bool>,
    #[prost(message, optional, tag = "10")]
    pub asof_desc: ::core::option::Option<super::plan_common::AsOfJoinDesc>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct TemporalJoinNode {
    #[prost(enumeration = "super::plan_common::JoinType", tag = "1")]
    pub join_type: i32,
    #[prost(int32, repeated, tag = "2")]
    pub left_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(int32, repeated, tag = "3")]
    pub right_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(bool, repeated, tag = "4")]
    pub null_safe: ::prost::alloc::vec::Vec<bool>,
    #[prost(message, optional, tag = "5")]
    pub condition: ::core::option::Option<super::expr::ExprNode>,
    /// The output indices of current node
    #[prost(uint32, repeated, tag = "6")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
    /// The table desc of the lookup side table.
    #[prost(message, optional, tag = "7")]
    pub table_desc: ::core::option::Option<super::plan_common::StorageTableDesc>,
    /// The output indices of the lookup side table
    #[prost(uint32, repeated, tag = "8")]
    pub table_output_indices: ::prost::alloc::vec::Vec<u32>,
    /// The state table used for non-append-only temporal join.
    #[prost(message, optional, tag = "9")]
    pub memo_table: ::core::option::Option<super::catalog::Table>,
    /// If it is a nested lool temporal join
    #[prost(bool, tag = "10")]
    pub is_nested_loop: bool,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DynamicFilterNode {
    #[prost(uint32, tag = "1")]
    pub left_key: u32,
    /// Must be one of <, <=, >, >=
    #[prost(message, optional, tag = "2")]
    pub condition: ::core::option::Option<super::expr::ExprNode>,
    /// Left table stores all states with predicate possibly not NULL.
    #[prost(message, optional, tag = "3")]
    pub left_table: ::core::option::Option<super::catalog::Table>,
    /// Right table stores single value from RHS of predicate.
    #[prost(message, optional, tag = "4")]
    pub right_table: ::core::option::Option<super::catalog::Table>,
    /// If the right side's change always make the condition more relaxed.
    /// In other words, make more record in the left side satisfy the condition.
    /// If this is true, we need to store LHS records which do not match the condition in the internal table.
    /// When the condition changes, we will tell downstream to insert the LHS records which now match the condition.
    /// If this is false, we need to store RHS records which match the condition in the internal table.
    /// When the condition changes, we will tell downstream to delete the LHS records which now no longer match the condition.
    #[deprecated]
    #[prost(bool, tag = "5")]
    pub condition_always_relax: bool,
}
/// Delta join with two indexes. This is a pseudo plan node generated on frontend. On meta
/// service, it will be rewritten into lookup joins.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DeltaIndexJoinNode {
    #[prost(enumeration = "super::plan_common::JoinType", tag = "1")]
    pub join_type: i32,
    #[prost(int32, repeated, tag = "2")]
    pub left_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(int32, repeated, tag = "3")]
    pub right_key: ::prost::alloc::vec::Vec<i32>,
    #[prost(message, optional, tag = "4")]
    pub condition: ::core::option::Option<super::expr::ExprNode>,
    /// Table id of the left index.
    #[prost(uint32, tag = "7")]
    pub left_table_id: u32,
    /// Table id of the right index.
    #[prost(uint32, tag = "8")]
    pub right_table_id: u32,
    /// Info about the left index
    #[prost(message, optional, tag = "9")]
    pub left_info: ::core::option::Option<ArrangementInfo>,
    /// Info about the right index
    #[prost(message, optional, tag = "10")]
    pub right_info: ::core::option::Option<ArrangementInfo>,
    /// the output indices of current node
    #[prost(uint32, repeated, tag = "11")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct HopWindowNode {
    #[prost(uint32, tag = "1")]
    pub time_col: u32,
    #[prost(message, optional, tag = "2")]
    pub window_slide: ::core::option::Option<super::data::Interval>,
    #[prost(message, optional, tag = "3")]
    pub window_size: ::core::option::Option<super::data::Interval>,
    #[prost(uint32, repeated, tag = "4")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
    #[prost(message, repeated, tag = "5")]
    pub window_start_exprs: ::prost::alloc::vec::Vec<super::expr::ExprNode>,
    #[prost(message, repeated, tag = "6")]
    pub window_end_exprs: ::prost::alloc::vec::Vec<super::expr::ExprNode>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct MergeNode {
    /// **WARNING**: Use this field with caution.
    ///
    /// `upstream_actor_id` stored in the plan node in `Fragment` meta model cannot be directly used.
    /// See `compose_fragment`.
    /// The field is deprecated because the upstream actor info is provided separately instead of
    /// injected here in the node.
    #[deprecated]
    #[prost(uint32, repeated, packed = "false", tag = "1")]
    pub upstream_actor_id: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, tag = "2")]
    pub upstream_fragment_id: u32,
    /// Type of the upstream dispatcher. If there's always one upstream according to this
    /// type, the compute node may use the `ReceiverExecutor` as an optimization.
    #[prost(enumeration = "DispatcherType", tag = "3")]
    pub upstream_dispatcher_type: i32,
    /// The schema of input columns. TODO: remove this field.
    #[prost(message, repeated, tag = "4")]
    pub fields: ::prost::alloc::vec::Vec<super::plan_common::Field>,
}
/// passed from frontend to meta, used by fragmenter to generate `MergeNode`
/// and maybe `DispatcherNode` later.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ExchangeNode {
    #[prost(message, optional, tag = "1")]
    pub strategy: ::core::option::Option<DispatchStrategy>,
}
/// StreamScanNode reads data from upstream table first, and then pass all events to downstream.
/// It always these 2 inputs in the following order:
/// 1. A MergeNode (as a placeholder) of upstream.
/// 2. A BatchPlanNode for the snapshot read.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamScanNode {
    #[prost(uint32, tag = "1")]
    pub table_id: u32,
    /// The columns from the upstream table that'll be internally required by this stream scan node.
    /// - For non-backfill stream scan node, it's the same as the output columns.
    /// - For backfill stream scan node, there're additionally primary key columns.
    #[prost(int32, repeated, tag = "2")]
    pub upstream_column_ids: ::prost::alloc::vec::Vec<i32>,
    /// The columns to be output by this stream scan node. The index is based on the internal required columns.
    /// - For non-backfill stream scan node, it's simply all the columns.
    /// - For backfill stream scan node, this strips the primary key columns if they're unnecessary.
    #[prost(uint32, repeated, tag = "3")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
    /// Generally, the barrier needs to be rearranged during the MV creation process, so that data can
    /// be flushed to shared buffer periodically, instead of making the first epoch from batch query extra
    /// large. However, in some cases, e.g., shared state, the barrier cannot be rearranged in StreamScanNode.
    /// StreamScanType is used to decide which implementation for the StreamScanNode.
    #[prost(enumeration = "StreamScanType", tag = "4")]
    pub stream_scan_type: i32,
    /// / The state table used by Backfill operator for persisting internal state
    #[prost(message, optional, tag = "5")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    /// The upstream materialized view info used by backfill.
    /// Used iff `ChainType::Backfill`.
    #[prost(message, optional, tag = "7")]
    pub table_desc: ::core::option::Option<super::plan_common::StorageTableDesc>,
    /// The backfill rate limit for the stream scan node.
    #[prost(uint32, optional, tag = "8")]
    pub rate_limit: ::core::option::Option<u32>,
    /// Snapshot read every N barriers
    #[deprecated]
    #[prost(uint32, tag = "9")]
    pub snapshot_read_barrier_interval: u32,
    /// The state table used by ArrangementBackfill to replicate upstream mview's state table.
    /// Used iff `ChainType::ArrangementBackfill`.
    #[prost(message, optional, tag = "10")]
    pub arrangement_table: ::core::option::Option<super::catalog::Table>,
    #[prost(uint64, optional, tag = "11")]
    pub snapshot_backfill_epoch: ::core::option::Option<u64>,
}
/// Config options for CDC backfill
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct StreamCdcScanOptions {
    /// Whether skip the backfill and only consume from upstream.
    #[prost(bool, tag = "1")]
    pub disable_backfill: bool,
    #[prost(uint32, tag = "2")]
    pub snapshot_barrier_interval: u32,
    #[prost(uint32, tag = "3")]
    pub snapshot_batch_size: u32,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamCdcScanNode {
    #[prost(uint32, tag = "1")]
    pub table_id: u32,
    /// The columns from the upstream table that'll be internally required by this stream scan node.
    /// Contains Primary Keys and Output columns.
    #[prost(int32, repeated, tag = "2")]
    pub upstream_column_ids: ::prost::alloc::vec::Vec<i32>,
    /// Strips the primary key columns if they're unnecessary.
    #[prost(uint32, repeated, tag = "3")]
    pub output_indices: ::prost::alloc::vec::Vec<u32>,
    /// The state table used by CdcBackfill operator for persisting internal state
    #[prost(message, optional, tag = "4")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    /// The external table that will be backfilled for CDC.
    #[prost(message, optional, tag = "5")]
    pub cdc_table_desc: ::core::option::Option<super::plan_common::ExternalTableDesc>,
    /// The backfill rate limit for the stream cdc scan node.
    #[prost(uint32, optional, tag = "6")]
    pub rate_limit: ::core::option::Option<u32>,
    /// Whether skip the backfill and only consume from upstream.
    /// keep it for backward compatibility, new stream plan will use `options.disable_backfill`
    #[prost(bool, tag = "7")]
    pub disable_backfill: bool,
    #[prost(message, optional, tag = "8")]
    pub options: ::core::option::Option<StreamCdcScanOptions>,
}
/// BatchPlanNode is used for mv on mv snapshot read.
/// BatchPlanNode is supposed to carry a batch plan that can be optimized with the streaming plan_common.
/// Currently, streaming to batch push down is not yet supported, BatchPlanNode is simply a table scan.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BatchPlanNode {
    #[prost(message, optional, tag = "1")]
    pub table_desc: ::core::option::Option<super::plan_common::StorageTableDesc>,
    #[prost(int32, repeated, tag = "2")]
    pub column_ids: ::prost::alloc::vec::Vec<i32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ArrangementInfo {
    /// Order key of the arrangement, including order by columns and pk from the materialize
    /// executor.
    #[prost(message, repeated, tag = "1")]
    pub arrange_key_orders: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    /// Column descs of the arrangement
    #[prost(message, repeated, tag = "2")]
    pub column_descs: ::prost::alloc::vec::Vec<super::plan_common::ColumnDesc>,
    /// Used to build storage table by stream lookup join of delta join.
    #[prost(message, optional, tag = "4")]
    pub table_desc: ::core::option::Option<super::plan_common::StorageTableDesc>,
    /// Output index columns
    #[prost(uint32, repeated, tag = "5")]
    pub output_col_idx: ::prost::alloc::vec::Vec<u32>,
}
/// Special node for shared state, which will only be produced in fragmenter. ArrangeNode will
/// produce a special Materialize executor, which materializes data for downstream to query.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ArrangeNode {
    /// Info about the arrangement
    #[prost(message, optional, tag = "1")]
    pub table_info: ::core::option::Option<ArrangementInfo>,
    /// Hash key of the materialize node, which is a subset of pk.
    #[prost(uint32, repeated, tag = "2")]
    pub distribution_key: ::prost::alloc::vec::Vec<u32>,
    /// Used for internal table states.
    #[prost(message, optional, tag = "3")]
    pub table: ::core::option::Option<super::catalog::Table>,
}
/// Special node for shared state. LookupNode will join an arrangement with a stream.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct LookupNode {
    /// Join key of the arrangement side
    #[prost(int32, repeated, tag = "1")]
    pub arrange_key: ::prost::alloc::vec::Vec<i32>,
    /// Join key of the stream side
    #[prost(int32, repeated, tag = "2")]
    pub stream_key: ::prost::alloc::vec::Vec<i32>,
    /// Whether to join the current epoch of arrangement
    #[prost(bool, tag = "3")]
    pub use_current_epoch: bool,
    /// Sometimes we need to re-order the output data to meet the requirement of schema.
    /// By default, lookup executor will produce `<arrangement side, stream side>`. We
    /// will then apply the column mapping to the combined result.
    #[prost(int32, repeated, tag = "4")]
    pub column_mapping: ::prost::alloc::vec::Vec<i32>,
    /// Info about the arrangement
    #[prost(message, optional, tag = "7")]
    pub arrangement_table_info: ::core::option::Option<ArrangementInfo>,
    #[prost(oneof = "lookup_node::ArrangementTableId", tags = "5, 6")]
    pub arrangement_table_id: ::core::option::Option<lookup_node::ArrangementTableId>,
}
/// Nested message and enum types in `LookupNode`.
pub mod lookup_node {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
    pub enum ArrangementTableId {
        /// Table Id of the arrangement (when created along with join plan)
        #[prost(uint32, tag = "5")]
        TableId(u32),
        /// Table Id of the arrangement (when using index)
        #[prost(uint32, tag = "6")]
        IndexId(u32),
    }
}
/// WatermarkFilter needs to filter the upstream data by the water mark.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct WatermarkFilterNode {
    /// The watermark descs
    #[prost(message, repeated, tag = "1")]
    pub watermark_descs: ::prost::alloc::vec::Vec<super::catalog::WatermarkDesc>,
    /// The tables used to persist watermarks, the key is vnode.
    #[prost(message, repeated, tag = "2")]
    pub tables: ::prost::alloc::vec::Vec<super::catalog::Table>,
}
/// Acts like a merger, but on different inputs.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct UnionNode {}
/// Special node for shared state. Merge and align barrier from upstreams. Pipe inputs in order.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct LookupUnionNode {
    #[prost(uint32, repeated, tag = "1")]
    pub order: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ExpandNode {
    #[prost(message, repeated, tag = "1")]
    pub column_subsets: ::prost::alloc::vec::Vec<expand_node::Subset>,
}
/// Nested message and enum types in `ExpandNode`.
pub mod expand_node {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct Subset {
        #[prost(uint32, repeated, tag = "1")]
        pub column_indices: ::prost::alloc::vec::Vec<u32>,
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ProjectSetNode {
    #[prost(message, repeated, tag = "1")]
    pub select_list: ::prost::alloc::vec::Vec<super::expr::ProjectSetSelectItem>,
    /// this two field is expressing a list of usize pair, which means when project receives a
    /// watermark with `watermark_input_cols\[i\]` column index, it should derive a new watermark
    /// with `watermark_output_cols\[i\]`th expression
    #[prost(uint32, repeated, tag = "2")]
    pub watermark_input_cols: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, repeated, tag = "3")]
    pub watermark_expr_indices: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, repeated, tag = "4")]
    pub nondecreasing_exprs: ::prost::alloc::vec::Vec<u32>,
}
/// Sorts inputs and outputs ordered data based on watermark.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SortNode {
    /// Persists data above watermark.
    #[prost(message, optional, tag = "1")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    /// Column index of watermark to perform sorting.
    #[prost(uint32, tag = "2")]
    pub sort_column_index: u32,
}
/// Merges two streams from streaming and batch for data manipulation.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DmlNode {
    /// Id of the table on which DML performs.
    #[prost(uint32, tag = "1")]
    pub table_id: u32,
    /// Version of the table.
    #[prost(uint64, tag = "3")]
    pub table_version_id: u64,
    /// Column descriptions of the table.
    #[prost(message, repeated, tag = "2")]
    pub column_descs: ::prost::alloc::vec::Vec<super::plan_common::ColumnDesc>,
    #[prost(uint32, optional, tag = "4")]
    pub rate_limit: ::core::option::Option<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct RowIdGenNode {
    #[prost(uint64, tag = "1")]
    pub row_id_index: u64,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct NowModeUpdateCurrent {}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct NowModeGenerateSeries {
    #[prost(message, optional, tag = "1")]
    pub start_timestamp: ::core::option::Option<super::data::Datum>,
    #[prost(message, optional, tag = "2")]
    pub interval: ::core::option::Option<super::data::Datum>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct NowNode {
    /// Persists emitted 'now'.
    #[prost(message, optional, tag = "1")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(oneof = "now_node::Mode", tags = "101, 102")]
    pub mode: ::core::option::Option<now_node::Mode>,
}
/// Nested message and enum types in `NowNode`.
pub mod now_node {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Oneof)]
    pub enum Mode {
        #[prost(message, tag = "101")]
        UpdateCurrent(super::NowModeUpdateCurrent),
        #[prost(message, tag = "102")]
        GenerateSeries(super::NowModeGenerateSeries),
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ValuesNode {
    #[prost(message, repeated, tag = "1")]
    pub tuples: ::prost::alloc::vec::Vec<values_node::ExprTuple>,
    #[prost(message, repeated, tag = "2")]
    pub fields: ::prost::alloc::vec::Vec<super::plan_common::Field>,
}
/// Nested message and enum types in `ValuesNode`.
pub mod values_node {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct ExprTuple {
        #[prost(message, repeated, tag = "1")]
        pub cells: ::prost::alloc::vec::Vec<super::super::expr::ExprNode>,
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DedupNode {
    #[prost(message, optional, tag = "1")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(uint32, repeated, tag = "2")]
    pub dedup_column_indices: ::prost::alloc::vec::Vec<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct NoOpNode {}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct EowcOverWindowNode {
    #[prost(message, repeated, tag = "1")]
    pub calls: ::prost::alloc::vec::Vec<super::expr::WindowFunction>,
    #[prost(uint32, repeated, tag = "2")]
    pub partition_by: ::prost::alloc::vec::Vec<u32>,
    /// use `repeated` in case of future extension, now only one column is allowed
    #[prost(message, repeated, tag = "3")]
    pub order_by: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    #[prost(message, optional, tag = "4")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct OverWindowNode {
    #[prost(message, repeated, tag = "1")]
    pub calls: ::prost::alloc::vec::Vec<super::expr::WindowFunction>,
    #[prost(uint32, repeated, tag = "2")]
    pub partition_by: ::prost::alloc::vec::Vec<u32>,
    #[prost(message, repeated, tag = "3")]
    pub order_by: ::prost::alloc::vec::Vec<super::common::ColumnOrder>,
    #[prost(message, optional, tag = "4")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(enumeration = "OverWindowCachePolicy", tag = "5")]
    pub cache_policy: i32,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
pub struct LocalApproxPercentileNode {
    #[prost(double, tag = "1")]
    pub base: f64,
    #[prost(uint32, tag = "2")]
    pub percentile_index: u32,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct GlobalApproxPercentileNode {
    #[prost(double, tag = "1")]
    pub base: f64,
    #[prost(double, tag = "2")]
    pub quantile: f64,
    #[prost(message, optional, tag = "3")]
    pub bucket_state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(message, optional, tag = "4")]
    pub count_state_table: ::core::option::Option<super::catalog::Table>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct RowMergeNode {
    #[prost(message, optional, tag = "1")]
    pub lhs_mapping: ::core::option::Option<super::catalog::ColIndexMapping>,
    #[prost(message, optional, tag = "2")]
    pub rhs_mapping: ::core::option::Option<super::catalog::ColIndexMapping>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct SyncLogStoreNode {
    #[prost(message, optional, tag = "1")]
    pub log_store_table: ::core::option::Option<super::catalog::Table>,
    #[prost(uint32, tag = "2")]
    pub pause_duration_ms: u32,
    #[prost(uint32, tag = "3")]
    pub buffer_size: u32,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct MaterializedExprsNode {
    #[prost(message, repeated, tag = "1")]
    pub exprs: ::prost::alloc::vec::Vec<super::expr::ExprNode>,
    #[prost(message, optional, tag = "2")]
    pub state_table: ::core::option::Option<super::catalog::Table>,
    #[prost(uint32, optional, tag = "3")]
    pub state_clean_col_idx: ::core::option::Option<u32>,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamNode {
    /// The id for the operator. This is local per mview.
    /// TODO: should better be a uint32.
    #[prost(uint64, tag = "1")]
    pub operator_id: u64,
    /// Child node in plan aka. upstream nodes in the streaming DAG
    #[prost(message, repeated, tag = "3")]
    pub input: ::prost::alloc::vec::Vec<StreamNode>,
    #[prost(uint32, repeated, tag = "2")]
    pub stream_key: ::prost::alloc::vec::Vec<u32>,
    #[prost(bool, tag = "24")]
    pub append_only: bool,
    #[prost(string, tag = "18")]
    pub identity: ::prost::alloc::string::String,
    /// The schema of the plan node
    #[prost(message, repeated, tag = "19")]
    pub fields: ::prost::alloc::vec::Vec<super::plan_common::Field>,
    #[prost(
        oneof = "stream_node::NodeBody",
        tags = "100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127, 128, 129, 130, 131, 132, 133, 134, 135, 136, 137, 138, 139, 140, 142, 143, 144, 145, 146, 147, 148, 149"
    )]
    pub node_body: ::core::option::Option<stream_node::NodeBody>,
}
/// Nested message and enum types in `StreamNode`.
pub mod stream_node {
    #[derive(prost_helpers::AnyPB)]
    #[derive(::enum_as_inner::EnumAsInner, ::strum::Display, ::strum::EnumDiscriminants)]
    #[derive(Clone, PartialEq, ::prost::Oneof)]
    pub enum NodeBody {
        #[prost(message, tag = "100")]
        Source(::prost::alloc::boxed::Box<super::SourceNode>),
        #[prost(message, tag = "101")]
        Project(::prost::alloc::boxed::Box<super::ProjectNode>),
        #[prost(message, tag = "102")]
        Filter(::prost::alloc::boxed::Box<super::FilterNode>),
        #[prost(message, tag = "103")]
        Materialize(::prost::alloc::boxed::Box<super::MaterializeNode>),
        #[prost(message, tag = "104")]
        StatelessSimpleAgg(::prost::alloc::boxed::Box<super::SimpleAggNode>),
        #[prost(message, tag = "105")]
        SimpleAgg(::prost::alloc::boxed::Box<super::SimpleAggNode>),
        #[prost(message, tag = "106")]
        HashAgg(::prost::alloc::boxed::Box<super::HashAggNode>),
        #[prost(message, tag = "107")]
        AppendOnlyTopN(::prost::alloc::boxed::Box<super::TopNNode>),
        #[prost(message, tag = "108")]
        HashJoin(::prost::alloc::boxed::Box<super::HashJoinNode>),
        #[prost(message, tag = "109")]
        TopN(::prost::alloc::boxed::Box<super::TopNNode>),
        #[prost(message, tag = "110")]
        HopWindow(::prost::alloc::boxed::Box<super::HopWindowNode>),
        #[prost(message, tag = "111")]
        Merge(::prost::alloc::boxed::Box<super::MergeNode>),
        #[prost(message, tag = "112")]
        Exchange(::prost::alloc::boxed::Box<super::ExchangeNode>),
        #[prost(message, tag = "113")]
        StreamScan(::prost::alloc::boxed::Box<super::StreamScanNode>),
        #[prost(message, tag = "114")]
        BatchPlan(::prost::alloc::boxed::Box<super::BatchPlanNode>),
        #[prost(message, tag = "115")]
        Lookup(::prost::alloc::boxed::Box<super::LookupNode>),
        #[prost(message, tag = "116")]
        Arrange(::prost::alloc::boxed::Box<super::ArrangeNode>),
        #[prost(message, tag = "117")]
        LookupUnion(::prost::alloc::boxed::Box<super::LookupUnionNode>),
        #[prost(message, tag = "118")]
        Union(super::UnionNode),
        #[prost(message, tag = "119")]
        DeltaIndexJoin(::prost::alloc::boxed::Box<super::DeltaIndexJoinNode>),
        #[prost(message, tag = "120")]
        Sink(::prost::alloc::boxed::Box<super::SinkNode>),
        #[prost(message, tag = "121")]
        Expand(::prost::alloc::boxed::Box<super::ExpandNode>),
        #[prost(message, tag = "122")]
        DynamicFilter(::prost::alloc::boxed::Box<super::DynamicFilterNode>),
        #[prost(message, tag = "123")]
        ProjectSet(::prost::alloc::boxed::Box<super::ProjectSetNode>),
        #[prost(message, tag = "124")]
        GroupTopN(::prost::alloc::boxed::Box<super::GroupTopNNode>),
        #[prost(message, tag = "125")]
        Sort(::prost::alloc::boxed::Box<super::SortNode>),
        #[prost(message, tag = "126")]
        WatermarkFilter(::prost::alloc::boxed::Box<super::WatermarkFilterNode>),
        #[prost(message, tag = "127")]
        Dml(::prost::alloc::boxed::Box<super::DmlNode>),
        #[prost(message, tag = "128")]
        RowIdGen(::prost::alloc::boxed::Box<super::RowIdGenNode>),
        #[prost(message, tag = "129")]
        Now(::prost::alloc::boxed::Box<super::NowNode>),
        #[prost(message, tag = "130")]
        AppendOnlyGroupTopN(::prost::alloc::boxed::Box<super::GroupTopNNode>),
        #[prost(message, tag = "131")]
        TemporalJoin(::prost::alloc::boxed::Box<super::TemporalJoinNode>),
        #[prost(message, tag = "132")]
        BarrierRecv(::prost::alloc::boxed::Box<super::BarrierRecvNode>),
        #[prost(message, tag = "133")]
        Values(::prost::alloc::boxed::Box<super::ValuesNode>),
        #[prost(message, tag = "134")]
        AppendOnlyDedup(::prost::alloc::boxed::Box<super::DedupNode>),
        #[prost(message, tag = "135")]
        NoOp(super::NoOpNode),
        #[prost(message, tag = "136")]
        EowcOverWindow(::prost::alloc::boxed::Box<super::EowcOverWindowNode>),
        #[prost(message, tag = "137")]
        OverWindow(::prost::alloc::boxed::Box<super::OverWindowNode>),
        #[prost(message, tag = "138")]
        StreamFsFetch(::prost::alloc::boxed::Box<super::StreamFsFetchNode>),
        #[prost(message, tag = "139")]
        StreamCdcScan(::prost::alloc::boxed::Box<super::StreamCdcScanNode>),
        #[prost(message, tag = "140")]
        CdcFilter(::prost::alloc::boxed::Box<super::CdcFilterNode>),
        #[prost(message, tag = "142")]
        SourceBackfill(::prost::alloc::boxed::Box<super::SourceBackfillNode>),
        #[prost(message, tag = "143")]
        Changelog(::prost::alloc::boxed::Box<super::ChangeLogNode>),
        #[prost(message, tag = "144")]
        LocalApproxPercentile(
            ::prost::alloc::boxed::Box<super::LocalApproxPercentileNode>,
        ),
        #[prost(message, tag = "145")]
        GlobalApproxPercentile(
            ::prost::alloc::boxed::Box<super::GlobalApproxPercentileNode>,
        ),
        #[prost(message, tag = "146")]
        RowMerge(::prost::alloc::boxed::Box<super::RowMergeNode>),
        #[prost(message, tag = "147")]
        AsOfJoin(::prost::alloc::boxed::Box<super::AsOfJoinNode>),
        #[prost(message, tag = "148")]
        SyncLogStore(::prost::alloc::boxed::Box<super::SyncLogStoreNode>),
        #[prost(message, tag = "149")]
        MaterializedExprs(::prost::alloc::boxed::Box<super::MaterializedExprsNode>),
    }
}
/// The method to map the upstream columns in the dispatcher before dispatching.
///
/// - For intra-job exchange, typically the upstream and downstream columns are the same. `indices`
///    will be `0..len` and `types` will be empty.
///
/// - For inter-job exchange,
///    * if the downstream only requires a subset of the upstream columns, `indices` will be the
///      indices of the required columns in the upstream columns.
///    * if some columns are added to the upstream, `indices` will help to maintain the same schema
///      from the view of the downstream.
///    * if some columns are altered to different (composite) types, `types` will be used to convert
///      the upstream columns to the downstream columns to maintain the same schema.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DispatchOutputMapping {
    /// Indices of the columns to output.
    #[prost(uint32, repeated, tag = "1")]
    pub indices: ::prost::alloc::vec::Vec<u32>,
    /// Besides the indices, we may also need to convert the types of some columns.
    /// - If no type conversion is needed, this field will be empty.
    /// - If type conversion is needed, this will have the same length as `indices`. Only columns with
    ///    type conversion will have `upstream` and `downstream` field set.
    #[prost(message, repeated, tag = "2")]
    pub types: ::prost::alloc::vec::Vec<dispatch_output_mapping::TypePair>,
}
/// Nested message and enum types in `DispatchOutputMapping`.
pub mod dispatch_output_mapping {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct TypePair {
        #[prost(message, optional, tag = "1")]
        pub upstream: ::core::option::Option<super::super::data::DataType>,
        #[prost(message, optional, tag = "2")]
        pub downstream: ::core::option::Option<super::super::data::DataType>,
    }
}
/// The property of an edge in the fragment graph.
/// This is essientially a "logical" version of `Dispatcher`. See the doc of `Dispatcher` for more details.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct DispatchStrategy {
    #[prost(enumeration = "DispatcherType", tag = "1")]
    pub r#type: i32,
    #[prost(uint32, repeated, tag = "2")]
    pub dist_key_indices: ::prost::alloc::vec::Vec<u32>,
    #[prost(message, optional, tag = "3")]
    pub output_mapping: ::core::option::Option<DispatchOutputMapping>,
}
/// A dispatcher redistribute messages.
/// We encode both the type and other usage information in the proto.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Dispatcher {
    #[prost(enumeration = "DispatcherType", tag = "1")]
    pub r#type: i32,
    /// Indices of the columns to be used for hashing.
    /// For dispatcher types other than HASH, this is ignored.
    #[prost(uint32, repeated, tag = "2")]
    pub dist_key_indices: ::prost::alloc::vec::Vec<u32>,
    /// The method to map the upstream columns in the dispatcher before dispatching.
    #[prost(message, optional, tag = "6")]
    pub output_mapping: ::core::option::Option<DispatchOutputMapping>,
    /// The hash mapping for consistent hash.
    /// For dispatcher types other than HASH, this is ignored.
    #[prost(message, optional, tag = "3")]
    pub hash_mapping: ::core::option::Option<ActorMapping>,
    /// Dispatcher can be uniquely identified by a combination of actor id and dispatcher id.
    /// This is exactly the same as its downstream fragment id.
    #[prost(uint64, tag = "4")]
    pub dispatcher_id: u64,
    /// Number of downstreams decides how many endpoints a dispatcher should dispatch.
    #[prost(uint32, repeated, tag = "5")]
    pub downstream_actor_id: ::prost::alloc::vec::Vec<u32>,
}
/// A StreamActor is a running fragment of the overall stream graph,
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamActor {
    #[prost(uint32, tag = "1")]
    pub actor_id: u32,
    #[prost(uint32, tag = "2")]
    pub fragment_id: u32,
    #[prost(message, repeated, tag = "4")]
    pub dispatcher: ::prost::alloc::vec::Vec<Dispatcher>,
    /// Vnodes that the executors in this actor own.
    /// If the fragment is a singleton, this field will not be set and leave a `None`.
    #[prost(message, optional, tag = "8")]
    pub vnode_bitmap: ::core::option::Option<super::common::Buffer>,
    /// The SQL definition of this materialized view. Used for debugging only.
    #[prost(string, tag = "9")]
    pub mview_definition: ::prost::alloc::string::String,
    /// Provide the necessary context, e.g. session info like time zone, for the actor.
    #[prost(message, optional, tag = "10")]
    pub expr_context: ::core::option::Option<super::plan_common::ExprContext>,
}
/// The streaming context associated with a stream plan
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamContext {
    /// The timezone associated with the streaming plan. Only applies to MV for now.
    #[prost(string, tag = "1")]
    pub timezone: ::prost::alloc::string::String,
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BackfillOrder {
    #[prost(map = "uint32, message", tag = "1")]
    pub order: ::std::collections::HashMap<u32, super::common::Uint32Vector>,
}
/// Representation of a graph of stream fragments.
/// Generated by the fragmenter in the frontend, only used in DDL requests and never persisted.
///
/// For the persisted form, see `TableFragments`.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct StreamFragmentGraph {
    /// all the fragments in the graph.
    #[prost(map = "uint32, message", tag = "1")]
    pub fragments: ::std::collections::HashMap<
        u32,
        stream_fragment_graph::StreamFragment,
    >,
    /// edges between fragments.
    #[prost(message, repeated, tag = "2")]
    pub edges: ::prost::alloc::vec::Vec<stream_fragment_graph::StreamFragmentEdge>,
    #[prost(uint32, repeated, tag = "3")]
    pub dependent_table_ids: ::prost::alloc::vec::Vec<u32>,
    #[prost(uint32, tag = "4")]
    pub table_ids_cnt: u32,
    #[prost(message, optional, tag = "5")]
    pub ctx: ::core::option::Option<StreamContext>,
    /// If none, default parallelism will be applied.
    #[prost(message, optional, tag = "6")]
    pub parallelism: ::core::option::Option<stream_fragment_graph::Parallelism>,
    /// Specified max parallelism, i.e., expected vnode count for the graph.
    ///
    /// The scheduler on the meta service will use this as a hint to decide the vnode count
    /// for each fragment.
    ///
    /// Note that the actual vnode count may be different from this value.
    /// For example, a no-shuffle exchange between current fragment graph and an existing
    /// upstream fragment graph requires two fragments to be in the same distribution,
    /// thus the same vnode count.
    #[prost(uint32, tag = "7")]
    pub max_parallelism: u32,
    /// The backfill order strategy for the fragments.
    #[prost(message, optional, tag = "8")]
    pub backfill_order: ::core::option::Option<BackfillOrder>,
}
/// Nested message and enum types in `StreamFragmentGraph`.
pub mod stream_fragment_graph {
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct StreamFragment {
        /// 0-based on frontend, and will be rewritten to global id on meta.
        #[prost(uint32, tag = "1")]
        pub fragment_id: u32,
        /// root stream node in this fragment.
        #[prost(message, optional, tag = "2")]
        pub node: ::core::option::Option<super::StreamNode>,
        /// Bitwise-OR of `FragmentTypeFlag`s
        #[prost(uint32, tag = "3")]
        pub fragment_type_mask: u32,
        /// Mark whether this fragment requires exactly one actor.
        /// Note: if this is `false`, the fragment may still be a singleton according to the scheduler.
        /// One should check `meta.Fragment.distribution_type` for the final result.
        #[prost(bool, tag = "4")]
        pub requires_singleton: bool,
        /// Number of table ids (stateful states) for this fragment.
        #[prost(uint32, tag = "5")]
        pub table_ids_cnt: u32,
        /// Mark the upstream table ids of this fragment, Used for fragments with `StreamScan`s.
        #[prost(uint32, repeated, tag = "6")]
        pub upstream_table_ids: ::prost::alloc::vec::Vec<u32>,
    }
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, PartialEq, ::prost::Message)]
    pub struct StreamFragmentEdge {
        /// Dispatch strategy for the fragment.
        #[prost(message, optional, tag = "1")]
        pub dispatch_strategy: ::core::option::Option<super::DispatchStrategy>,
        /// A unique identifier of this edge. Generally it should be exchange node's operator id. When
        /// rewriting fragments into delta joins or when inserting 1-to-1 exchange, there will be
        /// virtual links generated.
        #[prost(uint64, tag = "3")]
        pub link_id: u64,
        #[prost(uint32, tag = "4")]
        pub upstream_id: u32,
        #[prost(uint32, tag = "5")]
        pub downstream_id: u32,
    }
    #[derive(prost_helpers::AnyPB)]
    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
    pub struct Parallelism {
        #[prost(uint64, tag = "1")]
        pub parallelism: u64,
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum SinkLogStoreType {
    /// / Default value is the normal in memory log store to be backward compatible with the previously unset value
    Unspecified = 0,
    KvLogStore = 1,
    InMemoryLogStore = 2,
}
impl SinkLogStoreType {
    /// String value of the enum field names used in the ProtoBuf definition.
    ///
    /// The values are not transformed in any way and thus are considered stable
    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
    pub fn as_str_name(&self) -> &'static str {
        match self {
            Self::Unspecified => "SINK_LOG_STORE_TYPE_UNSPECIFIED",
            Self::KvLogStore => "SINK_LOG_STORE_TYPE_KV_LOG_STORE",
            Self::InMemoryLogStore => "SINK_LOG_STORE_TYPE_IN_MEMORY_LOG_STORE",
        }
    }
    /// Creates an enum from field names used in the ProtoBuf definition.
    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
        match value {
            "SINK_LOG_STORE_TYPE_UNSPECIFIED" => Some(Self::Unspecified),
            "SINK_LOG_STORE_TYPE_KV_LOG_STORE" => Some(Self::KvLogStore),
            "SINK_LOG_STORE_TYPE_IN_MEMORY_LOG_STORE" => Some(Self::InMemoryLogStore),
            _ => None,
        }
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(prost_helpers::Version)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum AggNodeVersion {
    Unspecified = 0,
    /// <https://github.com/risingwavelabs/risingwave/issues/12140#issuecomment-1776289808>
    Issue12140 = 1,
    /// <https://github.com/risingwavelabs/risingwave/issues/13465#issuecomment-1821016508>
    Issue13465 = 2,
}
impl AggNodeVersion {
    /// String value of the enum field names used in the ProtoBuf definition.
    ///
    /// The values are not transformed in any way and thus are considered stable
    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
    pub fn as_str_name(&self) -> &'static str {
        match self {
            Self::Unspecified => "AGG_NODE_VERSION_UNSPECIFIED",
            Self::Issue12140 => "AGG_NODE_VERSION_ISSUE_12140",
            Self::Issue13465 => "AGG_NODE_VERSION_ISSUE_13465",
        }
    }
    /// Creates an enum from field names used in the ProtoBuf definition.
    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
        match value {
            "AGG_NODE_VERSION_UNSPECIFIED" => Some(Self::Unspecified),
            "AGG_NODE_VERSION_ISSUE_12140" => Some(Self::Issue12140),
            "AGG_NODE_VERSION_ISSUE_13465" => Some(Self::Issue13465),
            _ => None,
        }
    }
}
/// Decides which kind of Executor will be used
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum StreamScanType {
    Unspecified = 0,
    /// ChainExecutor
    Chain = 1,
    /// RearrangedChainExecutor
    Rearrange = 2,
    /// BackfillExecutor
    Backfill = 3,
    /// ChainExecutor with upstream_only = true
    UpstreamOnly = 4,
    /// ArrangementBackfillExecutor
    ArrangementBackfill = 5,
    /// SnapshotBackfillExecutor
    SnapshotBackfill = 6,
    /// SnapshotBackfillExecutor
    CrossDbSnapshotBackfill = 7,
}
impl StreamScanType {
    /// String value of the enum field names used in the ProtoBuf definition.
    ///
    /// The values are not transformed in any way and thus are considered stable
    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
    pub fn as_str_name(&self) -> &'static str {
        match self {
            Self::Unspecified => "STREAM_SCAN_TYPE_UNSPECIFIED",
            Self::Chain => "STREAM_SCAN_TYPE_CHAIN",
            Self::Rearrange => "STREAM_SCAN_TYPE_REARRANGE",
            Self::Backfill => "STREAM_SCAN_TYPE_BACKFILL",
            Self::UpstreamOnly => "STREAM_SCAN_TYPE_UPSTREAM_ONLY",
            Self::ArrangementBackfill => "STREAM_SCAN_TYPE_ARRANGEMENT_BACKFILL",
            Self::SnapshotBackfill => "STREAM_SCAN_TYPE_SNAPSHOT_BACKFILL",
            Self::CrossDbSnapshotBackfill => {
                "STREAM_SCAN_TYPE_CROSS_DB_SNAPSHOT_BACKFILL"
            }
        }
    }
    /// Creates an enum from field names used in the ProtoBuf definition.
    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
        match value {
            "STREAM_SCAN_TYPE_UNSPECIFIED" => Some(Self::Unspecified),
            "STREAM_SCAN_TYPE_CHAIN" => Some(Self::Chain),
            "STREAM_SCAN_TYPE_REARRANGE" => Some(Self::Rearrange),
            "STREAM_SCAN_TYPE_BACKFILL" => Some(Self::Backfill),
            "STREAM_SCAN_TYPE_UPSTREAM_ONLY" => Some(Self::UpstreamOnly),
            "STREAM_SCAN_TYPE_ARRANGEMENT_BACKFILL" => Some(Self::ArrangementBackfill),
            "STREAM_SCAN_TYPE_SNAPSHOT_BACKFILL" => Some(Self::SnapshotBackfill),
            "STREAM_SCAN_TYPE_CROSS_DB_SNAPSHOT_BACKFILL" => {
                Some(Self::CrossDbSnapshotBackfill)
            }
            _ => None,
        }
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum OverWindowCachePolicy {
    Unspecified = 0,
    Full = 1,
    Recent = 2,
    RecentFirstN = 3,
    RecentLastN = 4,
}
impl OverWindowCachePolicy {
    /// String value of the enum field names used in the ProtoBuf definition.
    ///
    /// The values are not transformed in any way and thus are considered stable
    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
    pub fn as_str_name(&self) -> &'static str {
        match self {
            Self::Unspecified => "OVER_WINDOW_CACHE_POLICY_UNSPECIFIED",
            Self::Full => "OVER_WINDOW_CACHE_POLICY_FULL",
            Self::Recent => "OVER_WINDOW_CACHE_POLICY_RECENT",
            Self::RecentFirstN => "OVER_WINDOW_CACHE_POLICY_RECENT_FIRST_N",
            Self::RecentLastN => "OVER_WINDOW_CACHE_POLICY_RECENT_LAST_N",
        }
    }
    /// Creates an enum from field names used in the ProtoBuf definition.
    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
        match value {
            "OVER_WINDOW_CACHE_POLICY_UNSPECIFIED" => Some(Self::Unspecified),
            "OVER_WINDOW_CACHE_POLICY_FULL" => Some(Self::Full),
            "OVER_WINDOW_CACHE_POLICY_RECENT" => Some(Self::Recent),
            "OVER_WINDOW_CACHE_POLICY_RECENT_FIRST_N" => Some(Self::RecentFirstN),
            "OVER_WINDOW_CACHE_POLICY_RECENT_LAST_N" => Some(Self::RecentLastN),
            _ => None,
        }
    }
}
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum DispatcherType {
    Unspecified = 0,
    /// Dispatch by hash key, hashed by consistent hash.
    Hash = 1,
    /// Broadcast to all downstreams.
    ///
    /// Note a broadcast cannot be represented as multiple simple dispatchers, since they are
    /// different when we update dispatchers during scaling.
    Broadcast = 2,
    /// Only one downstream.
    Simple = 3,
    /// A special kind of exchange that doesn't involve shuffle. The upstream actor will be directly
    /// piped into the downstream actor, if there are the same number of actors. If number of actors
    /// are not the same, should use hash instead. Should be only used when distribution is the same.
    NoShuffle = 4,
}
impl DispatcherType {
    /// String value of the enum field names used in the ProtoBuf definition.
    ///
    /// The values are not transformed in any way and thus are considered stable
    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
    pub fn as_str_name(&self) -> &'static str {
        match self {
            Self::Unspecified => "DISPATCHER_TYPE_UNSPECIFIED",
            Self::Hash => "DISPATCHER_TYPE_HASH",
            Self::Broadcast => "DISPATCHER_TYPE_BROADCAST",
            Self::Simple => "DISPATCHER_TYPE_SIMPLE",
            Self::NoShuffle => "DISPATCHER_TYPE_NO_SHUFFLE",
        }
    }
    /// Creates an enum from field names used in the ProtoBuf definition.
    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
        match value {
            "DISPATCHER_TYPE_UNSPECIFIED" => Some(Self::Unspecified),
            "DISPATCHER_TYPE_HASH" => Some(Self::Hash),
            "DISPATCHER_TYPE_BROADCAST" => Some(Self::Broadcast),
            "DISPATCHER_TYPE_SIMPLE" => Some(Self::Simple),
            "DISPATCHER_TYPE_NO_SHUFFLE" => Some(Self::NoShuffle),
            _ => None,
        }
    }
}
/// Indicates whether the fragment contains some special kind of nodes.
#[derive(prost_helpers::AnyPB)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum FragmentTypeFlag {
    FragmentUnspecified = 0,
    Source = 1,
    Mview = 2,
    Sink = 4,
    /// TODO: Remove this and insert a `BarrierRecv` instead.
    Now = 8,
    /// Include StreamScan and StreamCdcScan
    StreamScan = 16,
    BarrierRecv = 32,
    Values = 64,
    Dml = 128,
    CdcFilter = 256,
    SourceScan = 1024,
    SnapshotBackfillStreamScan = 2048,
    /// Note: this flag is not available in old fragments, so only suitable for debugging purpose.
    FsFetch = 4096,
    CrossDbSnapshotBackfillStreamScan = 8192,
}
impl FragmentTypeFlag {
    /// String value of the enum field names used in the ProtoBuf definition.
    ///
    /// The values are not transformed in any way and thus are considered stable
    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
    pub fn as_str_name(&self) -> &'static str {
        match self {
            Self::FragmentUnspecified => "FRAGMENT_TYPE_FLAG_FRAGMENT_UNSPECIFIED",
            Self::Source => "FRAGMENT_TYPE_FLAG_SOURCE",
            Self::Mview => "FRAGMENT_TYPE_FLAG_MVIEW",
            Self::Sink => "FRAGMENT_TYPE_FLAG_SINK",
            Self::Now => "FRAGMENT_TYPE_FLAG_NOW",
            Self::StreamScan => "FRAGMENT_TYPE_FLAG_STREAM_SCAN",
            Self::BarrierRecv => "FRAGMENT_TYPE_FLAG_BARRIER_RECV",
            Self::Values => "FRAGMENT_TYPE_FLAG_VALUES",
            Self::Dml => "FRAGMENT_TYPE_FLAG_DML",
            Self::CdcFilter => "FRAGMENT_TYPE_FLAG_CDC_FILTER",
            Self::SourceScan => "FRAGMENT_TYPE_FLAG_SOURCE_SCAN",
            Self::SnapshotBackfillStreamScan => {
                "FRAGMENT_TYPE_FLAG_SNAPSHOT_BACKFILL_STREAM_SCAN"
            }
            Self::FsFetch => "FRAGMENT_TYPE_FLAG_FS_FETCH",
            Self::CrossDbSnapshotBackfillStreamScan => {
                "FRAGMENT_TYPE_FLAG_CROSS_DB_SNAPSHOT_BACKFILL_STREAM_SCAN"
            }
        }
    }
    /// Creates an enum from field names used in the ProtoBuf definition.
    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
        match value {
            "FRAGMENT_TYPE_FLAG_FRAGMENT_UNSPECIFIED" => Some(Self::FragmentUnspecified),
            "FRAGMENT_TYPE_FLAG_SOURCE" => Some(Self::Source),
            "FRAGMENT_TYPE_FLAG_MVIEW" => Some(Self::Mview),
            "FRAGMENT_TYPE_FLAG_SINK" => Some(Self::Sink),
            "FRAGMENT_TYPE_FLAG_NOW" => Some(Self::Now),
            "FRAGMENT_TYPE_FLAG_STREAM_SCAN" => Some(Self::StreamScan),
            "FRAGMENT_TYPE_FLAG_BARRIER_RECV" => Some(Self::BarrierRecv),
            "FRAGMENT_TYPE_FLAG_VALUES" => Some(Self::Values),
            "FRAGMENT_TYPE_FLAG_DML" => Some(Self::Dml),
            "FRAGMENT_TYPE_FLAG_CDC_FILTER" => Some(Self::CdcFilter),
            "FRAGMENT_TYPE_FLAG_SOURCE_SCAN" => Some(Self::SourceScan),
            "FRAGMENT_TYPE_FLAG_SNAPSHOT_BACKFILL_STREAM_SCAN" => {
                Some(Self::SnapshotBackfillStreamScan)
            }
            "FRAGMENT_TYPE_FLAG_FS_FETCH" => Some(Self::FsFetch),
            "FRAGMENT_TYPE_FLAG_CROSS_DB_SNAPSHOT_BACKFILL_STREAM_SCAN" => {
                Some(Self::CrossDbSnapshotBackfillStreamScan)
            }
            _ => None,
        }
    }
}