risingwave_frontend/optimizer/rule/stream/
stream_project_merge_rule.rs1use super::prelude::*;
16use crate::optimizer::plan_node::{PlanTreeNodeUnary, StreamProject};
17use crate::optimizer::rule::ProjectMergeRule;
18
19pub struct StreamProjectMergeRule {}
21impl Rule<Stream> for StreamProjectMergeRule {
22 fn apply(&self, plan: PlanRef) -> Option<PlanRef> {
23 let outer_project = plan.as_stream_project()?;
24 let input = outer_project.input();
25 let inner_project = input.as_stream_project()?;
26
27 let mut core = outer_project.core().clone();
28 core.exprs = ProjectMergeRule::merge_project_exprs(
29 outer_project.exprs(),
30 inner_project.exprs(),
31 true,
32 )?;
33 core.input = inner_project.input();
34
35 let noop_update_hint = outer_project.noop_update_hint() || inner_project.noop_update_hint();
37
38 Some(
39 StreamProject::new(core)
40 .with_noop_update_hint(noop_update_hint)
41 .into(),
42 )
43 }
44}
45
46impl StreamProjectMergeRule {
47 pub fn create() -> BoxedRule {
48 Box::new(StreamProjectMergeRule {})
49 }
50}