feat(stream): add SQL MATCH_RECOGNIZE as a pure matcher over WatermarkSort

[dahankzter]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Supersedes #25899, implementing the architecture @chenzl25 proposed in review: split ordering from matching. MATCH_RECOGNIZE becomes a pure ordered-input matcher, and all ordering concerns (buffering, out-of-order arrival, watermark release) live in the sort operator that already owns them.

Architecture

StreamExchange HASH(partition keys)   -- exact key order: the matcher's state table hashes in PARTITION BY order
  └─ StreamWatermarkSort              -- full ORDER BY, releases rows strictly below the watermark
       └─ StreamMatchRecognize        -- same fragment (no exchange in between, order preserved)

• StreamEowcSort → StreamWatermarkSort: renamed (it is watermark-driven, not EOWC-specific) and extended with secondary order columns — a state-table PK extension after the sort column — so it realizes the full ORDER BY, not just the leading time column.
• The matcher is order-oblivious: it feeds rows to the NFA on arrival and emits each match as soon as it is decidable. A match is final when no earlier position could still produce a leftmost-preferred match, and no more-preferred path from its own start can be completed by future rows — or when its WITHIN deadline has strictly passed the watermark, which decides both at once. Results appear when they are decidable, not at window close; the EOWC-only restriction from feat(stream): add SQL MATCH_RECOGNIZE (row pattern recognition) #25899 is gone.
• State shrinks to one internal table (rows still referenced by live partial matches). The two NFA-snapshot tables from feat(stream): add SQL MATCH_RECOGNIZE (row pattern recognition) #25899 are dropped; recovery rebuilds matcher state by re-feeding the retained rows, and rescaling works the same way. seq, the PK tiebreaker for equal ORDER BY keys, is a per-actor monotonic counter seeded above every retained row on rebuild, so ties are re-fed in arrival order and replay is deterministic.
• Watermark handling in the matcher is reduced to WITHIN finality and dead-prefix pruning; it does not forward watermarks.
• Pathological patterns are metered everywhere: every NFA walk — including the match finder — runs under a per-visit scan budget (scoped per row / per partition visit, so one bad partition cannot starve the others) with a soundness-gated memo of failed positions. Budget exhaustion is never converted into a verdict: nothing is emitted, frozen, or evicted on partial information; the condition is reported once per pass and retried.
• MatchRecognizeInputMode proto enum: EVENT_TIME implemented here; PROCESSING_TIME reserved for a follow-up, and the executor rejects any other mode rather than running against input whose ordering guarantee does not hold.
• Rows with a NULL leading order key are filtered out below the sort at plan time, mirroring event-time processing dropping NULL-rowtime rows.

SQL scope (unchanged from #25899 v1)

Streaming, append-only input, ONE ROW PER MATCH; PARTITION BY + watermarked ORDER BY; PATTERN with concat, alternation, grouping, quantifiers * + ? {n,m} (greedy + reluctant), PERMUTE; DEFINE with running navigation (PREV/NEXT/FIRST/LAST, cross-variable refs); MEASURES incl. CLASSIFIER(), SUBSET, aggregates; AFTER MATCH SKIP variants; WITHIN.

Known costs and planned follow-ups

• A partition holding an open partial match re-scans its live window per arriving row; incrementalizing the provisional tail is the main planned perf follow-up. Related constant-factor work: per-row predicate caching, WITHIN deadline precompute, label interning.
• Each watermark visits every partition; a per-partition wakeup frontier (deadline index) would make that proportional to the partitions that need attention.
• Retained rows are resident in executor memory (bounded by WITHIN; a partial without WITHIN is retained until its partition decides it — noted at CREATE); wiring this into memory accounting is a follow-up.
• The sort and the matcher each persist a row once (buffer vs retained rows) — the cost of the ordering/matching split.
• PROCESSING_TIME input mode: enum reserved, planned follow-up.
• Empty matches are not emitted: PATTERN (a*) yields no summary row on non-matching input (SQL:2016 would emit one; Flink rejects such patterns).
• ALL ROWS PER MATCH, MATCH_NUMBER(), pattern anchors ^/$, and batch execution are not implemented (same as feat(stream): add SQL MATCH_RECOGNIZE (row pattern recognition) #25899).
• The design doc (docs/dev/src/design/match-recognize.md) rewrite for this architecture and operator metrics land as follow-ups on this PR.

Tests

• 45 planner cases (including multi-column PARTITION BY distribution pinning), parser tests, 86 stream unit tests including a 200-seed randomized operation-sequence oracle checking the incremental matcher against the batch scan and unit tests for the emission-finality gate.
• 29 e2e SLT files (e2e_test/streaming/match_recognize*.slt) covering skip modes, navigation, WITHIN boundaries, alternation preference (held vs superseded vs killed), PERMUTE listing-order preference via CLASSIFIER(), and state-table introspection; plus a recovery serial (kill/recover, including a multi-row partial over tied order keys) and rescale (ALTER ... SET PARALLELISM) — all green on a local cluster.
• fmt / clippy clean.

Relevant CI coverage a maintainer may want to enable (I can't add labels as an external contributor): e2e + recovery deterministic simulation, SQLSmith, and user-facing-changes (new SQL syntax).

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary. (No permission — see the note above.)
• I have added fuzzing tests or opened an issue to track them. (SQLSmith recognizes the new syntax but does not yet generate it; happy to open a tracking issue.)

Documentation

• My PR needs documentation updates.

Release note

RisingWave now supports the SQL MATCH_RECOGNIZE clause (row pattern recognition) on streaming, append-only inputs with a watermarked ORDER BY column. Patterns are matched per partition in event-time order, and each match is emitted as soon as it is decidable. Supported: ONE ROW PER MATCH, quantified patterns with PERMUTE, DEFINE with navigation functions, MEASURES with CLASSIFIER() and aggregates, AFTER MATCH SKIP variants, and WITHIN time bounds.