risingwave_frontend/optimizer/rule/
project_merge_rule.rs1use super::prelude::{PlanRef, *};
16use crate::expr::{ExprImpl, ExprRewriter, ExprVisitor};
17use crate::optimizer::plan_expr_visitor::InputRefCounter;
18use crate::optimizer::plan_node::*;
19use crate::utils::Substitute;
20
21pub struct ProjectMergeRule {}
23impl ProjectMergeRule {
24 pub fn merge_project_exprs(
28 outer_project_exprs: &[ExprImpl],
29 inner_project_exprs: &[ExprImpl],
30 keep_common_sub_func: bool,
31 ) -> Option<Vec<ExprImpl>> {
32 if keep_common_sub_func {
33 let mut input_ref_counter = InputRefCounter::default();
34 for expr in outer_project_exprs {
35 input_ref_counter.visit_expr(expr);
36 }
37 for (index, count) in &input_ref_counter.counter {
38 if *count > 1
39 && !matches!(
40 inner_project_exprs[*index],
41 ExprImpl::InputRef(_)
42 | ExprImpl::CorrelatedInputRef(_)
43 | ExprImpl::Literal(_)
44 )
45 {
46 return None;
47 }
48 }
49 }
50
51 let mut subst = Substitute {
52 mapping: inner_project_exprs.to_vec(),
53 };
54 let exprs = outer_project_exprs
55 .iter()
56 .cloned()
57 .map(|expr| subst.rewrite_expr(expr))
58 .collect();
59 Some(exprs)
60 }
61}
62
63impl ProjectMergeRule {
64 pub fn create() -> BoxedRule {
65 Box::new(ProjectMergeRule {})
66 }
67}
68
69impl Rule<Logical> for ProjectMergeRule {
70 fn apply(&self, plan: PlanRef) -> Option<PlanRef> {
71 let outer_project: &LogicalProject = plan.as_logical_project()?;
72 let input = outer_project.input();
73 let inner_project: &LogicalProject = input.as_logical_project()?;
74
75 let exprs = Self::merge_project_exprs(outer_project.exprs(), inner_project.exprs(), true)?;
76 Some(LogicalProject::new(inner_project.input(), exprs).into())
77 }
78}