krishiv_sql/
grammar.rs

1#![forbid(unsafe_code)]
2//! SQL grammar and **engine-dimensioned** feature matrix for Krishiv.
3//!
4//! Provides a machine-readable inventory of which SQL dialect features are
5//! supported, and — crucially — *in which of the three execution engines*:
6//! batch (DataFusion planner + Krishiv extensions), streaming (continuous
7//! window compiler), and incremental (IVM / krishiv-delta). A single feature is
8//! frequently "supported in batch, partial in streaming, n/a in incremental",
9//! so each [`FeatureEntry`] carries a per-engine [`FeatureStatus`] rather than
10//! one global status — otherwise "measured coverage" silently means *batch*
11//! coverage (Phase 60).
12//!
13//! The public SQL reference page is **generated** from this matrix
14//! ([`generate_reference_markdown`]), never hand-written, and a CI drift guard
15//! (see `coverage.rs`) asserts the checked-in page matches and that every
16//! non-`n/a` engine cell is backed by an executable coverage case.
17
18/// Support status for a single SQL feature **in one engine**.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum FeatureStatus {
21    /// Fully supported in the current release.
22    Supported,
23    /// Partially supported; the `note` field explains the gap.
24    Partial,
25    /// Planned for a future release.
26    Planned,
27    /// Not applicable to this engine.
28    NotApplicable,
29}
30
31impl FeatureStatus {
32    pub fn as_str(self) -> &'static str {
33        match self {
34            Self::Supported => "supported",
35            Self::Partial => "partial",
36            Self::Planned => "planned",
37            Self::NotApplicable => "n/a",
38        }
39    }
40
41    /// A non-`n/a`, non-`planned` cell — i.e. one that must be backed by an
42    /// executable coverage case (the matrix-to-test CI rule).
43    pub fn is_claimed(self) -> bool {
44        matches!(self, Self::Supported | Self::Partial)
45    }
46}
47
48impl std::fmt::Display for FeatureStatus {
49    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
50        f.write_str(self.as_str())
51    }
52}
53
54/// The three Krishiv execution engines a feature can be dimensioned across.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub enum Engine {
57    Batch,
58    Streaming,
59    Incremental,
60}
61
62impl Engine {
63    pub const ALL: [Engine; 3] = [Engine::Batch, Engine::Streaming, Engine::Incremental];
64
65    pub fn as_str(self) -> &'static str {
66        match self {
67            Engine::Batch => "batch",
68            Engine::Streaming => "streaming",
69            Engine::Incremental => "incremental",
70        }
71    }
72}
73
74/// An execution surface a feature can be reached from. A `supported` engine
75/// cell alone cannot say *where* the feature runs (audit §9b): several real,
76/// tested streaming operators are reachable only from the embedded
77/// `StreamingDataFrame`/process API, while the distributed `stream:loop`
78/// runtime executes only `WindowExecutionSpec` shapes (windows, window-join,
79/// CEP) and the SQL front door compiles a further subset.
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum Placement {
82    /// The embedded Rust `StreamingDataFrame` / process API and its Python mirror.
83    EmbeddedApi,
84    /// The SQL front door (text SQL parse → plan → execute).
85    Sql,
86    /// The distributed runtime (`stream:loop` / partitioned batch fragments).
87    Distributed,
88}
89
90impl Placement {
91    pub fn as_str(self) -> &'static str {
92        match self {
93            Self::EmbeddedApi => "embedded API",
94            Self::Sql => "SQL front door",
95            Self::Distributed => "distributed runtime",
96        }
97    }
98}
99
100/// A single entry in the Krishiv SQL feature matrix, dimensioned per engine.
101#[derive(Debug, Clone)]
102pub struct FeatureEntry {
103    /// Stable identifier (e.g. `"select.distinct"`).
104    pub id: &'static str,
105    /// Broad feature category (e.g. `"SELECT"`, `"JOIN"`, `"DML"`).
106    pub category: &'static str,
107    /// Human-readable description.
108    pub description: &'static str,
109    /// Support status in the batch engine.
110    pub batch: FeatureStatus,
111    /// Support status in the streaming (continuous) engine.
112    pub streaming: FeatureStatus,
113    /// Support status in the incremental (IVM) engine.
114    pub incremental: FeatureStatus,
115    /// Optional clarifying note (gap description, limitations, workarounds).
116    pub note: Option<&'static str>,
117    /// Where the feature actually executes. `None` means the feature is
118    /// reachable from every surface its engine cells imply (the common case).
119    /// `Some(..)` restricts the claim — e.g. embedded-API-only operators —
120    /// and is rendered into the generated reference page.
121    pub placement: Option<&'static [Placement]>,
122}
123
124impl FeatureEntry {
125    const fn new(
126        id: &'static str,
127        category: &'static str,
128        description: &'static str,
129        batch: FeatureStatus,
130        streaming: FeatureStatus,
131        incremental: FeatureStatus,
132    ) -> Self {
133        Self {
134            id,
135            category,
136            description,
137            batch,
138            streaming,
139            incremental,
140            note: None,
141            placement: None,
142        }
143    }
144
145    /// A batch-only feature: `n/a` in the streaming and incremental engines.
146    const fn batch_only(
147        id: &'static str,
148        category: &'static str,
149        description: &'static str,
150        batch: FeatureStatus,
151    ) -> Self {
152        Self::new(id, category, description, batch, NA, NA)
153    }
154
155    const fn with_note(mut self, note: &'static str) -> Self {
156        self.note = Some(note);
157        self
158    }
159
160    /// Restrict where this feature executes (placement honesty, audit §9b).
161    const fn placed(mut self, placement: &'static [Placement]) -> Self {
162        self.placement = Some(placement);
163        self
164    }
165
166    /// True when the entry's execution surface is restricted below what its
167    /// engine cells imply (i.e. it carries an explicit placement set that
168    /// excludes at least one surface).
169    pub fn placement_restricted(&self) -> bool {
170        matches!(self.placement, Some(p) if p.len() < 3)
171    }
172
173    /// Status for a given engine.
174    pub fn status_for(&self, engine: Engine) -> FeatureStatus {
175        match engine {
176            Engine::Batch => self.batch,
177            Engine::Streaming => self.streaming,
178            Engine::Incremental => self.incremental,
179        }
180    }
181}
182
183impl std::fmt::Display for FeatureEntry {
184    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
185        write!(
186            f,
187            "[batch:{} streaming:{} incremental:{}] {} — {}",
188            self.batch, self.streaming, self.incremental, self.id, self.description
189        )?;
190        if let Some(note) = self.note {
191            write!(f, " ({note})")?;
192        }
193        Ok(())
194    }
195}
196
197// ── Feature matrix ────────────────────────────────────────────────────────────
198
199/// Return the complete Krishiv SQL feature matrix.
200pub fn feature_matrix() -> &'static [FeatureEntry] {
201    FEATURES
202}
203
204/// Return only entries matching `category` (case-insensitive prefix match).
205pub fn features_for_category(category: &str) -> Vec<&'static FeatureEntry> {
206    let cat_upper = category.to_uppercase();
207    FEATURES
208        .iter()
209        .filter(|e| e.category.to_uppercase().starts_with(&cat_upper))
210        .collect()
211}
212
213/// Return entries whose **batch** status equals `status`. (Batch is the
214/// primary column; use [`FeatureEntry::status_for`] for the other engines.)
215pub fn features_by_status(status: FeatureStatus) -> Vec<&'static FeatureEntry> {
216    FEATURES.iter().filter(|e| e.batch == status).collect()
217}
218
219/// Render the public "Krishiv SQL feature matrix" reference page from the
220/// matrix. This is the single source of truth; the checked-in markdown is
221/// regenerated from here and drift-guarded in CI.
222pub fn generate_reference_markdown() -> String {
223    let mut out = String::new();
224    out.push_str("# Krishiv SQL feature matrix\n\n");
225    out.push_str(
226        "_Generated from `krishiv-sql/src/grammar.rs` — do not edit by hand._\n\n\
227         Each feature is dimensioned across the three Krishiv execution engines: \
228         **batch** (DataFusion + extensions), **streaming** (continuous windows), \
229         and **incremental** (IVM). `n/a` means the feature does not apply to that \
230         engine.\n\n",
231    );
232
233    // Stable category order = order of first appearance in the matrix.
234    let mut categories: Vec<&'static str> = Vec::new();
235    for e in FEATURES {
236        if !categories.contains(&e.category) {
237            categories.push(e.category);
238        }
239    }
240
241    for cat in categories {
242        out.push_str(&format!("## {cat}\n\n"));
243        out.push_str("| Feature | Description | Batch | Streaming | Incremental | Notes |\n");
244        out.push_str("|---|---|---|---|---|---|\n");
245        for e in FEATURES.iter().filter(|e| e.category == cat) {
246            let mut notes = String::new();
247            if let Some(placement) = e.placement {
248                let surfaces: Vec<&str> = placement.iter().map(|p| p.as_str()).collect();
249                notes.push_str(&format!("**placement: {} only.** ", surfaces.join(" + ")));
250            }
251            notes.push_str(e.note.unwrap_or(""));
252            out.push_str(&format!(
253                "| `{}` | {} | {} | {} | {} | {} |\n",
254                e.id,
255                e.description,
256                e.batch,
257                e.streaming,
258                e.incremental,
259                notes.trim_end()
260            ));
261        }
262        out.push('\n');
263    }
264
265    // Placement honesty (audit §9b): the embedded-only operator tier has no SQL
266    // spellings, so it has no rows above — but a user who builds on these
267    // operators embedded cannot run that job distributed, and this page must
268    // say so rather than stay silent. (Operators that DO have a SQL matrix row,
269    // like streaming.dedup, carry the placement marker on their row instead.)
270    out.push_str("## Embedded-API-only streaming operators\n\n");
271    out.push_str(
272        "These operators are real and tested but reachable **only** from the embedded \
273         Rust `StreamingDataFrame`/process API and its Python mirror — they are not \
274         compiled from SQL, and the distributed `stream:loop` runtime executes only \
275         `WindowExecutionSpec` shapes (windows, window-join, CEP). A job built on them \
276         cannot run distributed today.\n\n",
277    );
278    for (name, desc) in EMBEDDED_ONLY_OPERATORS {
279        out.push_str(&format!("- **{name}** — {desc}\n"));
280    }
281    out.push('\n');
282    out
283}
284
285/// The embedded-API-only streaming operator tier (audit §9b). These have no
286/// SQL spelling, so they carry no [`FeatureEntry`] row; they are published on
287/// the generated reference page so the placement restriction is stated
288/// somewhere a user will read it.
289pub const EMBEDDED_ONLY_OPERATORS: &[(&str, &str)] = &[
290    (
291        "temporal join",
292        "event-time temporal (versioned lookup) join between two streams",
293    ),
294    (
295        "side outputs",
296        "route late/rejected/tagged rows to a secondary stream",
297    ),
298    (
299        "broadcast state",
300        "low-volume control stream broadcast to all tasks of a keyed stream",
301    ),
302    (
303        "connected streams",
304        "two-input operators sharing state across both inputs",
305    ),
306    (
307        "ProcessFunction + timers",
308        "per-key user logic with registered event/processing-time timers",
309    ),
310];
311
312/// Render the "Krishiv SQL vs Spark SQL" dialect honesty / migration page from
313/// the matrix + the documented per-feature semantic differences. Generated,
314/// never hand-written: what maps 1:1, what differs semantically, what is absent.
315pub fn generate_honesty_markdown() -> String {
316    let mut out = String::new();
317    out.push_str("# Krishiv SQL vs Spark SQL\n\n");
318    out.push_str(
319        "_Generated from `krishiv-sql/src/grammar.rs` — do not edit by hand._\n\n\
320         Krishiv targets Spark-SQL reference parity as a **measured** number. This page \
321         is the honest ledger: what maps 1:1, what differs semantically, and what is \
322         absent, derived from the feature matrix.\n\n",
323    );
324
325    // Documented semantic differences (the correctness-trap items an alias layer
326    // must surface loudly rather than silently mis-behave on).
327    out.push_str("## Documented semantic differences\n\n");
328    out.push_str(
329        "- **`date_format(ts, fmt)` pattern letters.** Krishiv uses **Spark/Java** \
330         `DateTimeFormatter` letters (`yyyy-MM-dd`), not chrono/strftime (`%Y-%m-%d`). \
331         Supported letters translate exactly; unsupported letters (era `G`, timezone \
332         `z`/`X`) raise a clear error instead of emitting wrong output.\n\
333         - **`exists(array, x -> …)`.** The `exists(` spelling is shadowed by the \
334         EXISTS-subquery keyword in the parser; use `any_match(array, x -> …)` (the \
335         byte-identical implementation) for the Spark higher-order `exists`.\n\
336         - **Lambda / array-literal syntax.** The SQL front door parses with a \
337         lambda-capable dialect so `transform(arr, x -> …)` and `[1, 2, 3]` work; \
338         the array constructor is `make_array(...)` / `[...]` (Spark's `array(...)` \
339         maps to these).\n\
340         - **ANSI mode, integral division, NULL ordering** follow DataFusion \
341         semantics, which match Spark ANSI mode for the covered surface; divergences \
342         are tracked as matrix notes.\n\n",
343    );
344
345    // 1:1 — supported in batch with no divergence note.
346    out.push_str("## Maps 1:1 (supported, no semantic caveat)\n\n");
347    for e in FEATURES
348        .iter()
349        .filter(|e| e.batch == FeatureStatus::Supported && e.note.is_none())
350    {
351        out.push_str(&format!("- `{}` — {}\n", e.id, e.description));
352    }
353    out.push('\n');
354
355    // Partial / caveated.
356    out.push_str("## Supported with caveats (partial or noted)\n\n");
357    for e in FEATURES.iter().filter(|e| {
358        (e.batch == FeatureStatus::Partial) || (e.batch.is_claimed() && e.note.is_some())
359    }) {
360        out.push_str(&format!(
361            "- `{}` — {} _({})_\n",
362            e.id,
363            e.description,
364            e.note.unwrap_or("partial")
365        ));
366    }
367    out.push('\n');
368
369    // Absent / planned.
370    out.push_str("## Absent (planned — itemized shortfall)\n\n");
371    for e in FEATURES.iter().filter(|e| {
372        e.batch == FeatureStatus::Planned
373            && e.streaming != FeatureStatus::Supported
374            && e.incremental != FeatureStatus::Supported
375    }) {
376        out.push_str(&format!(
377            "- `{}` — {} _({})_\n",
378            e.id,
379            e.description,
380            e.note.unwrap_or("planned")
381        ));
382    }
383    out.push('\n');
384    out
385}
386
387const S: FeatureStatus = FeatureStatus::Supported;
388const P: FeatureStatus = FeatureStatus::Partial;
389const PL: FeatureStatus = FeatureStatus::Planned;
390const NA: FeatureStatus = FeatureStatus::NotApplicable;
391
392static FEATURES: &[FeatureEntry] = &[
393    // ── SELECT ────────────────────────────────────────────────────────────────
394    // The shared relational core: fully supported in batch; in streaming it is
395    // usable only inside the windowed continuous plan (Partial); IVM maintains
396    // it via krishiv-delta with a DiffBased recompute fallback (Partial).
397    FeatureEntry::new("select.projection", "SELECT", "Column projection and aliases", S, P, P),
398    FeatureEntry::new("select.star", "SELECT", "SELECT * expansion", S, P, P),
399    FeatureEntry::new("select.distinct", "SELECT", "SELECT DISTINCT deduplication", S, NA, P),
400    FeatureEntry::new("select.where", "SELECT", "WHERE predicate filtering", S, P, P),
401    FeatureEntry::new(
402        "select.order_by",
403        "SELECT",
404        "ORDER BY with ASC/DESC and NULLS FIRST/LAST",
405        S,
406        NA,
407        NA,
408    )
409    .with_note("streaming/IVM: unbounded ordering is not a maintainable operator"),
410    FeatureEntry::new("select.limit_offset", "SELECT", "LIMIT / OFFSET pagination", S, NA, NA),
411    FeatureEntry::new("select.having", "SELECT", "HAVING post-aggregation filter", S, P, P),
412    FeatureEntry::new(
413        "select.case",
414        "SELECT",
415        "CASE WHEN … THEN … ELSE … END expressions",
416        S,
417        P,
418        P,
419    ),
420    FeatureEntry::new("select.cast", "SELECT", "CAST(expr AS type) and TRY_CAST", S, P, P),
421    FeatureEntry::new(
422        "select.subquery_scalar",
423        "SELECT",
424        "Scalar subqueries in projection/predicate",
425        S,
426        NA,
427        P,
428    ),
429    FeatureEntry::new(
430        "select.subquery_exists",
431        "SELECT",
432        "EXISTS / NOT EXISTS correlated subqueries",
433        S,
434        NA,
435        P,
436    ),
437    FeatureEntry::new("select.subquery_in", "SELECT", "IN / NOT IN subqueries", S, NA, P),
438    FeatureEntry::new("select.values", "SELECT", "VALUES clause for inline data", S, NA, P),
439    // ── GROUP BY ─────────────────────────────────────────────────────────────
440    FeatureEntry::new("groupby.basic", "GROUP BY", "Basic GROUP BY column list", S, P, P)
441        .with_note("streaming: only inside a window TVF (windowed aggregation)"),
442    FeatureEntry::new("groupby.rollup", "GROUP BY", "ROLLUP grouping sets", S, NA, P),
443    FeatureEntry::new("groupby.cube", "GROUP BY", "CUBE grouping sets", S, NA, P),
444    FeatureEntry::new("groupby.grouping_sets", "GROUP BY", "Explicit GROUPING SETS", S, NA, P),
445    FeatureEntry::new(
446        "groupby.grouping_function",
447        "GROUP BY",
448        "GROUPING() function for NULL disambiguation",
449        S,
450        NA,
451        P,
452    ),
453    // ── JOIN ─────────────────────────────────────────────────────────────────
454    FeatureEntry::new("join.inner", "JOIN", "INNER JOIN (equi and non-equi)", S, NA, P),
455    FeatureEntry::new("join.left_outer", "JOIN", "LEFT OUTER JOIN", S, NA, P),
456    FeatureEntry::new("join.right_outer", "JOIN", "RIGHT OUTER JOIN", S, NA, P),
457    FeatureEntry::new("join.full_outer", "JOIN", "FULL OUTER JOIN", S, NA, P),
458    FeatureEntry::new("join.cross", "JOIN", "CROSS JOIN", S, NA, P),
459    FeatureEntry::new("join.natural", "JOIN", "NATURAL JOIN (column-name matching)", S, NA, P),
460    FeatureEntry::new("join.using", "JOIN", "JOIN … USING (column_list)", S, NA, P),
461    FeatureEntry::new("join.lateral", "JOIN", "LATERAL JOIN / CROSS JOIN LATERAL", S, NA, NA),
462    FeatureEntry::new(
463        "join.interval",
464        "JOIN",
465        "Streaming interval join on event-time bounds",
466        PL,
467        P,
468        NA,
469    )
470    .with_note(
471        "DataFrame-only today (audit §9b): the interval-join operator has no SQL planning path, \
472         so batch SQL cannot express it (Planned); the streaming operator exists (Partial). \
473         Corrected from the prior over-claim of batch Supported.",
474    ),
475    FeatureEntry::new(
476        "join.temporal_as_of",
477        "JOIN",
478        "Temporal AS OF point-in-time join",
479        PL,
480        NA,
481        NA,
482    )
483    .with_note(
484        "no SQL temporal-join planning path: `lakehouse/as_of.rs` is table time-travel \
485         (temporal.as_of), not a temporal join. Marked Planned rather than the prior Supported.",
486    ),
487    FeatureEntry::new(
488        "join.broadcast_hint",
489        "JOIN",
490        "/*+ BROADCAST(t) */ optimizer hint",
491        P,
492        NA,
493        NA,
494    )
495    .with_note("hint parsed and recorded; broadcast decision is cost-based (see hints.* entries)"),
496    // ── HINTS (Phase 60 statement completion) ───────────────────────────────
497    FeatureEntry::new(
498        "hints.join_strategy",
499        "HINTS",
500        "/*+ MERGE|SHUFFLE_HASH|BROADCAST(t) */ join-strategy hints",
501        P,
502        NA,
503        NA,
504    )
505    .with_note("parsed always; honored where the executor supports the strategy (Phase 52/54), recorded either way"),
506    FeatureEntry::new(
507        "hints.repartition",
508        "HINTS",
509        "/*+ REPARTITION(n)|COALESCE(n) */ partitioning hints",
510        P,
511        NA,
512        NA,
513    )
514    .with_note("parsed and recorded; applied where the distributed planner supports it"),
515    // ── WINDOW FUNCTIONS ─────────────────────────────────────────────────────
516    FeatureEntry::new("window.over", "WINDOW", "OVER () window function clauses", S, NA, NA),
517    FeatureEntry::new("window.partition_by", "WINDOW", "PARTITION BY inside OVER", S, NA, NA),
518    FeatureEntry::new("window.order_by", "WINDOW", "ORDER BY inside OVER", S, NA, NA),
519    FeatureEntry::new("window.rows_range", "WINDOW", "ROWS / RANGE frame specification", S, NA, NA),
520    FeatureEntry::new(
521        "window.rank_dense_rank",
522        "WINDOW",
523        "RANK(), DENSE_RANK(), ROW_NUMBER()",
524        S,
525        NA,
526        NA,
527    ),
528    FeatureEntry::new("window.lead_lag", "WINDOW", "LEAD() and LAG()", S, NA, NA),
529    FeatureEntry::new(
530        "window.first_last_value",
531        "WINDOW",
532        "FIRST_VALUE() and LAST_VALUE()",
533        S,
534        NA,
535        NA,
536    ),
537    FeatureEntry::new("window.nth_value", "WINDOW", "NTH_VALUE()", S, NA, NA),
538    FeatureEntry::new("window.ntile", "WINDOW", "NTILE(n)", S, NA, NA),
539    FeatureEntry::new(
540        "window.cume_dist_percent",
541        "WINDOW",
542        "CUME_DIST() and PERCENT_RANK()",
543        S,
544        NA,
545        NA,
546    ),
547    FeatureEntry::new(
548        "window.tumble",
549        "WINDOW",
550        "TUMBLE(col, interval) streaming window",
551        S,
552        S,
553        P,
554    )
555    .with_note("batch rewrites the TVF to scalar UDFs (streaming_tvf.rs); streaming compiles it natively"),
556    FeatureEntry::new("window.hop", "WINDOW", "HOP(col, slide, size) sliding window", S, S, P),
557    FeatureEntry::new("window.session", "WINDOW", "Session window on inactivity gap", S, S, NA),
558    // ── CTE ──────────────────────────────────────────────────────────────────
559    FeatureEntry::new("cte.non_recursive", "CTE", "WITH … AS (…) non-recursive CTEs", S, P, P),
560    FeatureEntry::new(
561        "cte.recursive",
562        "CTE",
563        "WITH RECURSIVE … (UNION ALL base + recursive)",
564        S,
565        NA,
566        NA,
567    ),
568    FeatureEntry::new("cte.multiple", "CTE", "Multiple CTEs in one WITH clause", S, P, P),
569    // ── SET OPERATIONS ────────────────────────────────────────────────────────
570    FeatureEntry::new("set.union_all", "SET", "UNION ALL", S, P, P),
571    FeatureEntry::new("set.union_distinct", "SET", "UNION (DISTINCT)", S, NA, P),
572    FeatureEntry::new("set.intersect", "SET", "INTERSECT", S, NA, P),
573    FeatureEntry::new("set.except", "SET", "EXCEPT", S, NA, P),
574    // ── LATERAL / UNNEST ─────────────────────────────────────────────────────
575    FeatureEntry::batch_only("lateral.unnest", "LATERAL", "UNNEST(array_col) in FROM clause", S),
576    FeatureEntry::batch_only(
577        "lateral.generate_series",
578        "LATERAL",
579        "generate_series() table function",
580        S,
581    ),
582    FeatureEntry::batch_only(
583        "lateral.cross_join_unnest",
584        "LATERAL",
585        "CROSS JOIN UNNEST(…) AS t(col)",
586        S,
587    ),
588    // ── PIVOT / UNPIVOT ───────────────────────────────────────────────────────
589    FeatureEntry::batch_only("pivot.pivot", "PIVOT", "PIVOT(agg FOR col IN (v1, v2, …))", S),
590    FeatureEntry::batch_only("pivot.unpivot", "PIVOT", "UNPIVOT(value FOR col IN (c1, c2, …))", S),
591    // ── FUNCTIONS: JSON (Phase 60) ───────────────────────────────────────────
592    FeatureEntry::batch_only(
593        "functions.json.get_json_object",
594        "FUNCTIONS",
595        "get_json_object(json, path) Spark JSONPath extraction",
596        S,
597    ),
598    FeatureEntry::batch_only(
599        "functions.json.json_array_length",
600        "FUNCTIONS",
601        "json_array_length(json) top-level array element count",
602        S,
603    ),
604    FeatureEntry::batch_only(
605        "functions.json.from_to_json",
606        "FUNCTIONS",
607        "from_json / to_json struct⇄JSON conversion",
608        PL,
609    )
610    .with_note(
611        "requires a typed arrow⇄JSON converter + a Spark-DDL schema parser with Spark's \
612         version-specific null-field/timestamp rules; itemized shortfall, not shipped approximate",
613    ),
614    FeatureEntry::batch_only(
615        "functions.json.json_tuple",
616        "FUNCTIONS",
617        "json_tuple(json, k1, k2, …) multi-key extraction (generator)",
618        PL,
619    )
620    .with_note("needs table-generating/LATERAL VIEW machinery; use get_json_object per key today"),
621    FeatureEntry::batch_only(
622        "functions.json.schema_of_json",
623        "FUNCTIONS",
624        "schema_of_json(json) infer a DDL schema string",
625        PL,
626    ),
627    // ── FUNCTIONS: higher-order array lambdas (Phase 60) ─────────────────────
628    FeatureEntry::batch_only(
629        "functions.hof.transform",
630        "FUNCTIONS",
631        "transform(array, x -> …) — Spark alias for array_transform",
632        S,
633    ),
634    FeatureEntry::batch_only(
635        "functions.hof.filter",
636        "FUNCTIONS",
637        "filter(array, x -> …) — Spark alias for array_filter",
638        S,
639    ),
640    FeatureEntry::batch_only(
641        "functions.hof.exists",
642        "FUNCTIONS",
643        "exists / any_match(array, x -> …) predicate-any",
644        P,
645    )
646    .with_note(
647        "any_match is reachable; the `exists(...)` spelling is shadowed by the EXISTS-subquery \
648         keyword in the parser (documented dialect difference)",
649    ),
650    FeatureEntry::batch_only(
651        "functions.hof.forall",
652        "FUNCTIONS",
653        "forall(array, x -> …) predicate-all (new, exact all-match)",
654        S,
655    ),
656    FeatureEntry::batch_only(
657        "functions.hof.aggregate_zip_map",
658        "FUNCTIONS",
659        "aggregate/reduce, zip_with, map_filter, transform_keys/values",
660        PL,
661    )
662    .with_note("require DataFusion's multi-step lambda / map-lambda protocol; itemized shortfall"),
663    // ── FUNCTIONS: Spark scalar alias layer (Phase 60) ───────────────────────
664    FeatureEntry::batch_only(
665        "functions.spark.nvl",
666        "FUNCTIONS",
667        "nvl / nvl2 null-coalescing (DataFusion-native, exact)",
668        S,
669    ),
670    FeatureEntry::batch_only(
671        "functions.spark.substring_index",
672        "FUNCTIONS",
673        "substring_index(str, delim, count) (DataFusion-native, exact)",
674        S,
675    ),
676    FeatureEntry::batch_only(
677        "functions.spark.date_format",
678        "FUNCTIONS",
679        "date_format(ts, fmt) with **Spark** pattern letters (yyyy-MM-dd)",
680        S,
681    )
682    .with_note(
683        "supported Spark pattern letters translate exactly to chrono; unsupported letters \
684         (era/timezone) error clearly rather than emitting wrong output. Differs from \
685         DataFusion's chrono-pattern date_format — see honesty page.",
686    ),
687    FeatureEntry::batch_only(
688        "functions.spark.crc32",
689        "FUNCTIONS",
690        "crc32(expr) IEEE CRC-32 as BIGINT (exact)",
691        S,
692    ),
693    FeatureEntry::batch_only(
694        "functions.spark.hash_generators",
695        "FUNCTIONS",
696        "xxhash64, stack, posexplode, inline",
697        PL,
698    )
699    .with_note(
700        "xxhash64 needs byte-exact replication of Spark's seed-42 typed hashing; \
701         stack/posexplode/inline need generator machinery — itemized shortfall",
702    ),
703    // ── DML ──────────────────────────────────────────────────────────────────
704    FeatureEntry::batch_only("dml.copy_to", "DML", "COPY (query) TO 'path' (FORMAT …)", S)
705        .with_note("inherited from DataFusion's native parser/planner; no Krishiv-side code involved"),
706    FeatureEntry::new("dml.insert_into", "DML", "INSERT INTO table SELECT …", S, NA, NA),
707    FeatureEntry::batch_only(
708        "dml.insert_overwrite",
709        "DML",
710        "INSERT OVERWRITE (full partition replace)",
711        S,
712    ),
713    FeatureEntry::batch_only("dml.delete", "DML", "DELETE FROM table WHERE …", P)
714        .with_note("supported on Iceberg tables; in-memory and Parquet tables require rewrite"),
715    FeatureEntry::batch_only("dml.update", "DML", "UPDATE table SET col = … WHERE …", P)
716        .with_note("supported on Iceberg tables via MERGE rewrite"),
717    FeatureEntry::batch_only(
718        "dml.merge",
719        "DML",
720        "MERGE INTO target USING source ON … WHEN MATCHED …",
721        S,
722    ),
723    FeatureEntry::batch_only(
724        "dml.iceberg_merge",
725        "DML",
726        "Atomic Iceberg MERGE with row-level deletes",
727        S,
728    ),
729    FeatureEntry::batch_only("dml.truncate", "DML", "TRUNCATE TABLE (Iceberg + memory)", PL)
730        .with_note("itemized shortfall: TRUNCATE is not yet wired for memory/Iceberg session tables"),
731    // ── DDL ──────────────────────────────────────────────────────────────────
732    FeatureEntry::batch_only(
733        "ddl.create_external_table",
734        "DDL",
735        "CREATE EXTERNAL TABLE … STORED AS …",
736        S,
737    ),
738    FeatureEntry::batch_only("ddl.create_view", "DDL", "CREATE VIEW name AS SELECT …", S),
739    FeatureEntry::batch_only(
740        "ddl.create_function",
741        "DDL",
742        "CREATE FUNCTION … LANGUAGE SQL|PYTHON",
743        S,
744    ),
745    FeatureEntry::batch_only("ddl.drop_table", "DDL", "DROP TABLE [IF EXISTS]", S),
746    FeatureEntry::batch_only("ddl.drop_view", "DDL", "DROP VIEW [IF EXISTS]", S),
747    FeatureEntry::batch_only("ddl.create_table_as", "DDL", "CREATE TABLE … AS SELECT (CTAS)", S)
748        .with_note("durable Iceberg landing (G17) when the target resolves to a registered Iceberg catalog; session table otherwise"),
749    FeatureEntry::batch_only(
750        "ddl.partitioned_by",
751        "DDL",
752        "CREATE TABLE … PARTITIONED BY (col | bucket/truncate/year/month/day/hour(col)) AS SELECT",
753        S,
754    )
755    .with_note("Iceberg catalog tables only; transforms follow the Iceberg partition spec"),
756    FeatureEntry::batch_only("ddl.alter_table", "DDL", "ALTER TABLE ADD/DROP COLUMN, RENAME", P)
757        .with_note("Iceberg schema evolution via ALTER TABLE is supported"),
758    FeatureEntry::batch_only("ddl.create_schema", "DDL", "CREATE SCHEMA name", S)
759        .with_note("inherited from DataFusion's native catalog; no Krishiv-side code involved"),
760    FeatureEntry::new(
761        "ddl.create_materialized_view",
762        "DDL",
763        "CREATE [OR REPLACE] MATERIALIZED VIEW … AS SELECT → IVM view (REFRESH/DROP)",
764        NA,
765        NA,
766        S,
767    )
768    .with_note(
769        "Phase 60 SQL-DDL-for-IVM: ANSI/Spark synonym routed onto the same IVM engine as \
770         CREATE MATERIALIZED INCREMENTAL VIEW; REFRESH/DROP MATERIALIZED VIEW lifecycle; \
771         engine primitive under the platform's governed pipelines",
772    ),
773    FeatureEntry::new(
774        "ddl.create_streaming_table",
775        "DDL",
776        "CREATE [OR REPLACE] STREAMING TABLE … AS SELECT → continuous job",
777        NA,
778        PL,
779        NA,
780    )
781    .with_note(
782        "Phase 60: SQL front door + planner validation land (the body lowers through the shared \
783         streaming compiler); continuous-job execution is coordinator-gated — a cluster-attached \
784         session submits the validated plan via the continuous-stream registration API",
785    ),
786    FeatureEntry::batch_only(
787        "ddl.live_table",
788        "DDL",
789        "CREATE / REFRESH / DROP LIVE TABLE via session.sql()",
790        S,
791    ),
792    // ── CONNECTOR DDL (Phase 60) ─────────────────────────────────────────────
793    FeatureEntry::batch_only(
794        "ddl.connector_source_sink",
795        "DDL",
796        "CREATE SOURCE/SINK … WITH (connector=…) resolved through the connector registry",
797        P,
798    )
799    .with_note(
800        "registry-backed dispatch replacing the parquet-only hardcoded factory (audit §8b); \
801         supported kinds come from connector descriptors, unsupported kinds fail loudly",
802    ),
803    // ── SESSION / CONFIG STATEMENTS (Phase 60) ───────────────────────────────
804    FeatureEntry::batch_only("stmt.set_reset", "SESSION", "SET / RESET / SET TIMEZONE session config", S)
805        .with_note("DataFusion-native session config"),
806    FeatureEntry::batch_only("stmt.use", "SESSION", "USE [CATALOG|SCHEMA] current-namespace", S)
807        .with_note("Phase 60: mutates the session default catalog/schema"),
808    FeatureEntry::batch_only(
809        "stmt.cache",
810        "SESSION",
811        "CACHE / UNCACHE / CLEAR CACHE TABLE (session materialization)",
812        PL,
813    )
814    .with_note("itemized shortfall: needs a session-scoped materialization + provider swap/restore"),
815    // ── SHOW / DESCRIBE (Phase 60) ───────────────────────────────────────────
816    FeatureEntry::batch_only(
817        "show.tables_databases_functions",
818        "SHOW",
819        "SHOW TABLES | DATABASES | SCHEMAS | FUNCTIONS | COLUMNS",
820        P,
821    )
822    .with_note(
823        "TABLES/FUNCTIONS/COLUMNS are DataFusion-native; DATABASES/SCHEMAS added in Phase 60 \
824         (information_schema.schemata). SHOW PARTITIONS (Iceberg) and SHOW VIEWS remain the gap.",
825    ),
826    FeatureEntry::batch_only(
827        "describe.function_database_query",
828        "DESCRIBE",
829        "DESCRIBE FUNCTION | DATABASE | QUERY",
830        PL,
831    )
832    .with_note("DESCRIBE <table> is native; FUNCTION/DATABASE/QUERY are the itemized shortfall"),
833    // ── TEMPORAL ─────────────────────────────────────────────────────────────
834    FeatureEntry::batch_only("temporal.as_of", "TEMPORAL", "AS OF TIMESTAMP point-in-time queries", S),
835    FeatureEntry::new(
836        "temporal.match_recognize",
837        "TEMPORAL",
838        "MATCH_RECOGNIZE pattern matching over ordered rows",
839        P,
840        P,
841        NA,
842    )
843    .with_note(
844        "streaming CEP subset: PARTITION BY / ORDER BY / PATTERN (…) / WITHIN <duration>; \
845         DEFINE (pattern-variable predicates) and MEASURES (computed output) clauses are the \
846         remaining gap vs Oracle/Flink's full grammar",
847    ),
848    FeatureEntry::batch_only(
849        "temporal.system_time",
850        "TEMPORAL",
851        "FOR SYSTEM_TIME AS OF (Iceberg time-travel)",
852        P,
853    )
854    .with_note("alias for AS OF on Iceberg tables"),
855    // ── PREPARED STATEMENTS ───────────────────────────────────────────────────
856    FeatureEntry::batch_only(
857        "prepared.create",
858        "PREPARED",
859        "CREATE PREPARED STATEMENT via Flight SQL action",
860        S,
861    ),
862    FeatureEntry::batch_only("prepared.execute", "PREPARED", "Execute prepared statement by handle", S),
863    FeatureEntry::batch_only(
864        "prepared.close",
865        "PREPARED",
866        "CLOSE PREPARED STATEMENT to release server memory",
867        S,
868    ),
869    FeatureEntry::batch_only(
870        "prepared.parameters",
871        "PREPARED",
872        "Positional parameter binding ($1, $2, …)",
873        S,
874    )
875    .with_note("local PreparedStatement::bind and Flight SQL DoPut parameter batches"),
876    FeatureEntry::batch_only(
877        "prepared.sql_text",
878        "PREPARED",
879        "PREPARE name AS …; EXECUTE name(…); DEALLOCATE name",
880        S,
881    )
882    .with_note("inherited from DataFusion's native parser/planner (session-scoped named plans)"),
883    // ── OPERATION CONTROL ────────────────────────────────────────────────────
884    FeatureEntry::new("operation.id", "OPERATION", "Operation IDs for query tracking", S, S, S),
885    FeatureEntry::new("operation.cancel", "OPERATION", "Cancel a running operation by ID", S, S, S),
886    FeatureEntry::new("operation.timeout", "OPERATION", "Per-query execution timeout", S, NA, NA),
887    FeatureEntry::new(
888        "operation.progress",
889        "OPERATION",
890        "Query progress reporting via QueryHandle",
891        S,
892        S,
893        S,
894    ),
895    // ── ERROR HANDLING ────────────────────────────────────────────────────────
896    FeatureEntry::batch_only("error.sqlstate", "ERROR", "SQLSTATE codes on error responses", S),
897    FeatureEntry::batch_only("error.error_position", "ERROR", "Source line/column in error messages", P)
898        .with_note("DataFusion provides message but not structured position"),
899    // ── FLIGHT SQL ────────────────────────────────────────────────────────────
900    FeatureEntry::batch_only(
901        "flight.get_flight_info",
902        "FLIGHT SQL",
903        "GetFlightInfo for statement execution",
904        S,
905    ),
906    FeatureEntry::batch_only("flight.do_get", "FLIGHT SQL", "DoGet streaming result delivery", S),
907    FeatureEntry::batch_only(
908        "flight.prepared_statements",
909        "FLIGHT SQL",
910        "Prepared statement create/execute/close",
911        S,
912    ),
913    FeatureEntry::batch_only(
914        "flight.do_action",
915        "FLIGHT SQL",
916        "DoAction for custom Krishiv operations",
917        S,
918    ),
919    FeatureEntry::batch_only(
920        "flight.get_sql_info",
921        "FLIGHT SQL",
922        "GetSqlInfo capability introspection",
923        S,
924    ),
925    FeatureEntry::batch_only("flight.auth", "FLIGHT SQL", "Bearer token authentication", S),
926    FeatureEntry::batch_only("flight.policy", "FLIGHT SQL", "Table-level access policy enforcement", S),
927    FeatureEntry::batch_only(
928        "flight.transactions",
929        "FLIGHT SQL",
930        "BEGIN/COMMIT/ROLLBACK transactions",
931        P,
932    )
933    .with_note("Flight SQL BeginTransaction/EndTransaction actions; SQL BEGIN/COMMIT not routed"),
934    FeatureEntry::batch_only(
935        "flight.schemas",
936        "FLIGHT SQL",
937        "GetDbSchemas / GetTables catalog introspection",
938        P,
939    )
940    .with_note("tables listed via Krishiv catalog; schema introspection via get_sql_info"),
941    // ── STREAMING SQL ─────────────────────────────────────────────────────────
942    FeatureEntry::new(
943        "streaming.continuous_select",
944        "STREAMING",
945        "Continuous SELECT over unbounded input",
946        NA,
947        S,
948        NA,
949    ),
950    FeatureEntry::new(
951        "streaming.window_agg",
952        "STREAMING",
953        "Windowed aggregations over streaming input",
954        NA,
955        S,
956        P,
957    ),
958    FeatureEntry::new(
959        "streaming.watermark",
960        "STREAMING",
961        "Event-time watermarks for late-data handling",
962        NA,
963        S,
964        NA,
965    ),
966    FeatureEntry::new(
967        "streaming.interval_join",
968        "STREAMING",
969        "Streaming-to-streaming interval join",
970        NA,
971        S,
972        NA,
973    )
974    .placed(&[Placement::EmbeddedApi, Placement::Distributed])
975    .with_note(
976        "no SQL planning path; embedded PerKeyIntervalJoin, and distributed only as the \
977         watermark window-join WindowExecutionSpec shape",
978    ),
979    FeatureEntry::new("streaming.cep", "STREAMING", "MATCH_RECOGNIZE CEP over streaming input", NA, S, NA),
980    FeatureEntry::new(
981        "streaming.dedup",
982        "STREAMING",
983        "Streaming deduplication (dropDuplicates)",
984        NA,
985        S,
986        NA,
987    )
988    .placed(&[Placement::EmbeddedApi])
989    .with_note(
990        "embedded API only — not compiled from SQL and not a distributed \
991         stream:loop shape (audit §9b)",
992    ),
993    FeatureEntry::new(
994        "streaming.sink_modes",
995        "STREAMING",
996        "Append / Update / Complete output modes",
997        NA,
998        S,
999        P,
1000    ),
1001    // ── INTROSPECTION ─────────────────────────────────────────────────────────
1002    FeatureEntry::batch_only(
1003        "introspection.describe",
1004        "INTROSPECTION",
1005        "DESCRIBE / DESC / SHOW COLUMNS table schema",
1006        S,
1007    ),
1008    FeatureEntry::batch_only(
1009        "introspection.explain",
1010        "INTROSPECTION",
1011        "EXPLAIN [LOGICAL|PHYSICAL|ANALYZE] query plans",
1012        S,
1013    ),
1014    FeatureEntry::batch_only(
1015        "introspection.information_schema",
1016        "INTROSPECTION",
1017        "information_schema.{tables,columns,views,df_settings,routines,parameters,schemata}",
1018        S,
1019    ),
1020];
1021
1022#[cfg(test)]
1023mod tests {
1024    use super::*;
1025
1026    #[test]
1027    fn feature_matrix_is_non_empty() {
1028        assert!(!feature_matrix().is_empty());
1029    }
1030
1031    #[test]
1032    fn all_ids_are_unique() {
1033        let mut seen = std::collections::HashSet::new();
1034        for e in feature_matrix() {
1035            assert!(seen.insert(e.id), "duplicate feature id: {}", e.id);
1036        }
1037    }
1038
1039    #[test]
1040    fn every_entry_has_at_least_one_non_na_engine() {
1041        // A row that is n/a in all three engines is meaningless.
1042        for e in feature_matrix() {
1043            assert!(
1044                e.batch != NA || e.streaming != NA || e.incremental != NA,
1045                "feature {} is n/a in every engine",
1046                e.id
1047            );
1048        }
1049    }
1050
1051    #[test]
1052    fn features_for_category_returns_subset() {
1053        let join_features = features_for_category("JOIN");
1054        assert!(!join_features.is_empty());
1055        for f in &join_features {
1056            assert!(f.category.to_uppercase().starts_with("JOIN"), "{}", f.id);
1057        }
1058    }
1059
1060    #[test]
1061    fn features_by_status_supported_is_non_empty() {
1062        assert!(!features_by_status(FeatureStatus::Supported).is_empty());
1063    }
1064
1065    #[test]
1066    fn feature_entry_display_includes_id_and_engines() {
1067        let entry = feature_matrix()
1068            .iter()
1069            .find(|e| e.id == "window.tumble")
1070            .unwrap();
1071        let s = entry.to_string();
1072        assert!(s.contains("window.tumble"));
1073        assert!(s.contains("batch:supported"));
1074        assert!(s.contains("streaming:supported"));
1075    }
1076
1077    #[test]
1078    fn generated_reference_has_all_three_engine_columns() {
1079        let md = generate_reference_markdown();
1080        assert!(md.contains("| Batch | Streaming | Incremental |"));
1081        // Spot-check a per-engine divergence is rendered.
1082        assert!(md.contains("`window.tumble`"));
1083        assert!(md.contains("`functions.hof.forall`"));
1084    }
1085
1086    #[test]
1087    fn drift_check_ctas_is_supported_not_partial() {
1088        // Regression: CTAS was marked Partial after G17 shipped durable Iceberg CTAS.
1089        let ctas = feature_matrix()
1090            .iter()
1091            .find(|e| e.id == "ddl.create_table_as")
1092            .unwrap();
1093        assert_eq!(ctas.batch, FeatureStatus::Supported);
1094    }
1095
1096    #[test]
1097    fn drift_check_interval_join_has_no_batch_sql_path() {
1098        // Regression: join.interval was over-claimed as batch Supported; it is
1099        // DataFrame-only with no SQL planning path (audit §9b).
1100        let ij = feature_matrix()
1101            .iter()
1102            .find(|e| e.id == "join.interval")
1103            .unwrap();
1104        assert_eq!(ij.batch, FeatureStatus::Planned);
1105        assert_eq!(ij.streaming, FeatureStatus::Partial);
1106    }
1107
1108    #[test]
1109    fn embedded_only_operators_carry_the_placement_marker() {
1110        // Regression (audit §9b): "supported" must say WHERE. dedup is
1111        // embedded-API-only; interval join has no SQL planning path and is
1112        // distributed only as the watermark window-join shape.
1113        let dedup = feature_matrix()
1114            .iter()
1115            .find(|e| e.id == "streaming.dedup")
1116            .unwrap();
1117        assert_eq!(dedup.placement, Some(&[Placement::EmbeddedApi][..]));
1118        assert!(dedup.placement_restricted());
1119
1120        let ij = feature_matrix()
1121            .iter()
1122            .find(|e| e.id == "streaming.interval_join")
1123            .unwrap();
1124        let p = ij
1125            .placement
1126            .expect("streaming.interval_join must carry a placement set");
1127        assert!(p.contains(&Placement::EmbeddedApi) && p.contains(&Placement::Distributed));
1128        assert!(
1129            !p.contains(&Placement::Sql),
1130            "no SQL planning path exists for interval join"
1131        );
1132    }
1133
1134    #[test]
1135    fn placement_restricted_entries_always_explain_themselves() {
1136        // A restricted placement without a note is a claim without a reason.
1137        for e in feature_matrix() {
1138            if e.placement_restricted() {
1139                assert!(
1140                    e.note.is_some(),
1141                    "feature {} restricts placement but has no note",
1142                    e.id
1143                );
1144            }
1145        }
1146    }
1147
1148    #[test]
1149    fn generated_reference_renders_placement_and_embedded_only_ledger() {
1150        let md = generate_reference_markdown();
1151        assert!(md.contains("**placement: embedded API only.**"));
1152        assert!(md.contains("**placement: embedded API + distributed runtime only.**"));
1153        assert!(md.contains("## Embedded-API-only streaming operators"));
1154        for (name, _) in EMBEDDED_ONLY_OPERATORS {
1155            assert!(md.contains(name), "embedded-only ledger is missing {name}");
1156        }
1157    }
1158}
krishiv_sql/grammar.rs

krishiv_sql/
grammar.rs
