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        /// Declared retention from the trailing `WITH (...)`.
134        retention: Option<crate::parser::MaterializedViewRetention>,
135        /// Declared freshness from the trailing `WITH (...)`: the most a committed read of the
136        /// view may lag.
137        freshness: Option<std::time::Duration>,
138    },
139
140    /// `CREATE STREAM` for a named streaming pipeline.
141    CreateStream {
142        /// Stream name.
143        name: ObjectName,
144        /// Backing query.
145        query: Box<StreamingStatement>,
146        /// Optional output strategy.
147        emit_clause: Option<EmitClause>,
148        /// Replace an existing stream.
149        or_replace: bool,
150        /// Ignore an existing stream.
151        if_not_exists: bool,
152        /// Raw query text between `AS` and `EMIT`.
153        query_sql: String,
154        /// `RETAIN_HISTORY` cap in bytes from the trailing `WITH (...)`.
155        retention_bytes: Option<u64>,
156    },
157
158    /// `DROP STREAM`.
159    DropStream {
160        /// Stream to drop.
161        name: ObjectName,
162        /// Ignore a missing stream.
163        if_exists: bool,
164        /// Also drop dependent objects.
165        cascade: bool,
166    },
167
168    /// `ALTER SOURCE`.
169    AlterSource {
170        /// Source to alter.
171        name: ObjectName,
172        /// Requested alteration.
173        operation: AlterSourceOperation,
174    },
175
176    /// `INSERT INTO` a streaming source or table.
177    InsertInto {
178        /// Target source or table.
179        table_name: ObjectName,
180        /// Empty when the statement does not name columns.
181        columns: Vec<Ident>,
182        /// Rows to insert.
183        values: Vec<Vec<Expr>>,
184    },
185
186    /// `CREATE LOOKUP TABLE`.
187    CreateLookupTable(Box<super::lookup_table::CreateLookupTableStatement>),
188
189    /// `DROP LOOKUP TABLE`.
190    DropLookupTable {
191        /// Lookup table to drop.
192        name: ObjectName,
193        /// Ignore a missing table.
194        if_exists: bool,
195    },
196
197    /// Trigger an immediate checkpoint.
198    Checkpoint,
199
200    /// `RESTORE FROM CHECKPOINT <id>`.
201    RestoreCheckpoint {
202        /// Durable checkpoint identifier.
203        checkpoint_id: u64,
204    },
205
206    /// `SUBSCRIBE <name> [WHERE ...] [WITH (...)]`.
207    Subscribe(Box<SubscribeStatement>),
208
209    /// `DECLARE <name> [NO SCROLL] CURSOR [WITHOUT HOLD] FOR SUBSCRIBE …`
210    ///
211    /// Forward-only cursor over a SUBSCRIBE, scoped to the current SimpleQuery
212    /// connection. SCROLL, BINARY, WITH HOLD, INSENSITIVE, and ASENSITIVE are
213    /// rejected at parse time.
214    DeclareCursorForSubscribe {
215        /// Cursor identifier.
216        name: Ident,
217        /// `NO SCROLL` was explicit in the source. We never emit SCROLL.
218        no_scroll: bool,
219        /// The SUBSCRIBE body the cursor wraps.
220        subscribe: Box<SubscribeStatement>,
221    },
222}
223
224/// `SUBSCRIBE <name> [AS OF EPOCH n] [WHERE <fragment>] [WITH (...)]`.
225#[derive(Debug, Clone, PartialEq)]
226pub struct SubscribeStatement {
227    /// Target stream or materialized view.
228    pub name: ObjectName,
229    /// Raw WHERE fragment, compiled by the engine against the target schema.
230    pub filter_sql: Option<String>,
231    /// `AS OF EPOCH n`: replay everything emitted strictly after barrier `n`.
232    pub as_of_epoch: Option<u64>,
233    /// Reserved `WITH` options.
234    pub options: HashMap<String, String>,
235}
236
237/// Operations for ALTER SOURCE statements.
238#[derive(Debug, Clone, PartialEq)]
239pub enum AlterSourceOperation {
240    /// Add a new column: `ALTER SOURCE name ADD COLUMN col_name data_type`
241    AddColumn {
242        /// Column to add.
243        column_def: ColumnDef,
244    },
245    /// Set source properties: `ALTER SOURCE name SET ('key' = 'value', ...)`
246    SetProperties {
247        /// Properties to replace.
248        properties: HashMap<String, String>,
249    },
250}
251
252/// Format specification for serialization (e.g., FORMAT JSON, FORMAT AVRO).
253#[derive(Debug, Clone, PartialEq)]
254pub struct FormatSpec {
255    /// For example, `JSON`, `AVRO`, or `PROTOBUF`.
256    pub format_type: String,
257    /// Additional format options (from WITH clause after FORMAT).
258    pub options: HashMap<String, String>,
259}
260
261/// Parsed `CREATE SOURCE` statement.
262#[derive(Debug, Clone, PartialEq)]
263pub struct CreateSourceStatement {
264    /// Source name.
265    pub name: ObjectName,
266    /// Declared columns.
267    pub columns: Vec<ColumnDef>,
268    /// Primary-key columns, empty when no key is declared.
269    pub primary_key: Vec<Ident>,
270    /// Optional event-time watermark.
271    pub watermark: Option<WatermarkDef>,
272    /// Source runtime options from the trailing `WITH` clause.
273    pub with_options: HashMap<String, String>,
274    /// Replace an existing source.
275    pub or_replace: bool,
276    /// Ignore an existing source.
277    pub if_not_exists: bool,
278    /// Connector type (e.g., "KAFKA") from `FROM KAFKA (...)` syntax
279    pub connector_type: Option<String>,
280    /// Connector-specific options (from `FROM KAFKA (...)`)
281    pub connector_options: HashMap<String, String>,
282    /// Format specification (e.g., `FORMAT JSON`)
283    pub format: Option<FormatSpec>,
284}
285
286/// Parsed `CREATE SINK` statement.
287#[derive(Debug, Clone, PartialEq)]
288pub struct CreateSinkStatement {
289    /// Sink name.
290    pub name: ObjectName,
291    /// Sink input.
292    pub from: SinkFrom,
293    /// Replace an existing sink.
294    pub or_replace: bool,
295    /// Ignore an existing sink.
296    pub if_not_exists: bool,
297    /// Optional row filter.
298    pub filter: Option<Expr>,
299    /// Connector type (e.g., "KAFKA") from `INTO KAFKA (...)` syntax
300    pub connector_type: Option<String>,
301    /// Connector-specific options (from `INTO KAFKA (...)`)
302    pub connector_options: HashMap<String, String>,
303    /// Format specification (e.g., `FORMAT JSON`)
304    pub format: Option<FormatSpec>,
305}
306
307/// Input selected by a sink.
308#[derive(Debug, Clone, PartialEq)]
309pub enum SinkFrom {
310    /// A named table or source.
311    Table(ObjectName),
312    /// A query result.
313    Query(Box<StreamingStatement>),
314}
315
316/// Parsed watermark definition.
317#[derive(Debug, Clone, PartialEq)]
318pub struct WatermarkDef {
319    /// Event-time column.
320    pub column: Ident,
321    /// Watermark expression (e.g., column - INTERVAL '5' SECOND).
322    /// `None` when `WATERMARK FOR col` is used without `AS expr`,
323    /// meaning watermark advances via `source.watermark()` with zero delay.
324    pub expression: Option<Expr>,
325}
326
327/// SQL configuration for events that arrive after their window closes.
328#[derive(Debug, Clone, PartialEq, Default)]
329pub struct LateDataClause {
330    /// For example, `INTERVAL '1' HOUR`.
331    pub allowed_lateness: Option<Box<Expr>>,
332    /// Destination for late events.
333    pub side_output: Option<String>,
334}
335
336impl LateDataClause {
337    /// Uses the given lateness without a side output.
338    #[must_use]
339    pub fn with_allowed_lateness(lateness: Expr) -> Self {
340        Self {
341            allowed_lateness: Some(Box::new(lateness)),
342            side_output: None,
343        }
344    }
345
346    /// Uses the given lateness and side output.
347    #[must_use]
348    pub fn with_side_output(lateness: Expr, side_output: String) -> Self {
349        Self {
350            allowed_lateness: Some(Box::new(lateness)),
351            side_output: Some(side_output),
352        }
353    }
354
355    /// Uses the default lateness with a side output.
356    #[must_use]
357    pub fn side_output_only(side_output: String) -> Self {
358        Self {
359            allowed_lateness: None,
360            side_output: Some(side_output),
361        }
362    }
363
364    /// # Errors
365    ///
366    /// Returns `ParseError::WindowError` if the interval cannot be parsed.
367    pub fn to_allowed_lateness(&self) -> Result<Duration, ParseError> {
368        match &self.allowed_lateness {
369            Some(expr) => WindowRewriter::parse_interval_to_duration(expr),
370            None => Ok(Duration::ZERO),
371        }
372    }
373
374    #[must_use]
375    /// Whether late events have a side-output destination.
376    pub fn has_side_output(&self) -> bool {
377        self.side_output.is_some()
378    }
379
380    #[must_use]
381    /// The configured side-output destination.
382    pub fn get_side_output(&self) -> Option<&str> {
383        self.side_output.as_deref()
384    }
385}
386
387/// Runtime output strategy for a streaming operator.
388#[derive(Debug, Clone, PartialEq)]
389pub enum EmitStrategy {
390    /// Emit when the watermark passes a window end.
391    OnWatermark,
392    /// Emit only when a window closes.
393    OnWindowClose,
394    /// Emit at a fixed interval.
395    Periodic(Duration),
396    /// Emit after every state change.
397    OnUpdate,
398    /// Emit Z-set changelog records.
399    Changelog,
400    /// Suppress intermediate results.
401    FinalOnly,
402}
403
404/// Parsed `EMIT` strategy.
405#[derive(Debug, Clone, PartialEq)]
406pub enum EmitClause {
407    /// Emit when the watermark passes the window end.
408    AfterWatermark,
409
410    /// Emit only when the window closes; this is distinct from `AfterWatermark`.
411    OnWindowClose,
412
413    /// Emit intermediate results periodically and final results on watermark.
414    Periodically {
415        /// Parsed interval expression.
416        interval: Box<Expr>,
417    },
418
419    /// Emit after every state change.
420    OnUpdate,
421
422    /// Emit Z-set changelog weights, including retraction pairs for updates.
423    Changes,
424
425    /// Emit only finalized results and drop data arriving after window close.
426    Final,
427}
428
429impl std::fmt::Display for EmitClause {
430    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
431        match self {
432            EmitClause::AfterWatermark => write!(f, "EMIT AFTER WATERMARK"),
433            EmitClause::OnWindowClose => write!(f, "EMIT ON WINDOW CLOSE"),
434            EmitClause::Periodically { interval } => write!(f, "EMIT EVERY {interval}"),
435            EmitClause::OnUpdate => write!(f, "EMIT ON UPDATE"),
436            EmitClause::Changes => write!(f, "EMIT CHANGES"),
437            EmitClause::Final => write!(f, "EMIT FINAL"),
438        }
439    }
440}
441
442impl EmitClause {
443    /// # Errors
444    ///
445    /// Returns `ParseError::WindowError` if the periodic interval cannot be parsed.
446    pub fn to_emit_strategy(&self) -> Result<EmitStrategy, ParseError> {
447        match self {
448            EmitClause::AfterWatermark => Ok(EmitStrategy::OnWatermark),
449            EmitClause::OnWindowClose => Ok(EmitStrategy::OnWindowClose),
450            EmitClause::Periodically { interval } => {
451                let duration = WindowRewriter::parse_interval_to_duration(interval)?;
452                Ok(EmitStrategy::Periodic(duration))
453            }
454            EmitClause::OnUpdate => Ok(EmitStrategy::OnUpdate),
455            EmitClause::Changes => Ok(EmitStrategy::Changelog),
456            EmitClause::Final => Ok(EmitStrategy::FinalOnly),
457        }
458    }
459
460    /// Whether the strategy can emit retractions.
461    #[must_use]
462    pub fn requires_changelog(&self) -> bool {
463        matches!(self, EmitClause::Changes | EmitClause::OnUpdate)
464    }
465
466    /// Whether the strategy produces no retractions.
467    #[must_use]
468    pub fn is_append_only(&self) -> bool {
469        matches!(
470            self,
471            EmitClause::OnWindowClose | EmitClause::Final | EmitClause::AfterWatermark
472        )
473    }
474
475    /// Whether output depends on source watermark advancement.
476    ///
477    /// `OnWindowClose`, `Final`, and `AfterWatermark` all depend on watermark
478    /// advancement to trigger window closure. Without a watermark, timers will
479    /// never fire and windows will never close.
480    #[must_use]
481    pub fn requires_watermark(&self) -> bool {
482        matches!(
483            self,
484            EmitClause::OnWindowClose | EmitClause::Final | EmitClause::AfterWatermark
485        )
486    }
487}
488
489/// Parsed streaming window function.
490#[derive(Debug, Clone, PartialEq)]
491pub enum WindowFunction {
492    /// `TUMBLE(column, interval [, offset])`.
493    Tumble {
494        /// Event-time expression.
495        time_column: Box<Expr>,
496        /// Window size.
497        interval: Box<Expr>,
498        /// Optional timezone-alignment offset.
499        offset: Option<Box<Expr>>,
500    },
501    /// `HOP(column, slide, size [, offset])`.
502    Hop {
503        /// Event-time expression.
504        time_column: Box<Expr>,
505        /// How often a new window begins.
506        slide_interval: Box<Expr>,
507        /// Window size.
508        window_interval: Box<Expr>,
509        /// Optional timezone-alignment offset.
510        offset: Option<Box<Expr>>,
511    },
512    /// `SESSION(column, gap)`.
513    Session {
514        /// Event-time expression.
515        time_column: Box<Expr>,
516        /// Maximum gap between events in one session.
517        gap_interval: Box<Expr>,
518    },
519    /// `CUMULATE(column, step, size)`.
520    Cumulate {
521        /// Event-time expression.
522        time_column: Box<Expr>,
523        /// Window growth increment.
524        step_interval: Box<Expr>,
525        /// Maximum window size.
526        max_size_interval: Box<Expr>,
527    },
528}
529
530#[cfg(test)]
531mod tests;
laminar_sql/parser/statements/mod.rs

laminar_sql/parser/statements/
mod.rs
