risingwave_stream/from_proto/
changelog.rs1use risingwave_pb::stream_plan::ChangeLogNode;
16use risingwave_storage::StateStore;
17
18use super::ExecutorBuilder;
19use crate::error::StreamResult;
20use crate::executor::{ChangeLogExecutor, Executor};
21use crate::task::ExecutorParams;
22
23pub struct ChangeLogExecutorBuilder;
24
25impl ExecutorBuilder for ChangeLogExecutorBuilder {
26 type Node = ChangeLogNode;
27
28 async fn new_boxed_executor(
29 params: ExecutorParams,
30 node: &Self::Node,
31 _store: impl StateStore,
32 ) -> StreamResult<Executor> {
33 let [input]: [_; 1] = params.input.try_into().unwrap();
34
35 let vnodes = params
36 .vnode_bitmap
37 .expect("vnodes not set for changelog executor");
38 let vnode_count = params.actor_context.vnode_count;
39 let distribution_keys = node
40 .distribution_keys
41 .iter()
42 .map(|k| *k as usize)
43 .collect::<Vec<_>>();
44 let exec = ChangeLogExecutor::new(
45 params.actor_context,
46 input,
47 node.need_op,
48 vnode_count,
49 vnodes,
50 distribution_keys,
51 );
52 Ok((params.info, exec).into())
53 }
54}