laminar_sql/parser/
lookup_table.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 {
400    use super::*;
401    use crate::parser::StreamingParser;
402    use crate::parser::StreamingStatement;
403
404    /// Helper to parse SQL and return the first statement.
405    fn parse_one(sql: &str) -> StreamingStatement {
406        let stmts = StreamingParser::parse_sql(sql).unwrap();
407        assert_eq!(stmts.len(), 1, "Expected exactly 1 statement");
408        stmts.into_iter().next().unwrap()
409    }
410
411    #[test]
412    fn test_parse_basic_create_lookup_table() {
413        let stmt = parse_one(
414            "CREATE LOOKUP TABLE instruments (
415                symbol VARCHAR NOT NULL,
416                name VARCHAR,
417                PRIMARY KEY (symbol)
418            ) WITH (
419                'connector' = 'catalog-source',
420                'endpoint' = 'https://catalog.example'
421            )",
422        );
423        match stmt {
424            StreamingStatement::CreateLookupTable(lt) => {
425                assert_eq!(lt.name.to_string(), "instruments");
426                assert_eq!(lt.columns.len(), 2);
427                assert_eq!(lt.primary_key, vec!["symbol"]);
428                assert!(!lt.or_replace);
429                assert!(!lt.if_not_exists);
430                assert_eq!(
431                    lt.with_options.get("connector"),
432                    Some(&"catalog-source".to_string())
433                );
434            }
435            _ => panic!("Expected CreateLookupTable, got {stmt:?}"),
436        }
437    }
438
439    #[test]
440    fn test_parse_or_replace_and_if_not_exists() {
441        let stmt = parse_one(
442            "CREATE OR REPLACE LOOKUP TABLE IF NOT EXISTS dims (
443                id INT,
444                PRIMARY KEY (id)
445            ) WITH (
446                'connector' = 'static'
447            )",
448        );
449        match stmt {
450            StreamingStatement::CreateLookupTable(lt) => {
451                assert!(lt.or_replace);
452                assert!(lt.if_not_exists);
453            }
454            _ => panic!("Expected CreateLookupTable, got {stmt:?}"),
455        }
456    }
457
458    #[test]
459    fn test_parse_with_primary_key() {
460        let stmt = parse_one(
461            "CREATE LOOKUP TABLE t (
462                a INT,
463                b VARCHAR,
464                c FLOAT,
465                PRIMARY KEY (a, b)
466            ) WITH ('connector' = 'static')",
467        );
468        match stmt {
469            StreamingStatement::CreateLookupTable(lt) => {
470                assert_eq!(lt.primary_key, vec!["a", "b"]);
471                assert_eq!(lt.columns.len(), 3);
472            }
473            _ => panic!("Expected CreateLookupTable, got {stmt:?}"),
474        }
475    }
476
477    #[test]
478    fn test_parse_with_clause_properties() {
479        let stmt = parse_one(
480            "CREATE LOOKUP TABLE t (
481                id INT,
482                PRIMARY KEY (id)
483            ) WITH (
484                'connector' = 'mock-direct',
485                'connection' = 'endpoint=localhost',
486                'strategy' = 'on-demand',
487                'cache.memory' = '512mb',
488                'pushdown' = 'auto'
489            )",
490        );
491        match stmt {
492            StreamingStatement::CreateLookupTable(lt) => {
493                let props = validate_properties(&lt.with_options).unwrap();
494                assert_eq!(
495                    props.connector,
496                    LookupConnector::External("mock-direct".into())
497                );
498                assert_eq!(
499                    lt.with_options.get("connection").map(String::as_str),
500                    Some("endpoint=localhost")
501                );
502                assert_eq!(props.strategy, LookupStrategy::OnDemand);
503                assert_eq!(props.cache_memory, Some(ByteSize(512 * 1024 * 1024)));
504                assert_eq!(props.pushdown_mode, PushdownMode::Auto);
505            }
506            _ => panic!("Expected CreateLookupTable, got {stmt:?}"),
507        }
508    }
509
510    #[test]
511    fn test_parse_drop_lookup_table() {
512        let stmt = parse_one("DROP LOOKUP TABLE instruments");
513        match stmt {
514            StreamingStatement::DropLookupTable { name, if_exists } => {
515                assert_eq!(name.to_string(), "instruments");
516                assert!(!if_exists);
517            }
518            _ => panic!("Expected DropLookupTable, got {stmt:?}"),
519        }
520    }
521
522    #[test]
523    fn test_parse_drop_lookup_table_if_exists() {
524        let stmt = parse_one("DROP LOOKUP TABLE IF EXISTS instruments");
525        match stmt {
526            StreamingStatement::DropLookupTable { name, if_exists } => {
527                assert_eq!(name.to_string(), "instruments");
528                assert!(if_exists);
529            }
530            _ => panic!("Expected DropLookupTable, got {stmt:?}"),
531        }
532    }
533
534    #[test]
535    fn test_byte_size_parsing() {
536        assert_eq!(
537            ByteSize::parse("512mb").unwrap(),
538            ByteSize(512 * 1024 * 1024)
539        );
540        assert_eq!(
541            ByteSize::parse("1gb").unwrap(),
542            ByteSize(1024 * 1024 * 1024)
543        );
544        assert_eq!(ByteSize::parse("10kb").unwrap(), ByteSize(10 * 1024));
545        assert_eq!(ByteSize::parse("100b").unwrap(), ByteSize(100));
546        assert_eq!(ByteSize::parse("1024").unwrap(), ByteSize(1024));
547        assert_eq!(
548            ByteSize::parse("2tb").unwrap(),
549            ByteSize(2 * 1024 * 1024 * 1024 * 1024)
550        );
551    }
552
553    #[test]
554    fn lookup_connector_parsing_is_provider_neutral() {
555        assert_eq!(
556            LookupConnector::parse(" STATIC ").unwrap(),
557            LookupConnector::Static
558        );
559        assert_eq!(
560            LookupConnector::parse("CuStOm-SrC").unwrap(),
561            LookupConnector::External("custom-src".to_string())
562        );
563        assert_eq!(
564            LookupConnector::parse("memory").unwrap(),
565            LookupConnector::External("memory".to_string())
566        );
567        assert_eq!(
568            LookupConnector::parse("alpha_lookup").unwrap(),
569            LookupConnector::External("alpha_lookup".to_string())
570        );
571        assert!(LookupConnector::parse("  ").is_err());
572    }
573
574    #[test]
575    fn test_error_missing_columns() {
576        let result =
577            StreamingParser::parse_sql("CREATE LOOKUP TABLE t () WITH ('connector' = 'static')");
578        assert!(result.is_err());
579    }
580
581    #[test]
582    fn test_error_missing_with_clause() {
583        let result = StreamingParser::parse_sql("CREATE LOOKUP TABLE t (id INT, PRIMARY KEY (id))");
584        assert!(result.is_err());
585    }
586
587    #[test]
588    fn test_error_invalid_property() {
589        let mut options = HashMap::new();
590        options.insert("connector".to_string(), "catalog-source".to_string());
591        options.insert("strategy".to_string(), "invalid-strategy".to_string());
592        let result = validate_properties(&options);
593        assert!(result.is_err());
594    }
595
596    #[test]
597    fn lookup_strategy_and_cache_surface_is_strict() {
598        assert_eq!(
599            LookupStrategy::parse("replicated").unwrap(),
600            LookupStrategy::Replicated
601        );
602        assert_eq!(
603            LookupStrategy::parse("on-demand").unwrap(),
604            LookupStrategy::OnDemand
605        );
606        for unsupported in [
607            "partitioned",
608            "sharded",
609            "full",
610            "poll",
611            "snapshot",
612            "cdc",
613            "on_demand",
614            "lazy",
615            "manual",
616        ] {
617            assert!(LookupStrategy::parse(unsupported).is_err(), "{unsupported}");
618        }
619
620        let mut replicated_cache = HashMap::from([
621            ("connector".to_string(), "catalog-source".to_string()),
622            ("strategy".to_string(), "replicated".to_string()),
623            ("cache.memory".to_string(), "1mb".to_string()),
624        ]);
625        assert!(validate_properties(&replicated_cache).is_err());
626        replicated_cache.insert("strategy".into(), "on-demand".into());
627        assert!(validate_properties(&replicated_cache).is_ok());
628        replicated_cache.insert("cache.disk".into(), "1gb".into());
629        assert!(validate_properties(&replicated_cache).is_err());
630    }
631}
laminar_sql/parser/lookup_table.rs

laminar_sql/parser/
lookup_table.rs
