risingwave_frontend/optimizer/plan_node/
stream_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 risingwave_common::catalog::Field;
16use risingwave_common::types::DataType;
17use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
18use risingwave_pb::stream_plan::stream_node::NodeBody;
19
20use super::stream::prelude::*;
21use super::utils::TableCatalogBuilder;
22use super::{ExprRewritable, ExprVisitable, PlanBase, PlanRef, Stream, TryToStreamPb, generic};
23use crate::TableCatalog;
24use crate::expr::{Expr, ExprRewriter, ExprVisitor};
25use crate::optimizer::plan_node::utils::impl_distill_by_unit;
26use crate::optimizer::property::{Distribution, MonotonicityMap, WatermarkColumns};
27use crate::scheduler::SchedulerResult;
28use crate::stream_fragmenter::BuildFragmentGraphState;
29
30/// `StreamMatchRecognize` implements [`super::Stream`] for a SQL `MATCH_RECOGNIZE` (row pattern
31/// recognition) operation.
32#[derive(Debug, Clone, PartialEq, Eq, Hash)]
33pub struct StreamMatchRecognize {
34    pub base: PlanBase<Stream>,
35    core: generic::MatchRecognize<PlanRef<Stream>>,
36}
37
38impl StreamMatchRecognize {
39    pub fn new(core: generic::MatchRecognize<PlanRef<Stream>>) -> Self {
40        // ONE ROW PER MATCH emits one row per completed match over append-only input, so the output
41        // is append-only. The output schema is the partition-by columns followed by the measures, so
42        // the partition key occupies the leading `n_part` output columns; the input was sharded by
43        // the partition key (see `to_stream`), so the output is hash-sharded on those columns.
44        let n_part = core.partition_by.len();
45        let dist = Distribution::HashShard((0..n_part).collect());
46        let base = PlanBase::new_stream_with_core(
47            &core,
48            dist,
49            StreamKind::AppendOnly,
50            // The operator emits only FINAL matches — each output row is decided once and never
51            // revised or retracted — so it already satisfies Emit-On-Window-Close semantics (it
52            // just doesn't wait for a window to close: followed/terminal matches emit at arrival,
53            // only WITHIN-finality waits for the watermark). Declaring it lets `EMIT ON WINDOW
54            // CLOSE` queries name that behavior explicitly instead of being rejected.
55            true,
56            WatermarkColumns::new(),
57            MonotonicityMap::new(),
58        );
59        Self { base, core }
60    }
61
62    /// Per-partition buffered-row state table. Layout:
63    ///   `[ seq (i64) , <input columns..> ]`
64    /// keyed by (partition columns, ORDER BY columns, seq). Keying by the order columns keeps the
65    /// state physically sorted by (partition, order key), so the watermark pass can scan it in PK
66    /// order — rows arrive grouped by partition and already ordered within each partition — and
67    /// process one partition at a time without an in-memory sort. The executor buffers the raw input
68    /// row per live row and restores the buffer from here on recovery. `seq` is a per-actor monotonic
69    /// id that breaks ties between rows with equal ORDER BY keys; consumed rows are deleted after the
70    /// scan. DEFINE predicates and MEASURES are both evaluated at match time from the stored input
71    /// rows, so neither is persisted. The partition and order-key columns are columns of the stored
72    /// input row, so they are not stored separately.
73    ///
74    /// The whole input row is stored for simplicity. A future optimization could project to only the
75    /// columns actually referenced (partition / order keys plus the columns read by DEFINE and
76    /// MEASURES), shrinking per-row state at the cost of a column-remapping layer between the stored
77    /// row and the slot indices. Deferred until state size warrants it.
78    fn infer_state_table(&self) -> TableCatalog {
79        let mut tbl_builder = TableCatalogBuilder::default();
80        let input_fields = self.core.input.schema().fields().to_vec();
81        let partition_indices = self
82            .core
83            .partition_key_indices()
84            .expect("partition keys validated to be columns");
85
86        // seq
87        tbl_builder.add_column(&Field::with_name(DataType::Int64, "seq"));
88        // raw input columns (offset 1)
89        for f in &input_fields {
90            tbl_builder.add_column(f);
91        }
92
93        // pk: partition columns, then the ORDER BY columns, then seq (the tiebreaker for equal
94        // order keys). Keying by the order columns makes the state physically sorted by
95        // (partition, order key): a PK-order scan yields each partition's rows already ordered, so
96        // the watermark pass needs no in-memory sort and can stream one partition at a time (holding
97        // only the current partition's rows resident) instead of loading a whole vnode at once.
98        let partition_positions: Vec<usize> = partition_indices.iter().map(|i| 1 + i).collect();
99        let order_positions: Vec<usize> = self
100            .core
101            .order_key_indices()
102            .expect("order keys validated to be columns")
103            .iter()
104            .map(|i| 1 + i)
105            .collect();
106        for &p in &partition_positions {
107            tbl_builder.add_order_column(p, OrderType::ascending());
108        }
109        for &o in &order_positions {
110            // Skip an order column already in the partition prefix — it is already a key column, and
111            // adding it twice would be redundant.
112            if !partition_positions.contains(&o) {
113                tbl_builder.add_order_column(o, OrderType::ascending());
114            }
115        }
116        tbl_builder.add_order_column(0, OrderType::ascending());
117        // Distribute the state by the partition columns so each actor owns its partitions' state.
118        // read_prefix_len_hint = 0: the watermark scan iterates each owned vnode with an empty prefix
119        // (it cannot compute a vnode from an empty prefix), so we must not assert a prefix length.
120        tbl_builder.build(partition_positions, 0)
121    }
122
123    fn input(&self) -> PlanRef<Stream> {
124        self.core.input.clone()
125    }
126
127    fn clone_with_input(&self, input: PlanRef<Stream>) -> Self {
128        let mut core = self.core.clone();
129        core.input = input;
130        Self::new(core)
131    }
132}
133
134impl_plan_tree_node_for_unary! { Stream, StreamMatchRecognize }
135impl_distill_by_unit!(StreamMatchRecognize, core, "StreamMatchRecognize");
136
137impl TryToStreamPb for StreamMatchRecognize {
138    fn try_to_stream_prost_body(
139        &self,
140        state: &mut BuildFragmentGraphState,
141    ) -> SchedulerResult<NodeBody> {
142        use risingwave_pb::stream_plan::*;
143
144        let retract = self.stream_kind().is_retract();
145
146        // PARTITION BY / ORDER BY were validated to be plain columns in `to_stream`.
147        let partition_by = self
148            .core
149            .partition_key_indices()
150            .expect("partition keys validated to be columns")
151            .into_iter()
152            .map(|i| i as u32)
153            .collect();
154        // ORDER BY is carried as `ColumnOrder` (like every other ordered streaming node). v1 only
155        // supports the default ascending order — non-ascending is rejected in the binder — so each
156        // key is emitted ascending; the executor (and `from_proto`) assert that on the way back.
157        let order_by = self
158            .core
159            .order_key_indices()
160            .expect("order keys validated to be columns")
161            .into_iter()
162            .map(|i| ColumnOrder::new(i, OrderType::ascending()).to_protobuf())
163            .collect();
164
165        let measures = self
166            .core
167            .measures
168            .iter()
169            .map(|m| {
170                let expr = m
171                    .expr
172                    .to_expr_proto_checked_pure(retract, "match_recognize measure")?;
173                let slots = m
174                    .slots
175                    .iter()
176                    .map(|s| MatchRecognizeMeasureSlot {
177                        // The binder's kind IS the wire enum; no conversion layer.
178                        kind: s.kind as i32,
179                        vars: s.vars.clone(),
180                        col_idx: s.col_idx as u32,
181                        data_type: Some(s.data_type.to_protobuf()),
182                        agg_call: s.agg.as_ref().map(|a| a.to_protobuf()),
183                    })
184                    .collect();
185                Ok(MatchRecognizeMeasure {
186                    expr: Some(expr),
187                    name: m.name.clone(),
188                    slots,
189                })
190            })
191            .collect::<crate::error::Result<Vec<_>>>()?;
192
193        let defines = self
194            .core
195            .defines
196            .iter()
197            .map(|d| {
198                let condition = d
199                    .definition
200                    .to_expr_proto_checked_pure(retract, "match_recognize define")?;
201                let slots = d
202                    .slots
203                    .iter()
204                    .map(|s| MatchRecognizeDefineSlot {
205                        // The binder's kind IS the wire enum; no conversion layer.
206                        kind: s.kind as i32,
207                        vars: s.vars.clone(),
208                        col_idx: s.col_idx as u32,
209                        offset: s.offset as u32,
210                    })
211                    .collect();
212                Ok(MatchRecognizeDefine {
213                    symbol: d.symbol.clone(),
214                    condition: Some(condition),
215                    slots,
216                })
217            })
218            .collect::<crate::error::Result<Vec<_>>>()?;
219
220        let state_table = self
221            .infer_state_table()
222            .with_id(state.gen_table_id_wrapped())
223            .to_internal_table_prost();
224        Ok(NodeBody::MatchRecognize(Box::new(MatchRecognizeNode {
225            partition_by,
226            order_by,
227            measures,
228            defines,
229            pattern_node: Some(lower_pattern(&self.core.pattern)?),
230            state_table: Some(state_table),
231            input_mode: MatchRecognizeInputMode::EventTime as i32,
232            after_match_skip: {
233                use risingwave_pb::stream_plan::MatchRecognizeAfterMatchSkip as PbSkip;
234                use risingwave_pb::stream_plan::match_recognize_after_match_skip::Mode;
235                use risingwave_sqlparser::ast::AfterMatchSkip;
236                Some(match &self.core.after_match_skip {
237                    Some(AfterMatchSkip::ToNextRow) => PbSkip {
238                        mode: Mode::ToNextRow as i32,
239                        target: None,
240                    },
241                    Some(AfterMatchSkip::ToFirst(sym)) => PbSkip {
242                        mode: Mode::ToFirst as i32,
243                        target: Some(sym.real_value()),
244                    },
245                    Some(AfterMatchSkip::ToLast(sym)) => PbSkip {
246                        mode: Mode::ToLast as i32,
247                        target: Some(sym.real_value()),
248                    },
249                    // `PAST LAST ROW`, explicit or defaulted. Exhaustive on purpose: a future AST
250                    // variant must fail to compile here, not silently lower to PAST LAST ROW.
251                    Some(AfterMatchSkip::PastLastRow) | None => PbSkip {
252                        mode: Mode::PastLastRow as i32,
253                        target: None,
254                    },
255                })
256            },
257            within: self
258                .core
259                .within
260                .as_ref()
261                .map(|w| w.to_expr_proto_checked_pure(retract, "match_recognize within"))
262                .transpose()?,
263            within_deadline: self
264                .core
265                .within_deadline
266                .as_ref()
267                .map(|w| w.to_expr_proto_checked_pure(retract, "match_recognize within deadline"))
268                .transpose()?,
269        })))
270    }
271}
272
273impl ExprRewritable<Stream> for StreamMatchRecognize {
274    fn has_rewritable_expr(&self) -> bool {
275        true
276    }
277
278    fn rewrite_exprs(&self, r: &mut dyn ExprRewriter) -> PlanRef<Stream> {
279        let mut core = self.core.clone();
280        core.rewrite_exprs(r);
281        Self {
282            base: self.base.clone_with_new_plan_id(),
283            core,
284        }
285        .into()
286    }
287}
288
289impl ExprVisitable for StreamMatchRecognize {
290    fn visit_exprs(&self, v: &mut dyn ExprVisitor) {
291        self.core.visit_exprs(v)
292    }
293}
294
295/// Lower a bound `MatchRecognizePattern` (the `sqlparser` AST) into the structured pattern proto
296/// consumed by the executor. Parenthesized groups are flattened (the executor pattern has no group
297/// node); anchors (`^`, `$`) and exclusions (`{- ... -}`) are rejected here as they are not yet
298/// supported. This replaces the previous text round-trip (`Display` ↔ a hand-rolled parser).
299fn lower_pattern(
300    pattern: &risingwave_sqlparser::ast::MatchRecognizePattern,
301) -> crate::error::Result<risingwave_pb::stream_plan::MatchRecognizePatternNode> {
302    use risingwave_common::bail_not_implemented;
303    use risingwave_pb::stream_plan::match_recognize_pattern_node::Node;
304    use risingwave_pb::stream_plan::{
305        MatchRecognizePatternNode, MatchRecognizePatternSeq, MatchRecognizePermutePattern,
306        MatchRecognizeQuantifiedPattern,
307    };
308    use risingwave_sqlparser::ast::{MatchRecognizePattern as Pat, MatchRecognizeSymbol as Sym};
309
310    fn named(symbol: &Sym) -> crate::error::Result<String> {
311        match symbol {
312            Sym::Named(ident) => Ok(ident.real_value()),
313            Sym::Start | Sym::End => {
314                bail_not_implemented!("row pattern anchors (^, $) in MATCH_RECOGNIZE")
315            }
316        }
317    }
318
319    fn node(n: Node) -> risingwave_pb::stream_plan::MatchRecognizePatternNode {
320        MatchRecognizePatternNode { node: Some(n) }
321    }
322
323    fn lower_seq(
324        patterns: &[Pat],
325    ) -> crate::error::Result<risingwave_pb::stream_plan::MatchRecognizePatternSeq> {
326        Ok(MatchRecognizePatternSeq {
327            patterns: patterns
328                .iter()
329                .map(lower_pattern)
330                .collect::<crate::error::Result<Vec<_>>>()?,
331        })
332    }
333
334    match pattern {
335        Pat::Symbol(symbol) => Ok(node(Node::Var(named(symbol)?))),
336        Pat::Exclude(_) => {
337            bail_not_implemented!("row pattern exclusions ({{- ... -}}) in MATCH_RECOGNIZE")
338        }
339        // PERMUTE expands to the alternation of all n! orderings of its variables, so the NFA grows
340        // factorially. The arity cap that bounds that is enforced at bind time, together with the
341        // quantifier bounds — see `validate_pattern` in `binder::relation::match_recognize`.
342        Pat::Permute(symbols) => Ok(node(Node::Permute(MatchRecognizePermutePattern {
343            vars: symbols
344                .iter()
345                .map(named)
346                .collect::<crate::error::Result<Vec<_>>>()?,
347        }))),
348        Pat::Concat(patterns) => Ok(node(Node::Concat(lower_seq(patterns)?))),
349        Pat::Alternation(patterns) => Ok(node(Node::Alternation(lower_seq(patterns)?))),
350        // A parenthesized group is purely syntactic grouping; flatten it away.
351        Pat::Group(inner) => lower_pattern(inner),
352        Pat::Repetition(inner, quantifier, reluctant) => Ok(node(Node::Quantified(Box::new(
353            MatchRecognizeQuantifiedPattern {
354                inner: Some(Box::new(lower_pattern(inner)?)),
355                quantifier: Some(lower_quantifier(quantifier)),
356                reluctant: *reluctant,
357            },
358        )))),
359    }
360}
361
362/// Map a `RepetitionQuantifier` to the proto quantifier. `*`, `+`, `?` map to their dedicated
363/// kinds; the `{...}` forms all map to `RANGE` with an explicit `min` and an optional `max`.
364///
365/// The bounds are validated at bind time (`binder::relation::match_recognize`: `validate_pattern`),
366/// which is the last point at which a bad bound can be reported to the author: the NFA is expanded
367/// from these bounds in `Nfa::compile`, on the compute node, when the actor is built — long after the
368/// statement has been acknowledged.
369fn lower_quantifier(
370    quantifier: &risingwave_sqlparser::ast::RepetitionQuantifier,
371) -> risingwave_pb::stream_plan::MatchRecognizeQuantifier {
372    use risingwave_pb::stream_plan::MatchRecognizeQuantifier;
373    use risingwave_pb::stream_plan::match_recognize_quantifier::Kind;
374    use risingwave_sqlparser::ast::RepetitionQuantifier as Q;
375
376    let (kind, min, max) = match quantifier {
377        Q::ZeroOrMore => (Kind::Star, 0, None),
378        Q::OneOrMore => (Kind::Plus, 0, None),
379        Q::AtMostOne => (Kind::Question, 0, None),
380        Q::Exactly(n) => (Kind::Range, *n, Some(*n)),
381        Q::AtLeast(n) => (Kind::Range, *n, None),
382        Q::AtMost(m) => (Kind::Range, 0, Some(*m)),
383        Q::Range(n, m) => (Kind::Range, *n, Some(*m)),
384    };
385    MatchRecognizeQuantifier {
386        kind: kind as i32,
387        min,
388        max,
389    }
390}
risingwave_frontend/optimizer/plan_node/stream_match_recognize.rs

risingwave_frontend/optimizer/plan_node/
stream_match_recognize.rs
