risingwave_stream/executor/match_recognize/nfa.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
15//! Row-pattern NFA for `MATCH_RECOGNIZE`.
16//!
17//! A `Pattern` (the supported v1 subset of the SQL `PATTERN` clause) is compiled to a
18//! Thompson-construction NFA whose labelled transitions are pattern variables. The simulation
19//! consumes a sequence of rows, where each row is summarised by the set of pattern variables whose
20//! `DEFINE` predicate it satisfies, and finds the greedy longest match from a start position
21//! (`ONE ROW PER MATCH` + `AFTER MATCH SKIP PAST LAST ROW`).
22//!
23//! Variable→predicate evaluation and the streaming/state layer live elsewhere; this module is pure
24//! and deterministic so it can be unit-tested without a cluster.
25
26// `BTreeSet` is used only by the test-only reference matchers and the unit tests; the streaming
27// matcher walks transitions directly with [`Visited`] guards (a `u64` bitmask for automata of at
28// most 64 states, a `HashSet` fallback beyond that).
29#[cfg(test)]
30use std::collections::BTreeSet;
31use std::collections::HashSet;
32
33use crate::executor::error::StreamExecutorResult;
34
35/// Cursor for [`Nfa::next_match`]: the next start position the scan will try. A fresh scan starts
36/// at 0; [`Nfa::next_match`] advances it by the `AFTER MATCH SKIP` mode on every match returned
37/// (or by one row past a non-matching start), so pulling repeatedly enumerates exactly the match
38/// sequence [`Nfa::find_matches_dynamic`] would collect.
39#[derive(Debug, Default)]
40pub struct MatchScan {
41 next_start: usize,
42 /// End of the contiguous run of starts, from where this scan began, proven MATCHLESS FOREVER:
43 /// their walks found no accept and never reached the boundary, so every path from them died
44 /// on a row below it — and those rows are immutable, so no arrival can revive them. The
45 /// incremental matcher reads this back and begins its next rescan past them.
46 matchless_upto: usize,
47}
48
49impl MatchScan {
50 pub fn new() -> Self {
51 Self::default()
52 }
53
54 /// A scan beginning at `start`, with nothing proven yet.
55 pub fn starting_at(start: usize) -> Self {
56 Self {
57 next_start: start,
58 matchless_upto: start,
59 }
60 }
61
62 /// See the `matchless_upto` field.
63 pub fn matchless_upto(&self) -> usize {
64 self.matchless_upto
65 }
66
67 /// The next start the pull loop would explore.
68 ///
69 /// This is NOT a "fully scanned prefix" marker, and must not be used as one: on a hit the cursor
70 /// jumps to `skip.next_pos`, so under `PAST LAST ROW` every start strictly inside the match it
71 /// just returned was never evaluated. What it does guarantee, since the budget fix below, is the
72 /// narrower property that a start whose walk the budget ABORTED is not advanced past — so the
73 /// cursor never claims a verdict the walk did not reach.
74 ///
75 /// No production caller reads this today; it exists for the emit-on-update port, which needs to
76 /// resume a pull, and for the test that pins the abort behaviour.
77 pub fn next_start(&self) -> usize {
78 self.next_start
79 }
80}
81
82/// Per-visit budget on NFA walk steps — predicate evaluations AND edges taken (row consumptions
83/// and ε-transitions) — shared by every walk of one partition visit (matching, eviction liveness,
84/// extension probing). ε-edges are charged deliberately: metering only predicate evaluations left
85/// ε-traversal free, so a large NFA could spend arbitrary CPU per metered evaluation and the budget
86/// was not actually a CPU bound.
87///
88/// The matcher is a backtracking DFS whose worst case is exponential in the pattern for
89/// pathological shapes (`(a? a? … a? b)` over a run of `a`-rows — the classic catastrophic
90/// regex-backtracking family), and [`MAX_PATTERN_NFA_STATES`-style caps bound *space*, not time.
91/// [`Memo`] removes the blowup entirely for path-independent patterns; for the rest, this budget
92/// is the hard backstop: when it runs out, the walk STOPS — it never fakes a verdict (a fabricated
93/// `false` in the liveness walker would evict rows of a live match, the exact bug class the
94/// decision-wait semantics exist to prevent). The caller must treat everything undecided
95/// conservatively (emit nothing more, evict nothing more, report once, retry next watermark), so a
96/// pathological pattern degrades to bounded CPU per visit and an observable report instead of
97/// pinning a compute node.
98#[derive(Debug)]
99pub struct ScanBudget {
100 remaining: usize,
101 /// Set once the budget runs out; sticky for the rest of the visit.
102 pub hit: bool,
103}
104
105impl ScanBudget {
106 pub fn new(evaluations: usize) -> Self {
107 Self {
108 remaining: evaluations,
109 hit: false,
110 }
111 }
112
113 /// Account one edge taken by a walk (an ε-transition or a row consumption). Returns `false` —
114 /// and latches `hit` — once the budget is spent.
115 #[must_use]
116 pub fn step(&mut self) -> bool {
117 self.charge()
118 }
119
120 /// No practical limit — for callers (tests, the collect wrapper) that need the historical
121 /// unbounded behaviour.
122 pub fn unlimited() -> Self {
123 Self::new(usize::MAX)
124 }
125
126 /// Account one predicate evaluation. Returns `false` — and latches `hit` — once exhausted.
127 fn charge(&mut self) -> bool {
128 if self.remaining == 0 {
129 self.hit = true;
130 return false;
131 }
132 self.remaining -= 1;
133 true
134 }
135}
136
137/// Per-start `(state, position)` failure memo for the backtracking walkers.
138///
139/// Soundness: an entry is recorded ONLY for recursion entered through a *consuming* transition —
140/// that recursion always starts with a fresh visited-set, so its outcome is context-free — and
141/// only when the pattern's verdicts are path-independent (no `DEFINE` slot reads the running label
142/// assignment; see the executor's `memoizable` flag). Within one start, a verdict then depends
143/// only on `(var, pos)` (`labels.len()` is `pos - start`, so `WITHIN` and the match-start offset
144/// are position-determined), so a failed `(state, pos)` fails identically on every re-entry. That
145/// re-entry via different consumption prefixes is exactly the exponential blowup; memoizing it
146/// makes a start's scan polynomial. ε-level failures are NOT recorded: they can be artifacts of
147/// the cycle-cutting visited-set and are not context-free.
148struct Memo {
149 /// Failure sets indexed by position offset from the memo's start; one [`Visited`] per position
150 /// (bitmask for small automata, set fallback beyond).
151 failed: Vec<Visited>,
152 n_states: usize,
153 base: usize,
154}
155
156impl Memo {
157 fn new(base: usize, _n_rows: usize, n_states: usize) -> Self {
158 Self {
159 // `slot()` grows on demand; an eager reservation over the whole suffix would malloc
160 // O(suffix) per walk INSTANCE — quadratic traffic per rescan — for walks that mostly
161 // die within a few positions.
162 failed: Vec::new(),
163 n_states,
164 base,
165 }
166 }
167
168 fn slot(&mut self, pos: usize) -> &mut Visited {
169 let idx = pos - self.base;
170 while self.failed.len() <= idx {
171 self.failed.push(Visited::new(self.n_states));
172 }
173 &mut self.failed[idx]
174 }
175
176 fn is_failed(&mut self, state: StateId, pos: usize) -> bool {
177 self.slot(pos).contains(state)
178 }
179
180 fn record_failure(&mut self, state: StateId, pos: usize) {
181 self.slot(pos).insert(state);
182 }
183}
184
185/// Decides whether the row at a physical position can be bound to a pattern variable, given the
186/// variables already bound to the earlier rows of the in-progress match. This is how `DEFINE`
187/// predicates are evaluated during matching: a predicate may reference the current row, its physical
188/// neighbours (`PREV`/`NEXT`), and the running values of other pattern variables (e.g. `A.price`),
189/// so membership cannot be precomputed independently of the match path.
190pub trait CandidateMatcher {
191 /// `labels[k]` is the variable bound to the match's `k`-th row; the candidate is the row at
192 /// `pos = match_start + labels.len()`. The returned future is `Send` so the matcher composes
193 /// with the (boxed, `Send`) executor stream.
194 ///
195 /// **Contract for callers:** membership MUST be queried *before* `var` is appended to `labels`,
196 /// so `labels` covers only the already-bound rows and never the candidate. Consequently a matcher
197 /// that resolves running navigation over `var` itself must treat `var` as the implicit trailing
198 /// label. Every caller MUST follow the same rule: [the finder] and [the eviction walker] share one
199 /// matcher, so a caller that pushed first would make the two disagree about which rows satisfy a
200 /// variable — and eviction would then delete rows the matcher still needs.
201 ///
202 /// [the finder]: Nfa::find_matches_dynamic
203 /// [the eviction walker]: Nfa::reaches_boundary_alive
204 fn matches(
205 &self,
206 var: &str,
207 pos: usize,
208 labels: &[String],
209 ) -> impl std::future::Future<Output = StreamExecutorResult<bool>> + Send;
210}
211
212/// A quantifier applied to a sub-pattern. Greedy semantics only (v1).
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub enum Quantifier {
215 /// `*`
216 Star,
217 /// `+`
218 Plus,
219 /// `?`
220 Question,
221 /// `{n}`, `{n,}`, `{n,m}`, `{,m}`. `min` defaults to 0, `max` is `None` for unbounded.
222 Range { min: u32, max: Option<u32> },
223}
224
225/// The supported v1 subset of a row pattern.
226#[derive(Debug, Clone, PartialEq, Eq)]
227pub enum Pattern {
228 /// A pattern variable, e.g. `A`.
229 Var(String),
230 /// Concatenation, e.g. `A B C`.
231 Concat(Vec<Pattern>),
232 /// Alternation, e.g. `A | B`.
233 Alt(Vec<Pattern>),
234 /// A quantified sub-pattern, e.g. `A+`. The bool is `reluctant` (`A+?` prefers fewer matches).
235 Quantified(Box<Pattern>, Quantifier, bool),
236 /// `PERMUTE(a, b, ...)` — expanded to the alternation of all orderings.
237 Permute(Vec<String>),
238}
239
240type StateId = usize;
241
242#[derive(Debug, Clone)]
243enum Transition {
244 /// An ε-transition (consumes no row).
245 Epsilon(StateId),
246 /// Consume a row that satisfies pattern variable `var`, moving to `target`.
247 OnVar { var: String, target: StateId },
248}
249
250/// A Thompson-construction NFA with a single start and single accept state.
251#[derive(Debug, Clone)]
252pub struct Nfa {
253 states: Vec<Vec<Transition>>,
254 start: StateId,
255 accept: StateId,
256 /// Per state: whether `accept` is reachable from it, with every predicate assumed satisfiable.
257 /// This is the static half of [`Nfa::may_extend`]: a consuming transition at the row boundary
258 /// whose target cannot reach `accept` can never contribute a longer match, no matter what rows
259 /// arrive. Computed once at compile with one reverse BFS.
260 reach_accept: Vec<bool>,
261 /// Fewest rows any accepting path consumes, with every predicate assumed satisfiable — the
262 /// shortest match the pattern admits (`0` when it accepts the empty match). A start with fewer
263 /// rows than this before the boundary cannot complete, so the finder skips it without a walk
264 /// (see [`Nfa::next_match`]). Computed once at compile with one 0-1 BFS.
265 min_match_rows: usize,
266 /// Most rows any path from `start` consumes, when the automaton is acyclic (no `*`, `+` or
267 /// unbounded range) — `None` when it has a cycle and a path can consume without bound. From a
268 /// position with MORE rows than this before the boundary, no path can reach the boundary while
269 /// still inside the automaton: the position is dead without a walk (see
270 /// [`Nfa::reaches_boundary_alive`]). Computed once at compile with one DFS.
271 max_match_rows: Option<usize>,
272}
273
274impl Nfa {
275 /// Compile a [`Pattern`] into an NFA.
276 pub fn compile(pattern: &Pattern) -> Self {
277 let mut builder = NfaBuilder { states: Vec::new() };
278 let frag = builder.build(pattern);
279 let reach_accept = Self::compute_reach_accept(&builder.states, frag.accept);
280 let max_match_rows = Self::compute_max_match_rows(&builder.states, frag.start);
281 let min_match_rows = Self::compute_min_match_rows(&builder.states, frag.start, frag.accept);
282 Nfa {
283 states: builder.states,
284 start: frag.start,
285 accept: frag.accept,
286 reach_accept,
287 min_match_rows,
288 max_match_rows,
289 }
290 }
291
292 /// See the `max_match_rows` field.
293 pub fn max_match_rows(&self) -> Option<usize> {
294 self.max_match_rows
295 }
296
297 /// See the `max_match_rows` field: the longest path from `start` counting consuming edges,
298 /// over ANY path (not only accepting ones — a path that dies still consumed its rows), or
299 /// `None` if a cycle is reachable. One iterative DFS with a tri-state mark, so a 100k-state
300 /// automaton does not recurse.
301 fn compute_max_match_rows(states: &[Vec<Transition>], start: StateId) -> Option<usize> {
302 #[derive(Clone, Copy, PartialEq, Eq)]
303 enum Mark {
304 New,
305 Active,
306 Done,
307 }
308 let mut mark = vec![Mark::New; states.len()];
309 // Longest consumption from each finished state.
310 let mut longest = vec![0usize; states.len()];
311 let mut stack: Vec<(StateId, usize)> = vec![(start, 0)];
312 mark[start] = Mark::Active;
313 while let Some(&(s, edge)) = stack.last() {
314 if let Some(t) = states[s].get(edge) {
315 stack.last_mut().expect("just peeked").1 += 1;
316 let next = match t {
317 Transition::Epsilon(next) => *next,
318 Transition::OnVar { target, .. } => *target,
319 };
320 match mark[next] {
321 // A state still on the DFS path is reachable from itself: a cycle.
322 Mark::Active => return None,
323 Mark::New => {
324 mark[next] = Mark::Active;
325 stack.push((next, 0));
326 }
327 Mark::Done => {}
328 }
329 } else {
330 stack.pop();
331 mark[s] = Mark::Done;
332 longest[s] = states[s]
333 .iter()
334 .map(|t| match t {
335 Transition::Epsilon(next) => longest[*next],
336 Transition::OnVar { target, .. } => longest[*target] + 1,
337 })
338 .max()
339 .unwrap_or(0);
340 }
341 }
342 Some(longest[start])
343 }
344
345 /// See the `min_match_rows` field.
346 pub fn min_match_rows(&self) -> usize {
347 self.min_match_rows
348 }
349
350 /// See the `min_match_rows` field: shortest path from `start` to `accept` where ε-edges cost
351 /// nothing and consuming edges cost one row (0-1 BFS, linear in states + transitions).
352 fn compute_min_match_rows(
353 states: &[Vec<Transition>],
354 start: StateId,
355 accept: StateId,
356 ) -> usize {
357 let mut dist = vec![usize::MAX; states.len()];
358 let mut queue = std::collections::VecDeque::from([start]);
359 dist[start] = 0;
360 while let Some(s) = queue.pop_front() {
361 let d = dist[s];
362 for t in &states[s] {
363 let (next, cost) = match t {
364 Transition::Epsilon(next) => (*next, 0),
365 Transition::OnVar { target, .. } => (*target, 1),
366 };
367 if d + cost < dist[next] {
368 dist[next] = d + cost;
369 if cost == 0 {
370 queue.push_front(next);
371 } else {
372 queue.push_back(next);
373 }
374 }
375 }
376 }
377 debug_assert_ne!(
378 dist[accept],
379 usize::MAX,
380 "accept must be reachable by construction"
381 );
382 dist[accept]
383 }
384
385 /// See the `reach_accept` field. Linear in states + transitions (one reverse BFS).
386 fn compute_reach_accept(states: &[Vec<Transition>], accept: StateId) -> Vec<bool> {
387 let n = states.len();
388 let mut rev: Vec<Vec<StateId>> = vec![Vec::new(); n];
389 for (s, ts) in states.iter().enumerate() {
390 for t in ts {
391 match t {
392 Transition::Epsilon(next) => rev[*next].push(s),
393 Transition::OnVar { target, .. } => rev[*target].push(s),
394 }
395 }
396 }
397 let mut reach = vec![false; n];
398 let mut stack = vec![accept];
399 reach[accept] = true;
400 while let Some(s) = stack.pop() {
401 for &p in &rev[s] {
402 if !reach[p] {
403 reach[p] = true;
404 stack.push(p);
405 }
406 }
407 }
408 reach
409 }
410
411 /// The set of states reachable from `states` via ε-transitions (inclusive). Only the test-only
412 /// reference matchers (e.g. [`Nfa::longest_match`]) use the explicit closure; the streaming
413 /// matcher walks transitions directly, so this is gated out of the release binary.
414 #[cfg(test)]
415 fn epsilon_closure(&self, states: impl IntoIterator<Item = StateId>) -> BTreeSet<StateId> {
416 let mut closure: BTreeSet<StateId> = BTreeSet::new();
417 let mut stack: Vec<StateId> = states.into_iter().collect();
418 while let Some(s) = stack.pop() {
419 if !closure.insert(s) {
420 continue;
421 }
422 for t in &self.states[s] {
423 if let Transition::Epsilon(next) = t {
424 stack.push(*next);
425 }
426 }
427 }
428 closure
429 }
430
431 /// Greedy longest match starting at `rows[start]`. `rows[i]` is the set of pattern variables
432 /// whose `DEFINE` predicate row `i` satisfies. Returns the exclusive end index of the longest
433 /// match (so `start..end` are the matched rows), or `None` if no match starts at `start`.
434 ///
435 /// An empty match (the pattern accepts zero rows, e.g. `A*`) returns `Some(start)`.
436 ///
437 /// Test-only: the streaming executor matches via [`Nfa::find_matches_dynamic`]. This precomputed
438 /// satisfied-set variant is kept as the simple reference the dynamic matcher is checked against,
439 /// and to unit-test NFA construction directly. Gated out of the release binary.
440 #[cfg(test)]
441 pub fn longest_match(&self, rows: &[BTreeSet<String>], start: usize) -> Option<usize> {
442 let mut current = self.epsilon_closure([self.start]);
443 let mut longest = current.contains(&self.accept).then_some(start);
444
445 let mut pos = start;
446 while pos < rows.len() && !current.is_empty() {
447 let row = &rows[pos];
448 let mut next: BTreeSet<StateId> = BTreeSet::new();
449 for &s in ¤t {
450 for t in &self.states[s] {
451 if let Transition::OnVar { var, target } = t
452 && row.contains(var)
453 {
454 next.insert(*target);
455 }
456 }
457 }
458 if next.is_empty() {
459 break;
460 }
461 current = self.epsilon_closure(next);
462 pos += 1;
463 if current.contains(&self.accept) {
464 longest = Some(pos);
465 }
466 }
467 longest
468 }
469}
470
471/// A single match span over the row sequence: `start..end` (end exclusive) are the matched rows.
472/// Test-only: produced by the reference matcher [`Nfa::find_matches`].
473#[cfg(test)]
474#[derive(Debug, Clone, Copy, PartialEq, Eq)]
475pub struct MatchSpan {
476 pub start: usize,
477 pub end: usize,
478}
479
480/// Where the scan resumes after a match (the `AFTER MATCH SKIP` strategy).
481#[derive(Debug, Clone, PartialEq, Eq)]
482pub enum SkipMode {
483 /// `AFTER MATCH SKIP PAST LAST ROW`: resume past the match's last row (non-overlapping).
484 PastLastRow,
485 /// `AFTER MATCH SKIP TO NEXT ROW`: resume at the row after the match's first row (overlapping).
486 ToNextRow,
487 /// `AFTER MATCH SKIP TO FIRST <var>`: resume at the first row labeled `var`.
488 ToFirst(String),
489 /// `AFTER MATCH SKIP TO LAST <var>`: resume at the last row labeled `var`.
490 ToLast(String),
491}
492
493/// Why a variable-targeted `AFTER MATCH SKIP` could not resume where the query asked, and which
494/// weaker strategy the resume position fell back to. Returned by [`SkipMode::next_pos`] next to the
495/// position instead of being reported here: this module stays pure (no error reporter, no executor
496/// types), and the executor — which owns the actor's `EvalErrorReport` — decides what to do with it.
497#[derive(Debug, Clone, Copy, PartialEq, Eq)]
498pub enum SkipDegradation {
499 /// The target variable is bound to no row of the match, so there is no row to resume at. The
500 /// scan resumed past the match's last row, i.e. as `SKIP PAST LAST ROW`.
501 TargetAbsent,
502 /// The target resolves to the match's own first row, so resuming there would re-find the same
503 /// match forever. The scan resumed one row later, i.e. as `SKIP TO NEXT ROW`.
504 TargetAtMatchStart,
505}
506
507impl SkipDegradation {
508 /// The user-facing diagnostic for this degradation under `skip`: the target variable, what could
509 /// not be resolved about it, and the strategy actually applied. The *clause* is not repeated here
510 /// — the caller supplies it separately via [`SkipMode::clause_name`], so the rendered message
511 /// names the mode once (see `report_skip_degradation_once` in `executor.rs`).
512 ///
513 /// Deliberately carries no row, match or partition identity: the cause is a property of the
514 /// query, not of one row, which is what lets the executor report it once per watermark pass
515 /// instead of once per match.
516 pub fn describe(&self, skip: &SkipMode) -> String {
517 // Unreachable for the variable-less modes: only the targeted arms of `next_pos` can produce a
518 // degradation. A placeholder rather than an `unwrap` — a diagnostic path must not panic.
519 let target = skip.target_var().unwrap_or("?");
520 match self {
521 SkipDegradation::TargetAbsent => format!(
522 "target variable `{target}` is bound to no row of the match, so there is no row to \
523 resume at; the scan resumed past the match's last row instead (degraded to SKIP \
524 PAST LAST ROW)"
525 ),
526 SkipDegradation::TargetAtMatchStart => format!(
527 "target variable `{target}` resolves to the match's own first row, so resuming there \
528 would re-find the same match forever; the scan resumed at the row after the match's \
529 first row instead (degraded to SKIP TO NEXT ROW)"
530 ),
531 }
532 }
533}
534
535impl SkipMode {
536 /// The `AFTER MATCH SKIP` clause as SQL spells it, *without* the target variable — the mode alone.
537 /// `&'static str` so it can be the `name` of an `ExprError::InvalidParam`; the target variable is
538 /// named by [`SkipDegradation::describe`] instead, so a rendered diagnostic states the mode once.
539 pub fn clause_name(&self) -> &'static str {
540 match self {
541 SkipMode::PastLastRow => "AFTER MATCH SKIP PAST LAST ROW",
542 SkipMode::ToNextRow => "AFTER MATCH SKIP TO NEXT ROW",
543 SkipMode::ToFirst(_) => "AFTER MATCH SKIP TO FIRST",
544 SkipMode::ToLast(_) => "AFTER MATCH SKIP TO LAST",
545 }
546 }
547
548 /// The pattern variable a `SKIP TO FIRST|LAST` resumes at; `None` for the variable-less modes,
549 /// which are also the only ones that can never degrade.
550 pub fn target_var(&self) -> Option<&str> {
551 match self {
552 SkipMode::PastLastRow | SkipMode::ToNextRow => None,
553 SkipMode::ToFirst(var) | SkipMode::ToLast(var) => Some(var),
554 }
555 }
556
557 /// The position the scan resumes at after a match spanning `[start, end)` with per-row `labels`
558 /// (`labels[i]` is the variable bound to `rows[start + i]`), plus a [`SkipDegradation`] when that
559 /// position is not the one the query asked for. Always returns `> start` so the scan makes
560 /// progress.
561 ///
562 /// Two cases have no valid resume row, and both are data-dependent — the same query degrades or
563 /// not depending on which rows arrive:
564 ///
565 /// * the target variable is bound to no row of this match (`(a b?)` matching only `a`, with
566 /// `SKIP TO LAST b`), so the resume position falls back to the match end — silently becoming
567 /// `SKIP PAST LAST ROW`;
568 /// * the target resolves to the match's own first row (`SKIP TO FIRST` of the pattern's leading
569 /// variable), which would re-find the same match forever, so it is clamped to `start + 1` —
570 /// silently becoming `SKIP TO NEXT ROW`.
571 ///
572 /// The SQL standard prescribes a runtime error for both (Oracle raises ORA-62511 / ORA-62512;
573 /// Flink likewise). This implementation deliberately keeps the degradation and **reports** it
574 /// instead of raising it: an error here would abort the actor over a data-dependent condition,
575 /// and since the materialized view is already committed, every recovery attempt would replay the
576 /// same rows and die again — a recoverable query turned into a crash loop. No RisingWave
577 /// streaming operator fails an actor for a data-dependent condition; every hard error in this
578 /// operator is a contract or plan violation (non-append-only input, an unknown slot kind), which
579 /// recovery cannot fix either way. So the degradation is made *visible* rather than fatal: the
580 /// executor routes the returned diagnostic to the actor's `EvalErrorReport`, the same surface
581 /// expression evaluation errors already use (the `stream_expr_error` log and the
582 /// `user_compute_error` metric). See `report_skip_degradation_once` in `executor.rs` for how the
583 /// message is rendered — including the surface's fixed log prefix — and for the volume policy.
584 pub fn next_pos(
585 &self,
586 start: usize,
587 end: usize,
588 labels: &[String],
589 ) -> (usize, Option<SkipDegradation>) {
590 // Resolve a variable-targeted skip: `found` is the target's index within `labels`. Only these
591 // modes can degrade; the variable-less ones always have a valid resume row.
592 let resolve = |found: Option<usize>| match found {
593 // Index 0 is the match's own first row, so resuming there makes no progress.
594 Some(0) => (start + 1, Some(SkipDegradation::TargetAtMatchStart)),
595 Some(j) => (start + j, None),
596 None => (end.max(start + 1), Some(SkipDegradation::TargetAbsent)),
597 };
598 match self {
599 SkipMode::PastLastRow => (end.max(start + 1), None),
600 SkipMode::ToNextRow => (start + 1, None),
601 SkipMode::ToFirst(var) => resolve(labels.iter().position(|l| l == var)),
602 SkipMode::ToLast(var) => resolve(labels.iter().rposition(|l| l == var)),
603 }
604 }
605}
606
607impl Nfa {
608 /// Find all matches over `rows` under `ONE ROW PER MATCH` with the given `AFTER MATCH SKIP`
609 /// strategy: scan left to right; at each position take the greedy longest match; on a non-empty
610 /// match, record it and resume per `skip`; otherwise advance by one row.
611 ///
612 /// Empty matches (a pattern that accepts zero rows, e.g. `A*` on a non-matching row) are not
613 /// emitted and advance the scan by one, so the scan always terminates.
614 ///
615 /// Test-only reference matcher (see [`Nfa::longest_match`]); gated out of the release binary.
616 #[cfg(test)]
617 pub fn find_matches(&self, rows: &[BTreeSet<String>], skip: &SkipMode) -> Vec<MatchSpan> {
618 let mut matches = Vec::new();
619 let mut i = 0;
620 while i < rows.len() {
621 if let Some(end) = self.longest_match(rows, i)
622 && end > i
623 {
624 matches.push(MatchSpan { start: i, end });
625 // `find_matches` is label-less; the variable-targeted skips resolve like
626 // `PAST LAST ROW` here. `find_matches_labeled` applies them precisely.
627 i = match skip {
628 SkipMode::ToNextRow => i + 1,
629 _ => end,
630 };
631 } else {
632 i += 1;
633 }
634 }
635 matches
636 }
637}
638
639/// A match span together with the pattern variable assigned to each matched row.
640/// `labels[i]` is the variable that `rows[start + i]` was matched as.
641#[derive(Debug, Clone, PartialEq, Eq)]
642pub struct LabeledMatch {
643 pub start: usize,
644 pub end: usize,
645 pub labels: Vec<String>,
646}
647
648/// Visited-state guard for one traversal position of the dynamic matcher. ε-transitions keep the
649/// position, so each consumed row starts a fresh set (see [`Nfa::walk`]); these sets are opened
650/// O(rows × branches) times per partition visit, so their allocation cost matters. The
651/// common case — an automaton with at most 64 states — is a single `u64` bitmask (no allocation,
652/// membership is a bit test); larger automata (deep `PERMUTE` expansions) fall back to a `HashSet`.
653enum Visited {
654 Small(u64),
655 Large(HashSet<StateId>),
656}
657
658impl Visited {
659 fn new(n_states: usize) -> Self {
660 if n_states <= 64 {
661 Visited::Small(0)
662 } else {
663 Visited::Large(HashSet::new())
664 }
665 }
666
667 /// Marks `s` visited; returns whether it was newly inserted (mirrors `HashSet::insert`).
668 fn insert(&mut self, s: StateId) -> bool {
669 match self {
670 Visited::Small(bits) => {
671 let mask = 1u64 << s;
672 let newly = *bits & mask == 0;
673 *bits |= mask;
674 newly
675 }
676 Visited::Large(set) => set.insert(s),
677 }
678 }
679
680 fn remove(&mut self, s: StateId) {
681 match self {
682 Visited::Small(bits) => *bits &= !(1u64 << s),
683 Visited::Large(set) => {
684 set.remove(&s);
685 }
686 }
687 }
688
689 fn contains(&self, s: StateId) -> bool {
690 match self {
691 Visited::Small(bits) => *bits & (1u64 << s) != 0,
692 Visited::Large(set) => set.contains(&s),
693 }
694 }
695
696 /// Empties the set for reuse as a fresh scope (keeps the `HashSet` fallback's allocation).
697 fn clear(&mut self) {
698 match self {
699 Visited::Small(bits) => *bits = 0,
700 Visited::Large(set) => set.clear(),
701 }
702 }
703}
704
705impl Nfa {
706 /// Greedy longest match starting at `rows[start]`, returning the per-row variable assignment
707 /// along the chosen accepting path (the variable each consumed row was matched as). This is
708 /// what `MEASURES` navigation (`FIRST`/`LAST`), `CLASSIFIER()`, and aggregates over matched
709 /// rows consume. Returns `(end, labels)` where `labels.len() == end - start`, or `None`.
710 #[cfg(test)]
711 pub fn longest_match_labeled(
712 &self,
713 rows: &[BTreeSet<String>],
714 start: usize,
715 ) -> Option<(usize, Vec<String>)> {
716 let mut visited: HashSet<(StateId, usize)> = HashSet::new();
717 self.longest_from(rows, self.start, start, &mut visited)
718 }
719
720 /// Recursive longest-accepting-path search. `visited` guards against ε-cycles on the current
721 /// path (it tracks `(state, pos)` and is unwound on backtrack). Among continuations the one
722 /// reaching the furthest `end` wins; ties keep the first in transition order, making the label
723 /// assignment deterministic.
724 #[cfg(test)]
725 fn longest_from(
726 &self,
727 rows: &[BTreeSet<String>],
728 state: StateId,
729 pos: usize,
730 visited: &mut HashSet<(StateId, usize)>,
731 ) -> Option<(usize, Vec<String>)> {
732 if !visited.insert((state, pos)) {
733 return None;
734 }
735 let mut best: Option<(usize, Vec<String>)> =
736 (state == self.accept).then(|| (pos, Vec::new()));
737 for t in &self.states[state] {
738 let candidate = match t {
739 Transition::Epsilon(next) => self.longest_from(rows, *next, pos, visited),
740 Transition::OnVar { var, target } => {
741 if pos < rows.len() && rows[pos].contains(var) {
742 self.longest_from(rows, *target, pos + 1, visited).map(
743 |(end, mut labels)| {
744 labels.insert(0, var.clone());
745 (end, labels)
746 },
747 )
748 } else {
749 None
750 }
751 }
752 };
753 if let Some((end, labels)) = candidate
754 && best.as_ref().is_none_or(|(b, _)| end > *b)
755 {
756 best = Some((end, labels));
757 }
758 }
759 visited.remove(&(state, pos));
760 best
761 }
762
763 /// Like [`Nfa::find_matches`] but returns each match with its per-row variable labels.
764 /// Test-only reference matcher; the streaming executor uses [`Nfa::find_matches_dynamic`].
765 #[cfg(test)]
766 pub fn find_matches_labeled(
767 &self,
768 rows: &[BTreeSet<String>],
769 skip: &SkipMode,
770 ) -> Vec<LabeledMatch> {
771 let mut matches = Vec::new();
772 let mut i = 0;
773 while i < rows.len() {
774 if let Some((end, labels)) = self.longest_match_labeled(rows, i)
775 && end > i
776 {
777 let start = i;
778 // Test-only matcher: the diagnostic is dropped (there is no actor to report to).
779 (i, _) = skip.next_pos(start, end, &labels);
780 matches.push(LabeledMatch { start, end, labels });
781 } else {
782 i += 1;
783 }
784 }
785 matches
786 }
787
788 /// Like `find_matches_labeled`, but membership is decided by an async [`CandidateMatcher`]
789 /// instead of precomputed satisfied-sets, so `DEFINE` predicates with row-pattern navigation can
790 /// be evaluated against the running match. `n_rows` is the number of (sorted) rows to scan. This
791 /// is the only matcher the streaming executor uses.
792 pub async fn find_matches_dynamic(
793 &self,
794 n_rows: usize,
795 matcher: &(impl CandidateMatcher + Sync),
796 skip: &SkipMode,
797 ) -> StreamExecutorResult<Vec<LabeledMatch>> {
798 let mut matches = Vec::new();
799 let mut scan = MatchScan::new();
800 let mut budget = ScanBudget::unlimited();
801 while let Some(m) = self
802 .next_match(&mut scan, n_rows, matcher, skip, &mut budget, false)
803 .await?
804 {
805 matches.push(m);
806 }
807 Ok(matches)
808 }
809
810 /// Pull the next match at or after `scan`'s cursor, advancing the cursor by the skip mode.
811 /// Returns `None` once the cursor passes `n_rows`.
812 ///
813 /// This is the streaming form of [`Nfa::find_matches_dynamic`], which is a thin collect over
814 /// it. The executor's emit loop pulls instead of collecting so that stopping — at the first
815 /// boundary match that must be held for maximality — stops the *scan*, not just the emission:
816 /// nothing past the held match is computed (it would be recomputed from scratch on the next
817 /// watermark anyway) and at most one match is resident at a time. Collecting is worst-case
818 /// quadratic in live rows: under an overlapping skip mode a greedy `(a+)` over `n` qualifying
819 /// rows yields `n` matches whose label vectors sum to `O(n^2)` strings, all materialized before
820 /// the first one is examined.
821 pub async fn next_match(
822 &self,
823 scan: &mut MatchScan,
824 n_rows: usize,
825 matcher: &(impl CandidateMatcher + Sync),
826 skip: &SkipMode,
827 budget: &mut ScanBudget,
828 memoize: bool,
829 ) -> StreamExecutorResult<Option<LabeledMatch>> {
830 while scan.next_start < n_rows && !budget.hit {
831 let i = scan.next_start;
832 // Fewer rows than the shortest match before the boundary: no accepting path exists
833 // from `i` within `n_rows`, which is exactly the verdict a walk would reach after
834 // consuming its way to the boundary — so skip the walk and take the verdict. This is
835 // what keeps a rescan over a long pending run of a chain pattern (`a{600}`) from
836 // walking every start to the boundary, Θ(k²) per rescan. (Not a matchless start: the
837 // walk it stands in for is blocked at the boundary, and more rows may complete it.)
838 if n_rows - i < self.min_match_rows() {
839 scan.next_start += 1;
840 continue;
841 }
842 // The memo is per START: within one start a verdict depends only on `(var, pos)`
843 // (given path-independence), so it must not leak across starts, where `labels.len()`
844 // differs for the same position.
845 let mut memo = memoize.then(|| Memo::new(i, n_rows, self.states.len()));
846 let mut reached_boundary = false;
847 let found = self
848 .walk(
849 Goal::Accept {
850 n_rows,
851 reached_boundary: &mut reached_boundary,
852 },
853 i,
854 matcher,
855 budget,
856 memo.as_mut(),
857 )
858 .await?;
859 let found_empty = match found {
860 Some((end, labels)) if end > i => {
861 // The diagnostic is dropped here on purpose: the executor recomputes the
862 // resume position for the matches it actually *emits* and reports from there,
863 // so a match that this scan finds but the emit path holds back or skips is not
864 // reported twice (nor reported at all until it is emitted).
865 (scan.next_start, _) = skip.next_pos(i, end, &labels);
866 return Ok(Some(LabeledMatch {
867 start: i,
868 end,
869 labels,
870 }));
871 }
872 Some(_) => true,
873 None => false,
874 };
875 // Only advance past a start with a real verdict. The walk returns `None` for two
876 // different reasons — "no match from here" and "the budget died mid-walk, no
877 // verdict" — and advancing on the second leaves the cursor claiming a verdict the walk
878 // never reached: `(a b) | a` over two `a` rows with a budget of 1 dies inside start 0
879 // before ever trying the second alternative, which matches there.
880 //
881 // Inert for this operator (the loop condition already stops on `budget.hit`, and nothing
882 // here reads the cursor afterwards); it matters to a caller that resumes a pull.
883 if budget.hit {
884 break;
885 }
886 // A start whose walk found no accept and never reached the boundary is matchless
887 // FOREVER: every path from it died on a row below the boundary, and those rows are
888 // immutable, so no arrival can revive it. Record the contiguous run of such starts (see
889 // `MatchScan::matchless_upto`) — the finder's cross-visit memory, the counterpart of
890 // the freeze's proven-dead prefix: a broken run of `r` rows costs its Θ(r²) once,
891 // amortised across visits, instead of on every rescan forever. An empty match proves
892 // nothing about the other paths (the walk stopped at its first verdict), so it does
893 // not count.
894 if !found_empty && !reached_boundary && scan.matchless_upto == i {
895 scan.matchless_upto = i + 1;
896 }
897 scan.next_start += 1;
898 }
899 Ok(None)
900 }
901
902 /// Whether a match starting at `pos` is still *live* at the safe boundary `n_rows`: there exists
903 /// a path that consumes the safe rows `pos..n_rows` and reaches the boundary while still inside
904 /// the automaton (not yet accepted), so a future row could extend it into a complete match. Used
905 /// to evict rows that can no longer be part of any match.
906 ///
907 /// This is strictly stronger than "can `pos` begin the pattern": for `(a b)` over `[a, x]` where
908 /// `x` matches neither, the `a` *can* begin the pattern, but every path dies on `x` before the
909 /// boundary, so the start is dead and must be evictable. A lone `[a]` (boundary right after `a`),
910 /// in contrast, is kept because a future `b` may still complete it.
911 pub async fn reaches_boundary_alive(
912 &self,
913 pos: usize,
914 n_rows: usize,
915 matcher: &(impl CandidateMatcher + Sync),
916 budget: &mut ScanBudget,
917 memoize: bool,
918 ) -> StreamExecutorResult<bool> {
919 // Acyclic automaton: no path consumes more than `max_match_rows` rows, so with more rows
920 // than that before the boundary no path can reach it — dead, without a walk. This is what
921 // makes freezing a long chain match (`a{600}`) cheap: every position of its region has the
922 // rest of the match, and more, ahead of it. (With EXACTLY that many rows the walk runs: a
923 // match ending on the boundary keeps its start alive.)
924 if let Some(max) = self.max_match_rows
925 && n_rows - pos > max
926 {
927 return Ok(false);
928 }
929 let mut memo = memoize.then(|| Memo::new(pos, n_rows, self.states.len()));
930 Ok(self
931 .walk(
932 Goal::Boundary { n_rows },
933 pos,
934 matcher,
935 budget,
936 memo.as_mut(),
937 )
938 .await?
939 .is_some())
940 }
941
942 /// Whether the automaton is a fixed-length linear chain: every state has at most one
943 /// outgoing transition (plain concatenations like `(a b c)` — no alternation, no
944 /// quantifiers, no PERMUTE). For such patterns an accepted match consumed the only path there
945 /// is, so no more-preferred extension can exist and [`Nfa::may_extend`] is statically `false`
946 /// — the emission gate can skip the probe entirely.
947 pub fn is_linear(&self) -> bool {
948 self.states.iter().all(|ts| ts.len() <= 1)
949 }
950
951 /// Whether the finder's *preferred* result for the match starting at `start` could change if
952 /// more rows arrived past the boundary `end`.
953 ///
954 /// The finder ([`Nfa::next_match`]) returns the first accepting path in transition
955 /// order: greedy quantifiers try their consume edge before their exit edge, reluctant ones the
956 /// reverse, and ordered alternation tries branches as listed. A lower-priority path can
957 /// therefore NEVER override an accepting higher-priority one, no matter what rows arrive — for
958 /// `PATTERN (A (B | B C))` the first-listed `B` alternative wins even if a `C` shows up later.
959 /// So the question is not "could any NFA path consume more" but "could a path the finder
960 /// prefers over the current result become accepting".
961 ///
962 /// This walk mirrors the finder's own traversal exactly and stops at its first accept — the
963 /// preferred result. It answers `true` iff, strictly before that accept in preference order,
964 /// some consuming transition was blocked by the row boundary while its target can still reach
965 /// `accept` ([`Nfa::reach_accept`], unknown rows assumed satisfiable): exactly the paths that
966 /// arriving rows could turn into a more-preferred accepting result. Everything explored after
967 /// the accept is lower-priority and irrelevant.
968 ///
969 /// `false` means the preferred result is **terminal**: it cannot change, so a boundary match is
970 /// final and must be emitted — holding it would starve an idle partition forever (the frontier
971 /// recompute finds neither a future row nor, without `WITHIN`, a deadline, and drops the
972 /// partition). `true` means the standard maximality wait applies.
973 pub async fn may_extend(
974 &self,
975 start: usize,
976 end: usize,
977 matcher: &(impl CandidateMatcher + Sync),
978 budget: &mut ScanBudget,
979 memoize: bool,
980 ) -> StreamExecutorResult<bool> {
981 let mut blocked = false;
982 let mut memo = memoize.then(|| Memo::new(start, end, self.states.len()));
983 let accepted = self
984 .walk(
985 Goal::AcceptOrBlocked {
986 end,
987 blocked: &mut blocked,
988 },
989 start,
990 matcher,
991 budget,
992 memo.as_mut(),
993 )
994 .await?
995 .is_some();
996 // A spent budget means the walk may have stopped before the preferred accept: `blocked`
997 // then under-approximates, and the only safe answer is "may extend" (hold — never emit on
998 // partial information).
999 if budget.hit {
1000 return Ok(true);
1001 }
1002 // Called for a match the finder produced over these same rows, so an accepting path within
1003 // `end` exists and `accepted` holds; if a caller ever probes a non-match, every explored
1004 // path is by definition higher-priority than the (absent) result, so `blocked` is still
1005 // the right answer.
1006 debug_assert!(accepted, "may_extend probed a non-accepting span");
1007 Ok(blocked)
1008 }
1009
1010 /// The one traversal behind the finder, the liveness check and the extension probe: a
1011 /// depth-first search from `self.start` at row `start_pos`, in transition order, with `DEFINE`
1012 /// predicates evaluated over the running match (`path`, the labels bound so far, is threaded to
1013 /// the matcher). Returns the row position at which `goal` was met together with the labels of
1014 /// the path that met it; `None` is "no such path" — or, with `budget.hit` latched, "undecided".
1015 ///
1016 /// Iterative, over an explicit heap stack. The recursive walkers this replaced — boxed
1017 /// `async_recursion` frames, polled through the real thread stack once per consumed row and
1018 /// once per ε-edge — needed a hard depth cap to avoid overflowing it, and the cap made any
1019 /// match spanning more than a couple of hundred rows permanently undecidable: unlike the
1020 /// budget it did not reset between visits, so every refresh died at the same depth. Every push
1021 /// is charged to the budget, so the stack holds at most `budget` frames — at the executor's
1022 /// 2^20 steps, about 32 megabytes of 32-byte frames in the worst case — and the budget is the
1023 /// only bound on a walk.
1024 ///
1025 /// The discipline, identical for all three goals:
1026 /// - transitions are tried in the order the builder emitted them, and the FIRST verdict wins —
1027 /// greedy quantifiers list their consume edge before their exit edge, reluctant ones the
1028 /// reverse, alternation as written;
1029 /// - a [`Visited`] scope per row position cuts ε-cycles: ε-edges keep the position and the
1030 /// scope, a consumed row opens a fresh one, so distinct label assignments may reach the same
1031 /// state at the next row;
1032 /// - the budget is charged once per predicate evaluation and once per edge taken, and a spent
1033 /// budget ends the walk without a verdict and without recording anything;
1034 /// - failures are memoized only at consumption boundaries — a consumed frame starts with a fresh
1035 /// scope, so its outcome is context-free — and never for a budget-aborted walk, which would
1036 /// turn a transient abort into a permanent wrong verdict (see [`Memo`]).
1037 async fn walk(
1038 &self,
1039 mut goal: Goal<'_>,
1040 start_pos: usize,
1041 matcher: &(impl CandidateMatcher + Sync),
1042 budget: &mut ScanBudget,
1043 mut memo: Option<&mut Memo>,
1044 ) -> StreamExecutorResult<Option<(usize, Vec<String>)>> {
1045 let n_states = self.states.len();
1046 let mut path: Vec<String> = Vec::new();
1047 // One visited scope per consumption level; `scopes[depth]` is the live one. A walk opens a
1048 // scope per consumed row, so they are cleared and reused rather than reallocated.
1049 let mut scopes: Vec<Visited> = vec![Visited::new(n_states)];
1050 let mut depth = 0usize;
1051 let mut stack = vec![Frame::new(self.start, start_pos, false)];
1052
1053 loop {
1054 let Some(top) = stack.last_mut() else {
1055 return Ok(None);
1056 };
1057 if !top.entered {
1058 top.entered = true;
1059 // A walk entered with an already-spent budget aborts without deciding anything
1060 // (the caller must treat everything undecided conservatively; see `ScanBudget`).
1061 // Exhaustion MID-walk never reaches this point: every `step`/`charge` that
1062 // latches `hit` returns from `walk` directly below — which is what lets
1063 // `pop_failed` record a failure unconditionally.
1064 if budget.hit {
1065 return Ok(None);
1066 }
1067 let (state, pos) = (top.state, top.pos);
1068 match goal.enter(self, state, pos) {
1069 Enter::Verdict => return Ok(Some((pos, path))),
1070 Enter::Dead => {
1071 Self::pop_failed(
1072 &mut stack,
1073 &mut scopes,
1074 &mut depth,
1075 &mut path,
1076 memo.as_deref_mut(),
1077 budget,
1078 );
1079 continue;
1080 }
1081 Enter::Explore => {}
1082 }
1083 if scopes[depth].insert(state) {
1084 top.inserted = true;
1085 } else {
1086 Self::pop_failed(
1087 &mut stack,
1088 &mut scopes,
1089 &mut depth,
1090 &mut path,
1091 memo.as_deref_mut(),
1092 budget,
1093 );
1094 continue;
1095 }
1096 }
1097 let (state, pos) = (top.state, top.pos);
1098 let Some(t) = self.states[state].get(top.next_edge) else {
1099 // Every transition tried and none met the goal: this frame fails.
1100 Self::pop_failed(
1101 &mut stack,
1102 &mut scopes,
1103 &mut depth,
1104 &mut path,
1105 memo.as_deref_mut(),
1106 budget,
1107 );
1108 continue;
1109 };
1110 top.next_edge += 1;
1111 match t {
1112 Transition::Epsilon(next) => {
1113 if !budget.step() {
1114 return Ok(None);
1115 }
1116 stack.push(Frame::new(*next, pos, false));
1117 }
1118 Transition::OnVar { var, target } => {
1119 if !goal.may_consume(self, *target, pos) {
1120 continue;
1121 }
1122 if !budget.charge() {
1123 return Ok(None);
1124 }
1125 if !matcher.matches(var, pos, &path).await? {
1126 continue;
1127 }
1128 // Consumption boundary: the frame pushed below starts with a fresh visited
1129 // scope, so its outcome is context-free — the failure memo is checked here and
1130 // recorded on its exit, never at ε-level, where a failure can be a cycle-cut
1131 // artifact.
1132 if memo
1133 .as_deref_mut()
1134 .is_some_and(|m| m.is_failed(*target, pos + 1))
1135 {
1136 continue;
1137 }
1138 if !budget.step() {
1139 return Ok(None);
1140 }
1141 path.push(var.clone());
1142 depth += 1;
1143 if depth == scopes.len() {
1144 scopes.push(Visited::new(n_states));
1145 } else {
1146 scopes[depth].clear();
1147 }
1148 stack.push(Frame::new(*target, pos + 1, true));
1149 }
1150 }
1151 }
1152 }
1153
1154 /// Pop the top frame as a failure: unmark its state in its scope, and if it was entered by
1155 /// consuming a row, close that scope, drop its label, and memoize the failure.
1156 ///
1157 /// A budget-aborted walk never gets here — every latch site returns from `walk` directly — so
1158 /// the failure being recorded is always a proven one, by construction rather than by a guard.
1159 fn pop_failed(
1160 stack: &mut Vec<Frame>,
1161 scopes: &mut [Visited],
1162 depth: &mut usize,
1163 path: &mut Vec<String>,
1164 memo: Option<&mut Memo>,
1165 budget: &ScanBudget,
1166 ) {
1167 debug_assert!(
1168 !budget.hit,
1169 "a spent budget returns from `walk`; it never pops a frame"
1170 );
1171 // Callers hold the top frame, so the stack is non-empty; an empty stack here would be a
1172 // walk-invariant violation, and letting the loop end on it (no verdict) beats panicking an
1173 // actor over it.
1174 let Some(frame) = stack.pop() else {
1175 return;
1176 };
1177 if frame.inserted {
1178 scopes[*depth].remove(frame.state);
1179 }
1180 if frame.consumed {
1181 debug_assert!(*depth > 0, "a consumed frame always opened a scope");
1182 *depth = depth.saturating_sub(1);
1183 path.pop();
1184 if let Some(m) = memo {
1185 m.record_failure(frame.state, frame.pos);
1186 }
1187 }
1188 }
1189}
1190
1191/// What a [`Nfa::walk`] is looking for. The three questions the executor asks of the automaton share
1192/// one traversal and differ only in when a frame is a verdict and in what a consuming edge may do at
1193/// the row boundary.
1194enum Goal<'a> {
1195 /// The first accepting path in preference order — the match finder ([`Nfa::next_match`]).
1196 /// Consuming edges stop at `n_rows`. `reached_boundary` records whether any frame was entered
1197 /// AT `n_rows`: a walk that finds no accept and never got there died entirely on rows below
1198 /// the boundary, which no arrival can change — the start is matchless forever.
1199 Accept {
1200 n_rows: usize,
1201 reached_boundary: &'a mut bool,
1202 },
1203 /// Whether some path consumes the safe suffix up to `n_rows` while still inside the automaton
1204 /// ([`Nfa::reaches_boundary_alive`]). Reaching `n_rows` is the verdict; reaching `accept` before
1205 /// it is a complete (already-finalized) match, not a live partial one, and fails that path.
1206 Boundary { n_rows: usize },
1207 /// The first accept, bounded by `end` ([`Nfa::may_extend`]). A consuming edge at `end` cannot
1208 /// be evaluated — the row does not exist yet; if its target can still reach `accept`
1209 /// ([`Nfa::reach_accept`]) it is recorded in `blocked`, since a future row could fire it and
1210 /// produce a more-preferred result than any accept found later in the walk.
1211 AcceptOrBlocked { end: usize, blocked: &'a mut bool },
1212}
1213
1214/// A frame's verdict on entry.
1215enum Enter {
1216 /// The goal is met at this frame.
1217 Verdict,
1218 /// This path can no longer meet the goal.
1219 Dead,
1220 /// Keep walking.
1221 Explore,
1222}
1223
1224impl Goal<'_> {
1225 fn enter(&mut self, nfa: &Nfa, state: StateId, pos: usize) -> Enter {
1226 if let Goal::Accept {
1227 n_rows,
1228 reached_boundary,
1229 } = self
1230 && pos == *n_rows
1231 {
1232 **reached_boundary = true;
1233 }
1234 match self {
1235 // The single accept state is terminal: reaching it completes the match here.
1236 Goal::Accept { .. } | Goal::AcceptOrBlocked { .. } if state == nfa.accept => {
1237 Enter::Verdict
1238 }
1239 Goal::Boundary { n_rows } if pos == *n_rows => Enter::Verdict,
1240 Goal::Boundary { .. } if state == nfa.accept => Enter::Dead,
1241 _ => Enter::Explore,
1242 }
1243 }
1244
1245 /// Whether a consuming edge into `target` may be evaluated at `pos`.
1246 fn may_consume(&mut self, nfa: &Nfa, target: StateId, pos: usize) -> bool {
1247 match self {
1248 Goal::Accept { n_rows, .. } | Goal::Boundary { n_rows } => pos < *n_rows,
1249 Goal::AcceptOrBlocked { end, blocked } => {
1250 if pos == *end {
1251 if nfa.reach_accept[target] {
1252 **blocked = true;
1253 }
1254 false
1255 } else {
1256 true
1257 }
1258 }
1259 }
1260 }
1261}
1262
1263/// One frame of the explicit walk stack: the state under exploration at row `pos`, and how far
1264/// through its transitions the walk has got.
1265struct Frame {
1266 state: StateId,
1267 pos: usize,
1268 /// Index of the next transition of `state` to try.
1269 next_edge: usize,
1270 /// Entered through a consuming transition: it opened a [`Visited`] scope and pushed a label,
1271 /// both undone on exit, and its failure is memoizable.
1272 consumed: bool,
1273 /// The entry checks have run (goal verdict, budget, visited).
1274 entered: bool,
1275 /// The frame marked `state` in its scope, and must unmark it on exit.
1276 inserted: bool,
1277}
1278
1279impl Frame {
1280 fn new(state: StateId, pos: usize, consumed: bool) -> Self {
1281 Self {
1282 state,
1283 pos,
1284 next_edge: 0,
1285 consumed,
1286 entered: false,
1287 inserted: false,
1288 }
1289 }
1290}
1291
1292/// A sub-NFA fragment with one entry and one exit state.
1293struct Fragment {
1294 start: StateId,
1295 accept: StateId,
1296}
1297
1298struct NfaBuilder {
1299 states: Vec<Vec<Transition>>,
1300}
1301
1302impl NfaBuilder {
1303 fn new_state(&mut self) -> StateId {
1304 self.states.push(Vec::new());
1305 self.states.len() - 1
1306 }
1307
1308 fn add_epsilon(&mut self, from: StateId, to: StateId) {
1309 self.states[from].push(Transition::Epsilon(to));
1310 }
1311
1312 fn add_on_var(&mut self, from: StateId, var: String, to: StateId) {
1313 self.states[from].push(Transition::OnVar { var, target: to });
1314 }
1315
1316 /// LOCKSTEP: the per-construct state counts below are mirrored by `estimate_nfa_states` in
1317 /// `frontend/src/binder/relation/match_recognize.rs`, which rejects a pattern whose expansion
1318 /// would exhaust memory here. No test can span the two crates, so changing how many states a
1319 /// construct allocates requires updating that estimator in the same change.
1320 fn build(&mut self, pattern: &Pattern) -> Fragment {
1321 match pattern {
1322 Pattern::Var(v) => {
1323 let start = self.new_state();
1324 let accept = self.new_state();
1325 self.add_on_var(start, v.clone(), accept);
1326 Fragment { start, accept }
1327 }
1328 Pattern::Concat(parts) => {
1329 if parts.is_empty() {
1330 let s = self.new_state();
1331 return Fragment {
1332 start: s,
1333 accept: s,
1334 };
1335 }
1336 let first = self.build(&parts[0]);
1337 let mut accept = first.accept;
1338 for p in &parts[1..] {
1339 let frag = self.build(p);
1340 self.add_epsilon(accept, frag.start);
1341 accept = frag.accept;
1342 }
1343 Fragment {
1344 start: first.start,
1345 accept,
1346 }
1347 }
1348 Pattern::Alt(alts) => {
1349 let start = self.new_state();
1350 let accept = self.new_state();
1351 for a in alts {
1352 let frag = self.build(a);
1353 self.add_epsilon(start, frag.start);
1354 self.add_epsilon(frag.accept, accept);
1355 }
1356 Fragment { start, accept }
1357 }
1358 Pattern::Quantified(inner, q, reluctant) => self.build_quantified(inner, q, *reluctant),
1359 Pattern::Permute(vars) => {
1360 // PERMUTE expands to the alternation of every ordering of the variables.
1361 let alts: Vec<Pattern> = permutations(vars)
1362 .into_iter()
1363 .map(|order| Pattern::Concat(order.into_iter().map(Pattern::Var).collect()))
1364 .collect();
1365 self.build(&Pattern::Alt(alts))
1366 }
1367 }
1368 }
1369
1370 /// LOCKSTEP with `estimate_nfa_states` — see [`Self::build`].
1371 fn build_quantified(&mut self, inner: &Pattern, q: &Quantifier, reluctant: bool) -> Fragment {
1372 match q {
1373 Quantifier::Star => self.build_star(inner, reluctant),
1374 Quantifier::Plus => {
1375 // inner followed by inner* (the repetition carries the reluctant preference)
1376 let first = self.build(inner);
1377 let star = self.build_star(inner, reluctant);
1378 self.add_epsilon(first.accept, star.start);
1379 Fragment {
1380 start: first.start,
1381 accept: star.accept,
1382 }
1383 }
1384 Quantifier::Question => {
1385 let start = self.new_state();
1386 let accept = self.new_state();
1387 let frag = self.build(inner);
1388 // Greedy orders take-the-inner before skip; reluctant orders skip first.
1389 if reluctant {
1390 self.add_epsilon(start, accept); // skip first
1391 self.add_epsilon(start, frag.start);
1392 } else {
1393 self.add_epsilon(start, frag.start);
1394 self.add_epsilon(start, accept); // skip
1395 }
1396 self.add_epsilon(frag.accept, accept);
1397 Fragment { start, accept }
1398 }
1399 Quantifier::Range { min, max } => self.build_range(inner, *min, *max, reluctant),
1400 }
1401 }
1402
1403 /// LOCKSTEP with `estimate_nfa_states` — see [`Self::build`].
1404 fn build_star(&mut self, inner: &Pattern, reluctant: bool) -> Fragment {
1405 let start = self.new_state();
1406 let accept = self.new_state();
1407 let frag = self.build(inner);
1408 // The matcher takes the first accepting path in edge order. Greedy emits the consume/loop
1409 // edge before the exit edge (longest match first); reluctant emits the exit edge first
1410 // (shortest match first).
1411 if reluctant {
1412 self.add_epsilon(start, accept); // zero occurrences first
1413 self.add_epsilon(start, frag.start);
1414 self.add_epsilon(frag.accept, accept); // exit before loop
1415 self.add_epsilon(frag.accept, frag.start);
1416 } else {
1417 self.add_epsilon(start, frag.start);
1418 self.add_epsilon(start, accept); // zero occurrences
1419 self.add_epsilon(frag.accept, frag.start); // loop
1420 self.add_epsilon(frag.accept, accept);
1421 }
1422 Fragment { start, accept }
1423 }
1424
1425 /// LOCKSTEP with `estimate_nfa_states` — see [`Self::build`]. The bounds reaching here are already
1426 /// capped at bind time precisely because this expansion is eager.
1427 fn build_range(
1428 &mut self,
1429 inner: &Pattern,
1430 min: u32,
1431 max: Option<u32>,
1432 reluctant: bool,
1433 ) -> Fragment {
1434 // Expand to `min` mandatory copies followed by either `*` (unbounded) or `max-min`
1435 // optional copies.
1436 let mut parts: Vec<Pattern> = Vec::new();
1437 for _ in 0..min {
1438 parts.push(inner.clone());
1439 }
1440 match max {
1441 None => parts.push(Pattern::Quantified(
1442 Box::new(inner.clone()),
1443 Quantifier::Star,
1444 reluctant,
1445 )),
1446 Some(max) => {
1447 for _ in min..max {
1448 parts.push(Pattern::Quantified(
1449 Box::new(inner.clone()),
1450 Quantifier::Question,
1451 reluctant,
1452 ));
1453 }
1454 }
1455 }
1456 self.build(&Pattern::Concat(parts))
1457 }
1458}
1459
1460/// All orderings of `items`. Only used for `PERMUTE`, which has a small arity in practice.
1461fn permutations(items: &[String]) -> Vec<Vec<String>> {
1462 if items.is_empty() {
1463 return vec![vec![]];
1464 }
1465 let mut out = Vec::new();
1466 for i in 0..items.len() {
1467 let mut rest = items.to_vec();
1468 let head = rest.remove(i);
1469 for mut tail in permutations(&rest) {
1470 tail.insert(0, head.clone());
1471 out.push(tail);
1472 }
1473 }
1474 out
1475}
1476
1477/// A [`CandidateMatcher`] backed by precomputed satisfied-sets — the dynamic driver should then
1478/// agree with [`Nfa::find_matches_labeled`]. Test-only, but `pub(crate)` so the sibling
1479/// `incremental` module's differential-oracle tests can reuse the exact same reference matcher as
1480/// `nfa`'s own tests.
1481#[cfg(test)]
1482pub(crate) struct SetMatcher {
1483 rows: Vec<BTreeSet<String>>,
1484}
1485
1486#[cfg(test)]
1487impl SetMatcher {
1488 pub(crate) fn new(rows: Vec<BTreeSet<String>>) -> Self {
1489 Self { rows }
1490 }
1491}
1492
1493#[cfg(test)]
1494impl CandidateMatcher for SetMatcher {
1495 async fn matches(
1496 &self,
1497 var: &str,
1498 pos: usize,
1499 _labels: &[String],
1500 ) -> StreamExecutorResult<bool> {
1501 Ok(self.rows[pos].contains(var))
1502 }
1503}
1504
1505#[cfg(test)]
1506mod tests {
1507 use super::*;
1508
1509 fn vars(s: &str) -> Pattern {
1510 Pattern::Var(s.to_owned())
1511 }
1512
1513 /// Build a row sequence from a string where each char names the single variable that row
1514 /// satisfies, e.g. "abc" -> [{a}, {b}, {c}].
1515 fn rows(seq: &str) -> Vec<BTreeSet<String>> {
1516 seq.chars()
1517 .map(|c| BTreeSet::from([c.to_string()]))
1518 .collect()
1519 }
1520
1521 #[test]
1522 fn concat_exact() {
1523 // A B C
1524 let p = Pattern::Concat(vec![vars("a"), vars("b"), vars("c")]);
1525 let nfa = Nfa::compile(&p);
1526 assert_eq!(nfa.longest_match(&rows("abc"), 0), Some(3));
1527 assert_eq!(nfa.longest_match(&rows("abx"), 0), None);
1528 assert_eq!(nfa.longest_match(&rows("ab"), 0), None);
1529 }
1530
1531 #[test]
1532 fn plus_is_greedy() {
1533 // A B+ C on a b b b c
1534 let p = Pattern::Concat(vec![
1535 vars("a"),
1536 Pattern::Quantified(Box::new(vars("b")), Quantifier::Plus, false),
1537 vars("c"),
1538 ]);
1539 let nfa = Nfa::compile(&p);
1540 assert_eq!(nfa.longest_match(&rows("abbbc"), 0), Some(5));
1541 // B+ requires at least one b.
1542 assert_eq!(nfa.longest_match(&rows("ac"), 0), None);
1543 }
1544
1545 #[test]
1546 fn question_optional() {
1547 // A B? C matches both "abc" and "ac"
1548 let p = Pattern::Concat(vec![
1549 vars("a"),
1550 Pattern::Quantified(Box::new(vars("b")), Quantifier::Question, false),
1551 vars("c"),
1552 ]);
1553 let nfa = Nfa::compile(&p);
1554 assert_eq!(nfa.longest_match(&rows("abc"), 0), Some(3));
1555 assert_eq!(nfa.longest_match(&rows("ac"), 0), Some(2));
1556 }
1557
1558 #[test]
1559 fn star_greedy_longest() {
1560 // A* on a a a -> greedy longest is 3
1561 let p = Pattern::Quantified(Box::new(vars("a")), Quantifier::Star, false);
1562 let nfa = Nfa::compile(&p);
1563 assert_eq!(nfa.longest_match(&rows("aaa"), 0), Some(3));
1564 // zero occurrences still matches (empty match).
1565 assert_eq!(nfa.longest_match(&rows("xyz"), 0), Some(0));
1566 }
1567
1568 #[test]
1569 fn alternation() {
1570 // (A | B) C
1571 let p = Pattern::Concat(vec![Pattern::Alt(vec![vars("a"), vars("b")]), vars("c")]);
1572 let nfa = Nfa::compile(&p);
1573 assert_eq!(nfa.longest_match(&rows("ac"), 0), Some(2));
1574 assert_eq!(nfa.longest_match(&rows("bc"), 0), Some(2));
1575 assert_eq!(nfa.longest_match(&rows("cc"), 0), None);
1576 }
1577
1578 #[test]
1579 fn range_bounds() {
1580 // A{2,3}
1581 let p = Pattern::Quantified(
1582 Box::new(vars("a")),
1583 Quantifier::Range {
1584 min: 2,
1585 max: Some(3),
1586 },
1587 false,
1588 );
1589 let nfa = Nfa::compile(&p);
1590 assert_eq!(nfa.longest_match(&rows("a"), 0), None); // need >= 2
1591 assert_eq!(nfa.longest_match(&rows("aa"), 0), Some(2));
1592 assert_eq!(nfa.longest_match(&rows("aaa"), 0), Some(3));
1593 assert_eq!(nfa.longest_match(&rows("aaaa"), 0), Some(3)); // capped at 3
1594 }
1595
1596 #[test]
1597 fn permute_any_order() {
1598 // PERMUTE(a, b)
1599 let p = Pattern::Permute(vec!["a".to_owned(), "b".to_owned()]);
1600 let nfa = Nfa::compile(&p);
1601 assert_eq!(nfa.longest_match(&rows("ab"), 0), Some(2));
1602 assert_eq!(nfa.longest_match(&rows("ba"), 0), Some(2));
1603 assert_eq!(nfa.longest_match(&rows("aa"), 0), None);
1604 }
1605
1606 #[test]
1607 fn match_from_offset() {
1608 // A B starting at index 1 of x a b
1609 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
1610 let nfa = Nfa::compile(&p);
1611 assert_eq!(nfa.longest_match(&rows("xab"), 1), Some(3));
1612 assert_eq!(nfa.longest_match(&rows("xab"), 0), None);
1613 }
1614
1615 fn spans(v: &[(usize, usize)]) -> Vec<MatchSpan> {
1616 v.iter()
1617 .map(|&(start, end)| MatchSpan { start, end })
1618 .collect()
1619 }
1620
1621 #[test]
1622 fn find_matches_skip_past_last_row() {
1623 // A B, repeated, with SKIP PAST LAST ROW -> non-overlapping matches.
1624 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
1625 let nfa = Nfa::compile(&p);
1626 assert_eq!(
1627 nfa.find_matches(&rows("ababab"), &SkipMode::PastLastRow),
1628 spans(&[(0, 2), (2, 4), (4, 6)])
1629 );
1630 }
1631
1632 #[test]
1633 fn find_matches_skip_to_next_row_overlaps() {
1634 // A+ with SKIP TO NEXT ROW: matches may overlap (resume at start+1).
1635 let p = Pattern::Quantified(Box::new(vars("a")), Quantifier::Plus, false);
1636 let nfa = Nfa::compile(&p);
1637 // "aaa": greedy A+ at 0->(0,3); to-next resumes at 1->(1,3); 2->(2,3).
1638 assert_eq!(
1639 nfa.find_matches(&rows("aaa"), &SkipMode::ToNextRow),
1640 spans(&[(0, 3), (1, 3), (2, 3)])
1641 );
1642 // PAST LAST ROW on the same input: single match.
1643 assert_eq!(
1644 nfa.find_matches(&rows("aaa"), &SkipMode::PastLastRow),
1645 spans(&[(0, 3)])
1646 );
1647 }
1648
1649 #[test]
1650 fn find_matches_greedy_then_resume() {
1651 // A B+ : greedy consumes all b's, then resumes past the match.
1652 let p = Pattern::Concat(vec![
1653 vars("a"),
1654 Pattern::Quantified(Box::new(vars("b")), Quantifier::Plus, false),
1655 ]);
1656 let nfa = Nfa::compile(&p);
1657 // a b b | a b -> (0,3) then (3,5)
1658 assert_eq!(
1659 nfa.find_matches(&rows("abbab"), &SkipMode::PastLastRow),
1660 spans(&[(0, 3), (3, 5)])
1661 );
1662 }
1663
1664 #[test]
1665 fn find_matches_skips_non_matching_rows() {
1666 // A B with junk rows between matches.
1667 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
1668 let nfa = Nfa::compile(&p);
1669 // x a b x x a b -> (1,3),(5,7)
1670 assert_eq!(
1671 nfa.find_matches(&rows("xabxxab"), &SkipMode::PastLastRow),
1672 spans(&[(1, 3), (5, 7)])
1673 );
1674 }
1675
1676 #[test]
1677 fn find_matches_empty_pattern_terminates() {
1678 // A* matches empty everywhere; empty matches are not emitted and the scan terminates.
1679 let p = Pattern::Quantified(Box::new(vars("a")), Quantifier::Star, false);
1680 let nfa = Nfa::compile(&p);
1681 // "aa b aa" -> greedy A* consumes runs of a, emits non-empty ones.
1682 assert_eq!(
1683 nfa.find_matches(&rows("aabaa"), &SkipMode::PastLastRow),
1684 spans(&[(0, 2), (3, 5)])
1685 );
1686 // all-non-matching -> no matches, terminates.
1687 assert_eq!(
1688 nfa.find_matches(&rows("xxx"), &SkipMode::PastLastRow),
1689 spans(&[])
1690 );
1691 }
1692
1693 fn lbl(s: &str) -> Vec<String> {
1694 s.chars().map(|c| c.to_string()).collect()
1695 }
1696
1697 #[test]
1698 fn labeled_concat() {
1699 // A B -> rows labelled a, b.
1700 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
1701 let nfa = Nfa::compile(&p);
1702 assert_eq!(
1703 nfa.longest_match_labeled(&rows("ab"), 0),
1704 Some((2, lbl("ab")))
1705 );
1706 }
1707
1708 #[test]
1709 fn labeled_plus_greedy() {
1710 // A B+ on a b b -> labels a, b, b (greedy consumes both b's).
1711 let p = Pattern::Concat(vec![
1712 vars("a"),
1713 Pattern::Quantified(Box::new(vars("b")), Quantifier::Plus, false),
1714 ]);
1715 let nfa = Nfa::compile(&p);
1716 assert_eq!(
1717 nfa.longest_match_labeled(&rows("abb"), 0),
1718 Some((3, lbl("abb")))
1719 );
1720 }
1721
1722 #[test]
1723 fn labeled_alternation() {
1724 // (A | B) C on b c -> labels b, c.
1725 let p = Pattern::Concat(vec![Pattern::Alt(vec![vars("a"), vars("b")]), vars("c")]);
1726 let nfa = Nfa::compile(&p);
1727 assert_eq!(
1728 nfa.longest_match_labeled(&rows("bc"), 0),
1729 Some((2, lbl("bc")))
1730 );
1731 }
1732
1733 #[test]
1734 fn labeled_permute() {
1735 // PERMUTE(a, b) on b a -> labels b, a.
1736 let p = Pattern::Permute(vec!["a".to_owned(), "b".to_owned()]);
1737 let nfa = Nfa::compile(&p);
1738 assert_eq!(
1739 nfa.longest_match_labeled(&rows("ba"), 0),
1740 Some((2, lbl("ba")))
1741 );
1742 }
1743
1744 #[test]
1745 fn find_matches_labeled_carries_labels() {
1746 // A B repeated -> two labelled matches.
1747 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
1748 let nfa = Nfa::compile(&p);
1749 assert_eq!(
1750 nfa.find_matches_labeled(&rows("abab"), &SkipMode::PastLastRow),
1751 vec![
1752 LabeledMatch {
1753 start: 0,
1754 end: 2,
1755 labels: lbl("ab")
1756 },
1757 LabeledMatch {
1758 start: 2,
1759 end: 4,
1760 labels: lbl("ab")
1761 },
1762 ]
1763 );
1764 }
1765
1766 #[test]
1767 fn skip_to_first_last_var() {
1768 // Pattern (a b b) over five rows that each satisfy both `a` and `b`, so matches can overlap.
1769 // The skip strategy decides where each next match starts:
1770 // PAST LAST ROW -> one match [0,3)
1771 // SKIP TO LAST b -> [0,3), [2,5) (resume at the match's last `b`)
1772 // SKIP TO FIRST b -> [0,3), [1,4), [2,5) (resume at the match's first `b`)
1773 let p = Pattern::Concat(vec![vars("a"), vars("b"), vars("b")]);
1774 let nfa = Nfa::compile(&p);
1775 let rows = vec![BTreeSet::from(["a".to_owned(), "b".to_owned()]); 5];
1776
1777 let starts = |skip: &SkipMode| {
1778 nfa.find_matches_labeled(&rows, skip)
1779 .into_iter()
1780 .map(|m| m.start)
1781 .collect::<Vec<_>>()
1782 };
1783 assert_eq!(starts(&SkipMode::PastLastRow), vec![0]);
1784 assert_eq!(starts(&SkipMode::ToLast("b".to_owned())), vec![0, 2]);
1785 assert_eq!(starts(&SkipMode::ToFirst("b".to_owned())), vec![0, 1, 2]);
1786 }
1787
1788 /// The two data-dependent cases where a variable-targeted skip has no valid resume row degrade
1789 /// (deliberately, instead of raising the runtime error the SQL standard prescribes — see
1790 /// [`SkipMode::next_pos`]). The degradation must be *visible*, so `next_pos` returns a
1791 /// diagnostic alongside the position, and the position itself must be unchanged: this is a
1792 /// visibility fix, not a behavior change.
1793 #[test]
1794 fn skip_target_degradations_are_reported() {
1795 // A match over rows 3..5, labelled `a` then `b`.
1796 let labels = lbl("ab");
1797
1798 // (a) The target is bound to no row of this match: nothing to resume at, so the scan resumes
1799 // past the match's last row — silently becoming SKIP PAST LAST ROW.
1800 assert_eq!(
1801 SkipMode::ToLast("c".to_owned()).next_pos(3, 5, &labels),
1802 (5, Some(SkipDegradation::TargetAbsent))
1803 );
1804 assert_eq!(
1805 SkipMode::ToFirst("c".to_owned()).next_pos(3, 5, &labels),
1806 (5, Some(SkipDegradation::TargetAbsent))
1807 );
1808
1809 // (b) The target resolves to the match's own first row: resuming there would re-find the
1810 // same match forever, so it is clamped one row on — silently becoming SKIP TO NEXT ROW.
1811 assert_eq!(
1812 SkipMode::ToFirst("a".to_owned()).next_pos(3, 5, &labels),
1813 (4, Some(SkipDegradation::TargetAtMatchStart))
1814 );
1815 assert_eq!(
1816 SkipMode::ToLast("a".to_owned()).next_pos(3, 5, &labels),
1817 (4, Some(SkipDegradation::TargetAtMatchStart))
1818 );
1819
1820 // A resolvable target reports nothing — note it can land on the same position as the clamped
1821 // case above, so the diagnostic (not the position) is what distinguishes them.
1822 assert_eq!(
1823 SkipMode::ToLast("b".to_owned()).next_pos(3, 5, &labels),
1824 (4, None)
1825 );
1826 // The variable-less modes cannot degrade.
1827 assert_eq!(SkipMode::PastLastRow.next_pos(3, 5, &labels), (5, None));
1828 assert_eq!(SkipMode::ToNextRow.next_pos(3, 5, &labels), (4, None));
1829 }
1830
1831 /// The diagnostic names the target variable and the strategy the resume position degraded to. The
1832 /// skip mode itself is named by the caller (`clause_name`), so it must NOT be repeated here — the
1833 /// rendered message would otherwise say `SKIP` twice.
1834 #[test]
1835 fn skip_degradation_describes_target_and_fallback() {
1836 let absent = SkipDegradation::TargetAbsent.describe(&SkipMode::ToLast("c".to_owned()));
1837 assert!(absent.contains("target variable `c`"), "{absent}");
1838 assert!(absent.contains("SKIP PAST LAST ROW"), "{absent}");
1839 assert!(!absent.contains("SKIP TO LAST"), "{absent}");
1840
1841 let at_start =
1842 SkipDegradation::TargetAtMatchStart.describe(&SkipMode::ToFirst("a".to_owned()));
1843 assert!(at_start.contains("target variable `a`"), "{at_start}");
1844 assert!(at_start.contains("SKIP TO NEXT ROW"), "{at_start}");
1845 assert!(!at_start.contains("SKIP TO FIRST"), "{at_start}");
1846 }
1847
1848 /// The clause name feeds the `ExprError` parameter name, so it must state the mode without the
1849 /// target variable (which `describe` names) — one mention of the mode per rendered message.
1850 #[test]
1851 fn skip_clause_name_and_target_var() {
1852 assert_eq!(
1853 SkipMode::ToLast("c".to_owned()).clause_name(),
1854 "AFTER MATCH SKIP TO LAST"
1855 );
1856 assert_eq!(
1857 SkipMode::ToFirst("a".to_owned()).clause_name(),
1858 "AFTER MATCH SKIP TO FIRST"
1859 );
1860 assert_eq!(
1861 SkipMode::PastLastRow.clause_name(),
1862 "AFTER MATCH SKIP PAST LAST ROW"
1863 );
1864 assert_eq!(
1865 SkipMode::ToNextRow.clause_name(),
1866 "AFTER MATCH SKIP TO NEXT ROW"
1867 );
1868
1869 assert_eq!(SkipMode::ToLast("c".to_owned()).target_var(), Some("c"));
1870 assert_eq!(SkipMode::ToFirst("a".to_owned()).target_var(), Some("a"));
1871 // The variable-less modes have no target — and can never degrade.
1872 assert_eq!(SkipMode::PastLastRow.target_var(), None);
1873 assert_eq!(SkipMode::ToNextRow.target_var(), None);
1874 }
1875
1876 #[test]
1877 fn row_satisfying_multiple_vars() {
1878 // Overlapping DEFINE predicates: a row can satisfy several variables.
1879 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
1880 let nfa = Nfa::compile(&p);
1881 let rows = vec![
1882 BTreeSet::from(["a".to_owned(), "b".to_owned()]),
1883 BTreeSet::from(["b".to_owned()]),
1884 ];
1885 assert_eq!(nfa.longest_match(&rows, 0), Some(2));
1886 }
1887
1888 /// [`Nfa::next_match`] is lazy: pulling one match must not evaluate predicates past that
1889 /// match's scan region, and pulling to exhaustion must enumerate exactly what
1890 /// [`Nfa::find_matches_dynamic`] collects. The emit loop relies on the first property to stop
1891 /// scanning at a held boundary match without paying for (or materializing) the rest.
1892 #[tokio::test]
1893 async fn next_match_is_lazy_and_equivalent() {
1894 use std::sync::atomic::{AtomicUsize, Ordering};
1895
1896 struct CountingMatcher {
1897 rows: Vec<BTreeSet<String>>,
1898 calls: AtomicUsize,
1899 }
1900 impl CandidateMatcher for CountingMatcher {
1901 async fn matches(
1902 &self,
1903 var: &str,
1904 pos: usize,
1905 _labels: &[String],
1906 ) -> StreamExecutorResult<bool> {
1907 self.calls.fetch_add(1, Ordering::Relaxed);
1908 Ok(self.rows[pos].contains(var))
1909 }
1910 }
1911
1912 let nfa = Nfa::compile(&Pattern::Concat(vec![vars("a"), vars("b")]));
1913 let r = rows("ababab");
1914
1915 // Pull ONE match, then compare predicate-evaluation counts against a full collect.
1916 let first_only = CountingMatcher {
1917 rows: r.clone(),
1918 calls: AtomicUsize::new(0),
1919 };
1920 let mut scan = MatchScan::new();
1921 let skip = SkipMode::PastLastRow;
1922 let first = nfa
1923 .next_match(
1924 &mut scan,
1925 r.len(),
1926 &first_only,
1927 &skip,
1928 &mut ScanBudget::unlimited(),
1929 false,
1930 )
1931 .await
1932 .unwrap()
1933 .expect("first match");
1934 assert_eq!((first.start, first.end), (0, 2));
1935 let one_pull = first_only.calls.load(Ordering::Relaxed);
1936
1937 let full = CountingMatcher {
1938 rows: r.clone(),
1939 calls: AtomicUsize::new(0),
1940 };
1941 let collected = nfa
1942 .find_matches_dynamic(r.len(), &full, &skip)
1943 .await
1944 .unwrap();
1945 let full_scan = full.calls.load(Ordering::Relaxed);
1946 assert!(
1947 one_pull < full_scan,
1948 "one pull ({one_pull} evaluations) must cost less than the full scan ({full_scan})"
1949 );
1950
1951 // Pulling to exhaustion enumerates exactly the collected sequence.
1952 let m = SetMatcher { rows: r.clone() };
1953 let mut scan = MatchScan::new();
1954 let mut pulled = Vec::new();
1955 while let Some(mm) = nfa
1956 .next_match(
1957 &mut scan,
1958 r.len(),
1959 &m,
1960 &skip,
1961 &mut ScanBudget::unlimited(),
1962 false,
1963 )
1964 .await
1965 .unwrap()
1966 {
1967 pulled.push(mm);
1968 }
1969 assert_eq!(pulled, collected);
1970 }
1971
1972 /// The catastrophic-backtracking family: `(a? a? … a? b)` over a run of `a`-rows costs
1973 /// exponentially many predicate evaluations per start unmemoized. The per-start `(state, pos)`
1974 /// failure memo (sound for path-independent verdicts, recorded only at consumption boundaries)
1975 /// must collapse that to polynomial — and must not change a single result.
1976 #[tokio::test]
1977 async fn memoization_defuses_catastrophic_backtracking() {
1978 use std::sync::atomic::{AtomicUsize, Ordering};
1979
1980 struct CountingMatcher {
1981 rows: Vec<BTreeSet<String>>,
1982 calls: AtomicUsize,
1983 }
1984 impl CandidateMatcher for CountingMatcher {
1985 async fn matches(
1986 &self,
1987 var: &str,
1988 pos: usize,
1989 _labels: &[String],
1990 ) -> StreamExecutorResult<bool> {
1991 self.calls.fetch_add(1, Ordering::Relaxed);
1992 Ok(self.rows[pos].contains(var))
1993 }
1994 }
1995
1996 // (a? ×16 b) over 20 `a`-rows and no `b`: every start fails, each exploring the
1997 // exponential subset lattice of which optionals consumed which rows.
1998 let mut parts: Vec<Pattern> = (0..16)
1999 .map(|_| Pattern::Quantified(Box::new(vars("a")), Quantifier::Question, false))
2000 .collect();
2001 parts.push(vars("b"));
2002 let nfa = Nfa::compile(&Pattern::Concat(parts));
2003 let r = rows(&"a".repeat(20));
2004
2005 let run = |memoize: bool| {
2006 let nfa = &nfa;
2007 let r = r.clone();
2008 async move {
2009 let m = CountingMatcher {
2010 rows: r.clone(),
2011 calls: AtomicUsize::new(0),
2012 };
2013 let mut budget = ScanBudget::unlimited();
2014 let mut scan = MatchScan::new();
2015 let mut out = Vec::new();
2016 while let Some(mm) = nfa
2017 .next_match(
2018 &mut scan,
2019 r.len(),
2020 &m,
2021 &SkipMode::PastLastRow,
2022 &mut budget,
2023 memoize,
2024 )
2025 .await
2026 .unwrap()
2027 {
2028 out.push(mm);
2029 }
2030 (out, m.calls.load(Ordering::Relaxed))
2031 }
2032 };
2033
2034 let (plain_out, plain_calls) = run(false).await;
2035 let (memo_out, memo_calls) = run(true).await;
2036 assert_eq!(plain_out, memo_out);
2037 assert!(
2038 memo_calls * 20 < plain_calls,
2039 "memoized scan ({memo_calls} evaluations) must be far below the backtracking scan \
2040 ({plain_calls})"
2041 );
2042 // And the memoized cost is genuinely polynomial-small for this size.
2043 assert!(memo_calls < 50_000, "memoized: {memo_calls}");
2044 }
2045
2046 /// A spent [`ScanBudget`] stops the walks without verdicts: the finder yields no further
2047 /// match (never a wrong one), the liveness walker answers "not alive" only alongside the
2048 /// sticky `hit` flag (which the executor must check before believing it), and the extension
2049 /// probe answers "may extend" (hold).
2050 #[tokio::test]
2051 async fn spent_budget_stops_without_verdicts() {
2052 let nfa = Nfa::compile(&Pattern::Concat(vec![vars("a"), vars("b")]));
2053 let r = rows("ab");
2054 let m = SetMatcher { rows: r.clone() };
2055
2056 // Unlimited: the match is found.
2057 let mut budget = ScanBudget::unlimited();
2058 let mut scan = MatchScan::new();
2059 let found = nfa
2060 .next_match(
2061 &mut scan,
2062 r.len(),
2063 &m,
2064 &SkipMode::PastLastRow,
2065 &mut budget,
2066 false,
2067 )
2068 .await
2069 .unwrap();
2070 assert!(found.is_some());
2071 assert!(!budget.hit);
2072
2073 // Zero budget: no match, hit latched.
2074 let mut budget = ScanBudget::new(0);
2075 let mut scan = MatchScan::new();
2076 let found = nfa
2077 .next_match(
2078 &mut scan,
2079 r.len(),
2080 &m,
2081 &SkipMode::PastLastRow,
2082 &mut budget,
2083 false,
2084 )
2085 .await
2086 .unwrap();
2087 assert!(found.is_none());
2088 assert!(budget.hit);
2089
2090 // Liveness under zero budget: answers false with hit latched — the executor treats that
2091 // as "undecided, retain", never as "dead".
2092 let mut budget = ScanBudget::new(0);
2093 let alive = nfa
2094 .reaches_boundary_alive(0, 1, &m, &mut budget, false)
2095 .await
2096 .unwrap();
2097 assert!(!alive);
2098 assert!(budget.hit);
2099
2100 // Extension probe under zero budget: conservative "may extend".
2101 let mut budget = ScanBudget::new(0);
2102 assert!(nfa.may_extend(0, 2, &m, &mut budget, false).await.unwrap());
2103 assert!(budget.hit);
2104 }
2105
2106 #[tokio::test]
2107 async fn dynamic_matches_static_for_set_predicate() {
2108 let p = Pattern::Concat(vec![vars("a"), vars("b")]);
2109 let nfa = Nfa::compile(&p);
2110 let r = rows("abab");
2111 let m = SetMatcher { rows: r.clone() };
2112 let dynamic = nfa
2113 .find_matches_dynamic(r.len(), &m, &SkipMode::PastLastRow)
2114 .await
2115 .unwrap();
2116 assert_eq!(
2117 dynamic,
2118 nfa.find_matches_labeled(&r, &SkipMode::PastLastRow)
2119 );
2120 }
2121
2122 /// [`Nfa::may_extend`]: a boundary match whose PREFERRED result cannot change is final (emit
2123 /// now); one where a higher-priority path is blocked on the row boundary — an open greedy
2124 /// quantifier, an earlier-listed longer alternation branch, an unexhausted range, an optional
2125 /// tail — must be held. The check follows the finder's preference order: lower-priority paths
2126 /// (a later-listed alternation branch, a reluctant loop's consume edge) can never override an
2127 /// accepting result and must not hold it.
2128 #[tokio::test]
2129 async fn may_extend_follows_the_finder_preference_order() {
2130 let m = |s: &str| SetMatcher { rows: rows(s) };
2131
2132 // Fixed (a b): after consuming both, nothing can extend — terminal.
2133 let fixed = Nfa::compile(&Pattern::Concat(vec![vars("a"), vars("b")]));
2134 assert!(
2135 !fixed
2136 .may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2137 .await
2138 .unwrap()
2139 );
2140
2141 // (a b+): the greedy loop's consume edge precedes its exit, so a future `b` would produce
2142 // a preferred (longer) result — held.
2143 let open = Nfa::compile(&Pattern::Concat(vec![
2144 vars("a"),
2145 Pattern::Quantified(Box::new(vars("b")), Quantifier::Plus, false),
2146 ]));
2147 assert!(
2148 open.may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2149 .await
2150 .unwrap()
2151 );
2152
2153 // (a (b | b c)): ordered alternation — the first-listed `b` branch already accepted, and a
2154 // later `c` cannot override it. Terminal, matching what the finder would return.
2155 let alt = Nfa::compile(&Pattern::Concat(vec![
2156 vars("a"),
2157 Pattern::Alt(vec![vars("b"), Pattern::Concat(vec![vars("b"), vars("c")])]),
2158 ]));
2159 assert!(
2160 !alt.may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2161 .await
2162 .unwrap()
2163 );
2164
2165 // (a (b c | b)): the LONGER branch is listed first, so a future `c` would produce a
2166 // preferred result — held. Preference direction, not structure, decides.
2167 let alt_rev = Nfa::compile(&Pattern::Concat(vec![
2168 vars("a"),
2169 Pattern::Alt(vec![Pattern::Concat(vec![vars("b"), vars("c")]), vars("b")]),
2170 ]));
2171 assert!(
2172 alt_rev
2173 .may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2174 .await
2175 .unwrap()
2176 );
2177
2178 // (a b+?): reluctant — the exit edge precedes the consume edge, so the short result is the
2179 // preferred one and future `b`s cannot change it. Terminal.
2180 let reluctant = Nfa::compile(&Pattern::Concat(vec![
2181 vars("a"),
2182 Pattern::Quantified(Box::new(vars("b")), Quantifier::Plus, true),
2183 ]));
2184 assert!(
2185 !reluctant
2186 .may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2187 .await
2188 .unwrap()
2189 );
2190
2191 // (a b{1,2}): one `b` leaves the range open; two exhaust it.
2192 let range = Nfa::compile(&Pattern::Concat(vec![
2193 vars("a"),
2194 Pattern::Quantified(
2195 Box::new(vars("b")),
2196 Quantifier::Range {
2197 min: 1,
2198 max: Some(2),
2199 },
2200 false,
2201 ),
2202 ]));
2203 assert!(
2204 range
2205 .may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2206 .await
2207 .unwrap()
2208 );
2209 assert!(
2210 !range
2211 .may_extend(0, 3, &m("abb"), &mut ScanBudget::unlimited(), false)
2212 .await
2213 .unwrap()
2214 );
2215
2216 // (a b?): the optional tail extends a bare [a]; a consumed (a, b) is terminal.
2217 let opt = Nfa::compile(&Pattern::Concat(vec![
2218 vars("a"),
2219 Pattern::Quantified(Box::new(vars("b")), Quantifier::Question, false),
2220 ]));
2221 assert!(
2222 opt.may_extend(0, 1, &m("a"), &mut ScanBudget::unlimited(), false)
2223 .await
2224 .unwrap()
2225 );
2226 assert!(
2227 !opt.may_extend(0, 2, &m("ab"), &mut ScanBudget::unlimited(), false)
2228 .await
2229 .unwrap()
2230 );
2231 }
2232
2233 #[tokio::test]
2234 async fn reaches_boundary_alive_evicts_dead_prefix() {
2235 // PATTERN (a b): a start is live only if a match from it can still reach the safe boundary.
2236 let nfa = Nfa::compile(&Pattern::Concat(vec![vars("a"), vars("b")]));
2237
2238 // `[a]` with the boundary right after it: the `a` is a live partial match — a future `b` may
2239 // complete it — so it must be retained.
2240 let m = SetMatcher { rows: rows("a") };
2241 assert!(
2242 nfa.reaches_boundary_alive(0, 1, &m, &mut ScanBudget::unlimited(), false)
2243 .await
2244 .unwrap()
2245 );
2246
2247 // `[a, x]` and `[a, x, x]` (x satisfies neither `a` nor `b`): the `a` can still *begin* the
2248 // pattern, but the following safe rows already block it from completing, so it is dead and
2249 // must be evictable. This is the case the previous `can_begin_at`-based predicate retained
2250 // forever.
2251 let m = SetMatcher { rows: rows("ax") };
2252 assert!(
2253 !nfa.reaches_boundary_alive(0, 2, &m, &mut ScanBudget::unlimited(), false)
2254 .await
2255 .unwrap()
2256 );
2257 let m = SetMatcher { rows: rows("axx") };
2258 assert!(
2259 !nfa.reaches_boundary_alive(0, 3, &m, &mut ScanBudget::unlimited(), false)
2260 .await
2261 .unwrap()
2262 );
2263
2264 // A later start can be the live one: in `[x, a]` row 0 is dead but row 1 (the `a`) is live.
2265 let m = SetMatcher { rows: rows("xa") };
2266 assert!(
2267 !nfa.reaches_boundary_alive(0, 2, &m, &mut ScanBudget::unlimited(), false)
2268 .await
2269 .unwrap()
2270 );
2271 assert!(
2272 nfa.reaches_boundary_alive(1, 2, &m, &mut ScanBudget::unlimited(), false)
2273 .await
2274 .unwrap()
2275 );
2276
2277 // A complete match sitting exactly at the boundary is not yet finalized (it needs a following
2278 // safe row to confirm maximality), so its start is still retained.
2279 let m = SetMatcher { rows: rows("ab") };
2280 assert!(
2281 nfa.reaches_boundary_alive(0, 2, &m, &mut ScanBudget::unlimited(), false)
2282 .await
2283 .unwrap()
2284 );
2285 }
2286
2287 /// A path-dependent matcher: `b` only matches once an `a` has been bound earlier in the match.
2288 /// This exercises threading the running labels into the predicate.
2289 struct NeedsPrecedingA;
2290 impl CandidateMatcher for NeedsPrecedingA {
2291 async fn matches(
2292 &self,
2293 var: &str,
2294 _pos: usize,
2295 labels: &[String],
2296 ) -> StreamExecutorResult<bool> {
2297 Ok(match var {
2298 "a" => true,
2299 "b" => labels.iter().any(|l| l == "a"),
2300 _ => false,
2301 })
2302 }
2303 }
2304
2305 #[tokio::test]
2306 async fn reluctant_quantifier_prefers_fewer() {
2307 // Three rows that each satisfy both `a` and `b`, so `a+ b` can stop early.
2308 let rows = vec![BTreeSet::from(["a".to_owned(), "b".to_owned()]); 3];
2309 let m = SetMatcher { rows: rows.clone() };
2310
2311 // Greedy `a+ b`: consume as many `a` as possible -> [0, 3) (a a b).
2312 let greedy = Nfa::compile(&Pattern::Concat(vec![
2313 Pattern::Quantified(Box::new(vars("a")), Quantifier::Plus, false),
2314 vars("b"),
2315 ]));
2316 assert_eq!(
2317 greedy
2318 .find_matches_dynamic(rows.len(), &m, &SkipMode::PastLastRow)
2319 .await
2320 .unwrap(),
2321 vec![LabeledMatch {
2322 start: 0,
2323 end: 3,
2324 labels: lbl("aab")
2325 }]
2326 );
2327
2328 // Reluctant `a+? b`: take the fewest `a` -> [0, 2) (a b), then [2, ...) finds nothing more.
2329 let reluctant = Nfa::compile(&Pattern::Concat(vec![
2330 Pattern::Quantified(Box::new(vars("a")), Quantifier::Plus, true),
2331 vars("b"),
2332 ]));
2333 assert_eq!(
2334 reluctant
2335 .find_matches_dynamic(rows.len(), &m, &SkipMode::PastLastRow)
2336 .await
2337 .unwrap(),
2338 vec![LabeledMatch {
2339 start: 0,
2340 end: 2,
2341 labels: lbl("ab")
2342 }]
2343 );
2344 }
2345
2346 /// `n` rows that each satisfy both `a` and `b`, so quantifier preference (not the predicates)
2347 /// decides how a match is split between variables.
2348 fn ab_rows(n: usize) -> Vec<BTreeSet<String>> {
2349 vec![BTreeSet::from(["a".to_owned(), "b".to_owned()]); n]
2350 }
2351
2352 fn plus(inner: Pattern, reluctant: bool) -> Pattern {
2353 Pattern::Quantified(Box::new(inner), Quantifier::Plus, reluctant)
2354 }
2355
2356 fn star(inner: Pattern, reluctant: bool) -> Pattern {
2357 Pattern::Quantified(Box::new(inner), Quantifier::Star, reluctant)
2358 }
2359
2360 #[tokio::test]
2361 async fn nested_reluctant_then_greedy_adjacent() {
2362 // `a*? a*` over three `a` rows. The reluctant first star takes as few as possible (zero) and
2363 // the greedy second star takes the rest, so the whole run is still consumed: [0, 3). This
2364 // guards against an empty-match or non-termination bug when two quantifiers over the same
2365 // variable sit adjacent with opposite preferences.
2366 let r = rows("aaa");
2367 let m = SetMatcher { rows: r.clone() };
2368 let nfa = Nfa::compile(&Pattern::Concat(vec![
2369 star(vars("a"), true),
2370 star(vars("a"), false),
2371 ]));
2372 assert_eq!(
2373 nfa.find_matches_dynamic(r.len(), &m, &SkipMode::PastLastRow)
2374 .await
2375 .unwrap(),
2376 vec![LabeledMatch {
2377 start: 0,
2378 end: 3,
2379 labels: lbl("aaa")
2380 }]
2381 );
2382 }
2383
2384 #[tokio::test]
2385 async fn nested_quantifier_preference_flips_split() {
2386 // Four rows that each satisfy both `a` and `b`, matched by `(<a-quant> b+)+`. The inner
2387 // first-variable quantifier's preference decides the split; the rest is greedy `b+`.
2388 let r = ab_rows(4);
2389 let m = SetMatcher { rows: r.clone() };
2390
2391 // Reluctant `a+?` takes the fewest `a` (one), then greedy `b+` takes the rest -> "abbb".
2392 let reluctant = Nfa::compile(&plus(
2393 Pattern::Concat(vec![plus(vars("a"), true), plus(vars("b"), false)]),
2394 false,
2395 ));
2396 assert_eq!(
2397 reluctant
2398 .find_matches_dynamic(r.len(), &m, &SkipMode::PastLastRow)
2399 .await
2400 .unwrap(),
2401 vec![LabeledMatch {
2402 start: 0,
2403 end: 4,
2404 labels: lbl("abbb")
2405 }]
2406 );
2407
2408 // Greedy `a+` takes as many `a` as it can while still leaving one row for the mandatory
2409 // `b+`, so it backtracks from four to three -> "aaab".
2410 let greedy = Nfa::compile(&plus(
2411 Pattern::Concat(vec![plus(vars("a"), false), plus(vars("b"), false)]),
2412 false,
2413 ));
2414 assert_eq!(
2415 greedy
2416 .find_matches_dynamic(r.len(), &m, &SkipMode::PastLastRow)
2417 .await
2418 .unwrap(),
2419 vec![LabeledMatch {
2420 start: 0,
2421 end: 4,
2422 labels: lbl("aaab")
2423 }]
2424 );
2425 }
2426
2427 #[tokio::test]
2428 async fn dynamic_threads_running_labels() {
2429 // (a b): `b` sees `a` in the running labels -> matches.
2430 let ab = Nfa::compile(&Pattern::Concat(vec![vars("a"), vars("b")]));
2431 let m = NeedsPrecedingA;
2432 assert_eq!(
2433 ab.find_matches_dynamic(2, &m, &SkipMode::PastLastRow)
2434 .await
2435 .unwrap(),
2436 vec![LabeledMatch {
2437 start: 0,
2438 end: 2,
2439 labels: lbl("ab")
2440 }]
2441 );
2442
2443 // (b a): `b` is first, the running labels are empty, so it cannot match -> no match.
2444 let ba = Nfa::compile(&Pattern::Concat(vec![vars("b"), vars("a")]));
2445 assert_eq!(
2446 ba.find_matches_dynamic(2, &m, &SkipMode::PastLastRow)
2447 .await
2448 .unwrap(),
2449 vec![]
2450 );
2451 }
2452
2453 /// A budget that dies inside a start must NOT leave the scan cursor past it: the cursor would
2454 /// then claim a verdict for a start whose walk never reached one. (It is not a fully-scanned
2455 /// marker either way — a hit jumps it past the whole match — but "aborted" and "decided" must
2456 /// stay distinguishable for a caller that resumes a pull.)
2457 #[tokio::test]
2458 async fn an_aborted_start_is_not_reported_as_scanned() {
2459 // `(a b) | a`: the preferred branch charges `a` then `b`; with a budget of 1 the walk dies
2460 // inside start 0 having never tried the second alternative, which DOES match there.
2461 let pat = Pattern::Alt(vec![
2462 Pattern::Concat(vec![Pattern::Var("a".into()), Pattern::Var("b".into())]),
2463 Pattern::Var("a".into()),
2464 ]);
2465 let nfa = Nfa::compile(&pat);
2466 let matcher = SetMatcher::new(vec![BTreeSet::from(["a".to_owned()]); 2]);
2467 let mut scan = MatchScan::new();
2468 let mut budget = ScanBudget::new(1);
2469 let found = nfa
2470 .next_match(
2471 &mut scan,
2472 2,
2473 &matcher,
2474 &SkipMode::PastLastRow,
2475 &mut budget,
2476 false,
2477 )
2478 .await
2479 .unwrap();
2480 assert!(
2481 found.is_none() && budget.hit,
2482 "the test needs the budget to die inside start 0"
2483 );
2484 assert_eq!(
2485 scan.next_start(),
2486 0,
2487 "start 0 was aborted with no verdict, so the cursor must still point at it"
2488 );
2489 }
2490
2491 #[test]
2492 fn max_match_rows_is_the_longest_path_or_none_for_a_cycle() {
2493 let var = |s: &str| Pattern::Var(s.to_owned());
2494 let concat = |s: &str| Pattern::Concat(s.split(' ').map(var).collect());
2495 let q = |p: Pattern, q: Quantifier| Pattern::Quantified(Box::new(p), q, false);
2496 let range = |min, max| Quantifier::Range { min, max };
2497 assert_eq!(Nfa::compile(&concat("a b c")).max_match_rows(), Some(3));
2498 assert_eq!(
2499 Nfa::compile(&Pattern::Alt(vec![concat("a b"), var("c")])).max_match_rows(),
2500 Some(2)
2501 );
2502 assert_eq!(
2503 Nfa::compile(&Pattern::Concat(vec![
2504 q(var("a"), Quantifier::Question),
2505 var("b")
2506 ]))
2507 .max_match_rows(),
2508 Some(2)
2509 );
2510 assert_eq!(
2511 Nfa::compile(&q(var("a"), range(2, Some(5)))).max_match_rows(),
2512 Some(5)
2513 );
2514 assert_eq!(
2515 Nfa::compile(&q(var("a"), range(600, Some(600)))).max_match_rows(),
2516 Some(600)
2517 );
2518 assert_eq!(
2519 Nfa::compile(&q(var("a"), Quantifier::Plus)).max_match_rows(),
2520 None
2521 );
2522 assert_eq!(
2523 Nfa::compile(&q(var("a"), Quantifier::Star)).max_match_rows(),
2524 None
2525 );
2526 assert_eq!(
2527 Nfa::compile(&q(var("a"), range(2, None))).max_match_rows(),
2528 None
2529 );
2530 }
2531
2532 /// With more rows to the boundary than an acyclic automaton can consume, a position is dead
2533 /// without a walk — a budget of 1 survives untouched. With exactly as many, the walk runs, and
2534 /// a match that ends on the boundary keeps its start alive.
2535 #[tokio::test]
2536 async fn far_from_the_boundary_an_acyclic_pattern_is_dead_without_a_walk() {
2537 let pat = Pattern::Quantified(
2538 Box::new(Pattern::Var("a".into())),
2539 Quantifier::Range {
2540 min: 600,
2541 max: Some(600),
2542 },
2543 false,
2544 );
2545 let nfa = Nfa::compile(&pat);
2546 let n_rows = 1300;
2547 let matcher = SetMatcher::new(vec![BTreeSet::from(["a".to_owned()]); n_rows]);
2548 let mut budget = ScanBudget::new(1);
2549 assert!(
2550 !nfa.reaches_boundary_alive(0, n_rows, &matcher, &mut budget, true)
2551 .await
2552 .unwrap()
2553 );
2554 assert!(!budget.hit, "the verdict must not have cost a walk");
2555 let mut budget = ScanBudget::unlimited();
2556 assert!(
2557 nfa.reaches_boundary_alive(700, n_rows, &matcher, &mut budget, true)
2558 .await
2559 .unwrap(),
2560 "a{{600}} from 700 accepts exactly on the boundary, which is alive"
2561 );
2562 }
2563
2564 /// A start whose walk finds no accept and never reaches the boundary died on rows that will
2565 /// never change: the scan proves it matchless forever. The proof runs contiguously from where
2566 /// the scan began, stops at a start that reached the boundary, and is not extended by a start
2567 /// the finder skipped as too short (that one is blocked at the boundary, not dead).
2568 #[tokio::test]
2569 async fn starts_that_die_below_the_boundary_are_proven_matchless() {
2570 let nfa = Nfa::compile(&Pattern::Concat(vec![
2571 Pattern::Var("a".into()),
2572 Pattern::Var("b".into()),
2573 ]));
2574 let scan_all = |seq: &str| {
2575 let nfa = &nfa;
2576 let matcher = SetMatcher::new(rows(seq));
2577 let n = seq.len();
2578 async move {
2579 let mut scan = MatchScan::new();
2580 let mut budget = ScanBudget::unlimited();
2581 let found = nfa
2582 .next_match(
2583 &mut scan,
2584 n,
2585 &matcher,
2586 &SkipMode::PastLastRow,
2587 &mut budget,
2588 false,
2589 )
2590 .await
2591 .unwrap();
2592 (found.map(|m| (m.start, m.end)), scan)
2593 }
2594 };
2595 // Starts 0, 1 and 2 all die below the boundary; start 3 is skipped as too short.
2596 let (found, scan) = scan_all("axaa").await;
2597 assert_eq!(found, None);
2598 assert_eq!(scan.matchless_upto(), 3);
2599 assert_eq!(scan.next_start(), 4);
2600 // Start 2 consumes `a` and stands at the boundary waiting for `b`: not matchless.
2601 let (found, scan) = scan_all("axa").await;
2602 assert_eq!(found, None);
2603 assert_eq!(scan.matchless_upto(), 2);
2604 // A match ends the contiguous run.
2605 let (found, scan) = scan_all("xab").await;
2606 assert_eq!(found, Some((1, 3)));
2607 assert_eq!(scan.matchless_upto(), 1);
2608 }
2609
2610 #[test]
2611 fn min_match_rows_is_the_shortest_accepting_path() {
2612 let var = |s: &str| Pattern::Var(s.to_owned());
2613 let concat = |s: &str| Pattern::Concat(s.split(' ').map(var).collect());
2614 let q = |p: Pattern, q: Quantifier| Pattern::Quantified(Box::new(p), q, false);
2615 assert_eq!(Nfa::compile(&concat("a b c")).min_match_rows(), 3);
2616 assert_eq!(
2617 Nfa::compile(&q(var("a"), Quantifier::Plus)).min_match_rows(),
2618 1
2619 );
2620 assert_eq!(
2621 Nfa::compile(&q(var("a"), Quantifier::Star)).min_match_rows(),
2622 0
2623 );
2624 assert_eq!(
2625 Nfa::compile(&Pattern::Concat(vec![
2626 q(var("a"), Quantifier::Question),
2627 var("b")
2628 ]))
2629 .min_match_rows(),
2630 1
2631 );
2632 assert_eq!(
2633 Nfa::compile(&Pattern::Alt(vec![concat("a b"), var("c")])).min_match_rows(),
2634 1
2635 );
2636 assert_eq!(
2637 Nfa::compile(&Pattern::Permute(vec!["a".into(), "b".into(), "c".into()]))
2638 .min_match_rows(),
2639 3
2640 );
2641 assert_eq!(
2642 Nfa::compile(&q(
2643 var("a"),
2644 Quantifier::Range {
2645 min: 600,
2646 max: Some(600),
2647 }
2648 ))
2649 .min_match_rows(),
2650 600
2651 );
2652 }
2653
2654 /// A start with fewer rows before the boundary than the shortest possible match is skipped
2655 /// without a walk — and the skip IS a verdict ("no match from here within these rows"), so the
2656 /// cursor advances past it exactly as it would after a failed walk. Here the whole buffer is
2657 /// shorter than `a{600}`: nothing is walked, and a budget of 1 survives untouched.
2658 #[tokio::test]
2659 async fn starts_too_close_to_the_boundary_are_skipped_without_a_walk() {
2660 let pat = Pattern::Quantified(
2661 Box::new(Pattern::Var("a".into())),
2662 Quantifier::Range {
2663 min: 600,
2664 max: Some(600),
2665 },
2666 false,
2667 );
2668 let nfa = Nfa::compile(&pat);
2669 let n_rows = 10;
2670 let matcher = SetMatcher::new(vec![BTreeSet::from(["a".to_owned()]); n_rows]);
2671 let mut scan = MatchScan::new();
2672 let mut budget = ScanBudget::new(1);
2673 let found = nfa
2674 .next_match(
2675 &mut scan,
2676 n_rows,
2677 &matcher,
2678 &SkipMode::PastLastRow,
2679 &mut budget,
2680 true,
2681 )
2682 .await
2683 .unwrap();
2684 assert!(found.is_none());
2685 assert!(!budget.hit, "no start was walked, so nothing was charged");
2686 assert_eq!(
2687 scan.next_start(),
2688 n_rows,
2689 "every start was skipped with a verdict, so the cursor passed them all"
2690 );
2691 }
2692
2693 /// The walkers used to recurse once per consumed row and once per ε-edge under a hard cap of
2694 /// 512 frames, so any match spanning more than a couple of hundred rows was permanently
2695 /// undecidable: the cap did not reset with the budget, and every visit died at the same depth.
2696 /// The walk is iterative now and only the budget bounds it. All three walkers reach this depth.
2697 #[tokio::test]
2698 async fn a_match_spanning_thousands_of_rows_is_decided() {
2699 let pat = Pattern::Quantified(Box::new(Pattern::Var("a".into())), Quantifier::Plus, false);
2700 let nfa = Nfa::compile(&pat);
2701 let n_rows = 2000;
2702 let matcher = SetMatcher::new(vec![BTreeSet::from(["a".to_owned()]); n_rows]);
2703
2704 let mut scan = MatchScan::new();
2705 let mut budget = ScanBudget::unlimited();
2706 let found = nfa
2707 .next_match(
2708 &mut scan,
2709 n_rows,
2710 &matcher,
2711 &SkipMode::PastLastRow,
2712 &mut budget,
2713 true,
2714 )
2715 .await
2716 .unwrap()
2717 .expect("a 2000-row greedy match must be found");
2718 assert_eq!((found.start, found.end), (0, n_rows));
2719 assert_eq!(found.labels.len(), n_rows);
2720 assert!(!budget.hit);
2721
2722 // The greedy `a+` is blocked at the boundary, so the match may still extend...
2723 let mut budget = ScanBudget::unlimited();
2724 assert!(
2725 nfa.may_extend(0, n_rows, &matcher, &mut budget, true)
2726 .await
2727 .unwrap()
2728 );
2729 assert!(!budget.hit);
2730 // ...and its start is alive there.
2731 let mut budget = ScanBudget::unlimited();
2732 assert!(
2733 nfa.reaches_boundary_alive(0, n_rows, &matcher, &mut budget, true)
2734 .await
2735 .unwrap()
2736 );
2737 assert!(!budget.hit);
2738 }
2739
2740 /// The reported case: the binder accepts repetition counts up to 1000, and `a{600}` compiles to
2741 /// a 600-copy chain the old depth cap could never walk to its accept. It must match exactly 600
2742 /// rows — no fewer, and not greedily more.
2743 #[tokio::test]
2744 async fn a_bounded_repetition_beyond_the_old_depth_cap_matches_exactly() {
2745 let pat = Pattern::Quantified(
2746 Box::new(Pattern::Var("a".into())),
2747 Quantifier::Range {
2748 min: 600,
2749 max: Some(600),
2750 },
2751 false,
2752 );
2753 let nfa = Nfa::compile(&pat);
2754 let spans = |n_rows: usize| {
2755 let nfa = &nfa;
2756 async move {
2757 let matcher = SetMatcher::new(vec![BTreeSet::from(["a".to_owned()]); n_rows]);
2758 nfa.find_matches_dynamic(n_rows, &matcher, &SkipMode::PastLastRow)
2759 .await
2760 .unwrap()
2761 .into_iter()
2762 .map(|m| (m.start, m.end))
2763 .collect::<Vec<_>>()
2764 }
2765 };
2766 assert_eq!(
2767 spans(599).await,
2768 vec![],
2769 "599 rows cannot complete a{{600}}"
2770 );
2771 assert_eq!(spans(600).await, vec![(0, 600)]);
2772 assert_eq!(
2773 spans(1300).await,
2774 vec![(0, 600), (600, 1200)],
2775 "two exact matches; the 100-row tail cannot complete a third"
2776 );
2777 }
2778}