risingwave_stream/from_proto/
filter.rs1use risingwave_expr::expr::build_non_strict_from_prost;
16use risingwave_pb::stream_plan::FilterNode;
17use risingwave_pb::stream_plan::stream_node::PbStreamKind;
18
19use super::*;
20use crate::executor::{FilterExecutor, UpsertFilterExecutor};
21
22pub struct FilterExecutorBuilder;
23
24impl ExecutorBuilder for FilterExecutorBuilder {
25 type Node = FilterNode;
26
27 async fn new_boxed_executor(
28 params: ExecutorParams,
29 node: &Self::Node,
30 _store: impl StateStore,
31 ) -> StreamResult<Executor> {
32 let [input]: [_; 1] = params.input.try_into().unwrap();
33 let search_condition =
34 build_non_strict_from_prost(node.get_search_condition()?, params.eval_error_report)?;
35
36 let exec = if let PbStreamKind::Upsert = input.stream_kind() {
37 UpsertFilterExecutor::new(params.actor_context, input, search_condition).boxed()
38 } else {
39 FilterExecutor::new(params.actor_context, input, search_condition).boxed()
40 };
41 Ok((params.info, exec).into())
42 }
43}