risingwave_stream/from_proto/
mod.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//! Build executor from protobuf.
16
17mod agg_common;
18mod append_only_dedup;
19mod asof_join;
20mod barrier_recv;
21mod batch_query;
22mod cdc_filter;
23mod changelog;
24mod dml;
25mod dynamic_filter;
26mod eowc_over_window;
27mod expand;
28mod filter;
29mod group_top_n;
30mod hash_agg;
31mod hash_join;
32mod hop_window;
33mod lookup;
34mod lookup_union;
35mod materialized_exprs;
36mod merge;
37mod mview;
38mod no_op;
39mod now;
40mod over_window;
41mod project;
42mod project_set;
43mod row_id_gen;
44mod simple_agg;
45mod sink;
46mod sort;
47mod source;
48mod source_backfill;
49mod stateless_simple_agg;
50mod stream_cdc_scan;
51mod stream_scan;
52mod temporal_join;
53mod top_n;
54mod union;
55mod upstream_sink_union;
56mod values;
57mod watermark_filter;
58
59mod row_merge;
60
61mod approx_percentile;
62
63mod sync_log_store;
64mod vector_index;
65
66// import for submodules
67use itertools::Itertools;
68use risingwave_pb::stream_plan::stream_node::NodeBody;
69use risingwave_pb::stream_plan::{StreamNode, TemporalJoinNode};
70use risingwave_storage::StateStore;
71
72use self::append_only_dedup::*;
73use self::approx_percentile::global::*;
74use self::approx_percentile::local::*;
75use self::asof_join::AsOfJoinExecutorBuilder;
76use self::barrier_recv::*;
77use self::batch_query::*;
78use self::cdc_filter::CdcFilterExecutorBuilder;
79use self::dml::*;
80use self::dynamic_filter::*;
81use self::eowc_over_window::*;
82use self::expand::*;
83use self::filter::*;
84use self::group_top_n::GroupTopNExecutorBuilder;
85use self::hash_agg::*;
86use self::hash_join::*;
87use self::hop_window::*;
88use self::lookup::*;
89use self::lookup_union::*;
90use self::materialized_exprs::MaterializedExprsExecutorBuilder;
91pub(crate) use self::merge::MergeExecutorBuilder;
92use self::mview::*;
93use self::no_op::*;
94use self::now::NowExecutorBuilder;
95use self::over_window::*;
96use self::project::*;
97use self::project_set::*;
98use self::row_id_gen::RowIdGenExecutorBuilder;
99use self::row_merge::*;
100use self::simple_agg::*;
101use self::sink::*;
102use self::sort::*;
103use self::source::*;
104use self::source_backfill::*;
105use self::stateless_simple_agg::*;
106use self::stream_cdc_scan::*;
107use self::stream_scan::*;
108use self::sync_log_store::*;
109use self::temporal_join::*;
110use self::top_n::*;
111use self::union::*;
112use self::upstream_sink_union::*;
113use self::watermark_filter::WatermarkFilterBuilder;
114use crate::error::StreamResult;
115use crate::executor::{Execute, Executor, ExecutorInfo};
116use crate::from_proto::changelog::ChangeLogExecutorBuilder;
117use crate::from_proto::values::ValuesExecutorBuilder;
118use crate::from_proto::vector_index::VectorIndexWriteExecutorBuilder;
119use crate::task::ExecutorParams;
120
121trait ExecutorBuilder {
122    type Node;
123
124    /// Create an [`Executor`] from [`StreamNode`].
125    async fn new_boxed_executor(
126        params: ExecutorParams,
127        node: &Self::Node,
128        store: impl StateStore,
129    ) -> StreamResult<Executor>;
130}
131
132macro_rules! build_executor {
133    ($source:expr, $node:expr, $store:expr, $($proto_type_name:path => $data_type:ty),* $(,)?) => {
134        match $node.get_node_body().unwrap() {
135            $(
136                $proto_type_name(node) => {
137                    <$data_type>::new_boxed_executor($source, node, $store).await
138                },
139            )*
140            NodeBody::Exchange(_) | NodeBody::DeltaIndexJoin(_) => unreachable!()
141        }
142    }
143}
144
145/// Create an executor from protobuf [`StreamNode`].
146pub async fn create_executor(
147    params: ExecutorParams,
148    node: &StreamNode,
149    store: impl StateStore,
150) -> StreamResult<Executor> {
151    build_executor! {
152        params,
153        node,
154        store,
155        NodeBody::Source => SourceExecutorBuilder,
156        NodeBody::Sink => SinkExecutorBuilder,
157        NodeBody::Project => ProjectExecutorBuilder,
158        NodeBody::TopN => TopNExecutorBuilder::<false>,
159        NodeBody::AppendOnlyTopN => TopNExecutorBuilder::<true>,
160        NodeBody::StatelessSimpleAgg => StatelessSimpleAggExecutorBuilder,
161        NodeBody::SimpleAgg => SimpleAggExecutorBuilder,
162        NodeBody::HashAgg => HashAggExecutorBuilder,
163        NodeBody::HashJoin => HashJoinExecutorBuilder,
164        NodeBody::HopWindow => HopWindowExecutorBuilder,
165        NodeBody::StreamScan => StreamScanExecutorBuilder,
166        NodeBody::StreamCdcScan => StreamCdcScanExecutorBuilder,
167        NodeBody::BatchPlan => BatchQueryExecutorBuilder,
168        NodeBody::Merge => MergeExecutorBuilder,
169        NodeBody::Materialize => MaterializeExecutorBuilder,
170        NodeBody::Filter => FilterExecutorBuilder,
171        NodeBody::CdcFilter => CdcFilterExecutorBuilder,
172        NodeBody::Arrange => ArrangeExecutorBuilder,
173        NodeBody::Lookup => LookupExecutorBuilder,
174        NodeBody::Union => UnionExecutorBuilder,
175        NodeBody::LookupUnion => LookupUnionExecutorBuilder,
176        NodeBody::Expand => ExpandExecutorBuilder,
177        NodeBody::DynamicFilter => DynamicFilterExecutorBuilder,
178        NodeBody::ProjectSet => ProjectSetExecutorBuilder,
179        NodeBody::GroupTopN => GroupTopNExecutorBuilder::<false>,
180        NodeBody::AppendOnlyGroupTopN => GroupTopNExecutorBuilder::<true>,
181        NodeBody::Sort => SortExecutorBuilder,
182        NodeBody::WatermarkFilter => WatermarkFilterBuilder,
183        NodeBody::Dml => DmlExecutorBuilder,
184        NodeBody::RowIdGen => RowIdGenExecutorBuilder,
185        NodeBody::Now => NowExecutorBuilder,
186        NodeBody::TemporalJoin => TemporalJoinExecutorBuilder,
187        NodeBody::Values => ValuesExecutorBuilder,
188        NodeBody::BarrierRecv => BarrierRecvExecutorBuilder,
189        NodeBody::AppendOnlyDedup => AppendOnlyDedupExecutorBuilder,
190        NodeBody::NoOp => NoOpExecutorBuilder,
191        NodeBody::EowcOverWindow => EowcOverWindowExecutorBuilder,
192        NodeBody::OverWindow => OverWindowExecutorBuilder,
193        NodeBody::StreamFsFetch => FsFetchExecutorBuilder,
194        NodeBody::SourceBackfill => SourceBackfillExecutorBuilder,
195        NodeBody::Changelog => ChangeLogExecutorBuilder,
196        NodeBody::GlobalApproxPercentile => GlobalApproxPercentileExecutorBuilder,
197        NodeBody::LocalApproxPercentile => LocalApproxPercentileExecutorBuilder,
198        NodeBody::RowMerge => RowMergeExecutorBuilder,
199        NodeBody::AsOfJoin => AsOfJoinExecutorBuilder,
200        NodeBody::SyncLogStore => SyncLogStoreExecutorBuilder,
201        NodeBody::MaterializedExprs => MaterializedExprsExecutorBuilder,
202        NodeBody::VectorIndexWrite => VectorIndexWriteExecutorBuilder,
203        NodeBody::UpstreamSinkUnion => UpstreamSinkUnionExecutorBuilder,
204    }
205}