krishiv_sql/
streaming_table_ddl.rs

1#![forbid(unsafe_code)]
2//! `CREATE STREAMING TABLE <name> AS <select>` — the SQL front door for a
3//! continuous streaming job (Phase 60 "SQL DDL for the other two engines").
4//!
5//! A SQL-only client (Flight SQL / JDBC / BI / the workbench) declares a
6//! continuous job as a table-producing statement, the way Databricks Delta Live
7//! Tables and Flink `CREATE TABLE … AS` do. The body must lower to a continuous
8//! plan through the *same* front door as every other streaming query
9//! ([`crate::streaming_window_plan::compile_streaming_window_sql`]), so an
10//! unsupported body fails at the planner ("cannot lower to a continuous plan"),
11//! not at a bespoke matcher.
12//!
13//! Parsing and validation are engine-local and unit-tested here. **Executing**
14//! the job — placing the operator on the streaming coordinator/executors — needs
15//! a running coordinator, so the pure [`crate::SqlEngine`] surfaces a clear
16//! "requires a streaming coordinator" error and a cluster-attached session
17//! submits the validated plan through the continuous-stream registration API.
18
19/// A parsed `CREATE [OR REPLACE] STREAMING TABLE <name> AS <query>` statement.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct StreamingTableDdl {
22    /// The streaming table (continuous job output) name.
23    pub name: String,
24    /// The windowed streaming `SELECT` that defines the job.
25    pub query: String,
26    /// Whether `OR REPLACE` was given.
27    pub or_replace: bool,
28}
29
30/// Parse a `CREATE [OR REPLACE] STREAMING TABLE <name> AS <query>` statement,
31/// or `None` if `sql` is not one.
32pub fn parse_create_streaming_table(sql: &str) -> Option<StreamingTableDdl> {
33    let trimmed = sql.trim().trim_end_matches(';').trim();
34    let upper = trimmed.to_ascii_uppercase();
35    let (prefix, or_replace) = if upper.starts_with("CREATE OR REPLACE STREAMING TABLE ") {
36        ("CREATE OR REPLACE STREAMING TABLE ", true)
37    } else if upper.starts_with("CREATE STREAMING TABLE ") {
38        ("CREATE STREAMING TABLE ", false)
39    } else {
40        return None;
41    };
42    let rest = trimmed.get(prefix.len()..)?;
43    // `<name> AS <query>` — the first top-level ` AS ` separates them.
44    let as_pos = rest.to_ascii_uppercase().find(" AS ")?;
45    let name = rest.get(..as_pos)?.trim().to_string();
46    let query = rest.get(as_pos + 4..)?.trim().to_string();
47    if name.is_empty() || query.is_empty() {
48        return None;
49    }
50    Some(StreamingTableDdl {
51        name,
52        query,
53        or_replace,
54    })
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60
61    #[test]
62    fn parses_create_streaming_table() {
63        let ddl = parse_create_streaming_table(
64            "CREATE STREAMING TABLE clicks_1m AS \
65             SELECT k, COUNT(*) AS c FROM TUMBLE(TABLE clicks, DESCRIPTOR(ts), 60000) \
66             GROUP BY k, window_start, window_end",
67        )
68        .expect("recognised");
69        assert_eq!(ddl.name, "clicks_1m");
70        assert!(!ddl.or_replace);
71        assert!(ddl.query.to_uppercase().starts_with("SELECT"));
72        assert!(ddl.query.contains("TUMBLE"));
73    }
74
75    #[test]
76    fn parses_or_replace() {
77        let ddl = parse_create_streaming_table("CREATE OR REPLACE STREAMING TABLE t AS SELECT 1")
78            .expect("recognised");
79        assert_eq!(ddl.name, "t");
80        assert!(ddl.or_replace);
81    }
82
83    #[test]
84    fn rejects_non_streaming_table_ddl() {
85        assert!(parse_create_streaming_table("CREATE TABLE t AS SELECT 1").is_none());
86        assert!(parse_create_streaming_table("SELECT 1").is_none());
87        assert!(parse_create_streaming_table("CREATE MATERIALIZED VIEW v AS SELECT 1").is_none());
88    }
89
90    #[test]
91    fn requires_name_and_query() {
92        assert!(parse_create_streaming_table("CREATE STREAMING TABLE t").is_none());
93        assert!(parse_create_streaming_table("CREATE STREAMING TABLE AS SELECT 1").is_none());
94    }
95}
krishiv_sql/streaming_table_ddl.rs

krishiv_sql/
streaming_table_ddl.rs
