krishiv_sql/
pipeline_ddl.rs

1//! Declarative pipeline DDL: `CREATE SOURCE` / `CREATE SINK` / `START PIPELINE`.
2//!
3//! This is the SQL surface for the Tier-2 pipeline layer, following the same
4//! lightweight prefix-match approach as [`crate::incremental_view`]. It is a
5//! **metadata layer only**: it parses and registers source/sink declarations.
6//! Execution of `START PIPELINE` happens in `krishiv-api`, which resolves the
7//! registered specs against a [`Session`](../../krishiv_api/struct.Session.html)
8//! and compiles them to `session.pipeline()…run()`.
9//!
10//! # Grammar
11//!
12//! ```sql
13//! CREATE SOURCE orders AS SELECT * FROM orders_raw;          -- bounded query source
14//! CREATE INCREMENTAL VIEW revenue AS SELECT ... FROM orders; -- transform (see incremental_view)
15//! CREATE SINK revenue_out FROM revenue;                      -- collect view output
16//! START PIPELINE revenue_out;                                -- run; returns sink output
17//! DROP SOURCE orders;  DROP SINK revenue_out;
18//! ```
19
20use std::collections::HashMap;
21use std::sync::RwLock;
22
23use crate::{SqlError, SqlResult};
24
25// ── Parsed statement ────────────────────────────────────────────────────────
26
27/// A parsed pipeline DDL statement.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum PipelineStatement {
30    /// `CREATE SOURCE <name> AS <query>` or `... FROM <CONNECTOR>(...)`.
31    CreateSource { name: String, source: SourceSpec },
32    /// `CREATE SINK <name> FROM <view> [INTO <CONNECTOR>(...)]`.
33    CreateSink {
34        name: String,
35        view: String,
36        connector: Option<ConnectorSpec>,
37    },
38    /// `START PIPELINE <sink_name>` — run the pipeline feeding `sink_name`.
39    StartPipeline { sink: String },
40    /// `REFRESH PIPELINE <sink_name> [FULL]` — re-run a pipeline; `full` resets
41    /// its persisted state first (Spark SDP `--full-refresh`).
42    RefreshPipeline { sink: String, full: bool },
43    /// `DROP SOURCE <name>`.
44    DropSource { name: String },
45    /// `DROP SINK <name>`.
46    DropSink { name: String },
47}
48
49// ── Registry ────────────────────────────────────────────────────────────────
50
51/// A connector reference: a kind (`parquet`, `kafka`, …) plus key/value options.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct ConnectorSpec {
54    pub kind: String,
55    pub options: HashMap<String, String>,
56}
57
58impl ConnectorSpec {
59    /// Fetch a required option, or a descriptive error if it is missing.
60    pub fn require(&self, key: &str) -> SqlResult<&str> {
61        self.options
62            .get(key)
63            .map(String::as_str)
64            .ok_or_else(|| SqlError::Unsupported {
65                feature: format!("connector '{}' requires option '{key}'", self.kind),
66            })
67    }
68}
69
70/// A declared source: either a bounded SQL query (fed as insertions) or a
71/// connector (e.g. `PARQUET(path='…')`).
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub enum SourceSpec {
74    /// `AS <query>` — rows from a bounded query.
75    Query(String),
76    /// `FROM <CONNECTOR>(...)` — rows pulled from a connector.
77    Connector(ConnectorSpec),
78}
79
80/// A declared sink: which view's output it collects, and optionally where it is
81/// written (a connector). With no connector, the output is returned as a result
82/// set by `START PIPELINE`.
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub struct SinkSpec {
85    pub view: String,
86    pub connector: Option<ConnectorSpec>,
87}
88
89/// SQL-layer registry of declared pipeline sources and sinks (metadata only).
90#[derive(Debug, Default)]
91pub struct PipelineRegistry {
92    sources: RwLock<HashMap<String, SourceSpec>>,
93    sinks: RwLock<HashMap<String, SinkSpec>>,
94}
95
96fn poisoned() -> SqlError {
97    SqlError::DataFusion {
98        message: "pipeline registry lock poisoned".into(),
99    }
100}
101
102impl PipelineRegistry {
103    pub fn new() -> Self {
104        Self::default()
105    }
106
107    pub fn register_source(&self, name: impl Into<String>, spec: SourceSpec) -> SqlResult<()> {
108        self.sources
109            .write()
110            .map_err(|_| poisoned())?
111            .insert(name.into(), spec);
112        Ok(())
113    }
114
115    pub fn register_sink(&self, name: impl Into<String>, spec: SinkSpec) -> SqlResult<()> {
116        self.sinks
117            .write()
118            .map_err(|_| poisoned())?
119            .insert(name.into(), spec);
120        Ok(())
121    }
122
123    pub fn source(&self, name: &str) -> SqlResult<Option<SourceSpec>> {
124        Ok(self
125            .sources
126            .read()
127            .map_err(|_| poisoned())?
128            .get(name)
129            .cloned())
130    }
131
132    pub fn sink(&self, name: &str) -> SqlResult<Option<SinkSpec>> {
133        Ok(self
134            .sinks
135            .read()
136            .map_err(|_| poisoned())?
137            .get(name)
138            .cloned())
139    }
140
141    /// View name backing the given sink, if the sink is registered.
142    /// Returns `None` for unknown sinks.
143    pub fn view_for_sink(&self, name: &str) -> SqlResult<Option<String>> {
144        Ok(self
145            .sinks
146            .read()
147            .map_err(|_| poisoned())?
148            .get(name)
149            .map(|spec| spec.view.clone()))
150    }
151
152    /// Names of all declared sinks.
153    pub fn sink_names(&self) -> SqlResult<Vec<String>> {
154        Ok(self
155            .sinks
156            .read()
157            .map_err(|_| poisoned())?
158            .keys()
159            .cloned()
160            .collect())
161    }
162
163    /// All declared source specs `(name, spec)`.
164    pub fn sources(&self) -> SqlResult<Vec<(String, SourceSpec)>> {
165        Ok(self
166            .sources
167            .read()
168            .map_err(|_| poisoned())?
169            .iter()
170            .map(|(k, v)| (k.clone(), v.clone()))
171            .collect())
172    }
173
174    pub fn remove_source(&self, name: &str) -> SqlResult<bool> {
175        Ok(self
176            .sources
177            .write()
178            .map_err(|_| poisoned())?
179            .remove(name)
180            .is_some())
181    }
182
183    pub fn remove_sink(&self, name: &str) -> SqlResult<bool> {
184        Ok(self
185            .sinks
186            .write()
187            .map_err(|_| poisoned())?
188            .remove(name)
189            .is_some())
190    }
191}
192
193// ── Parser ──────────────────────────────────────────────────────────────────
194
195/// Parse a pipeline DDL statement, or `Ok(None)` if it is not one.
196pub fn parse_pipeline_statement(sql: &str) -> SqlResult<Option<PipelineStatement>> {
197    let trimmed = sql.trim().trim_end_matches(';').trim();
198    let upper = trimmed.to_uppercase();
199
200    if upper.starts_with("CREATE SOURCE ") {
201        let rest = &trimmed["CREATE SOURCE ".len()..];
202        // `... AS <query>` (query source) or `... FROM <CONNECTOR>(...)`.
203        if let Some((name, query)) = split_keyword(rest, " AS ") {
204            require_nonempty(&name)?;
205            if query.trim().is_empty() {
206                return Err(SqlError::Unsupported {
207                    feature: "CREATE SOURCE requires a query after AS".into(),
208                });
209            }
210            return Ok(Some(PipelineStatement::CreateSource {
211                name,
212                source: SourceSpec::Query(query.trim().to_string()),
213            }));
214        }
215        if let Some((name, conn)) = split_keyword(rest, " FROM ") {
216            require_nonempty(&name)?;
217            return Ok(Some(PipelineStatement::CreateSource {
218                name,
219                source: SourceSpec::Connector(parse_connector_spec(&conn)?),
220            }));
221        }
222        return Err(SqlError::Unsupported {
223            feature: "CREATE SOURCE requires '<name> AS <query>' or '<name> FROM <connector>(...)'"
224                .into(),
225        });
226    }
227
228    if upper.starts_with("CREATE SINK ") {
229        let rest = &trimmed["CREATE SINK ".len()..];
230        let (name, after_from) =
231            split_keyword(rest, " FROM ").ok_or_else(|| SqlError::Unsupported {
232                feature: "CREATE SINK requires '<name> FROM <view>'".into(),
233            })?;
234        require_nonempty(&name)?;
235        // Optional `INTO <CONNECTOR>(...)` after the view.
236        let (view, connector) = if let Some((view, conn)) = split_keyword(&after_from, " INTO ") {
237            (view, Some(parse_connector_spec(&conn)?))
238        } else {
239            (after_from.trim().to_string(), None)
240        };
241        let view = view.trim().to_string();
242        require_nonempty(&view)?;
243        return Ok(Some(PipelineStatement::CreateSink {
244            name,
245            view,
246            connector,
247        }));
248    }
249
250    if upper.starts_with("START PIPELINE ") {
251        let sink = trimmed["START PIPELINE ".len()..].trim().to_string();
252        require_nonempty(&sink)?;
253        return Ok(Some(PipelineStatement::StartPipeline { sink }));
254    }
255
256    if upper.starts_with("REFRESH PIPELINE ") {
257        let rest = trimmed["REFRESH PIPELINE ".len()..].trim();
258        // Optional trailing FULL keyword.
259        let (sink, full) = match rest.to_uppercase().strip_suffix(" FULL") {
260            Some(_) => (rest[..rest.len() - " FULL".len()].trim().to_string(), true),
261            None => (rest.to_string(), false),
262        };
263        require_nonempty(&sink)?;
264        return Ok(Some(PipelineStatement::RefreshPipeline { sink, full }));
265    }
266
267    if upper.starts_with("DROP SOURCE ") {
268        let name = trimmed["DROP SOURCE ".len()..].trim().to_string();
269        require_nonempty(&name)?;
270        return Ok(Some(PipelineStatement::DropSource { name }));
271    }
272
273    if upper.starts_with("DROP SINK ") {
274        let name = trimmed["DROP SINK ".len()..].trim().to_string();
275        require_nonempty(&name)?;
276        return Ok(Some(PipelineStatement::DropSink { name }));
277    }
278
279    Ok(None)
280}
281
282/// Apply a CREATE/DROP pipeline DDL to the registry. `START PIPELINE` is **not**
283/// handled here (it needs the `krishiv-api` execution layer); it returns
284/// `Ok(None)` so the caller can intercept it.
285///
286/// Returns `Some(name)` if a CREATE/DROP statement was applied.
287pub fn execute_pipeline_ddl(registry: &PipelineRegistry, sql: &str) -> SqlResult<Option<String>> {
288    let Some(stmt) = parse_pipeline_statement(sql)? else {
289        return Ok(None);
290    };
291    match stmt {
292        PipelineStatement::CreateSource { name, source } => {
293            registry.register_source(name.clone(), source)?;
294            Ok(Some(name))
295        }
296        PipelineStatement::CreateSink {
297            name,
298            view,
299            connector,
300        } => {
301            registry.register_sink(name.clone(), SinkSpec { view, connector })?;
302            Ok(Some(name))
303        }
304        PipelineStatement::DropSource { name } => {
305            registry.remove_source(&name)?;
306            Ok(Some(name))
307        }
308        PipelineStatement::DropSink { name } => {
309            registry.remove_sink(&name)?;
310            Ok(Some(name))
311        }
312        // START / REFRESH PIPELINE are handled by the api layer, not the registry.
313        PipelineStatement::StartPipeline { .. } | PipelineStatement::RefreshPipeline { .. } => {
314            Ok(None)
315        }
316    }
317}
318
319// ── helpers ─────────────────────────────────────────────────────────────────
320
321/// Split `rest` on the first case-insensitive occurrence of `keyword`
322/// (e.g. " AS "), returning `(before_trimmed, after)`.
323fn split_keyword(rest: &str, keyword: &str) -> Option<(String, String)> {
324    let upper = rest.to_uppercase();
325    let key_upper = keyword.to_uppercase();
326    let idx = upper.find(&key_upper)?;
327    let before = rest[..idx].trim().to_string();
328    let after = rest[idx + keyword.len()..].to_string();
329    Some((before, after))
330}
331
332fn require_nonempty(s: &str) -> SqlResult<()> {
333    if s.trim().is_empty() {
334        Err(SqlError::EmptyTableName)
335    } else {
336        Ok(())
337    }
338}
339
340/// Parse a connector reference of the form `KIND(key='value', key2='value2')`
341/// or a bare registered connector name (`registered_sink`). Keys are
342/// lowercased; values are unquoted (single or double quotes).
343fn parse_connector_spec(s: &str) -> SqlResult<ConnectorSpec> {
344    let s = s.trim();
345    let Some(open) = s.find('(') else {
346        let kind = s.trim().to_lowercase();
347        require_nonempty(&kind)?;
348        return Ok(ConnectorSpec {
349            kind,
350            options: HashMap::new(),
351        });
352    };
353    let close = s.rfind(')').ok_or_else(|| SqlError::Unsupported {
354        feature: "connector spec missing closing ')'".into(),
355    })?;
356    if close < open {
357        return Err(SqlError::Unsupported {
358            feature: "connector spec has malformed parentheses".into(),
359        });
360    }
361    let kind = s[..open].trim().to_lowercase();
362    require_nonempty(&kind)?;
363
364    let mut options = HashMap::new();
365    for part in s[open + 1..close].split(',') {
366        let part = part.trim();
367        if part.is_empty() {
368            continue;
369        }
370        let (k, v) = part.split_once('=').ok_or_else(|| SqlError::Unsupported {
371            feature: format!("connector option '{part}' must be 'key=value'"),
372        })?;
373        let v = v.trim().trim_matches(['\'', '"']);
374        options.insert(k.trim().to_lowercase(), v.to_string());
375    }
376    Ok(ConnectorSpec { kind, options })
377}
378
379// ── tests ─────────────────────────────────────────────────────────────────────
380
381#[cfg(test)]
382mod tests {
383    use super::*;
384
385    #[test]
386    fn parse_create_source_query() {
387        let s = parse_pipeline_statement("CREATE SOURCE orders AS SELECT * FROM raw").unwrap();
388        assert_eq!(
389            s,
390            Some(PipelineStatement::CreateSource {
391                name: "orders".into(),
392                source: SourceSpec::Query("SELECT * FROM raw".into()),
393            })
394        );
395    }
396
397    #[test]
398    fn parse_create_source_connector() {
399        let s =
400            parse_pipeline_statement("CREATE SOURCE orders FROM PARQUET(path='/data/o.parquet')")
401                .unwrap();
402        let Some(PipelineStatement::CreateSource {
403            name,
404            source: SourceSpec::Connector(spec),
405        }) = s
406        else {
407            panic!("expected connector source");
408        };
409        assert_eq!(name, "orders");
410        assert_eq!(spec.kind, "parquet");
411        assert_eq!(spec.require("path").unwrap(), "/data/o.parquet");
412    }
413
414    #[test]
415    fn parse_bare_registered_connector_reference() {
416        let Some(PipelineStatement::CreateSink {
417            name,
418            view,
419            connector: Some(spec),
420        }) = parse_pipeline_statement("CREATE SINK out FROM revenue INTO registered_out").unwrap()
421        else {
422            panic!("expected connector sink");
423        };
424        assert_eq!(name, "out");
425        assert_eq!(view, "revenue");
426        assert_eq!(spec.kind, "registered_out");
427        assert!(spec.options.is_empty());
428    }
429
430    #[test]
431    fn parse_create_sink_plain_and_connector() {
432        // Plain (memory result) sink.
433        assert_eq!(
434            parse_pipeline_statement("CREATE SINK out FROM revenue;").unwrap(),
435            Some(PipelineStatement::CreateSink {
436                name: "out".into(),
437                view: "revenue".into(),
438                connector: None,
439            })
440        );
441        // Connector sink.
442        let Some(PipelineStatement::CreateSink {
443            name,
444            view,
445            connector: Some(spec),
446        }) = parse_pipeline_statement(
447            "CREATE SINK out FROM revenue INTO PARQUET(path='/o.parquet')",
448        )
449        .unwrap()
450        else {
451            panic!("expected connector sink");
452        };
453        assert_eq!((name.as_str(), view.as_str()), ("out", "revenue"));
454        assert_eq!(spec.kind, "parquet");
455        assert_eq!(spec.require("path").unwrap(), "/o.parquet");
456    }
457
458    #[test]
459    fn parse_start_and_drops() {
460        assert_eq!(
461            parse_pipeline_statement("START PIPELINE out").unwrap(),
462            Some(PipelineStatement::StartPipeline { sink: "out".into() })
463        );
464        assert_eq!(
465            parse_pipeline_statement("DROP SOURCE orders").unwrap(),
466            Some(PipelineStatement::DropSource {
467                name: "orders".into()
468            })
469        );
470        assert_eq!(
471            parse_pipeline_statement("DROP SINK out").unwrap(),
472            Some(PipelineStatement::DropSink { name: "out".into() })
473        );
474    }
475
476    #[test]
477    fn parse_refresh_pipeline() {
478        assert_eq!(
479            parse_pipeline_statement("REFRESH PIPELINE out").unwrap(),
480            Some(PipelineStatement::RefreshPipeline {
481                sink: "out".into(),
482                full: false
483            })
484        );
485        assert_eq!(
486            parse_pipeline_statement("REFRESH PIPELINE out FULL;").unwrap(),
487            Some(PipelineStatement::RefreshPipeline {
488                sink: "out".into(),
489                full: true
490            })
491        );
492    }
493
494    #[test]
495    fn non_pipeline_sql_returns_none() {
496        assert_eq!(parse_pipeline_statement("SELECT 1").unwrap(), None);
497    }
498
499    #[test]
500    fn registry_create_drop_roundtrip() {
501        let reg = PipelineRegistry::new();
502        execute_pipeline_ddl(&reg, "CREATE SOURCE orders AS SELECT * FROM raw").unwrap();
503        execute_pipeline_ddl(&reg, "CREATE SINK out FROM revenue").unwrap();
504        assert_eq!(
505            reg.source("orders").unwrap().unwrap(),
506            SourceSpec::Query("SELECT * FROM raw".into())
507        );
508        assert_eq!(reg.sink("out").unwrap().unwrap().view, "revenue");
509        // START PIPELINE is not consumed by the registry layer.
510        assert_eq!(
511            execute_pipeline_ddl(&reg, "START PIPELINE out").unwrap(),
512            None
513        );
514        assert!(
515            execute_pipeline_ddl(&reg, "DROP SOURCE orders")
516                .unwrap()
517                .is_some()
518        );
519        assert!(reg.source("orders").unwrap().is_none());
520    }
521}
krishiv_sql/pipeline_ddl.rs

krishiv_sql/
pipeline_ddl.rs
