risingwave_stream/from_proto/
mod.rs

1// Copyright 2022 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//! Build executor from protobuf.
16
17macro_rules! impl_stream_node_body {
18    ($variant:ident($node:ty) => $executor_builder:ty) => {
19        paste::paste! {
20            impl crate::from_proto::StreamNodeBody
21                for risingwave_pb::stream_plan::stream_node::[<$variant Variant>]
22            {
23                type Node = $node;
24                type ExecutorBuilder = $executor_builder;
25            }
26        }
27    };
28}
29
30mod agg_common;
31mod append_only_dedup;
32mod asof_join;
33mod barrier_recv;
34mod batch_query;
35mod cdc_filter;
36mod changelog;
37mod dml;
38mod dynamic_filter;
39mod eowc_gap_fill;
40mod eowc_over_window;
41mod expand;
42mod filter;
43mod gap_fill;
44mod group_top_n;
45mod hash_agg;
46mod hash_join;
47mod hop_window;
48mod iceberg_with_pk_index;
49mod locality_provider;
50mod lookup;
51mod lookup_union;
52mod match_recognize;
53mod materialized_exprs;
54mod merge;
55mod mview;
56mod no_op;
57mod now;
58mod over_window;
59mod project;
60mod project_set;
61mod row_id_gen;
62mod simple_agg;
63mod sink;
64mod sort;
65mod source;
66mod source_backfill;
67mod stateless_simple_agg;
68mod stream_cdc_scan;
69mod stream_scan;
70mod temporal_join;
71mod top_n;
72mod union;
73mod upstream_sink_union;
74mod values;
75mod watermark_filter;
76
77mod row_merge;
78
79mod approx_percentile;
80
81mod sync_log_store;
82mod vector_index_lookup_join;
83mod vector_index_write;
84
85// import for submodules
86use itertools::Itertools;
87use risingwave_common::dispatch_stream_node_body;
88use risingwave_pb::stream_plan::{self, StreamNode, TemporalJoinNode};
89use risingwave_storage::StateStore;
90
91pub(crate) use self::merge::MergeExecutorBuilder;
92use crate::error::StreamResult;
93use crate::executor::{Execute, Executor, ExecutorInfo};
94use crate::task::ExecutorParams;
95
96trait ExecutorBuilder {
97    type Node;
98
99    /// Create an [`Executor`] from [`StreamNode`].
100    async fn new_boxed_executor(
101        params: ExecutorParams,
102        node: &Self::Node,
103        store: impl StateStore,
104    ) -> StreamResult<Executor>;
105}
106
107trait StreamNodeBody {
108    type Node;
109    type ExecutorBuilder: ExecutorBuilder<Node = Self::Node>;
110}
111
112struct UnreachableExecutorBuilder<T>(std::marker::PhantomData<T>);
113
114impl<T> ExecutorBuilder for UnreachableExecutorBuilder<T> {
115    type Node = T;
116
117    async fn new_boxed_executor(
118        _params: ExecutorParams,
119        _node: &Self::Node,
120        _store: impl StateStore,
121    ) -> StreamResult<Executor> {
122        unreachable!()
123    }
124}
125
126impl_stream_node_body!(
127    Exchange(stream_plan::ExchangeNode) => UnreachableExecutorBuilder<stream_plan::ExchangeNode>
128);
129impl_stream_node_body!(
130    DeltaIndexJoin(stream_plan::DeltaIndexJoinNode) => UnreachableExecutorBuilder<stream_plan::DeltaIndexJoinNode>
131);
132
133macro_rules! create_executor {
134    ($params:expr, $node:expr, $store:expr) => {
135        dispatch_stream_node_body!($node.get_node_body().unwrap(), NodeVariant, node_body => {
136            <NodeVariant as StreamNodeBody>::ExecutorBuilder::new_boxed_executor(
137                $params, node_body, $store,
138            )
139            .await
140        })
141    };
142}
143
144/// Create an executor from protobuf [`StreamNode`].
145pub async fn create_executor(
146    params: ExecutorParams,
147    node: &StreamNode,
148    store: impl StateStore,
149) -> StreamResult<Executor> {
150    create_executor!(params, node, store)
151}
risingwave_stream/from_proto/mod.rs

risingwave_stream/from_proto/
mod.rs
