laminar_sql/translator/streaming_ddl/
mod.rs

1//! SQL DDL to streaming API translation.
2
3#[allow(clippy::disallowed_types)] // cold path: SQL translation
4use std::collections::HashMap;
5use std::sync::Arc;
6use std::time::Duration;
7
8use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
9use sqlparser::ast::{ColumnDef, DataType as SqlDataType};
10
11use laminar_core::streaming::config::{DEFAULT_BUFFER_SIZE, MAX_BUFFER_SIZE, MIN_BUFFER_SIZE};
12
13use crate::parser::ParseError;
14use crate::parser::{CreateSourceStatement, WatermarkDef};
15
16/// Watermark specification for a source.
17#[derive(Debug, Clone)]
18pub struct WatermarkSpec {
19    /// Column name for event time.
20    pub column: String,
21    /// Bounded out-of-orderness duration.
22    pub max_out_of_orderness: Duration,
23    /// Whether this is a processing-time watermark (`PROCTIME()`).
24    ///
25    /// When `true`, the runtime should use `ProcessingTimeGenerator`
26    /// instead of `BoundedOutOfOrdernessGenerator`.
27    pub is_processing_time: bool,
28}
29
30/// Configuration options for a streaming source.
31#[derive(Debug, Clone)]
32pub struct SourceConfigOptions {
33    /// Buffer size for the channel.
34    pub buffer_size: usize,
35}
36
37impl Default for SourceConfigOptions {
38    fn default() -> Self {
39        Self {
40            buffer_size: DEFAULT_BUFFER_SIZE,
41        }
42    }
43}
44
45/// Column definition for a streaming source.
46#[derive(Debug, Clone)]
47pub struct ColumnDefinition {
48    /// Column name.
49    pub name: String,
50    /// Arrow data type.
51    pub data_type: DataType,
52    /// Whether the column is nullable.
53    pub nullable: bool,
54}
55
56/// A validated streaming source definition.
57///
58/// This is the output of translating a `CreateSourceStatement` to a typed
59/// configuration that can be used to create runtime sources.
60#[derive(Debug, Clone)]
61pub struct SourceDefinition {
62    /// Source name.
63    pub name: String,
64    /// Column definitions.
65    pub columns: Vec<ColumnDefinition>,
66    /// Primary-key columns in declaration order.
67    pub primary_key: Vec<String>,
68    /// Arrow schema.
69    pub schema: SchemaRef,
70    /// Watermark specification, if defined.
71    pub watermark: Option<WatermarkSpec>,
72    /// Configuration options.
73    pub config: SourceConfigOptions,
74}
75
76impl TryFrom<CreateSourceStatement> for SourceDefinition {
77    type Error = ParseError;
78
79    fn try_from(stmt: CreateSourceStatement) -> Result<Self, Self::Error> {
80        translate_create_source(stmt)
81    }
82}
83
84/// Translates a CREATE SOURCE statement to a typed SourceDefinition.
85///
86/// # Errors
87///
88/// Returns `ParseError::ValidationError` if:
89/// - An invalid `buffer_size` is provided
90/// - Column types cannot be converted to Arrow types
91pub fn translate_create_source(
92    stmt: CreateSourceStatement,
93) -> Result<SourceDefinition, ParseError> {
94    let columns = convert_columns(&stmt.columns)?;
95    translate_create_source_with_columns(stmt, columns)
96}
97
98/// Translate a CREATE SOURCE statement using an already-resolved column
99/// list. Used by the DDL layer after `discover_schema` so `WATERMARK FOR`
100/// validates against the discovered columns rather than the SQL text.
101///
102/// # Errors
103///
104/// Returns `ParseError` from option validation or watermark parsing.
105pub fn translate_create_source_with_columns(
106    stmt: CreateSourceStatement,
107    mut columns: Vec<ColumnDefinition>,
108) -> Result<SourceDefinition, ParseError> {
109    let config = parse_source_options(&stmt.with_options)?;
110    if let Some(column) = columns
111        .iter()
112        .find(|column| is_reserved_operation_column(&column.name))
113    {
114        return Err(ParseError::ValidationError(format!(
115            "CREATE SOURCE column '{}' is reserved mutation metadata",
116            column.name
117        )));
118    }
119    validate_full_changelog_weight(&columns)?;
120    let (primary_key, primary_key_indices) = resolve_source_primary_key(&stmt, &columns)?;
121    for index in primary_key_indices {
122        if stmt.columns.is_empty() && columns[index].nullable {
123            return Err(ParseError::ValidationError(format!(
124                "CREATE SOURCE discovered PRIMARY KEY column '{}' must be non-nullable",
125                columns[index].name
126            )));
127        }
128        columns[index].nullable = false;
129    }
130
131    let fields: Vec<Field> = columns
132        .iter()
133        .map(|col| Field::new(&col.name, col.data_type.clone(), col.nullable))
134        .collect();
135    let schema = Arc::new(Schema::new(fields));
136
137    let watermark = if let Some(wm) = stmt.watermark {
138        Some(parse_watermark(&wm, &columns)?)
139    } else {
140        None
141    };
142
143    Ok(SourceDefinition {
144        name: stmt.name.to_string(),
145        columns,
146        primary_key,
147        schema,
148        watermark,
149        config,
150    })
151}
152
153fn resolve_source_primary_key(
154    stmt: &CreateSourceStatement,
155    columns: &[ColumnDefinition],
156) -> Result<(Vec<String>, Vec<usize>), ParseError> {
157    let mut names = Vec::with_capacity(stmt.primary_key.len());
158    let mut indices = Vec::with_capacity(stmt.primary_key.len());
159
160    for key in &stmt.primary_key {
161        let matches: Vec<_> = if stmt.columns.is_empty() {
162            columns
163                .iter()
164                .enumerate()
165                .filter(|(_, column)| discovered_identifier_matches(key, &column.name))
166                .map(|(index, _)| index)
167                .collect()
168        } else {
169            stmt.columns
170                .iter()
171                .enumerate()
172                .filter(|(_, column)| declared_identifier_matches(key, &column.name))
173                .map(|(index, _)| index)
174                .collect()
175        };
176        let [index] = matches.as_slice() else {
177            let reason = if matches.is_empty() {
178                "does not exist"
179            } else {
180                "is ambiguous"
181            };
182            return Err(ParseError::ValidationError(format!(
183                "CREATE SOURCE PRIMARY KEY column '{key}' {reason}"
184            )));
185        };
186        let column = columns.get(*index).ok_or_else(|| {
187            ParseError::ValidationError(
188                "CREATE SOURCE resolved column metadata does not match its declaration".into(),
189            )
190        })?;
191        if indices.contains(index) {
192            return Err(ParseError::ValidationError(format!(
193                "CREATE SOURCE PRIMARY KEY repeats column '{}'",
194                column.name
195            )));
196        }
197        if stmt.columns.get(*index).is_some_and(|declared| {
198            declared
199                .options
200                .iter()
201                .any(|option| matches!(&option.option, sqlparser::ast::ColumnOption::Null))
202        }) {
203            return Err(ParseError::ValidationError(format!(
204                "CREATE SOURCE PRIMARY KEY column '{}' cannot be declared NULL",
205                column.name
206            )));
207        }
208        names.push(column.name.clone());
209        indices.push(*index);
210    }
211
212    Ok((names, indices))
213}
214
215fn discovered_identifier_matches(identifier: &sqlparser::ast::Ident, column: &str) -> bool {
216    if identifier.quote_style.is_some() {
217        identifier.value == column
218    } else {
219        identifier.value.eq_ignore_ascii_case(column)
220    }
221}
222
223fn declared_identifier_matches(
224    reference: &sqlparser::ast::Ident,
225    declaration: &sqlparser::ast::Ident,
226) -> bool {
227    match (reference.quote_style, declaration.quote_style) {
228        (None, None) => reference.value.eq_ignore_ascii_case(&declaration.value),
229        (Some(_), Some(_)) => reference.value == declaration.value,
230        (None, Some(_)) => reference.value.to_ascii_lowercase() == declaration.value,
231        (Some(_), None) => reference.value == declaration.value.to_ascii_lowercase(),
232    }
233}
234
235fn is_reserved_operation_column(column: &str) -> bool {
236    ["_op", "__op"]
237        .iter()
238        .any(|reserved| column.eq_ignore_ascii_case(reserved))
239}
240
241fn validate_full_changelog_weight(columns: &[ColumnDefinition]) -> Result<(), ParseError> {
242    let weights = columns
243        .iter()
244        .enumerate()
245        .filter(|(_, column)| column.name.eq_ignore_ascii_case("__weight"))
246        .collect::<Vec<_>>();
247    let [] = weights.as_slice() else {
248        let [(index, column)] = weights.as_slice() else {
249            return Err(ParseError::ValidationError(
250                "CREATE SOURCE may declare at most one case-insensitive __weight column".into(),
251            ));
252        };
253        if column.name != "__weight"
254            || *index + 1 != columns.len()
255            || column.data_type != DataType::Int64
256            || column.nullable
257        {
258            return Err(ParseError::ValidationError(
259                "CREATE SOURCE full-changelog metadata must be one exact trailing non-null BIGINT __weight column"
260                    .into(),
261            ));
262        }
263        return Ok(());
264    };
265    Ok(())
266}
267
268/// Parses source options from WITH clause.
269fn parse_source_options(
270    options: &HashMap<String, String>,
271) -> Result<SourceConfigOptions, ParseError> {
272    let mut config = SourceConfigOptions::default();
273
274    for (key, value) in options {
275        match key.to_lowercase().as_str() {
276            "buffer_size" => {
277                config.buffer_size = parse_buffer_size(value)?;
278            }
279            _ => {
280                return Err(ParseError::ValidationError(format!(
281                    "unsupported CREATE SOURCE runtime option '{key}'; only 'buffer_size' is supported"
282                )));
283            }
284        }
285    }
286
287    Ok(config)
288}
289
290/// Parses buffer_size option.
291fn parse_buffer_size(value: &str) -> Result<usize, ParseError> {
292    let size: usize = value.parse().map_err(|_| {
293        ParseError::ValidationError(format!(
294            "invalid buffer_size: '{}' - must be a number",
295            value
296        ))
297    })?;
298
299    if size < MIN_BUFFER_SIZE {
300        return Err(ParseError::ValidationError(format!(
301            "buffer_size {} is too small - minimum is {}",
302            size, MIN_BUFFER_SIZE
303        )));
304    }
305
306    if size > MAX_BUFFER_SIZE {
307        return Err(ParseError::ValidationError(format!(
308            "buffer_size {} is too large - maximum is {}",
309            size, MAX_BUFFER_SIZE
310        )));
311    }
312
313    Ok(size)
314}
315
316/// Converts SQL column definitions to Arrow types.
317fn convert_columns(columns: &[ColumnDef]) -> Result<Vec<ColumnDefinition>, ParseError> {
318    columns.iter().map(convert_column).collect()
319}
320
321/// Converts a single SQL column definition to Arrow type.
322fn convert_column(col: &ColumnDef) -> Result<ColumnDefinition, ParseError> {
323    let data_type = sql_type_to_arrow(&col.data_type)?;
324
325    // Check for NOT NULL constraint
326    let nullable = !col
327        .options
328        .iter()
329        .any(|opt| matches!(opt.option, sqlparser::ast::ColumnOption::NotNull));
330
331    Ok(ColumnDefinition {
332        name: col.name.value.clone(),
333        data_type,
334        nullable,
335    })
336}
337
338/// Converts SQL data type to Arrow data type.
339///
340/// # Errors
341///
342/// Returns `ParseError::ValidationError` for unsupported SQL data types.
343pub fn sql_type_to_arrow(sql_type: &SqlDataType) -> Result<DataType, ParseError> {
344    match sql_type {
345        // Integer types
346        SqlDataType::TinyInt(_) => Ok(DataType::Int8),
347        SqlDataType::SmallInt(_) => Ok(DataType::Int16),
348        SqlDataType::Int(_) | SqlDataType::Integer(_) => Ok(DataType::Int32),
349        SqlDataType::BigInt(_) => Ok(DataType::Int64),
350
351        // Unsigned integer types - wrapped in Unsigned variant
352        // Note: sqlparser wraps unsigned types differently in different versions
353
354        // Floating point types
355        SqlDataType::Float(_) | SqlDataType::Real => Ok(DataType::Float32),
356        SqlDataType::Double(_) | SqlDataType::DoublePrecision => Ok(DataType::Float64),
357
358        // Decimal types
359        SqlDataType::Decimal(info) | SqlDataType::Numeric(info) => {
360            #[allow(clippy::cast_possible_truncation)] // Precision/scale are typically small values
361            let (precision, scale) = match info {
362                sqlparser::ast::ExactNumberInfo::PrecisionAndScale(p, s) => (*p as u8, *s as i8),
363                sqlparser::ast::ExactNumberInfo::Precision(p) => (*p as u8, 0),
364                sqlparser::ast::ExactNumberInfo::None => (38, 9), // Default precision/scale
365            };
366            Ok(DataType::Decimal128(precision, scale))
367        }
368
369        // String types (including JSON/UUID stored as strings)
370        SqlDataType::Char(_)
371        | SqlDataType::Character(_)
372        | SqlDataType::Varchar(_)
373        | SqlDataType::CharacterVarying(_)
374        | SqlDataType::Text
375        | SqlDataType::String(_)
376        | SqlDataType::JSON
377        | SqlDataType::JSONB
378        | SqlDataType::Uuid => Ok(DataType::Utf8),
379
380        // Binary types
381        SqlDataType::Binary(_)
382        | SqlDataType::Varbinary(_)
383        | SqlDataType::Blob(_)
384        | SqlDataType::Bytea => Ok(DataType::Binary),
385
386        // Boolean type
387        SqlDataType::Boolean | SqlDataType::Bool => Ok(DataType::Boolean),
388
389        // Date/time types
390        SqlDataType::Date => Ok(DataType::Date32),
391        SqlDataType::Time(_, _) => Ok(DataType::Time64(TimeUnit::Microsecond)),
392        SqlDataType::Timestamp(_, _) => Ok(DataType::Timestamp(TimeUnit::Microsecond, None)),
393
394        // Interval type
395        SqlDataType::Interval { .. } => Ok(DataType::Interval(
396            arrow::datatypes::IntervalUnit::MonthDayNano,
397        )),
398
399        // Array type: ARRAY<T>, T[], Array(T)
400        SqlDataType::Array(elem_def) => {
401            let item_type = match elem_def {
402                sqlparser::ast::ArrayElemTypeDef::AngleBracket(t)
403                | sqlparser::ast::ArrayElemTypeDef::SquareBracket(t, _)
404                | sqlparser::ast::ArrayElemTypeDef::Parenthesis(t) => sql_type_to_arrow(t)?,
405                sqlparser::ast::ArrayElemTypeDef::None => {
406                    return Err(ParseError::ValidationError(
407                        "ARRAY type requires element type, e.g. ARRAY<INT>".into(),
408                    ));
409                }
410            };
411            Ok(DataType::List(Arc::new(Field::new(
412                "item", item_type, true,
413            ))))
414        }
415
416        // Complex types (MAP, STRUCT, nested records) — use auto-discovery instead.
417        _ => Err(ParseError::ValidationError(format!(
418            "unsupported data type in hand-declared column: {sql_type:?} \
419             — use auto-discovery with an Avro source for complex types"
420        ))),
421    }
422}
423
424/// Checks if an expression is a `PROCTIME()` function call.
425fn is_proctime_call(expr: &sqlparser::ast::Expr) -> bool {
426    if let sqlparser::ast::Expr::Function(func) = expr {
427        if let Some(name) = func.name.0.last() {
428            return name.to_string().eq_ignore_ascii_case("proctime");
429        }
430    }
431    false
432}
433
434/// Parses watermark definition.
435fn parse_watermark(
436    wm: &WatermarkDef,
437    columns: &[ColumnDefinition],
438) -> Result<WatermarkSpec, ParseError> {
439    let column_name = wm.column.value.clone();
440
441    // Verify column exists and is a timestamp type
442    let col = columns
443        .iter()
444        .find(|c| c.name == column_name)
445        .ok_or_else(|| {
446            ParseError::ValidationError(format!(
447                "watermark column '{}' not found in column list",
448                column_name
449            ))
450        })?;
451
452    if !matches!(col.data_type, DataType::Timestamp(_, _)) {
453        return Err(ParseError::ValidationError(format!(
454            "watermark column '{}' must be a TIMESTAMP, found {:?}",
455            column_name, col.data_type
456        )));
457    }
458
459    // Check for PROCTIME() watermark expression
460    if let Some(expr) = &wm.expression {
461        if is_proctime_call(expr) {
462            return Ok(WatermarkSpec {
463                column: column_name,
464                max_out_of_orderness: Duration::ZERO,
465                is_processing_time: true,
466            });
467        }
468    }
469
470    // Parse the watermark expression to extract out-of-orderness.
471    // When expression is None (WATERMARK FOR col without AS), use zero delay.
472    let max_out_of_orderness = match &wm.expression {
473        Some(expr) => parse_watermark_expression(expr),
474        None => Duration::ZERO,
475    };
476
477    Ok(WatermarkSpec {
478        column: column_name,
479        max_out_of_orderness,
480        is_processing_time: false,
481    })
482}
483
484/// Parses watermark expression to extract the bounded out-of-orderness.
485fn parse_watermark_expression(expr: &sqlparser::ast::Expr) -> Duration {
486    use sqlparser::ast::Expr;
487
488    match expr {
489        Expr::BinaryOp { op, right, .. } => match op {
490            sqlparser::ast::BinaryOperator::Minus => parse_interval_expr(right),
491            _ => Duration::ZERO,
492        },
493        // If just the column name, assume zero lateness
494        Expr::Identifier(_) => Duration::ZERO,
495        // Default to 1 second for complex expressions
496        _ => Duration::from_secs(1),
497    }
498}
499
500/// Parses an interval expression to a Duration.
501fn parse_interval_expr(expr: &sqlparser::ast::Expr) -> Duration {
502    use sqlparser::ast::Expr;
503
504    let Expr::Interval(interval) = expr else {
505        return Duration::from_secs(1);
506    };
507
508    // Extract value and unit from interval
509    let value_str = match interval.value.as_ref() {
510        Expr::Value(v) => {
511            // v is ValueWithSpan, access the inner value
512            match &v.value {
513                sqlparser::ast::Value::SingleQuotedString(s) => s.clone(),
514                sqlparser::ast::Value::Number(n, _) => n.clone(),
515                _ => return Duration::from_secs(1),
516            }
517        }
518        _ => return Duration::from_secs(1),
519    };
520
521    let value: u64 = value_str.parse().unwrap_or(1);
522
523    // Determine unit
524    let unit = interval
525        .leading_field
526        .as_ref()
527        .map_or("second", |u| match u {
528            sqlparser::ast::DateTimeField::Microsecond => "microsecond",
529            sqlparser::ast::DateTimeField::Millisecond => "millisecond",
530            sqlparser::ast::DateTimeField::Minute => "minute",
531            sqlparser::ast::DateTimeField::Hour => "hour",
532            sqlparser::ast::DateTimeField::Day => "day",
533            _ => "second",
534        });
535
536    match unit {
537        "microsecond" | "microseconds" => Duration::from_micros(value),
538        "millisecond" | "milliseconds" => Duration::from_millis(value),
539        "minute" | "minutes" => Duration::from_secs(value * 60),
540        "hour" | "hours" => Duration::from_secs(value * 3600),
541        "day" | "days" => Duration::from_secs(value * 86400),
542        _ => Duration::from_secs(value),
543    }
544}
545
546#[cfg(test)]
547mod tests;
laminar_sql/translator/streaming_ddl/mod.rs

laminar_sql/translator/streaming_ddl/
mod.rs
