laminar_sql/parser/lookup_table/
mod.rs

1//! Parser for CREATE/DROP LOOKUP TABLE DDL statements.
2//!
3//! Lookup tables are dimension/reference tables used in enrichment joins.
4//! They can be backed by registered snapshot-capable or on-demand connectors
5//! with configurable caching and predicate pushdown strategies.
6
7#[allow(clippy::disallowed_types)] // cold path: SQL parsing
8use std::collections::HashMap;
9
10use sqlparser::ast::{ColumnDef, ObjectName};
11use sqlparser::keywords::Keyword;
12use sqlparser::parser::Parser;
13use sqlparser::tokenizer::Token;
14
15use super::tokenizer::{expect_custom_keyword, expect_statement_end, parse_with_options};
16use super::ParseError;
17
18/// CREATE LOOKUP TABLE statement.
19#[derive(Debug, Clone, PartialEq)]
20pub struct CreateLookupTableStatement {
21    /// Table name.
22    pub name: ObjectName,
23    /// Column definitions.
24    pub columns: Vec<ColumnDef>,
25    /// Primary key column names.
26    pub primary_key: Vec<String>,
27    /// WITH clause options.
28    pub with_options: HashMap<String, String>,
29    /// Whether OR REPLACE was specified.
30    pub or_replace: bool,
31    /// Whether IF NOT EXISTS was specified.
32    pub if_not_exists: bool,
33}
34
35/// Validated lookup table properties from the WITH clause.
36#[derive(Debug, Clone, PartialEq)]
37pub struct LookupTableProperties {
38    /// Lookup connector.
39    pub connector: LookupConnector,
40    /// Lookup strategy.
41    pub strategy: LookupStrategy,
42    /// In-memory cache size.
43    pub cache_memory: Option<ByteSize>,
44    /// Cache TTL in seconds.
45    pub cache_ttl: Option<u64>,
46    /// Predicate pushdown mode.
47    pub pushdown_mode: PushdownMode,
48}
49
50/// Connector backing a lookup table.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub enum LookupConnector {
53    /// Static in-memory data.
54    Static,
55    /// Name resolved against the frozen connector registry.
56    External(String),
57}
58
59impl LookupConnector {
60    /// Parse a lookup connector from a string.
61    ///
62    /// # Errors
63    ///
64    /// Returns `ParseError` if the connector name is empty.
65    pub fn parse(s: &str) -> Result<Self, ParseError> {
66        let connector = s.trim().to_ascii_lowercase();
67        if connector.is_empty() {
68            return Err(ParseError::ValidationError(
69                "connector name cannot be empty".to_string(),
70            ));
71        }
72        Ok(if connector == "static" {
73            Self::Static
74        } else {
75            Self::External(connector)
76        })
77    }
78
79    /// Canonical connector name used for registry lookup and plan metadata.
80    #[must_use]
81    pub fn as_str(&self) -> &str {
82        match self {
83            Self::Static => "static",
84            Self::External(name) => name,
85        }
86    }
87}
88
89impl std::fmt::Display for LookupConnector {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        f.write_str(self.as_str())
92    }
93}
94
95/// Lookup strategy for how table data is distributed/cached.
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
97pub enum LookupStrategy {
98    /// Full table replicated on every node.
99    #[default]
100    Replicated,
101    /// Rows fetched on demand (no pre-loading).
102    OnDemand,
103}
104
105impl LookupStrategy {
106    /// Parse a lookup strategy from a string.
107    ///
108    /// # Errors
109    ///
110    /// Returns `ParseError` if the strategy is unknown.
111    pub fn parse(s: &str) -> Result<Self, ParseError> {
112        match s.as_bytes() {
113            b"replicated" => Ok(Self::Replicated),
114            b"on-demand" => Ok(Self::OnDemand),
115            other => Err(ParseError::ValidationError(format!(
116                "unknown lookup strategy: '{}' (expected: replicated or on-demand)",
117                String::from_utf8_lossy(other)
118            ))),
119        }
120    }
121}
122
123/// Predicate pushdown mode.
124#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
125pub enum PushdownMode {
126    /// Automatically decide based on connector capabilities.
127    #[default]
128    Auto,
129    /// Always push predicates to the source.
130    Enabled,
131    /// Never push predicates to the source.
132    Disabled,
133}
134
135impl PushdownMode {
136    /// Parse a pushdown mode from a string.
137    ///
138    /// # Errors
139    ///
140    /// Returns `ParseError` if the mode is unknown.
141    pub fn parse(s: &str) -> Result<Self, ParseError> {
142        match s.to_lowercase().as_str() {
143            "auto" => Ok(Self::Auto),
144            "enabled" | "true" | "on" => Ok(Self::Enabled),
145            "disabled" | "false" | "off" => Ok(Self::Disabled),
146            other => Err(ParseError::ValidationError(format!(
147                "unknown pushdown mode: '{other}' (expected: auto, enabled, disabled)"
148            ))),
149        }
150    }
151}
152
153/// A parsed byte size (e.g., "512mb", "1gb", "10kb").
154#[derive(Debug, Clone, Copy, PartialEq, Eq)]
155pub struct ByteSize(pub u64);
156
157impl ByteSize {
158    /// Parse a human-readable byte size string.
159    ///
160    /// Supports suffixes: b, kb, mb, gb, tb (case-insensitive).
161    ///
162    /// # Errors
163    ///
164    /// Returns `ParseError` if the string cannot be parsed.
165    pub fn parse(s: &str) -> Result<Self, ParseError> {
166        let s = s.trim().to_lowercase();
167        let (num_str, multiplier) = if let Some(n) = s.strip_suffix("tb") {
168            (n, 1024 * 1024 * 1024 * 1024)
169        } else if let Some(n) = s.strip_suffix("gb") {
170            (n, 1024 * 1024 * 1024)
171        } else if let Some(n) = s.strip_suffix("mb") {
172            (n, 1024 * 1024)
173        } else if let Some(n) = s.strip_suffix("kb") {
174            (n, 1024)
175        } else if let Some(n) = s.strip_suffix('b') {
176            (n, 1)
177        } else {
178            // Assume bytes if no suffix
179            (s.as_str(), 1)
180        };
181
182        let num: u64 = num_str
183            .trim()
184            .parse()
185            .map_err(|_| ParseError::ValidationError(format!("invalid byte size: '{s}'")))?;
186
187        Ok(Self(num * multiplier))
188    }
189
190    /// Returns the size in bytes.
191    #[must_use]
192    pub fn as_bytes(&self) -> u64 {
193        self.0
194    }
195}
196
197/// Parse a CREATE LOOKUP TABLE statement.
198///
199/// Syntax:
200/// ```sql
201/// CREATE [OR REPLACE] LOOKUP TABLE [IF NOT EXISTS] <name> (
202///   <col> <type> [NOT NULL],
203///   ...
204///   PRIMARY KEY (<col>, ...)
205/// ) WITH (
206///   'connector' = 'catalog-source',
207///   'endpoint' = 'https://catalog.example',
208///   ...
209/// );
210/// ```
211///
212/// # Errors
213///
214/// Returns `ParseError` if the statement syntax is invalid.
215pub fn parse_create_lookup_table(
216    parser: &mut Parser,
217) -> Result<CreateLookupTableStatement, ParseError> {
218    // CREATE already consumed by the router; consume it here for standalone parsing
219    parser
220        .expect_keyword(Keyword::CREATE)
221        .map_err(ParseError::SqlParseError)?;
222
223    let or_replace = parser.parse_keywords(&[Keyword::OR, Keyword::REPLACE]);
224
225    expect_custom_keyword(parser, "LOOKUP")?;
226
227    parser
228        .expect_keyword(Keyword::TABLE)
229        .map_err(ParseError::SqlParseError)?;
230
231    let if_not_exists = parser.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
232
233    let name = parser
234        .parse_object_name(false)
235        .map_err(ParseError::SqlParseError)?;
236
237    // Parse column definitions: ( col1 TYPE, col2 TYPE, ..., PRIMARY KEY (col1, ...) )
238    parser
239        .expect_token(&Token::LParen)
240        .map_err(ParseError::SqlParseError)?;
241
242    let mut columns = Vec::new();
243    let mut primary_key = Vec::new();
244
245    loop {
246        // Check for PRIMARY KEY clause
247        if parser.parse_keywords(&[Keyword::PRIMARY, Keyword::KEY]) {
248            parser
249                .expect_token(&Token::LParen)
250                .map_err(ParseError::SqlParseError)?;
251
252            loop {
253                let ident = parser
254                    .parse_identifier()
255                    .map_err(ParseError::SqlParseError)?;
256                primary_key.push(ident.value);
257
258                if !parser.consume_token(&Token::Comma) {
259                    break;
260                }
261            }
262
263            parser
264                .expect_token(&Token::RParen)
265                .map_err(ParseError::SqlParseError)?;
266
267            // Consume optional trailing comma after PRIMARY KEY clause
268            let _ = parser.consume_token(&Token::Comma);
269        } else if parser.consume_token(&Token::RParen) {
270            // End of column definitions
271            break;
272        } else {
273            // Parse column definition
274            let col = parser
275                .parse_column_def()
276                .map_err(ParseError::SqlParseError)?;
277            columns.push(col);
278
279            // Comma or closing paren
280            if !parser.consume_token(&Token::Comma) {
281                parser
282                    .expect_token(&Token::RParen)
283                    .map_err(ParseError::SqlParseError)?;
284                break;
285            }
286        }
287    }
288
289    if columns.is_empty() {
290        return Err(ParseError::StreamingError(
291            "LOOKUP TABLE must have at least one column".to_string(),
292        ));
293    }
294
295    // Parse WITH clause
296    let with_options = parse_with_options(parser)?;
297    if with_options.is_empty() {
298        return Err(ParseError::StreamingError(
299            "LOOKUP TABLE requires a WITH clause".to_string(),
300        ));
301    }
302    expect_statement_end(parser)?;
303
304    Ok(CreateLookupTableStatement {
305        name,
306        columns,
307        primary_key,
308        with_options,
309        or_replace,
310        if_not_exists,
311    })
312}
313
314/// Parse a DROP LOOKUP TABLE statement.
315///
316/// Syntax: `DROP LOOKUP TABLE [IF EXISTS] <name>`
317///
318/// # Errors
319///
320/// Returns `ParseError` if the statement syntax is invalid.
321pub fn parse_drop_lookup_table(parser: &mut Parser) -> Result<(ObjectName, bool), ParseError> {
322    parser
323        .expect_keyword(Keyword::DROP)
324        .map_err(ParseError::SqlParseError)?;
325
326    expect_custom_keyword(parser, "LOOKUP")?;
327
328    parser
329        .expect_keyword(Keyword::TABLE)
330        .map_err(ParseError::SqlParseError)?;
331
332    let if_exists = parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
333
334    let name = parser
335        .parse_object_name(false)
336        .map_err(ParseError::SqlParseError)?;
337
338    Ok((name, if_exists))
339}
340
341/// Validate and extract typed properties from raw WITH options.
342///
343/// # Errors
344///
345/// Returns `ParseError` if required properties are missing or invalid.
346pub fn validate_properties<S: ::std::hash::BuildHasher>(
347    options: &HashMap<String, String, S>,
348) -> Result<LookupTableProperties, ParseError> {
349    let connector_str = options.get("connector").ok_or_else(|| {
350        ParseError::ValidationError("missing required property: 'connector'".to_string())
351    })?;
352    let connector = LookupConnector::parse(connector_str)?;
353
354    let strategy = match options.get("strategy") {
355        Some(s) => LookupStrategy::parse(s)?,
356        None => LookupStrategy::default(),
357    };
358
359    let cache_memory = options
360        .get("cache.memory")
361        .map(|s| ByteSize::parse(s))
362        .transpose()?;
363
364    if options.contains_key("cache.disk") {
365        return Err(ParseError::ValidationError(
366            "cache.disk is unsupported; on-demand lookup caching is memory-only".into(),
367        ));
368    }
369
370    let cache_ttl = options
371        .get("cache.ttl")
372        .map(|s| {
373            s.parse::<u64>()
374                .map_err(|_| ParseError::ValidationError(format!("invalid cache.ttl: '{s}'")))
375        })
376        .transpose()?;
377
378    let pushdown_mode = match options.get("pushdown") {
379        Some(s) => PushdownMode::parse(s)?,
380        None => PushdownMode::default(),
381    };
382
383    if strategy == LookupStrategy::Replicated && (cache_memory.is_some() || cache_ttl.is_some()) {
384        return Err(ParseError::ValidationError(
385            "cache.memory and cache.ttl require strategy = 'on-demand'".into(),
386        ));
387    }
388
389    Ok(LookupTableProperties {
390        connector,
391        strategy,
392        cache_memory,
393        cache_ttl,
394        pushdown_mode,
395    })
396}
397
398#[cfg(test)]
399mod tests;
laminar_sql/parser/lookup_table/mod.rs

laminar_sql/parser/lookup_table/
mod.rs
