risingwave_frontend/optimizer/plan_node/generic/
changelog.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use pretty_xmlish::{Pretty, Str, XmlNode};
16use risingwave_common::catalog::{Field, FieldDisplay, Schema};
17use risingwave_common::util::column_index_mapping::ColIndexMapping;
18
19use super::{DistillUnit, GenericPlanNode};
20use crate::OptimizerContextRef;
21use crate::optimizer::plan_node::stream::prelude::GenericPlanRef;
22use crate::optimizer::plan_node::utils::childless_record;
23use crate::optimizer::property::FunctionalDependencySet;
24use crate::utils::ColIndexMappingRewriteExt;
25
26pub const CHANGELOG_OP: &str = "changelog_op";
27pub const _CHANGELOG_ROW_ID: &str = "_changelog_row_id";
28#[derive(Debug, Clone, PartialEq, Eq, Hash)]
29pub struct ChangeLog<PlanRef> {
30    pub input: PlanRef,
31    /// Declared key column indices in the input schema.
32    pub key_indices: Option<Vec<usize>>,
33    /// Explicit vnode distribution columns, independent of keyed changelog processing.
34    pub distribution_key_indices: Option<Vec<usize>>,
35    // If there is no op in the output result, it is false, example 'create materialized view mv1 as with sub as changelog from t1 select v1 from sub;'
36    pub need_op: bool,
37    // Before rewrite. If there is no changelog_row_id in the output result, it is false.
38    // After rewrite. It is always true.
39    pub need_changelog_row_id: bool,
40}
41impl<PlanRef: GenericPlanRef> DistillUnit for ChangeLog<PlanRef> {
42    fn distill_with_name<'a>(&self, name: impl Into<Str<'a>>) -> XmlNode<'a> {
43        let mut fields = Vec::with_capacity(1);
44
45        if let Some(key_indices) = &self.key_indices {
46            let key = Pretty::Array(
47                key_indices
48                    .iter()
49                    .map(|&index| {
50                        let field = &self.input.schema()[index];
51
52                        Pretty::display(&FieldDisplay(field))
53                    })
54                    .collect(),
55            );
56
57            fields.push(("key", key));
58        }
59
60        if let Some(indices) = &self.distribution_key_indices {
61            fields.push((
62                "distribution_key",
63                Pretty::Array(
64                    indices
65                        .iter()
66                        .map(|&index| Pretty::display(&FieldDisplay(&self.input.schema()[index])))
67                        .collect(),
68                ),
69            ));
70        }
71
72        childless_record(name, fields)
73    }
74}
75impl<PlanRef: GenericPlanRef> ChangeLog<PlanRef> {
76    pub fn new(
77        input: PlanRef,
78        key_indices: Option<Vec<usize>>,
79        distribution_key_indices: Option<Vec<usize>>,
80        need_op: bool,
81        need_changelog_row_id: bool,
82    ) -> Self {
83        ChangeLog {
84            input,
85            key_indices,
86            distribution_key_indices,
87            need_op,
88            need_changelog_row_id,
89        }
90    }
91
92    pub fn clone_with_input<OtherPlanRef>(&self, input: OtherPlanRef) -> ChangeLog<OtherPlanRef> {
93        ChangeLog {
94            input,
95            key_indices: self.key_indices.clone(),
96            distribution_key_indices: self.distribution_key_indices.clone(),
97            need_op: self.need_op,
98            need_changelog_row_id: self.need_changelog_row_id,
99        }
100    }
101
102    pub fn i2o_col_mapping(&self) -> ColIndexMapping {
103        let mut map = vec![None; self.input.schema().len()];
104        (0..self.input.schema().len()).for_each(|i| map[i] = Some(i));
105        ColIndexMapping::new(map, self.schema().len())
106    }
107}
108impl<PlanRef: GenericPlanRef> GenericPlanNode for ChangeLog<PlanRef> {
109    fn schema(&self) -> Schema {
110        let mut fields = self.input.schema().fields.clone();
111        if self.need_op {
112            fields.push(Field::with_name(
113                risingwave_common::types::DataType::Int16,
114                CHANGELOG_OP,
115            ));
116        }
117        if self.need_changelog_row_id {
118            fields.push(Field::with_name(
119                risingwave_common::types::DataType::Serial,
120                _CHANGELOG_ROW_ID,
121            ));
122        }
123        Schema::new(fields)
124    }
125
126    fn stream_key(&self) -> Option<Vec<usize>> {
127        if self.need_changelog_row_id {
128            let keys = vec![self.schema().len() - 1];
129            Some(keys)
130        } else {
131            None
132        }
133    }
134
135    fn ctx(&self) -> OptimizerContextRef {
136        self.input.ctx()
137    }
138
139    fn functional_dependency(&self) -> FunctionalDependencySet {
140        let i2o = self.i2o_col_mapping();
141        i2o.rewrite_functional_dependency_set(self.input.functional_dependency().clone())
142    }
143}
risingwave_frontend/optimizer/plan_node/generic/changelog.rs

risingwave_frontend/optimizer/plan_node/generic/
changelog.rs
