laminar_sql/parser/statements/
mod.rs

1//! AST types for streaming SQL extensions and their runtime conversions.
2
3#[allow(clippy::disallowed_types)] // cold path: SQL parsing
4use std::collections::HashMap;
5use std::time::Duration;
6
7use sqlparser::ast::{ColumnDef, Expr, Ident, ObjectName, Statement};
8
9use super::join_parser::JoinAnalysis;
10use super::window_rewriter::WindowRewriter;
11use super::ParseError;
12
13/// Supported `SHOW` commands.
14#[derive(Debug, Clone, PartialEq)]
15pub enum ShowCommand {
16    /// List registered sources.
17    Sources,
18    /// List registered sinks.
19    Sinks,
20    /// List continuous queries.
21    Queries,
22    /// List materialized views.
23    MaterializedViews,
24    /// List named streams.
25    Streams,
26    /// List reference tables.
27    Tables,
28    /// Show checkpoint state.
29    CheckpointStatus,
30    /// Reconstruct source DDL.
31    CreateSource {
32        /// Source to reconstruct.
33        name: ObjectName,
34    },
35    /// Reconstruct sink DDL.
36    CreateSink {
37        /// Sink to reconstruct.
38        name: ObjectName,
39    },
40}
41
42/// SQL statements understood by the streaming parser.
43#[derive(Debug, Clone, PartialEq)]
44pub enum StreamingStatement {
45    /// A standard SQL statement without streaming syntax.
46    Standard(Box<sqlparser::ast::Statement>),
47
48    /// A multi-horizon temporal probe normalized to the canonical AS-OF query shape.
49    TemporalProbeQuery {
50        /// AST used for projection and non-join analysis.
51        statement: Box<Statement>,
52        /// Parsed temporal-join contract.
53        analysis: Box<JoinAnalysis>,
54    },
55
56    /// `CREATE SOURCE`.
57    CreateSource(Box<CreateSourceStatement>),
58    /// `CREATE SINK`.
59    CreateSink(Box<CreateSinkStatement>),
60    /// `CREATE CONTINUOUS QUERY`.
61    CreateContinuousQuery {
62        /// Query name.
63        name: ObjectName,
64        /// Streaming query body.
65        query: Box<StreamingStatement>,
66        /// Optional output strategy.
67        emit_clause: Option<EmitClause>,
68    },
69
70    /// `DROP SOURCE`.
71    DropSource {
72        /// Source to drop.
73        name: ObjectName,
74        /// Ignore a missing source.
75        if_exists: bool,
76        /// Also drop dependent streams and materialized views.
77        cascade: bool,
78    },
79
80    /// `DROP SINK`.
81    DropSink {
82        /// Sink to drop.
83        name: ObjectName,
84        /// Ignore a missing sink.
85        if_exists: bool,
86        /// Also drop dependent objects.
87        cascade: bool,
88    },
89
90    /// `DROP MATERIALIZED VIEW`.
91    DropMaterializedView {
92        /// View to drop.
93        name: ObjectName,
94        /// Ignore a missing view.
95        if_exists: bool,
96        /// Also drop dependent objects.
97        cascade: bool,
98    },
99
100    /// A `SHOW` command.
101    Show(ShowCommand),
102
103    /// `DESCRIBE` a streaming object.
104    Describe {
105        /// Object to describe.
106        name: ObjectName,
107        /// Include extended details.
108        extended: bool,
109    },
110
111    /// `EXPLAIN [ANALYZE]` a streaming query plan.
112    Explain {
113        /// Statement to explain.
114        statement: Box<StreamingStatement>,
115        /// Execute the query and collect metrics.
116        analyze: bool,
117    },
118
119    /// `CREATE MATERIALIZED VIEW`.
120    CreateMaterializedView {
121        /// View name.
122        name: ObjectName,
123        /// Backing query.
124        query: Box<StreamingStatement>,
125        /// Optional output strategy.
126        emit_clause: Option<EmitClause>,
127        /// Replace an existing view.
128        or_replace: bool,
129        /// Ignore an existing view.
130        if_not_exists: bool,
131        /// Raw query text between `AS` and `EMIT`.
132        query_sql: String,
133    },
134
135    /// `CREATE STREAM` for a named streaming pipeline.
136    CreateStream {
137        /// Stream name.
138        name: ObjectName,
139        /// Backing query.
140        query: Box<StreamingStatement>,
141        /// Optional output strategy.
142        emit_clause: Option<EmitClause>,
143        /// Replace an existing stream.
144        or_replace: bool,
145        /// Ignore an existing stream.
146        if_not_exists: bool,
147        /// Raw query text between `AS` and `EMIT`.
148        query_sql: String,
149        /// `RETAIN_HISTORY` cap in bytes from the trailing `WITH (...)`.
150        retention_bytes: Option<u64>,
151    },
152
153    /// `DROP STREAM`.
154    DropStream {
155        /// Stream to drop.
156        name: ObjectName,
157        /// Ignore a missing stream.
158        if_exists: bool,
159        /// Also drop dependent objects.
160        cascade: bool,
161    },
162
163    /// `ALTER SOURCE`.
164    AlterSource {
165        /// Source to alter.
166        name: ObjectName,
167        /// Requested alteration.
168        operation: AlterSourceOperation,
169    },
170
171    /// `INSERT INTO` a streaming source or table.
172    InsertInto {
173        /// Target source or table.
174        table_name: ObjectName,
175        /// Empty when the statement does not name columns.
176        columns: Vec<Ident>,
177        /// Rows to insert.
178        values: Vec<Vec<Expr>>,
179    },
180
181    /// `CREATE LOOKUP TABLE`.
182    CreateLookupTable(Box<super::lookup_table::CreateLookupTableStatement>),
183
184    /// `DROP LOOKUP TABLE`.
185    DropLookupTable {
186        /// Lookup table to drop.
187        name: ObjectName,
188        /// Ignore a missing table.
189        if_exists: bool,
190    },
191
192    /// Trigger an immediate checkpoint.
193    Checkpoint,
194
195    /// `RESTORE FROM CHECKPOINT <id>`.
196    RestoreCheckpoint {
197        /// Durable checkpoint identifier.
198        checkpoint_id: u64,
199    },
200
201    /// `SUBSCRIBE <name> [WHERE ...] [WITH (...)]`.
202    Subscribe(Box<SubscribeStatement>),
203
204    /// `DECLARE <name> [NO SCROLL] CURSOR [WITHOUT HOLD] FOR SUBSCRIBE …`
205    ///
206    /// Forward-only cursor over a SUBSCRIBE, scoped to the current SimpleQuery
207    /// connection. SCROLL, BINARY, WITH HOLD, INSENSITIVE, and ASENSITIVE are
208    /// rejected at parse time.
209    DeclareCursorForSubscribe {
210        /// Cursor identifier.
211        name: Ident,
212        /// `NO SCROLL` was explicit in the source. We never emit SCROLL.
213        no_scroll: bool,
214        /// The SUBSCRIBE body the cursor wraps.
215        subscribe: Box<SubscribeStatement>,
216    },
217}
218
219/// `SUBSCRIBE <name> [AS OF EPOCH n] [WHERE <fragment>] [WITH (...)]`.
220#[derive(Debug, Clone, PartialEq)]
221pub struct SubscribeStatement {
222    /// Target stream or materialized view.
223    pub name: ObjectName,
224    /// Raw WHERE fragment, compiled by the engine against the target schema.
225    pub filter_sql: Option<String>,
226    /// `AS OF EPOCH n`: replay everything emitted strictly after barrier `n`.
227    pub as_of_epoch: Option<u64>,
228    /// Reserved `WITH` options.
229    pub options: HashMap<String, String>,
230}
231
232/// Operations for ALTER SOURCE statements.
233#[derive(Debug, Clone, PartialEq)]
234pub enum AlterSourceOperation {
235    /// Add a new column: `ALTER SOURCE name ADD COLUMN col_name data_type`
236    AddColumn {
237        /// Column to add.
238        column_def: ColumnDef,
239    },
240    /// Set source properties: `ALTER SOURCE name SET ('key' = 'value', ...)`
241    SetProperties {
242        /// Properties to replace.
243        properties: HashMap<String, String>,
244    },
245}
246
247/// Format specification for serialization (e.g., FORMAT JSON, FORMAT AVRO).
248#[derive(Debug, Clone, PartialEq)]
249pub struct FormatSpec {
250    /// For example, `JSON`, `AVRO`, or `PROTOBUF`.
251    pub format_type: String,
252    /// Additional format options (from WITH clause after FORMAT).
253    pub options: HashMap<String, String>,
254}
255
256/// Parsed `CREATE SOURCE` statement.
257#[derive(Debug, Clone, PartialEq)]
258pub struct CreateSourceStatement {
259    /// Source name.
260    pub name: ObjectName,
261    /// Declared columns.
262    pub columns: Vec<ColumnDef>,
263    /// Primary-key columns, empty when no key is declared.
264    pub primary_key: Vec<Ident>,
265    /// Optional event-time watermark.
266    pub watermark: Option<WatermarkDef>,
267    /// Source runtime options from the trailing `WITH` clause.
268    pub with_options: HashMap<String, String>,
269    /// Replace an existing source.
270    pub or_replace: bool,
271    /// Ignore an existing source.
272    pub if_not_exists: bool,
273    /// Connector type (e.g., "KAFKA") from `FROM KAFKA (...)` syntax
274    pub connector_type: Option<String>,
275    /// Connector-specific options (from `FROM KAFKA (...)`)
276    pub connector_options: HashMap<String, String>,
277    /// Format specification (e.g., `FORMAT JSON`)
278    pub format: Option<FormatSpec>,
279}
280
281/// Parsed `CREATE SINK` statement.
282#[derive(Debug, Clone, PartialEq)]
283pub struct CreateSinkStatement {
284    /// Sink name.
285    pub name: ObjectName,
286    /// Sink input.
287    pub from: SinkFrom,
288    /// Replace an existing sink.
289    pub or_replace: bool,
290    /// Ignore an existing sink.
291    pub if_not_exists: bool,
292    /// Optional row filter.
293    pub filter: Option<Expr>,
294    /// Connector type (e.g., "KAFKA") from `INTO KAFKA (...)` syntax
295    pub connector_type: Option<String>,
296    /// Connector-specific options (from `INTO KAFKA (...)`)
297    pub connector_options: HashMap<String, String>,
298    /// Format specification (e.g., `FORMAT JSON`)
299    pub format: Option<FormatSpec>,
300}
301
302/// Input selected by a sink.
303#[derive(Debug, Clone, PartialEq)]
304pub enum SinkFrom {
305    /// A named table or source.
306    Table(ObjectName),
307    /// A query result.
308    Query(Box<StreamingStatement>),
309}
310
311/// Parsed watermark definition.
312#[derive(Debug, Clone, PartialEq)]
313pub struct WatermarkDef {
314    /// Event-time column.
315    pub column: Ident,
316    /// Watermark expression (e.g., column - INTERVAL '5' SECOND).
317    /// `None` when `WATERMARK FOR col` is used without `AS expr`,
318    /// meaning watermark advances via `source.watermark()` with zero delay.
319    pub expression: Option<Expr>,
320}
321
322/// SQL configuration for events that arrive after their window closes.
323#[derive(Debug, Clone, PartialEq, Default)]
324pub struct LateDataClause {
325    /// For example, `INTERVAL '1' HOUR`.
326    pub allowed_lateness: Option<Box<Expr>>,
327    /// Destination for late events.
328    pub side_output: Option<String>,
329}
330
331impl LateDataClause {
332    /// Uses the given lateness without a side output.
333    #[must_use]
334    pub fn with_allowed_lateness(lateness: Expr) -> Self {
335        Self {
336            allowed_lateness: Some(Box::new(lateness)),
337            side_output: None,
338        }
339    }
340
341    /// Uses the given lateness and side output.
342    #[must_use]
343    pub fn with_side_output(lateness: Expr, side_output: String) -> Self {
344        Self {
345            allowed_lateness: Some(Box::new(lateness)),
346            side_output: Some(side_output),
347        }
348    }
349
350    /// Uses the default lateness with a side output.
351    #[must_use]
352    pub fn side_output_only(side_output: String) -> Self {
353        Self {
354            allowed_lateness: None,
355            side_output: Some(side_output),
356        }
357    }
358
359    /// # Errors
360    ///
361    /// Returns `ParseError::WindowError` if the interval cannot be parsed.
362    pub fn to_allowed_lateness(&self) -> Result<Duration, ParseError> {
363        match &self.allowed_lateness {
364            Some(expr) => WindowRewriter::parse_interval_to_duration(expr),
365            None => Ok(Duration::ZERO),
366        }
367    }
368
369    #[must_use]
370    /// Whether late events have a side-output destination.
371    pub fn has_side_output(&self) -> bool {
372        self.side_output.is_some()
373    }
374
375    #[must_use]
376    /// The configured side-output destination.
377    pub fn get_side_output(&self) -> Option<&str> {
378        self.side_output.as_deref()
379    }
380}
381
382/// Runtime output strategy for a streaming operator.
383#[derive(Debug, Clone, PartialEq)]
384pub enum EmitStrategy {
385    /// Emit when the watermark passes a window end.
386    OnWatermark,
387    /// Emit only when a window closes.
388    OnWindowClose,
389    /// Emit at a fixed interval.
390    Periodic(Duration),
391    /// Emit after every state change.
392    OnUpdate,
393    /// Emit Z-set changelog records.
394    Changelog,
395    /// Suppress intermediate results.
396    FinalOnly,
397}
398
399/// Parsed `EMIT` strategy.
400#[derive(Debug, Clone, PartialEq)]
401pub enum EmitClause {
402    /// Emit when the watermark passes the window end.
403    AfterWatermark,
404
405    /// Emit only when the window closes; this is distinct from `AfterWatermark`.
406    OnWindowClose,
407
408    /// Emit intermediate results periodically and final results on watermark.
409    Periodically {
410        /// Parsed interval expression.
411        interval: Box<Expr>,
412    },
413
414    /// Emit after every state change.
415    OnUpdate,
416
417    /// Emit Z-set changelog weights, including retraction pairs for updates.
418    Changes,
419
420    /// Emit only finalized results and drop data arriving after window close.
421    Final,
422}
423
424impl std::fmt::Display for EmitClause {
425    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
426        match self {
427            EmitClause::AfterWatermark => write!(f, "EMIT AFTER WATERMARK"),
428            EmitClause::OnWindowClose => write!(f, "EMIT ON WINDOW CLOSE"),
429            EmitClause::Periodically { interval } => write!(f, "EMIT EVERY {interval}"),
430            EmitClause::OnUpdate => write!(f, "EMIT ON UPDATE"),
431            EmitClause::Changes => write!(f, "EMIT CHANGES"),
432            EmitClause::Final => write!(f, "EMIT FINAL"),
433        }
434    }
435}
436
437impl EmitClause {
438    /// # Errors
439    ///
440    /// Returns `ParseError::WindowError` if the periodic interval cannot be parsed.
441    pub fn to_emit_strategy(&self) -> Result<EmitStrategy, ParseError> {
442        match self {
443            EmitClause::AfterWatermark => Ok(EmitStrategy::OnWatermark),
444            EmitClause::OnWindowClose => Ok(EmitStrategy::OnWindowClose),
445            EmitClause::Periodically { interval } => {
446                let duration = WindowRewriter::parse_interval_to_duration(interval)?;
447                Ok(EmitStrategy::Periodic(duration))
448            }
449            EmitClause::OnUpdate => Ok(EmitStrategy::OnUpdate),
450            EmitClause::Changes => Ok(EmitStrategy::Changelog),
451            EmitClause::Final => Ok(EmitStrategy::FinalOnly),
452        }
453    }
454
455    /// Whether the strategy can emit retractions.
456    #[must_use]
457    pub fn requires_changelog(&self) -> bool {
458        matches!(self, EmitClause::Changes | EmitClause::OnUpdate)
459    }
460
461    /// Whether the strategy produces no retractions.
462    #[must_use]
463    pub fn is_append_only(&self) -> bool {
464        matches!(
465            self,
466            EmitClause::OnWindowClose | EmitClause::Final | EmitClause::AfterWatermark
467        )
468    }
469
470    /// Whether output depends on source watermark advancement.
471    ///
472    /// `OnWindowClose`, `Final`, and `AfterWatermark` all depend on watermark
473    /// advancement to trigger window closure. Without a watermark, timers will
474    /// never fire and windows will never close.
475    #[must_use]
476    pub fn requires_watermark(&self) -> bool {
477        matches!(
478            self,
479            EmitClause::OnWindowClose | EmitClause::Final | EmitClause::AfterWatermark
480        )
481    }
482}
483
484/// Parsed streaming window function.
485#[derive(Debug, Clone, PartialEq)]
486pub enum WindowFunction {
487    /// `TUMBLE(column, interval [, offset])`.
488    Tumble {
489        /// Event-time expression.
490        time_column: Box<Expr>,
491        /// Window size.
492        interval: Box<Expr>,
493        /// Optional timezone-alignment offset.
494        offset: Option<Box<Expr>>,
495    },
496    /// `HOP(column, slide, size [, offset])`.
497    Hop {
498        /// Event-time expression.
499        time_column: Box<Expr>,
500        /// How often a new window begins.
501        slide_interval: Box<Expr>,
502        /// Window size.
503        window_interval: Box<Expr>,
504        /// Optional timezone-alignment offset.
505        offset: Option<Box<Expr>>,
506    },
507    /// `SESSION(column, gap)`.
508    Session {
509        /// Event-time expression.
510        time_column: Box<Expr>,
511        /// Maximum gap between events in one session.
512        gap_interval: Box<Expr>,
513    },
514    /// `CUMULATE(column, step, size)`.
515    Cumulate {
516        /// Event-time expression.
517        time_column: Box<Expr>,
518        /// Window growth increment.
519        step_interval: Box<Expr>,
520        /// Maximum window size.
521        max_size_interval: Box<Expr>,
522    },
523}
524
525#[cfg(test)]
526mod tests;
laminar_sql/parser/statements/mod.rs

laminar_sql/parser/statements/
mod.rs
