risingwave_stream/from_proto/
mod.rs1mod 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
66use 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 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
145pub 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}