risingwave_stream/from_proto/
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::types::DataType;
16use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
17use risingwave_expr::expr::build_non_strict_from_prost;
18use risingwave_pb::stream_plan::{MatchRecognizeInputMode, MatchRecognizeNode};
19use risingwave_storage::StateStore;
20
21use super::ExecutorBuilder;
22use crate::common::table::state_table::StateTableBuilder;
23use crate::error::StreamResult;
24use crate::executor::Executor;
25use crate::executor::match_recognize::executor::{
26    CompiledDefine, CompiledMeasure, DeadlineErrorReport, MatchRecognizeExecutor,
27    MatchRecognizeExecutorArgs,
28};
29use crate::executor::match_recognize::nfa::{Nfa, SkipMode};
30use crate::executor::match_recognize::proto::pattern_from_protobuf;
31use crate::task::ExecutorParams;
32
33pub struct MatchRecognizeExecutorBuilder;
34
35impl_stream_node_body!(MatchRecognize(MatchRecognizeNode) => MatchRecognizeExecutorBuilder);
36
37impl ExecutorBuilder for MatchRecognizeExecutorBuilder {
38    type Node = MatchRecognizeNode;
39
40    async fn new_boxed_executor(
41        params: ExecutorParams,
42        node: &MatchRecognizeNode,
43        store: impl StateStore,
44    ) -> StreamResult<Executor> {
45        let [input]: [_; 1] = params.input.try_into().unwrap();
46
47        // This executor's entire correctness rests on the ordered-input contract the EVENT_TIME
48        // plan (an EowcSort upstream in the same fragment) provides. A different input mode —
49        // PROCESSING_TIME is reserved, unimplemented — must fail here, not silently run against
50        // rows whose ordering guarantee does not hold.
51        // An out-of-range wire value decodes as `Unspecified` through the accessor, which would
52        // silently run an unknown future mode as event-time — the one contract this executor's
53        // correctness rests on. Reject it like every other enum in this decode path; a raw 0
54        // (genuinely unset) is accepted as event-time since this frontend always writes it.
55        if node.input_mode != 0 && node.input_mode() == MatchRecognizeInputMode::Unspecified {
56            return Err(
57                anyhow::anyhow!("unknown MATCH_RECOGNIZE input mode: {}", node.input_mode).into(),
58            );
59        }
60        match node.input_mode() {
61            MatchRecognizeInputMode::Unspecified | MatchRecognizeInputMode::EventTime => {}
62            other => {
63                return Err(
64                    anyhow::anyhow!("unsupported MATCH_RECOGNIZE input mode: {other:?}").into(),
65                );
66            }
67        }
68
69        let partition_key_indices = node.partition_by.iter().map(|&i| i as usize).collect();
70        // ORDER BY is carried as `ColumnOrder`. v1 only supports the default ascending order (the
71        // binder rejects anything else); assert it here too so a non-ascending plan fails fast
72        // rather than being silently sorted ascending by the executor.
73        let order_key_indices = node
74            .order_by
75            .iter()
76            .map(|c| {
77                let co = ColumnOrder::from_protobuf(c);
78                if co.order_type != OrderType::ascending() {
79                    return Err(anyhow::anyhow!(
80                        "MATCH_RECOGNIZE only supports the default ascending ORDER BY, got {:?}",
81                        co.order_type
82                    )
83                    .into());
84                }
85                Ok(co.column_index)
86            })
87            .collect::<StreamResult<Vec<usize>>>()?;
88        // The executor reads the leading ORDER BY column unconditionally; an empty list is a
89        // corrupt plan and must fail here, not index-panic there.
90        if order_key_indices.is_empty() {
91            return Err(anyhow::anyhow!("MATCH_RECOGNIZE plan carries an empty ORDER BY").into());
92        }
93
94        let defines = node
95            .defines
96            .iter()
97            .map(|d| CompiledDefine::from_protobuf(d, params.eval_error_report.clone()))
98            .collect::<crate::executor::StreamExecutorResult<Vec<_>>>()?;
99        let measures = node
100            .measures
101            .iter()
102            .map(|m| CompiledMeasure::from_protobuf(m, params.eval_error_report.clone()))
103            .collect::<crate::executor::StreamExecutorResult<Vec<_>>>()?;
104
105        let pattern_node = node
106            .pattern_node
107            .as_ref()
108            .ok_or_else(|| anyhow::anyhow!("MATCH_RECOGNIZE node missing pattern"))?;
109        let pattern = pattern_from_protobuf(pattern_node)
110            .map_err(|e| anyhow::anyhow!("invalid MATCH_RECOGNIZE pattern: {e}"))?;
111        let nfa = Nfa::compile(&pattern);
112
113        // Fail fast on anything malformed rather than silently defaulting to PAST LAST ROW, which
114        // would mask a corrupt plan or a version skew.
115        let skip = {
116            use risingwave_pb::stream_plan::match_recognize_after_match_skip::Mode;
117            let pb_skip = node
118                .after_match_skip
119                .as_ref()
120                .ok_or_else(|| anyhow::anyhow!("MATCH_RECOGNIZE node missing after_match_skip"))?;
121            let target = || {
122                pb_skip.target.clone().ok_or_else(|| {
123                    anyhow::anyhow!("AFTER MATCH SKIP TO FIRST/LAST missing its target variable")
124                })
125            };
126            match pb_skip.mode() {
127                Mode::PastLastRow => SkipMode::PastLastRow,
128                Mode::ToNextRow => SkipMode::ToNextRow,
129                Mode::ToFirst => SkipMode::ToFirst(target()?),
130                Mode::ToLast => SkipMode::ToLast(target()?),
131                Mode::Unspecified => {
132                    return Err(anyhow::anyhow!(
133                        "invalid MATCH_RECOGNIZE after_match_skip mode: {}",
134                        pb_skip.mode
135                    )
136                    .into());
137                }
138            }
139        };
140
141        let within = node
142            .within
143            .as_ref()
144            .map(|e| build_non_strict_from_prost(e, params.eval_error_report.clone()))
145            .transpose()?;
146        // Over `DeadlineErrorReport`, not the actor's report directly: `first + bound` leaving the
147        // order key's range is the window that never closes, not a compute error to count and log
148        // per row. See `eval_deadline` in the executor.
149        let within_deadline = node
150            .within_deadline
151            .as_ref()
152            .map(|e| {
153                build_non_strict_from_prost(
154                    e,
155                    DeadlineErrorReport::new(params.eval_error_report.clone()),
156                )
157            })
158            .transpose()?;
159        // The two WITHIN expressions are a correctness-coupled pair, and the coupling tightened when
160        // the executor's span check started reading the cached deadline instead of evaluating the
161        // predicate: `within` present with `within_deadline` absent now rejects EVERY candidate, so
162        // the view would silently produce zero rows. The binder only ever emits both or neither
163        // (`lower_within`), so a plan carrying one is corrupt — fail loud, as the rest of this
164        // decoder does, rather than emitting nothing forever.
165        if within.is_some() != within_deadline.is_some() {
166            return Err(anyhow::anyhow!(
167                "MATCH_RECOGNIZE carries only one of the two WITHIN expressions \
168                 (predicate: {}, deadline: {}); the binder emits both or neither",
169                within.is_some(),
170                within_deadline.is_some(),
171            )
172            .into());
173        }
174        // The deadline is compared directly against the order key and the watermark
175        // (`ScalarRefImpl::default_cmp` panics across variants — an actor crash loop that recovery
176        // replays), and the span predicate is a boolean. The binder guarantees both
177        // (`lower_within`); re-state it here so a skewed or corrupt plan fails at build time.
178        let order_key_type = input
179            .schema()
180            .fields
181            .get(order_key_indices[0])
182            .map(|f| f.data_type())
183            .ok_or_else(|| {
184                anyhow::anyhow!(
185                    "MATCH_RECOGNIZE ORDER BY column {} is out of range for an input of {} columns",
186                    order_key_indices[0],
187                    input.schema().len()
188                )
189            })?;
190        if let Some(deadline) = &within_deadline
191            && deadline.return_type() != order_key_type
192        {
193            return Err(anyhow::anyhow!(
194                "MATCH_RECOGNIZE WITHIN deadline has type {} but the ORDER BY column has type {}; \
195                 the two are compared directly",
196                deadline.return_type(),
197                order_key_type,
198            )
199            .into());
200        }
201        if let Some(predicate) = &within
202            && predicate.return_type() != DataType::Boolean
203        {
204            return Err(anyhow::anyhow!(
205                "MATCH_RECOGNIZE WITHIN span predicate has type {}, expected boolean",
206                predicate.return_type(),
207            )
208            .into());
209        }
210
211        let vnode_bitmap = params.vnode_bitmap.clone().map(std::sync::Arc::new);
212        let state_table_catalog = node
213            .state_table
214            .as_ref()
215            .ok_or_else(|| anyhow::anyhow!("MATCH_RECOGNIZE node missing its state table"))?;
216        let state_table =
217            StateTableBuilder::new(state_table_catalog, store.clone(), vnode_bitmap.clone())
218                .forbid_preload_all_rows()
219                .build()
220                .await;
221        let exec = MatchRecognizeExecutor::new(MatchRecognizeExecutorArgs {
222            ctx: params.actor_context,
223            input,
224            schema: params.info.schema.clone(),
225            chunk_size: params.config.developer.chunk_size,
226            partition_key_indices,
227            order_key_indices,
228            measures,
229            defines,
230            within,
231            nfa,
232            skip,
233            eval_error_report: params.eval_error_report,
234            within_deadline,
235            state_table,
236        });
237
238        Ok((params.info, exec).into())
239    }
240}
risingwave_stream/from_proto/match_recognize.rs

risingwave_stream/from_proto/
match_recognize.rs
