risingwave_frontend/optimizer/plan_node/generic/
match_recognize.rs

1// Copyright 2026 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, Schema};
17use risingwave_common::types::DataType;
18use risingwave_sqlparser::ast::{AfterMatchSkip, MatchRecognizePattern, RowsPerMatch};
19
20use super::{DistillUnit, GenericPlanNode, GenericPlanRef};
21use crate::OptimizerContextRef;
22use crate::binder::{BoundMeasure, BoundSymbolDefinition, MeasureSlotKind};
23use crate::expr::{Expr, ExprDisplay, ExprImpl, ExprRewriter, ExprVisitor};
24use crate::optimizer::plan_node::ColIndexMapping;
25use crate::optimizer::plan_node::utils::childless_record;
26use crate::optimizer::property::FunctionalDependencySet;
27
28/// `MatchRecognize` is the generic core of a SQL `MATCH_RECOGNIZE` (row pattern recognition)
29/// operation. For `ONE ROW PER MATCH`, its output schema is the `PARTITION BY` columns followed by
30/// the `MEASURES` columns.
31#[derive(Debug, Clone, PartialEq, Eq, Hash)]
32pub struct MatchRecognize<PlanRef> {
33    pub input: PlanRef,
34    pub partition_by: Vec<ExprImpl>,
35    pub order_by: Vec<ExprImpl>,
36    pub measures: Vec<BoundMeasure>,
37    pub rows_per_match: Option<RowsPerMatch>,
38    pub after_match_skip: Option<AfterMatchSkip>,
39    pub pattern: MatchRecognizePattern,
40    pub defines: Vec<BoundSymbolDefinition>,
41    /// `WITHIN` span check over a synthetic `[last_order_key, first_order_key]` row; `None` if absent.
42    pub within: Option<ExprImpl>,
43    /// `WITHIN` deadline expr `first_order_key + interval` over a synthetic `[first_order_key]` row:
44    /// the watermark at which a partial starting at that row can no longer complete within the
45    /// bound. Used to wake an otherwise-idle partition so its expired partials are evicted. `None`
46    /// if there is no `WITHIN`.
47    pub within_deadline: Option<ExprImpl>,
48}
49
50impl<PlanRef: GenericPlanRef> GenericPlanNode for MatchRecognize<PlanRef> {
51    fn schema(&self) -> Schema {
52        let mut fields = Vec::with_capacity(self.partition_by.len() + self.measures.len() + 1);
53        for (i, e) in self.partition_by.iter().enumerate() {
54            fields.push(Field::with_name(e.return_type(), format!("partition_{i}")));
55        }
56        for m in &self.measures {
57            fields.push(Field::with_name(m.expr.return_type(), m.name.clone()));
58        }
59        // Hidden per-match id; the last column. See `stream_key`.
60        fields.push(Field::with_name(DataType::Int64, "_match_id"));
61        Schema::new(fields)
62    }
63
64    fn functional_dependency(&self) -> FunctionalDependencySet {
65        FunctionalDependencySet::new(self.partition_by.len() + self.measures.len() + 1)
66    }
67
68    fn stream_key(&self) -> Option<Vec<usize>> {
69        // ONE ROW PER MATCH emits one append-only row per match; a partition can contain many
70        // matches whose (partition + measures) output may be byte-identical, so those columns are
71        // not a key. The executor appends a hidden per-match id (the trailing column); the partition
72        // columns plus that id uniquely identify every emitted match. Keeping the partition columns
73        // in the key preserves partition-based sharding (no reshuffle before the downstream MV).
74        let match_id = self.partition_by.len() + self.measures.len();
75        let mut key: Vec<usize> = (0..self.partition_by.len()).collect();
76        key.push(match_id);
77        Some(key)
78    }
79
80    fn ctx(&self) -> OptimizerContextRef {
81        self.input.ctx()
82    }
83}
84
85impl<PlanRef: GenericPlanRef> crate::optimizer::plan_node::expr_visitable::ExprVisitable
86    for MatchRecognize<PlanRef>
87{
88    fn visit_exprs(&self, v: &mut dyn ExprVisitor) {
89        self.partition_by.iter().for_each(|e| v.visit_expr(e));
90        self.order_by.iter().for_each(|e| v.visit_expr(e));
91        self.measures.iter().for_each(|m| v.visit_expr(&m.expr));
92        self.defines
93            .iter()
94            .for_each(|d| v.visit_expr(&d.definition));
95        if let Some(within) = &self.within {
96            v.visit_expr(within);
97        }
98        if let Some(within_deadline) = &self.within_deadline {
99            v.visit_expr(within_deadline);
100        }
101    }
102}
103
104impl<PlanRef> MatchRecognize<PlanRef> {
105    /// Input column indices of `PARTITION BY`, or `None` if any key is not a plain column.
106    pub fn partition_key_indices(&self) -> Option<Vec<usize>> {
107        self.partition_by
108            .iter()
109            .map(|e| e.as_input_ref().map(|r| r.index()))
110            .collect()
111    }
112
113    /// Input column indices of `ORDER BY`, or `None` if any key is not a plain column.
114    pub fn order_key_indices(&self) -> Option<Vec<usize>> {
115        self.order_by
116            .iter()
117            .map(|e| e.as_input_ref().map(|r| r.index()))
118            .collect()
119    }
120
121    pub fn rewrite_exprs(&mut self, r: &mut dyn ExprRewriter) {
122        self.partition_by
123            .iter_mut()
124            .for_each(|e| *e = r.rewrite_expr(e.clone()));
125        self.order_by
126            .iter_mut()
127            .for_each(|e| *e = r.rewrite_expr(e.clone()));
128        self.measures
129            .iter_mut()
130            .for_each(|m| m.expr = r.rewrite_expr(m.expr.clone()));
131        self.defines
132            .iter_mut()
133            .for_each(|d| d.definition = r.rewrite_expr(d.definition.clone()));
134        if let Some(within) = &mut self.within {
135            *within = r.rewrite_expr(within.clone());
136        }
137        if let Some(within_deadline) = &mut self.within_deadline {
138            *within_deadline = r.rewrite_expr(within_deadline.clone());
139        }
140    }
141
142    pub fn rewrite_with_col_index_mapping(&mut self, mapping: &mut ColIndexMapping) {
143        self.partition_by
144            .iter_mut()
145            .for_each(|e| *e = mapping.rewrite_expr(e.clone()));
146        self.order_by
147            .iter_mut()
148            .for_each(|e| *e = mapping.rewrite_expr(e.clone()));
149        // Measure and DEFINE expressions are over synthetic per-match / per-candidate rows, not the
150        // plan input, so the input-column remapping applies to the slots' input column indices.
151        for m in &mut self.measures {
152            for slot in &mut m.slots {
153                if !matches!(slot.kind, MeasureSlotKind::Classifier) {
154                    slot.col_idx = mapping.map(slot.col_idx);
155                }
156            }
157        }
158        for d in &mut self.defines {
159            for slot in &mut d.slots {
160                slot.col_idx = mapping.map(slot.col_idx);
161            }
162        }
163    }
164}
165
166impl<PlanRef: GenericPlanRef> DistillUnit for MatchRecognize<PlanRef> {
167    fn distill_with_name<'a>(&self, name: impl Into<Str<'a>>) -> XmlNode<'a> {
168        // Schema-aware display (`t.ts`, not `$1`), like the sibling plan nodes.
169        let input_schema = self.input.schema();
170        let exprs = |es: &[ExprImpl]| {
171            Pretty::Array(
172                es.iter()
173                    .map(|e| {
174                        Pretty::display(&ExprDisplay {
175                            expr: e,
176                            input_schema,
177                        })
178                    })
179                    .collect(),
180            )
181        };
182        let measure_names: Vec<_> = self.measures.iter().map(|m| m.name.as_str()).collect();
183        let fields = vec![
184            ("partition_by", exprs(&self.partition_by)),
185            ("order_by", exprs(&self.order_by)),
186            ("measures", Pretty::debug(&measure_names)),
187            ("pattern", Pretty::display(&self.pattern)),
188        ];
189        childless_record(name, fields)
190    }
191}
risingwave_frontend/optimizer/plan_node/generic/match_recognize.rs

risingwave_frontend/optimizer/plan_node/generic/
match_recognize.rs
