risingwave_frontend/optimizer/plan_node/
logical_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 fixedbitset::FixedBitSet;
16use itertools::Itertools;
17use risingwave_expr::bail;
18use risingwave_sqlparser::ast::{AfterMatchSkip, MatchRecognizePattern, RowsPerMatch};
19
20use super::generic::GenericPlanRef;
21use super::stream::StreamPlanNodeMetadata;
22use super::{
23    ColPrunable, ColumnPruningContext, ExprRewritable, ExprVisitable, Logical,
24    LogicalPlanRef as PlanRef, LogicalProject, PlanBase, PlanTreeNodeUnary, PredicatePushdown,
25    PredicatePushdownContext, ToBatch, ToStream, ToStreamContext, gen_filter_and_pushdown, generic,
26};
27use crate::binder::{BoundMeasure, BoundSymbolDefinition, MeasureSlotKind};
28use crate::error::Result;
29use crate::expr::ExprImpl;
30use crate::optimizer::plan_node::utils::impl_distill_by_unit;
31use crate::utils::{ColIndexMapping, Condition};
32
33/// `LogicalMatchRecognize` implements [`super::Logical`] for a SQL `MATCH_RECOGNIZE` (row pattern
34/// recognition) operation.
35#[derive(Debug, Clone, PartialEq, Eq, Hash)]
36pub struct LogicalMatchRecognize {
37    pub base: PlanBase<super::Logical>,
38    core: generic::MatchRecognize<PlanRef>,
39}
40
41impl LogicalMatchRecognize {
42    #[allow(clippy::too_many_arguments)]
43    pub fn new(
44        input: PlanRef,
45        partition_by: Vec<ExprImpl>,
46        order_by: Vec<ExprImpl>,
47        measures: Vec<BoundMeasure>,
48        rows_per_match: Option<RowsPerMatch>,
49        after_match_skip: Option<AfterMatchSkip>,
50        pattern: MatchRecognizePattern,
51        defines: Vec<BoundSymbolDefinition>,
52        within: Option<ExprImpl>,
53        within_deadline: Option<ExprImpl>,
54    ) -> Self {
55        let core = generic::MatchRecognize {
56            input,
57            partition_by,
58            order_by,
59            measures,
60            rows_per_match,
61            after_match_skip,
62            pattern,
63            defines,
64            within,
65            within_deadline,
66        };
67        let base = PlanBase::new_logical_with_core(&core);
68        Self { base, core }
69    }
70
71    /// The set of input columns referenced by any expression in the clause.
72    fn input_required_cols(&self) -> FixedBitSet {
73        let input_col_num = self.core.input.schema().len();
74        let mut required = FixedBitSet::with_capacity(input_col_num);
75        for e in &self.core.partition_by {
76            required.union_with(&e.collect_input_refs(input_col_num));
77        }
78        for e in &self.core.order_by {
79            required.union_with(&e.collect_input_refs(input_col_num));
80        }
81        // Measure expressions are over the synthetic per-match row; the real input columns they read
82        // are recorded in the slots.
83        for m in &self.core.measures {
84            for slot in &m.slots {
85                if !matches!(slot.kind, MeasureSlotKind::Classifier) {
86                    required.insert(slot.col_idx);
87                }
88            }
89        }
90        // DEFINE predicates are over synthetic per-candidate rows; the real input columns they read
91        // are recorded in the slots.
92        for d in &self.core.defines {
93            for slot in &d.slots {
94                required.insert(slot.col_idx);
95            }
96        }
97        required
98    }
99}
100
101impl PlanTreeNodeUnary<Logical> for LogicalMatchRecognize {
102    fn input(&self) -> PlanRef {
103        self.core.input.clone()
104    }
105
106    fn clone_with_input(&self, input: PlanRef) -> Self {
107        Self::new(
108            input,
109            self.core.partition_by.clone(),
110            self.core.order_by.clone(),
111            self.core.measures.clone(),
112            self.core.rows_per_match.clone(),
113            self.core.after_match_skip.clone(),
114            self.core.pattern.clone(),
115            self.core.defines.clone(),
116            self.core.within.clone(),
117            self.core.within_deadline.clone(),
118        )
119    }
120}
121
122impl_plan_tree_node_for_unary! { Logical, LogicalMatchRecognize }
123impl_distill_by_unit!(LogicalMatchRecognize, core, "LogicalMatchRecognize");
124
125impl ColPrunable for LogicalMatchRecognize {
126    fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef {
127        // Prune the input down to the columns the clause's expressions actually reference.
128        let input_col_num = self.core.input.schema().len();
129        let input_required = self.input_required_cols();
130        let input_required_cols: Vec<_> = input_required.ones().collect();
131
132        let mut col_index_mapping =
133            ColIndexMapping::with_remaining_columns(&input_required_cols, input_col_num);
134
135        let mut new_core = self.core.clone();
136        new_core.input = self.core.input.prune_col(&input_required_cols, ctx);
137        new_core.rewrite_with_col_index_mapping(&mut col_index_mapping);
138
139        let node: PlanRef = Self {
140            base: PlanBase::new_logical_with_core(&new_core),
141            core: new_core,
142        }
143        .into();
144
145        // The node's own output is (partition cols + measures); project if the caller wants a subset.
146        let output_col_num = self.schema().len();
147        if required_cols == (0..output_col_num).collect_vec() {
148            node
149        } else {
150            LogicalProject::with_mapping(
151                node,
152                ColIndexMapping::with_remaining_columns(required_cols, output_col_num),
153            )
154            .into()
155        }
156    }
157}
158
159impl ExprRewritable<Logical> for LogicalMatchRecognize {
160    fn has_rewritable_expr(&self) -> bool {
161        true
162    }
163
164    fn rewrite_exprs(&self, r: &mut dyn crate::expr::ExprRewriter) -> PlanRef {
165        let mut core = self.core.clone();
166        core.rewrite_exprs(r);
167        Self {
168            base: PlanBase::new_logical_with_core(&core),
169            core,
170        }
171        .into()
172    }
173}
174
175impl ExprVisitable for LogicalMatchRecognize {
176    fn visit_exprs(&self, v: &mut dyn crate::expr::ExprVisitor) {
177        self.core.visit_exprs(v);
178    }
179}
180
181impl PredicatePushdown for LogicalMatchRecognize {
182    fn predicate_pushdown(
183        &self,
184        predicate: Condition,
185        ctx: &mut PredicatePushdownContext,
186    ) -> PlanRef {
187        // Output columns are computed (partition/measures), so do not push predicates through, but
188        // keep recursing so a share below this node receives a contribution from every parent.
189        gen_filter_and_pushdown(self, predicate, Condition::true_cond(), ctx)
190    }
191}
192
193impl ToBatch for LogicalMatchRecognize {
194    fn to_batch(&self) -> Result<super::BatchPlanRef> {
195        bail!("BatchMatchRecognize is not implemented yet")
196    }
197}
198
199impl ToStream for LogicalMatchRecognize {
200    fn to_stream(&self, ctx: &mut ToStreamContext) -> Result<super::StreamPlanRef> {
201        use super::{StreamEowcSort, StreamFilter, StreamMatchRecognize};
202        use crate::error::ErrorCode;
203        use crate::expr::{ExprType, FunctionCall, InputRef};
204        use crate::optimizer::property::RequiredDist;
205        use crate::utils::Condition;
206        // v1 restrictions: PARTITION BY / ORDER BY must be plain columns, PARTITION BY non-empty.
207        // `NotSupported(cause, hint)` throughout, matching this feature's binder-side validation.
208        if self.core.partition_key_indices().is_none() || self.core.order_key_indices().is_none() {
209            return Err(ErrorCode::NotSupported(
210                "MATCH_RECOGNIZE with an expression in PARTITION BY or ORDER BY".to_owned(),
211                "use plain column references; compute the expression in a view below and \
212                 partition/order by the resulting column"
213                    .to_owned(),
214            )
215            .into());
216        }
217        if self
218            .core
219            .partition_key_indices()
220            .expect("checked above")
221            .is_empty()
222        {
223            return Err(ErrorCode::NotSupported(
224                "MATCH_RECOGNIZE without a PARTITION BY".to_owned(),
225                "add PARTITION BY; for a global pattern, partition by a constant column computed \
226                 in a view below (all rows then match within one partition)"
227                    .to_owned(),
228            )
229            .into());
230        }
231        let order_indices = self.core.order_key_indices().expect("checked above");
232        let Some(&time_col) = order_indices.first() else {
233            bail!("MATCH_RECOGNIZE requires an ORDER BY clause");
234        };
235        let partition_key_indices = self.core.partition_key_indices().expect("checked above");
236
237        let stream_input = self.input().to_stream(ctx)?;
238        // The executor matches over an append-only sequence and emits insert-only results; it has no
239        // semantics for retracting or revising an already-emitted match, and the stream plan node
240        // declares append-only output. Reject a non-append-only input during planning so the user
241        // gets an error at `CREATE`, rather than the executor crashing on the first update/delete.
242        if !stream_input.append_only() {
243            return Err(ErrorCode::NotSupported(
244                "MATCH_RECOGNIZE over a non-append-only input (updates or deletes could revise \
245                 an already-emitted match)"
246                    .to_owned(),
247                "use an append-only source or table (e.g. CREATE TABLE ... APPEND ONLY)".to_owned(),
248            )
249            .into());
250        }
251        // Event-time contract: the executor buffers rows and finalises matches as the watermark on
252        // the leading ORDER BY column advances, so that column must carry a watermark. This mirrors
253        // Flink requiring a rowtime attribute on ORDER BY.
254        if !stream_input.watermark_columns().contains(time_col) {
255            return Err(ErrorCode::NotSupported(
256                "MATCH_RECOGNIZE without a watermark on the leading ORDER BY column".to_owned(),
257                "declare one on the source or table, e.g. WATERMARK FOR ts AS ts - INTERVAL '5' \
258                 SECOND"
259                    .to_owned(),
260            )
261            .into());
262        }
263        // A NULL leading order key has no event time: no watermark can ever release it from the
264        // sort (whose cache key requires it non-null), and it cannot be ordered against other
265        // rows. Filter such rows out below the sort, exactly as event-time processing drops
266        // NULL-rowtime rows.
267        let ts_type = stream_input.schema().fields()[time_col].data_type();
268        let not_null: ExprImpl = FunctionCall::new(
269            ExprType::IsNotNull,
270            vec![InputRef::new(time_col, ts_type).into()],
271        )?
272        .into();
273        let filter_core = generic::Filter {
274            predicate: Condition::with_expr(not_null),
275            input: stream_input,
276        };
277        let stream_input: super::StreamPlanRef = StreamFilter::new(filter_core).into();
278
279        // Ordered-input planning: hash-shard by the PARTITION BY key, then an EowcSort over
280        // the full ORDER BY (leading watermark column plus secondary order columns) so the matcher
281        // receives rows already in ORDER BY order, strictly below each forwarded watermark. Sort
282        // and matcher stay in the same fragment -- no exchange between them, which would destroy
283        // the ordering the sort just established. The matcher itself then owns only NFA state and
284        // match finalization.
285        //
286        // The requirement is the EXACT hash distribution over the partition columns in PARTITION
287        // BY order -- not `shard_by_key`. The matcher's state table hashes its distribution key in
288        // PARTITION BY order, so the rows must be physically routed by that same column order;
289        // `shard_by_key` would accept any subset in any order (and its enforcing exchange hashes
290        // in ascending column-index order), letting the row's routed vnode disagree with the vnode
291        // its state-table key computes -- a "vnode should not be accessed" panic on the first
292        // insert for `PARTITION BY (b, a)` or a pre-sharded input.
293        let stream_input = RequiredDist::hash_shard(&partition_key_indices)
294            .streaming_enforce_if_not_satisfies(stream_input)?;
295        let secondary_order: Vec<usize> = order_indices[1..].to_vec();
296        let sorted_input: super::StreamPlanRef =
297            StreamEowcSort::with_secondary_order(stream_input, time_col, secondary_order).into();
298        let core = generic::MatchRecognize {
299            input: sorted_input,
300            partition_by: self.core.partition_by.clone(),
301            order_by: self.core.order_by.clone(),
302            measures: self.core.measures.clone(),
303            rows_per_match: self.core.rows_per_match.clone(),
304            after_match_skip: self.core.after_match_skip.clone(),
305            pattern: self.core.pattern.clone(),
306            defines: self.core.defines.clone(),
307            within: self.core.within.clone(),
308            within_deadline: self.core.within_deadline.clone(),
309        };
310        Ok(StreamMatchRecognize::new(core).into())
311    }
312
313    fn logical_rewrite_for_stream(
314        &self,
315        ctx: &mut super::convert::RewriteStreamContext,
316    ) -> Result<(PlanRef, ColIndexMapping)> {
317        let (input, input_col_change) = self.core.input.logical_rewrite_for_stream(ctx)?;
318        let mut new_core = self.core.clone();
319        new_core.input = input;
320        let mut mapping = input_col_change;
321        new_core.rewrite_with_col_index_mapping(&mut mapping);
322        let node = Self {
323            base: PlanBase::new_logical_with_core(&new_core),
324            core: new_core,
325        };
326        // Output columns (partition + measures) are produced fresh by this node, so downstream sees
327        // an identity mapping over this node's own output schema.
328        let out_col_change = ColIndexMapping::identity(node.schema().len());
329        Ok((node.into(), out_col_change))
330    }
331}
risingwave_frontend/optimizer/plan_node/logical_match_recognize.rs

risingwave_frontend/optimizer/plan_node/
logical_match_recognize.rs
