Skip to main content

reddb_rql/parser/
ddl.rs

1//! DDL SQL Parser: CREATE TABLE, DROP TABLE, ALTER TABLE
2
3use super::error::ParseError;
4use super::Parser;
5use crate::ast::{
6    AlterOperation, AlterTableQuery, CreateCollectionQuery, CreateColumnDef, CreateTableQuery,
7    CreateVcsRefQuery, CreateVectorQuery, DropCollectionQuery, DropDocumentQuery, DropGraphQuery,
8    DropKvQuery, DropTableQuery, DropVcsRefQuery, DropVectorQuery, ExplainAlterQuery,
9    ExplainFormat, PartitionKind, PartitionSpec, QueryExpr, TruncateQuery, VcsRefKind,
10};
11use crate::lexer::Token;
12use reddb_types::catalog::{CollectionModel, SubscriptionDescriptor, SubscriptionOperation};
13use reddb_types::types::{SqlTypeName, TypeModifier, Value};
14
15impl<'a> Parser<'a> {
16    /// Parse: CREATE TABLE [IF NOT EXISTS] name (col1 TYPE [modifiers], ...)
17    pub fn parse_create_table_query(&mut self) -> Result<QueryExpr, ParseError> {
18        self.expect(Token::Create)?;
19        self.expect(Token::Table)?;
20
21        let if_not_exists = self.match_if_not_exists()?;
22        let name = self.expect_ident()?;
23
24        self.expect(Token::LParen)?;
25        let mut columns = Vec::new();
26        loop {
27            let col = self.parse_column_def()?;
28            columns.push(col);
29            if !self.consume(&Token::Comma)? {
30                break;
31            }
32        }
33        self.expect(Token::RParen)?;
34
35        let mut default_ttl_ms = None;
36        let mut context_index_fields = Vec::new();
37        let mut context_index_enabled = false;
38        let mut timestamps = false;
39        let mut subscriptions = Vec::new();
40
41        while self.consume(&Token::With)? {
42            if self.consume_ident_ci("EVENTS")? {
43                subscriptions.push(self.parse_subscription_descriptor(name.clone())?);
44            } else if self.consume_ident_ci("CONTEXT_INDEX")? {
45                context_index_enabled = self.parse_bool_assign()?;
46            } else if self.consume_ident_ci("CONTEXT")? {
47                // Consume INDEX token (reserved keyword)
48                if !self.consume(&Token::Index)? {
49                    return Err(ParseError::expected(
50                        vec!["INDEX"],
51                        self.peek(),
52                        self.position(),
53                    ));
54                }
55                self.expect(Token::On)?;
56                self.expect(Token::LParen)?;
57                loop {
58                    context_index_fields.push(self.expect_ident()?);
59                    if !self.consume(&Token::Comma)? {
60                        break;
61                    }
62                }
63                self.expect(Token::RParen)?;
64                context_index_enabled = true;
65            } else if self.consume_ident_ci("TIMESTAMPS")? {
66                timestamps = self.parse_bool_assign()?;
67            } else {
68                default_ttl_ms = self.parse_create_table_ttl_clause()?;
69            }
70        }
71
72        Ok(QueryExpr::CreateTable(CreateTableQuery {
73            collection_model: CollectionModel::Table,
74            name,
75            columns,
76            if_not_exists,
77            default_ttl_ms,
78            metrics_rollup_policies: Vec::new(),
79            context_index_fields,
80            context_index_enabled,
81            timestamps,
82            partition_by: None,
83            tenant_by: None,
84            append_only: false,
85            subscriptions,
86            analytics_config: Vec::new(),
87            vault_own_master_key: false,
88            ai_policy: None,
89        }))
90    }
91
92    /// Parse: DROP TABLE [IF EXISTS] name
93    pub fn parse_drop_table_query(&mut self) -> Result<QueryExpr, ParseError> {
94        self.expect(Token::Drop)?;
95        self.expect(Token::Table)?;
96        self.parse_drop_table_body()
97    }
98
99    pub fn parse_create_vcs_ref_body(&mut self, kind: VcsRefKind) -> Result<QueryExpr, ParseError> {
100        let name = self.expect_string_or_ident()?;
101        let target = match kind {
102            VcsRefKind::Branch => {
103                if self.consume(&Token::From)? {
104                    Some(self.expect_string_or_ident()?)
105                } else {
106                    None
107                }
108            }
109            VcsRefKind::Tag => {
110                if self.consume_ident_ci("AT")? {
111                    Some(self.expect_string_or_ident()?)
112                } else {
113                    None
114                }
115            }
116        };
117        Ok(QueryExpr::CreateVcsRef(CreateVcsRefQuery {
118            kind,
119            name,
120            target,
121        }))
122    }
123
124    pub fn parse_drop_vcs_ref_body(&mut self, kind: VcsRefKind) -> Result<QueryExpr, ParseError> {
125        let name = self.expect_string_or_ident()?;
126        Ok(QueryExpr::DropVcsRef(DropVcsRefQuery { kind, name }))
127    }
128
129    /// Parse the body of CREATE TABLE after CREATE TABLE has been consumed
130    pub fn parse_create_table_body(&mut self) -> Result<QueryExpr, ParseError> {
131        let if_not_exists = self.match_if_not_exists()?;
132        let name = self.expect_ident()?;
133
134        self.expect(Token::LParen)?;
135        let mut columns = Vec::new();
136        loop {
137            let col = self.parse_column_def()?;
138            columns.push(col);
139            if !self.consume(&Token::Comma)? {
140                break;
141            }
142        }
143        self.expect(Token::RParen)?;
144
145        let mut default_ttl_ms = None;
146        let mut context_index_fields = Vec::new();
147        let mut context_index_enabled = false;
148        let mut timestamps = false;
149        let mut tenant_by: Option<String> = None;
150        let mut append_only = false;
151        let mut subscriptions = Vec::new();
152        let mut ai_policy = reddb_types::catalog::AiPolicy::default();
153
154        while self.consume(&Token::With)? {
155            if self.consume_ident_ci("EVENTS")? {
156                subscriptions.push(self.parse_subscription_descriptor(name.clone())?);
157                continue;
158            }
159            // Accept both spellings:
160            //   WITH key = value
161            //   WITH (key = value, key = value)
162            // Postgres / ClickHouse use the parenthesised form; the
163            // bare form is our legacy shorthand. The parenthesised
164            // form collects options separated by commas until `)`.
165            let has_parens = self.consume(&Token::LParen)?;
166
167            loop {
168                if self.consume_ident_ci("CONTEXT_INDEX")? {
169                    context_index_enabled = self.parse_bool_assign()?;
170                } else if self.consume_ident_ci("CONTEXT")? {
171                    if !self.consume(&Token::Index)? {
172                        return Err(ParseError::expected(
173                            vec!["INDEX"],
174                            self.peek(),
175                            self.position(),
176                        ));
177                    }
178                    self.expect(Token::On)?;
179                    self.expect(Token::LParen)?;
180                    loop {
181                        context_index_fields.push(self.expect_ident()?);
182                        if !self.consume(&Token::Comma)? {
183                            break;
184                        }
185                    }
186                    self.expect(Token::RParen)?;
187                    context_index_enabled = true;
188                } else if self.consume_ident_ci("TIMESTAMPS")? {
189                    timestamps = self.parse_bool_assign()?;
190                } else if self.consume_ident_ci("EMBED")? {
191                    if ai_policy.embed.is_some() {
192                        return Err(ParseError::new(
193                            "duplicate EMBED clause in AI policy".to_string(),
194                            self.position(),
195                        ));
196                    }
197                    ai_policy.embed = Some(self.parse_ai_embed_policy()?);
198                } else if self.consume_ident_ci("MODERATE")? {
199                    if ai_policy.moderate.is_some() {
200                        return Err(ParseError::new(
201                            "duplicate MODERATE clause in AI policy".to_string(),
202                            self.position(),
203                        ));
204                    }
205                    ai_policy.moderate = Some(self.parse_ai_moderate_policy()?);
206                } else if self.consume_ident_ci("VISION")? {
207                    if ai_policy.vision.is_some() {
208                        return Err(ParseError::new(
209                            "duplicate VISION clause in AI policy".to_string(),
210                            self.position(),
211                        ));
212                    }
213                    ai_policy.vision = Some(self.parse_ai_vision_policy()?);
214                } else if self.consume_ident_ci("APPEND_ONLY")? {
215                    append_only = self.parse_bool_assign()?;
216                } else if self.consume_ident_ci("TENANT_BY")? {
217                    // `WITH (tenant_by = 'col')` form — accepts `=` optional
218                    // and expects a string literal column name.
219                    let _ = self.consume(&Token::Eq)?;
220                    let value = self.parse_literal_value()?;
221                    match value {
222                        Value::Text(col) => tenant_by = Some(col.to_string()),
223                        other => {
224                            return Err(ParseError::new(
225                                format!("WITH tenant_by expects a text literal, got {other:?}"),
226                                self.position(),
227                            ));
228                        }
229                    }
230                } else {
231                    default_ttl_ms = self.parse_create_table_ttl_clause()?;
232                }
233                if has_parens {
234                    if self.consume(&Token::Comma)? {
235                        continue;
236                    }
237                    self.expect(Token::RParen)?;
238                }
239                break;
240            }
241        }
242
243        // Optional `PARTITION BY RANGE|LIST|HASH (col)` clause (Phase 2.2).
244        let partition_by = if self.consume(&Token::Partition)? {
245            self.expect(Token::By)?;
246            let kind = if self.consume(&Token::Range)? {
247                PartitionKind::Range
248            } else if self.consume(&Token::List)? {
249                PartitionKind::List
250            } else if self.consume(&Token::Hash)? {
251                PartitionKind::Hash
252            } else {
253                return Err(ParseError::expected(
254                    vec!["RANGE", "LIST", "HASH"],
255                    self.peek(),
256                    self.position(),
257                ));
258            };
259            self.expect(Token::LParen)?;
260            let column = self.expect_ident()?;
261            self.expect(Token::RParen)?;
262            Some(PartitionSpec { kind, column })
263        } else {
264            None
265        };
266
267        // Shorthand: trailing `APPEND ONLY` keyword pair (PG / ClickHouse
268        // style). Accepted after partition spec / tenant spec / or on
269        // its own. `WITH (append_only = true)` is the other form and
270        // handled above.
271        if !append_only && self.consume_ident_ci("APPEND")? {
272            if !self.consume_ident_ci("ONLY")? {
273                return Err(ParseError::expected(
274                    vec!["ONLY"],
275                    self.peek(),
276                    self.position(),
277                ));
278            }
279            append_only = true;
280        }
281
282        // Shorthand: `TENANT BY (col)` or `TENANT BY (root.sub.path)`
283        // trailing clause (after partition spec if both are used).
284        //
285        // Dotted paths let non-table models declare tenancy over their
286        // natural nested structures — `metadata.tenant` for vectors,
287        // `payload.tenant` for queue messages, `tags.cluster` for
288        // timeseries, `properties.org` for graphs. The read-path
289        // resolver already navigates these paths via
290        // `resolve_runtime_document_path`; here we just store the
291        // dotted string and let the policy evaluator do the rest.
292        if tenant_by.is_none() && self.consume_ident_ci("TENANT")? {
293            self.expect(Token::By)?;
294            self.expect(Token::LParen)?;
295            // Allow keyword-idents (`metadata`, `type`, `data`) as
296            // column names — SQL treats them as bare identifiers in
297            // this context.
298            let mut path = self.expect_ident_or_keyword()?;
299            while self.consume(&Token::Dot)? {
300                let next = self.expect_ident_or_keyword()?;
301                path = format!("{path}.{next}");
302            }
303            self.expect(Token::RParen)?;
304            tenant_by = Some(path);
305        }
306
307        Ok(QueryExpr::CreateTable(CreateTableQuery {
308            collection_model: CollectionModel::Table,
309            name,
310            columns,
311            if_not_exists,
312            default_ttl_ms,
313            metrics_rollup_policies: Vec::new(),
314            context_index_fields,
315            context_index_enabled,
316            timestamps,
317            partition_by,
318            tenant_by,
319            append_only,
320            subscriptions,
321            analytics_config: Vec::new(),
322            vault_own_master_key: false,
323            ai_policy: (!ai_policy.is_empty()).then_some(ai_policy),
324        }))
325    }
326
327    /// Parse: EXPLAIN ALTER FOR CREATE TABLE name (...) [FORMAT JSON|SQL]
328    ///
329    /// Pure read: does not execute DDL. Returns a schema-diff rendering of the
330    /// difference between the table's current contract and the target CREATE
331    /// TABLE body.
332    pub fn parse_explain_alter_query(&mut self) -> Result<QueryExpr, ParseError> {
333        self.expect(Token::Explain)?;
334        self.expect(Token::Alter)?;
335        self.expect(Token::For)?;
336        self.expect(Token::Create)?;
337        self.expect(Token::Table)?;
338
339        let body = self.parse_create_table_body()?;
340        let target = match body {
341            QueryExpr::CreateTable(t) => t,
342            _ => {
343                return Err(ParseError::new(
344                    "EXPLAIN ALTER FOR CREATE TABLE body must be a CREATE TABLE statement"
345                        .to_string(),
346                    self.position(),
347                ));
348            }
349        };
350
351        let format = if self.consume(&Token::Format)? {
352            if self.consume(&Token::Json)? {
353                ExplainFormat::Json
354            } else if self.consume_ident_ci("SQL")? {
355                ExplainFormat::Sql
356            } else {
357                return Err(ParseError::expected(
358                    vec!["JSON", "SQL"],
359                    self.peek(),
360                    self.position(),
361                ));
362            }
363        } else {
364            ExplainFormat::Sql
365        };
366
367        Ok(QueryExpr::ExplainAlter(ExplainAlterQuery {
368            target,
369            format,
370        }))
371    }
372
373    /// Parse the body of DROP TABLE after DROP TABLE has been consumed
374    pub fn parse_drop_table_body(&mut self) -> Result<QueryExpr, ParseError> {
375        let if_exists = self.match_if_exists()?;
376        let name = self.parse_drop_collection_name()?;
377        Ok(QueryExpr::DropTable(DropTableQuery { name, if_exists }))
378    }
379
380    pub fn parse_drop_graph_body(&mut self) -> Result<QueryExpr, ParseError> {
381        let if_exists = self.match_if_exists()?;
382        let name = self.parse_drop_collection_name()?;
383        Ok(QueryExpr::DropGraph(DropGraphQuery { name, if_exists }))
384    }
385
386    pub fn parse_drop_vector_body(&mut self) -> Result<QueryExpr, ParseError> {
387        let if_exists = self.match_if_exists()?;
388        let name = self.parse_drop_collection_name()?;
389        Ok(QueryExpr::DropVector(DropVectorQuery { name, if_exists }))
390    }
391
392    pub fn parse_drop_document_body(&mut self) -> Result<QueryExpr, ParseError> {
393        let if_exists = self.match_if_exists()?;
394        let name = self.parse_drop_collection_name()?;
395        Ok(QueryExpr::DropDocument(DropDocumentQuery {
396            name,
397            if_exists,
398        }))
399    }
400
401    pub fn parse_create_keyed_body(
402        &mut self,
403        model: CollectionModel,
404    ) -> Result<QueryExpr, ParseError> {
405        let if_not_exists = self.match_if_not_exists()?;
406        let name = self.parse_drop_collection_name()?;
407        let vault_own_master_key =
408            if model == CollectionModel::Vault && self.consume(&Token::With)? {
409                if !self.consume_ident_ci("OWN")? {
410                    return Err(ParseError::expected(
411                        vec!["OWN"],
412                        self.peek(),
413                        self.position(),
414                    ));
415                }
416                if !self.consume_ident_ci("MASTER")? {
417                    return Err(ParseError::expected(
418                        vec!["MASTER"],
419                        self.peek(),
420                        self.position(),
421                    ));
422                }
423                if !self.consume(&Token::Key)? && !self.consume_ident_ci("KEY")? {
424                    return Err(ParseError::expected(
425                        vec!["KEY"],
426                        self.peek(),
427                        self.position(),
428                    ));
429                }
430                true
431            } else {
432                false
433            };
434        // `CREATE GRAPH <name> WITH ANALYTICS (...)` — the analytics opt-in
435        // is graph-only (issue #800). Other keyed models reject the clause so
436        // a misplaced `WITH ANALYTICS` fails loudly instead of being ignored.
437        let analytics_config = if model == CollectionModel::Graph && self.consume(&Token::With)? {
438            if !self.consume_ident_ci("ANALYTICS")? {
439                return Err(ParseError::expected(
440                    vec!["ANALYTICS"],
441                    self.peek(),
442                    self.position(),
443                ));
444            }
445            self.parse_analytics_clause()?
446        } else {
447            Vec::new()
448        };
449        Ok(QueryExpr::CreateTable(CreateTableQuery {
450            collection_model: model,
451            name,
452            columns: Vec::new(),
453            if_not_exists,
454            default_ttl_ms: None,
455            metrics_rollup_policies: Vec::new(),
456            context_index_fields: Vec::new(),
457            context_index_enabled: false,
458            timestamps: false,
459            partition_by: None,
460            tenant_by: None,
461            append_only: false,
462            subscriptions: Vec::new(),
463            analytics_config,
464            vault_own_master_key,
465            ai_policy: None,
466        }))
467    }
468
469    /// Parse the `( <output> [ ( <key> = <value> [, ...] ) ] [, ...] )` body of
470    /// a `WITH ANALYTICS` clause (issue #800). Recognised outputs are
471    /// `communities`, `components`, `centrality`; recognised options are
472    /// `using`, `resolution`, `max_iterations`, `tolerance`. Unknown output
473    /// names and option keys are rejected with a clear, structured error.
474    fn parse_analytics_clause(
475        &mut self,
476    ) -> Result<Vec<reddb_types::catalog::AnalyticsViewDescriptor>, ParseError> {
477        use reddb_types::catalog::{AnalyticsOutput, AnalyticsViewDescriptor};
478
479        self.expect(Token::LParen)?;
480        let mut views: Vec<AnalyticsViewDescriptor> = Vec::new();
481        loop {
482            let output_name = self.parse_analytics_output_name()?;
483            let output = AnalyticsOutput::from_str(&output_name).ok_or_else(|| {
484                ParseError::new(
485                    format!(
486                        "unknown analytics output '{output_name}': expected communities, components, or centrality"
487                    ),
488                    self.position(),
489                )
490            })?;
491            if views.iter().any(|view| view.output == output) {
492                return Err(ParseError::new(
493                    format!("duplicate analytics output '{output_name}'"),
494                    self.position(),
495                ));
496            }
497            let mut view = AnalyticsViewDescriptor {
498                output,
499                algorithm: None,
500                resolution: None,
501                max_iterations: None,
502                tolerance: None,
503            };
504            if self.consume(&Token::LParen)? {
505                loop {
506                    let key = self.parse_analytics_option_key()?;
507                    self.expect(Token::Eq)?;
508                    match key.as_str() {
509                        "using" => {
510                            view.algorithm =
511                                Some(self.expect_ident_or_keyword()?.to_ascii_lowercase());
512                        }
513                        "resolution" => view.resolution = Some(self.parse_float()?),
514                        "max_iterations" => view.max_iterations = Some(self.parse_integer()?),
515                        "tolerance" => view.tolerance = Some(self.parse_float()?),
516                        other => {
517                            return Err(ParseError::new(
518                                format!(
519                                    "unknown analytics option '{other}': expected using, resolution, max_iterations, or tolerance"
520                                ),
521                                self.position(),
522                            ))
523                        }
524                    }
525                    if !self.consume(&Token::Comma)? {
526                        break;
527                    }
528                }
529                self.expect(Token::RParen)?;
530            }
531            views.push(view);
532            if !self.consume(&Token::Comma)? {
533                break;
534            }
535        }
536        self.expect(Token::RParen)?;
537        if views.is_empty() {
538            return Err(ParseError::new(
539                "WITH ANALYTICS requires at least one output".to_string(),
540                self.position(),
541            ));
542        }
543        Ok(views)
544    }
545
546    /// Read one analytics output name, normalising the keyword-lexed outputs
547    /// (`components`, `centrality`) back to their lowercase spelling so they
548    /// compare uniformly with the ident-lexed `communities`.
549    fn parse_analytics_output_name(&mut self) -> Result<String, ParseError> {
550        match self.peek() {
551            Token::Components => {
552                self.advance()?;
553                Ok("components".to_string())
554            }
555            Token::Centrality => {
556                self.advance()?;
557                Ok("centrality".to_string())
558            }
559            _ => Ok(self.expect_ident()?.to_ascii_lowercase()),
560        }
561    }
562
563    /// Read one analytics option key, normalising the keyword-lexed keys
564    /// (`using`, `max_iterations`) back to their lowercase spelling.
565    fn parse_analytics_option_key(&mut self) -> Result<String, ParseError> {
566        match self.peek() {
567            Token::Using => {
568                self.advance()?;
569                Ok("using".to_string())
570            }
571            Token::MaxIterations => {
572                self.advance()?;
573                Ok("max_iterations".to_string())
574            }
575            _ => Ok(self.expect_ident()?.to_ascii_lowercase()),
576        }
577    }
578
579    pub fn parse_create_collection_model_body(
580        &mut self,
581        model: CollectionModel,
582    ) -> Result<QueryExpr, ParseError> {
583        self.parse_create_keyed_body(model)
584    }
585
586    pub fn parse_create_collection_body(&mut self) -> Result<QueryExpr, ParseError> {
587        let if_not_exists = self.match_if_not_exists()?;
588        let name = self.parse_drop_collection_name()?;
589        if !self.consume_ident_ci("KIND")? {
590            return Err(ParseError::expected(
591                vec!["KIND"],
592                self.peek(),
593                self.position(),
594            ));
595        }
596        let mut kind = self.expect_ident_or_keyword()?.to_ascii_lowercase();
597        while self.consume(&Token::Dot)? {
598            let part = self.expect_ident_or_keyword()?.to_ascii_lowercase();
599            kind.push('.');
600            kind.push_str(&part);
601        }
602        let (vector_dimension, vector_metric) = if kind == "vector.turbo" {
603            if !self.consume_ident_ci("DIM")? {
604                return Err(ParseError::expected(
605                    vec!["DIM"],
606                    self.peek(),
607                    self.position(),
608                ));
609            }
610            let dimension = self.parse_integer()?;
611            if dimension <= 0 {
612                return Err(ParseError::new(
613                    "VECTOR DIM must be a positive integer".to_string(),
614                    self.position(),
615                ));
616            }
617            let metric = if self.consume(&Token::Metric)? {
618                self.parse_distance_metric()?
619            } else {
620                reddb_types::distance::DistanceMetric::Cosine
621            };
622            (Some(dimension as usize), Some(metric))
623        } else {
624            (None, None)
625        };
626        let allowed_signers = if self.consume_ident_ci("SIGNED_BY")? {
627            self.parse_signed_by_list()?
628        } else {
629            Vec::new()
630        };
631        Ok(QueryExpr::CreateCollection(CreateCollectionQuery {
632            name,
633            kind,
634            if_not_exists,
635            vector_dimension,
636            vector_metric,
637            allowed_signers,
638        }))
639    }
640
641    /// Parse a single `'hex32'` string literal as a 32-byte Ed25519
642    /// pubkey. Used by `ALTER COLLECTION ... ADD|REVOKE SIGNER 'hex'`
643    /// (issue #522).
644    fn parse_single_signer_hex(&mut self) -> Result<[u8; 32], ParseError> {
645        let hex = match self.peek().clone() {
646            Token::String(s) => {
647                self.advance()?;
648                s
649            }
650            _ => {
651                return Err(ParseError::expected(
652                    vec!["string literal (ed25519 pubkey hex)"],
653                    self.peek(),
654                    self.position(),
655                ));
656            }
657        };
658        decode_hex_32(&hex).map_err(|msg| {
659            ParseError::new(
660                format!("SIGNER pubkey '{hex}' invalid: {msg}"),
661                self.position(),
662            )
663        })
664    }
665
666    /// Parse `( 'hex32', 'hex32', ... )` — Ed25519 pubkey list. Each entry
667    /// must decode to exactly 32 bytes. Used by both `CREATE COLLECTION ...
668    /// SIGNED_BY (...)` and (in a later iteration) `ALTER COLLECTION` signer
669    /// mutations. Issue #520.
670    fn parse_signed_by_list(&mut self) -> Result<Vec<[u8; 32]>, ParseError> {
671        self.expect(Token::LParen)?;
672        let mut out = Vec::new();
673        loop {
674            let hex = match self.peek().clone() {
675                Token::String(s) => {
676                    self.advance()?;
677                    s
678                }
679                _ => {
680                    return Err(ParseError::expected(
681                        vec!["string literal (ed25519 pubkey hex)"],
682                        self.peek(),
683                        self.position(),
684                    ));
685                }
686            };
687            let bytes = decode_hex_32(&hex).map_err(|msg| {
688                ParseError::new(
689                    format!("SIGNED_BY pubkey '{hex}' invalid: {msg}"),
690                    self.position(),
691                )
692            })?;
693            out.push(bytes);
694            if !self.consume(&Token::Comma)? {
695                break;
696            }
697        }
698        self.expect(Token::RParen)?;
699        if out.is_empty() {
700            return Err(ParseError::new(
701                "SIGNED_BY list must contain at least one pubkey".to_string(),
702                self.position(),
703            ));
704        }
705        Ok(out)
706    }
707
708    pub fn parse_create_vector_body(&mut self) -> Result<QueryExpr, ParseError> {
709        let if_not_exists = self.match_if_not_exists()?;
710        let name = self.parse_drop_collection_name()?;
711        if !self.consume_ident_ci("DIM")? {
712            return Err(ParseError::expected(
713                vec!["DIM"],
714                self.peek(),
715                self.position(),
716            ));
717        }
718        let dimension = self.parse_integer()?;
719        if dimension <= 0 {
720            return Err(ParseError::new(
721                "VECTOR DIM must be a positive integer".to_string(),
722                self.position(),
723            ));
724        }
725        let metric = if self.consume(&Token::Metric)? {
726            self.parse_distance_metric()?
727        } else {
728            reddb_types::distance::DistanceMetric::Cosine
729        };
730        Ok(QueryExpr::CreateVector(CreateVectorQuery {
731            name,
732            dimension: dimension as usize,
733            metric,
734            if_not_exists,
735        }))
736    }
737
738    pub fn parse_drop_keyed_body(
739        &mut self,
740        model: CollectionModel,
741    ) -> Result<QueryExpr, ParseError> {
742        let if_exists = self.match_if_exists()?;
743        let name = self.parse_drop_collection_name()?;
744        Ok(QueryExpr::DropKv(DropKvQuery {
745            name,
746            if_exists,
747            model,
748        }))
749    }
750
751    pub fn parse_drop_kv_body(&mut self) -> Result<QueryExpr, ParseError> {
752        self.parse_drop_keyed_body(CollectionModel::Kv)
753    }
754
755    pub fn parse_drop_collection_body(&mut self) -> Result<QueryExpr, ParseError> {
756        self.parse_drop_collection_model_body(None)
757    }
758
759    pub fn parse_drop_collection_model_body(
760        &mut self,
761        model: Option<CollectionModel>,
762    ) -> Result<QueryExpr, ParseError> {
763        let if_exists = self.match_if_exists()?;
764        let name = self.parse_drop_collection_name()?;
765        Ok(QueryExpr::DropCollection(DropCollectionQuery {
766            name,
767            if_exists,
768            model,
769        }))
770    }
771
772    pub fn parse_truncate_body(
773        &mut self,
774        model: Option<CollectionModel>,
775    ) -> Result<QueryExpr, ParseError> {
776        let if_exists = self.match_if_exists()?;
777        let name = self.parse_drop_collection_name()?;
778        Ok(QueryExpr::Truncate(TruncateQuery {
779            name,
780            model,
781            if_exists,
782        }))
783    }
784
785    pub(crate) fn parse_drop_collection_name(&mut self) -> Result<String, ParseError> {
786        let mut name = self.expect_ident()?;
787        while self.consume(&Token::Dot)? {
788            if self.consume(&Token::Star)? {
789                name.push_str(".*");
790                break;
791            }
792            let next = self.expect_ident_or_keyword()?;
793            name = format!("{name}.{next}");
794        }
795        Ok(name)
796    }
797
798    /// Parse: ALTER TABLE name ADD/DROP/RENAME COLUMN ...
799    ///
800    /// Also accepts `ALTER COLLECTION name ADD|REVOKE SIGNER 'hex'`
801    /// (issue #522) — collection-level signer registry mutations share
802    /// the AlterTable AST so the existing executor dispatch path picks
803    /// them up without a new top-level variant.
804    pub fn parse_alter_table_query(&mut self) -> Result<QueryExpr, ParseError> {
805        self.expect(Token::Alter)?;
806        if !self.consume(&Token::Table)?
807            && !self.consume(&Token::Collection)?
808            && !self.consume_ident_ci("COLLECTION")?
809        {
810            return Err(ParseError::expected(
811                vec!["TABLE", "COLLECTION"],
812                self.peek(),
813                self.position(),
814            ));
815        }
816        let name = self.expect_ident()?;
817
818        let mut operations = Vec::new();
819        loop {
820            let op = self.parse_alter_operation(&name)?;
821            operations.push(op);
822            if !self.consume(&Token::Comma)? {
823                break;
824            }
825        }
826
827        Ok(QueryExpr::AlterTable(AlterTableQuery { name, operations }))
828    }
829
830    /// Parse: `ALTER GRAPH <name> ADD ANALYTICS ( <output> [, ...] )`
831    /// and `ALTER GRAPH <name> DROP ANALYTICS <output>` (issue #801).
832    ///
833    /// Lifecycle management of the `WITH ANALYTICS` configuration declared at
834    /// `CREATE GRAPH` time (#800), without recreating the collection. Shares
835    /// the `AlterTable` AST so the existing executor dispatch path picks the
836    /// mutations up; the executor validates the target is a graph and that the
837    /// dropped output is actually enabled.
838    pub fn parse_alter_graph_query(&mut self) -> Result<QueryExpr, ParseError> {
839        self.expect(Token::Alter)?;
840        self.expect(Token::Graph)?;
841        let name = self.expect_ident()?;
842
843        let mut operations = Vec::new();
844        loop {
845            operations.push(self.parse_alter_graph_operation()?);
846            if !self.consume(&Token::Comma)? {
847                break;
848            }
849        }
850
851        Ok(QueryExpr::AlterTable(AlterTableQuery { name, operations }))
852    }
853
854    /// Parse a single `ALTER GRAPH` analytics operation: either
855    /// `ADD ANALYTICS ( ... )` or `DROP ANALYTICS <output>`.
856    fn parse_alter_graph_operation(&mut self) -> Result<AlterOperation, ParseError> {
857        if self.consume(&Token::Add)? {
858            if !self.consume_ident_ci("ANALYTICS")? {
859                return Err(ParseError::expected(
860                    vec!["ANALYTICS"],
861                    self.peek(),
862                    self.position(),
863                ));
864            }
865            // Reuse the `WITH ANALYTICS (...)` body grammar verbatim so the
866            // ADD form accepts the exact same outputs and options as CREATE.
867            let views = self.parse_analytics_clause()?;
868            Ok(AlterOperation::AddAnalytics(views))
869        } else if self.consume(&Token::Drop)? {
870            if !self.consume_ident_ci("ANALYTICS")? {
871                return Err(ParseError::expected(
872                    vec!["ANALYTICS"],
873                    self.peek(),
874                    self.position(),
875                ));
876            }
877            let output_name = self.parse_analytics_output_name()?;
878            let output = reddb_types::catalog::AnalyticsOutput::from_str(&output_name).ok_or_else(|| {
879                ParseError::new(
880                    format!(
881                        "unknown analytics output '{output_name}': expected communities, components, or centrality"
882                    ),
883                    self.position(),
884                )
885            })?;
886            Ok(AlterOperation::DropAnalytics(output))
887        } else {
888            Err(ParseError::expected(
889                vec!["ADD", "DROP"],
890                self.peek(),
891                self.position(),
892            ))
893        }
894    }
895
896    /// Parse a single ALTER TABLE operation
897    fn parse_alter_operation(&mut self, table_name: &str) -> Result<AlterOperation, ParseError> {
898        if self.consume(&Token::Add)? {
899            if self.consume_ident_ci("SUBSCRIPTION")? {
900                // ADD SUBSCRIPTION name TO queue [REDACT (...)] [WHERE ...]
901                // #1374 — a subscription name may be a keyword (e.g. `search`);
902                // accept keyword tokens as identifiers here (PG-style).
903                let sub_name = self.expect_ident_or_keyword()?;
904                let descriptor = self.parse_subscription_descriptor(table_name.to_string())?;
905                Ok(AlterOperation::AddSubscription {
906                    name: sub_name,
907                    descriptor,
908                })
909            } else if self.consume_ident_ci("SIGNER")? {
910                // ADD SIGNER 'hex_pubkey' — issue #522.
911                let pubkey = self.parse_single_signer_hex()?;
912                Ok(AlterOperation::AddSigner { pubkey })
913            } else {
914                // ADD COLUMN definition (COLUMN keyword is optional)
915                let _ = self.consume(&Token::Column)?;
916                let col_def = self.parse_column_def()?;
917                Ok(AlterOperation::AddColumn(col_def))
918            }
919        } else if self.consume_ident_ci("REVOKE")? {
920            // REVOKE SIGNER 'hex_pubkey' — issue #522.
921            if !self.consume_ident_ci("SIGNER")? {
922                return Err(ParseError::expected(
923                    vec!["SIGNER"],
924                    self.peek(),
925                    self.position(),
926                ));
927            }
928            let pubkey = self.parse_single_signer_hex()?;
929            Ok(AlterOperation::RevokeSigner { pubkey })
930        } else if self.consume(&Token::Drop)? {
931            if self.consume_ident_ci("SUBSCRIPTION")? {
932                // DROP SUBSCRIPTION name (may be a keyword, e.g. `search`)
933                let sub_name = self.expect_ident_or_keyword()?;
934                Ok(AlterOperation::DropSubscription { name: sub_name })
935            } else {
936                // DROP COLUMN name (COLUMN keyword is optional)
937                let _ = self.consume(&Token::Column)?;
938                let col_name = self.expect_ident()?;
939                Ok(AlterOperation::DropColumn(col_name))
940            }
941        } else if self.consume(&Token::Rename)? {
942            // RENAME COLUMN from TO to
943            let _ = self.consume(&Token::Column)?; // COLUMN keyword is optional
944            let from = self.expect_ident()?;
945            self.expect(Token::To)?;
946            let to = self.expect_ident()?;
947            Ok(AlterOperation::RenameColumn { from, to })
948        } else if self.consume(&Token::Attach)? {
949            // ATTACH PARTITION child FOR VALUES ...
950            self.expect(Token::Partition)?;
951            let child = self.expect_ident()?;
952            self.expect(Token::For)?;
953            // Accept `VALUES` as an ident since the grammar doesn't have it
954            // as a reserved keyword everywhere. Collect the remaining tokens
955            // as a raw bound string for round-trip persistence.
956            if !self.consume_ident_ci("VALUES")? && !self.consume(&Token::Values)? {
957                return Err(ParseError::expected(
958                    vec!["VALUES"],
959                    self.peek(),
960                    self.position(),
961                ));
962            }
963            let bound = self.collect_remaining_tokens_as_string()?;
964            Ok(AlterOperation::AttachPartition { child, bound })
965        } else if self.consume(&Token::Detach)? {
966            // DETACH PARTITION child
967            self.expect(Token::Partition)?;
968            let child = self.expect_ident()?;
969            Ok(AlterOperation::DetachPartition { child })
970        } else if self.consume(&Token::Enable)? {
971            // ENABLE EVENTS | ENABLE ROW LEVEL SECURITY | ENABLE TENANCY ON (col)
972            if self.consume_ident_ci("EVENTS")? {
973                Ok(AlterOperation::EnableEvents(
974                    self.parse_subscription_descriptor(table_name.to_string())?,
975                ))
976            } else if self.consume_ident_ci("TENANCY")? {
977                self.expect(Token::On)?;
978                self.expect(Token::LParen)?;
979                // Dotted paths allowed (`metadata.tenant`, `payload.org`).
980                let mut path = self.expect_ident_or_keyword()?;
981                while self.consume(&Token::Dot)? {
982                    let next = self.expect_ident_or_keyword()?;
983                    path = format!("{path}.{next}");
984                }
985                self.expect(Token::RParen)?;
986                Ok(AlterOperation::EnableTenancy { column: path })
987            } else {
988                self.expect(Token::Row)?;
989                self.expect(Token::Level)?;
990                self.expect(Token::Security)?;
991                Ok(AlterOperation::EnableRowLevelSecurity)
992            }
993        } else if self.consume(&Token::Disable)? {
994            // DISABLE EVENTS | DISABLE ROW LEVEL SECURITY | DISABLE TENANCY
995            if self.consume_ident_ci("EVENTS")? {
996                Ok(AlterOperation::DisableEvents)
997            } else if self.consume_ident_ci("TENANCY")? {
998                Ok(AlterOperation::DisableTenancy)
999            } else {
1000                self.expect(Token::Row)?;
1001                self.expect(Token::Level)?;
1002                self.expect(Token::Security)?;
1003                Ok(AlterOperation::DisableRowLevelSecurity)
1004            }
1005        } else if self.consume(&Token::Set)? || self.consume_ident_ci("SET")? {
1006            // SET APPEND_ONLY = true|false | SET VERSIONED = true|false
1007            // SET RETENTION <duration> (issue #580)
1008            if self.consume_ident_ci("APPEND_ONLY")? {
1009                let on = self.parse_bool_assign()?;
1010                Ok(AlterOperation::SetAppendOnly(on))
1011            } else if self.consume_ident_ci("VERSIONED")? {
1012                let on = self.parse_bool_assign()?;
1013                Ok(AlterOperation::SetVersioned(on))
1014            } else if self.consume(&Token::Retention)? {
1015                // `SET RETENTION <duration>` — reuse the same float+unit
1016                // grammar the timeseries CREATE clause uses so `7 DAYS`,
1017                // `30 m`, `1 h`, `90 d` all parse identically.
1018                let value = self.parse_float()?;
1019                let unit = self.parse_duration_unit()?;
1020                Ok(AlterOperation::SetRetention {
1021                    duration_ms: (value * unit) as u64,
1022                })
1023            } else {
1024                Err(ParseError::expected(
1025                    vec!["APPEND_ONLY", "VERSIONED", "RETENTION"],
1026                    self.peek(),
1027                    self.position(),
1028                ))
1029            }
1030        } else if self.consume_ident_ci("UNSET")? {
1031            // `UNSET RETENTION` — clears the declarative retention policy.
1032            if self.consume(&Token::Retention)? {
1033                Ok(AlterOperation::UnsetRetention)
1034            } else {
1035                Err(ParseError::expected(
1036                    vec!["RETENTION"],
1037                    self.peek(),
1038                    self.position(),
1039                ))
1040            }
1041        } else {
1042            Err(ParseError::expected(
1043                vec![
1044                    "ADD", "DROP", "RENAME", "ATTACH", "DETACH", "ENABLE", "DISABLE", "SET",
1045                    "UNSET",
1046                ],
1047                self.peek(),
1048                self.position(),
1049            ))
1050        }
1051    }
1052
1053    fn parse_subscription_descriptor(
1054        &mut self,
1055        source: String,
1056    ) -> Result<SubscriptionDescriptor, ParseError> {
1057        let mut ops_filter = Vec::new();
1058        if self.consume(&Token::LParen)? {
1059            loop {
1060                let op = if self.consume(&Token::Insert)? {
1061                    SubscriptionOperation::Insert
1062                } else if self.consume(&Token::Update)? {
1063                    SubscriptionOperation::Update
1064                } else if self.consume(&Token::Delete)? {
1065                    SubscriptionOperation::Delete
1066                } else {
1067                    return Err(ParseError::expected(
1068                        vec!["INSERT", "UPDATE", "DELETE"],
1069                        self.peek(),
1070                        self.position(),
1071                    ));
1072                };
1073                ops_filter.push(op);
1074                if !self.consume(&Token::Comma)? {
1075                    break;
1076                }
1077            }
1078            self.expect(Token::RParen)?;
1079        }
1080
1081        let target_queue = if self.consume(&Token::To)? {
1082            self.expect_ident()?
1083        } else {
1084            format!("{source}_events")
1085        };
1086
1087        let mut redact_fields = Vec::new();
1088        if self.consume_ident_ci("REDACT")? {
1089            self.expect(Token::LParen)?;
1090            loop {
1091                redact_fields.push(self.parse_dotted_redact_path()?);
1092                if !self.consume(&Token::Comma)? {
1093                    break;
1094                }
1095            }
1096            self.expect(Token::RParen)?;
1097        }
1098
1099        let where_filter = if self.consume(&Token::Where)? {
1100            Some(self.collect_subscription_where_filter()?)
1101        } else {
1102            None
1103        };
1104
1105        // ON ALL TENANTS: opt-in cluster-wide subscription (requires capability check at execution)
1106        let all_tenants = if self.consume(&Token::On)? {
1107            self.expect(Token::All)?;
1108            if !self.consume_ident_ci("TENANTS")? {
1109                return Err(ParseError::expected(
1110                    vec!["TENANTS"],
1111                    self.peek(),
1112                    self.position(),
1113                ));
1114            }
1115            true
1116        } else {
1117            false
1118        };
1119
1120        // REQUIRES CAPABILITY '...' — parsed and discarded; enforcement is at execution time
1121        if self.consume_ident_ci("REQUIRES")? {
1122            self.consume_ident_ci("CAPABILITY")?;
1123            // consume the capability string literal token
1124            self.advance()?;
1125        }
1126
1127        Ok(SubscriptionDescriptor {
1128            name: String::new(),
1129            source,
1130            target_queue,
1131            ops_filter,
1132            where_filter,
1133            redact_fields,
1134            enabled: true,
1135            all_tenants,
1136        })
1137    }
1138
1139    /// Parse a dotted redact path: `field`, `obj.field`, `obj.*.field`, etc.
1140    fn parse_dotted_redact_path(&mut self) -> Result<String, ParseError> {
1141        let mut parts = Vec::new();
1142        if self.consume(&Token::Star)? {
1143            parts.push("*".to_string());
1144        } else {
1145            parts.push(self.expect_ident_or_keyword()?);
1146        }
1147        while self.consume(&Token::Dot)? {
1148            if self.consume(&Token::Star)? {
1149                parts.push("*".to_string());
1150            } else {
1151                parts.push(self.expect_ident_or_keyword()?);
1152            }
1153        }
1154        Ok(parts.join("."))
1155    }
1156
1157    fn collect_subscription_where_filter(&mut self) -> Result<String, ParseError> {
1158        let mut parts = Vec::new();
1159        while !self.check(&Token::Eof) && !self.check(&Token::Comma) {
1160            parts.push(self.peek().to_string());
1161            self.advance()?;
1162        }
1163        if parts.is_empty() {
1164            return Err(ParseError::expected(
1165                vec!["predicate"],
1166                self.peek(),
1167                self.position(),
1168            ));
1169        }
1170        Ok(parts.join(" "))
1171    }
1172
1173    /// Capture remaining tokens as a display-joined string.
1174    ///
1175    /// Used by `ATTACH PARTITION ... FOR VALUES <bound>` to round-trip the
1176    /// bound clause into storage without needing a dedicated per-kind AST.
1177    fn collect_remaining_tokens_as_string(&mut self) -> Result<String, ParseError> {
1178        let mut parts: Vec<String> = Vec::new();
1179        while !self.check(&Token::Eof) && !self.check(&Token::Comma) {
1180            parts.push(self.peek().to_string());
1181            self.advance()?;
1182        }
1183        Ok(parts.join(" "))
1184    }
1185
1186    /// Parse a single column definition: name TYPE [NOT NULL] [DEFAULT=val] [COMPRESS:N] [UNIQUE] [PRIMARY KEY]
1187    fn parse_column_def(&mut self) -> Result<CreateColumnDef, ParseError> {
1188        let name = self.expect_column_ident()?;
1189        let sql_type = self.parse_column_type()?;
1190        let data_type = sql_type.to_string();
1191
1192        let mut def = CreateColumnDef {
1193            name,
1194            data_type,
1195            sql_type: sql_type.clone(),
1196            not_null: false,
1197            default: None,
1198            compress: None,
1199            unique: false,
1200            primary_key: false,
1201            enum_variants: sql_type.enum_variants().unwrap_or_default(),
1202            array_element: sql_type.array_element_type(),
1203            decimal_precision: sql_type.decimal_precision(),
1204        };
1205
1206        // Parse modifiers in any order
1207        loop {
1208            if self.match_not_null()? {
1209                def.not_null = true;
1210            } else if self.consume(&Token::Default)? {
1211                self.expect(Token::Eq)?;
1212                def.default = Some(self.parse_literal_string_for_ddl()?);
1213            } else if self.consume(&Token::Compress)? {
1214                self.expect(Token::Colon)?;
1215                def.compress = Some(self.parse_integer()? as u8);
1216            } else if self.consume(&Token::Unique)? {
1217                def.unique = true;
1218            } else if self.match_primary_key()? {
1219                def.primary_key = true;
1220            } else {
1221                break;
1222            }
1223        }
1224
1225        Ok(def)
1226    }
1227
1228    /// Parse column type: TEXT, INTEGER, EMAIL, ENUM('a','b','c'), ARRAY(TEXT), DECIMAL(2)
1229    fn parse_column_type(&mut self) -> Result<SqlTypeName, ParseError> {
1230        let type_name = self.expect_ident_or_keyword()?;
1231        if self.consume(&Token::LParen)? {
1232            let inner = self.parse_type_params()?;
1233            self.expect(Token::RParen)?;
1234            Ok(SqlTypeName::new(type_name).with_modifiers(inner))
1235        } else {
1236            Ok(SqlTypeName::new(type_name))
1237        }
1238    }
1239
1240    /// Parse type parameters inside parentheses: 'a','b' or TEXT or 2
1241    fn parse_type_params(&mut self) -> Result<Vec<TypeModifier>, ParseError> {
1242        let mut parts = Vec::new();
1243        loop {
1244            match self.peek().clone() {
1245                Token::String(s) => {
1246                    let s = s.clone();
1247                    self.advance()?;
1248                    parts.push(TypeModifier::StringLiteral(s));
1249                }
1250                Token::Integer(n) => {
1251                    self.advance()?;
1252                    parts.push(TypeModifier::Number(n as u32));
1253                }
1254                _ => {
1255                    parts.push(TypeModifier::Type(Box::new(self.parse_column_type()?)));
1256                }
1257            }
1258            if !self.consume(&Token::Comma)? {
1259                break;
1260            }
1261        }
1262        Ok(parts)
1263    }
1264
1265    /// Parse a literal string value for DDL DEFAULT expressions
1266    fn parse_literal_string_for_ddl(&mut self) -> Result<String, ParseError> {
1267        match self.peek().clone() {
1268            Token::String(s) => {
1269                let s = s.clone();
1270                self.advance()?;
1271                Ok(s)
1272            }
1273            Token::Integer(n) => {
1274                self.advance()?;
1275                Ok(n.to_string())
1276            }
1277            Token::Float(n) => {
1278                self.advance()?;
1279                Ok(n.to_string())
1280            }
1281            Token::True => {
1282                self.advance()?;
1283                Ok("true".to_string())
1284            }
1285            Token::False => {
1286                self.advance()?;
1287                Ok("false".to_string())
1288            }
1289            Token::Null => {
1290                self.advance()?;
1291                Ok("null".to_string())
1292            }
1293            ref other => Err(ParseError::expected(
1294                vec!["string", "number", "true", "false", "null"],
1295                other,
1296                self.position(),
1297            )),
1298        }
1299    }
1300
1301    fn check_ttl_keyword(&self) -> bool {
1302        matches!(self.peek(), Token::Ident(name) if name.eq_ignore_ascii_case("ttl"))
1303    }
1304
1305    /// Parse `= true` / `= false` after a `WITH <option>` keyword.
1306    /// Used for boolean table options like `WITH TIMESTAMPS = true`.
1307    fn parse_bool_assign(&mut self) -> Result<bool, ParseError> {
1308        self.expect(Token::Eq)?;
1309        match self.peek() {
1310            Token::True => {
1311                self.advance()?;
1312                Ok(true)
1313            }
1314            Token::False => {
1315                self.advance()?;
1316                Ok(false)
1317            }
1318            other => Err(ParseError::expected(
1319                vec!["true", "false"],
1320                other,
1321                self.position(),
1322            )),
1323        }
1324    }
1325
1326    /// Parse a parenthesised list of string literals: `('a', 'b', ...)`.
1327    /// Used for the `fields` / `outputs` AI-policy options. Requires at
1328    /// least one entry.
1329    fn parse_ai_string_list(&mut self) -> Result<Vec<String>, ParseError> {
1330        self.expect(Token::LParen)?;
1331        let mut out = Vec::new();
1332        loop {
1333            out.push(self.parse_string()?);
1334            if !self.consume(&Token::Comma)? {
1335                break;
1336            }
1337        }
1338        self.expect(Token::RParen)?;
1339        Ok(out)
1340    }
1341
1342    /// Parse a bare `true` / `false` token (after the `=` is consumed).
1343    fn parse_ai_bool(&mut self) -> Result<bool, ParseError> {
1344        match self.peek() {
1345            Token::True => {
1346                self.advance()?;
1347                Ok(true)
1348            }
1349            Token::False => {
1350                self.advance()?;
1351                Ok(false)
1352            }
1353            other => Err(ParseError::expected(
1354                vec!["true", "false"],
1355                other,
1356                self.position(),
1357            )),
1358        }
1359    }
1360
1361    /// Parse a single enum-ish word — either a quoted string or a bare
1362    /// identifier/keyword. Used for `degraded` / `on_reject` values.
1363    fn parse_ai_word(&mut self) -> Result<String, ParseError> {
1364        if matches!(self.peek(), Token::String(_)) {
1365            self.parse_string()
1366        } else {
1367            self.expect_ident_or_keyword()
1368        }
1369    }
1370
1371    /// `EMBED (fields = (...), provider = '..', model = '..')`.
1372    fn parse_ai_embed_policy(&mut self) -> Result<reddb_types::catalog::EmbedPolicy, ParseError> {
1373        self.expect(Token::LParen)?;
1374        let mut fields: Vec<String> = Vec::new();
1375        let mut provider: Option<String> = None;
1376        let mut model: Option<String> = None;
1377        loop {
1378            let key = self.expect_ident_or_keyword()?.to_ascii_lowercase();
1379            self.expect(Token::Eq)?;
1380            match key.as_str() {
1381                "fields" => fields = self.parse_ai_string_list()?,
1382                "provider" => provider = Some(self.parse_string()?),
1383                "model" => model = Some(self.parse_string()?),
1384                other => {
1385                    return Err(ParseError::new(
1386                        format!(
1387                            "unsupported EMBED policy option {other:?}; supported: fields, provider, model"
1388                        ),
1389                        self.position(),
1390                    ));
1391                }
1392            }
1393            if !self.consume(&Token::Comma)? {
1394                break;
1395            }
1396        }
1397        self.expect(Token::RParen)?;
1398        if fields.is_empty() {
1399            return Err(ParseError::new(
1400                "EMBED policy requires fields = ('<col>', ...)".to_string(),
1401                self.position(),
1402            ));
1403        }
1404        let provider = provider.ok_or_else(|| {
1405            ParseError::new(
1406                "EMBED policy requires provider = '<token>'".to_string(),
1407                self.position(),
1408            )
1409        })?;
1410        let model = model.ok_or_else(|| {
1411            ParseError::new(
1412                "EMBED policy requires model = '<name>'".to_string(),
1413                self.position(),
1414            )
1415        })?;
1416        Ok(reddb_types::catalog::EmbedPolicy {
1417            fields,
1418            provider,
1419            model,
1420        })
1421    }
1422
1423    /// `MODERATE (fields = (...), provider, model, sync, degraded, on_reject)`.
1424    fn parse_ai_moderate_policy(
1425        &mut self,
1426    ) -> Result<reddb_types::catalog::ModeratePolicy, ParseError> {
1427        use reddb_types::catalog::{ModerateDegradedMode, ModerateRejectAction};
1428        self.expect(Token::LParen)?;
1429        let mut fields: Vec<String> = Vec::new();
1430        let mut provider: Option<String> = None;
1431        let mut model: Option<String> = None;
1432        let mut sync_gate = false;
1433        let mut degraded_mode = ModerateDegradedMode::default();
1434        let mut reject_action = ModerateRejectAction::default();
1435        let mut hard_delete_on_reject = false;
1436        loop {
1437            let key = self.expect_ident_or_keyword()?.to_ascii_lowercase();
1438            self.expect(Token::Eq)?;
1439            match key.as_str() {
1440                "fields" => fields = self.parse_ai_string_list()?,
1441                "provider" => provider = Some(self.parse_string()?),
1442                "model" => model = Some(self.parse_string()?),
1443                "sync" | "sync_gate" => sync_gate = self.parse_ai_bool()?,
1444                "hard_delete" | "hard_delete_on_reject" => {
1445                    hard_delete_on_reject = self.parse_ai_bool()?
1446                }
1447                "degraded" | "degraded_mode" => {
1448                    let word = self.parse_ai_word()?;
1449                    degraded_mode = ModerateDegradedMode::from_str(&word).ok_or_else(|| {
1450                        ParseError::new(
1451                            format!(
1452                                "unsupported MODERATE degraded mode {word:?}; supported: open, closed"
1453                            ),
1454                            self.position(),
1455                        )
1456                    })?;
1457                }
1458                "on_reject" | "reject_action" => {
1459                    let word = self.parse_ai_word()?;
1460                    reject_action = ModerateRejectAction::from_str(&word).ok_or_else(|| {
1461                        ParseError::new(
1462                            format!(
1463                                "unsupported MODERATE reject action {word:?}; supported: reject, flag, redact"
1464                            ),
1465                            self.position(),
1466                        )
1467                    })?;
1468                }
1469                other => {
1470                    return Err(ParseError::new(
1471                        format!(
1472                            "unsupported MODERATE policy option {other:?}; supported: fields, provider, model, sync, degraded, on_reject, hard_delete"
1473                        ),
1474                        self.position(),
1475                    ));
1476                }
1477            }
1478            if !self.consume(&Token::Comma)? {
1479                break;
1480            }
1481        }
1482        self.expect(Token::RParen)?;
1483        if fields.is_empty() {
1484            return Err(ParseError::new(
1485                "MODERATE policy requires fields = ('<col>', ...)".to_string(),
1486                self.position(),
1487            ));
1488        }
1489        let provider = provider.ok_or_else(|| {
1490            ParseError::new(
1491                "MODERATE policy requires provider = '<token>'".to_string(),
1492                self.position(),
1493            )
1494        })?;
1495        let model = model.ok_or_else(|| {
1496            ParseError::new(
1497                "MODERATE policy requires model = '<name>'".to_string(),
1498                self.position(),
1499            )
1500        })?;
1501        Ok(reddb_types::catalog::ModeratePolicy {
1502            fields,
1503            provider,
1504            model,
1505            sync_gate,
1506            degraded_mode,
1507            reject_action,
1508            hard_delete_on_reject,
1509        })
1510    }
1511
1512    /// `VISION (image_field = '..', outputs = (...), provider, model)`.
1513    fn parse_ai_vision_policy(&mut self) -> Result<reddb_types::catalog::VisionPolicy, ParseError> {
1514        self.expect(Token::LParen)?;
1515        let mut image_field: Option<String> = None;
1516        let mut output_kinds: Vec<String> = Vec::new();
1517        let mut provider: Option<String> = None;
1518        let mut model: Option<String> = None;
1519        loop {
1520            let key = self.expect_ident_or_keyword()?.to_ascii_lowercase();
1521            self.expect(Token::Eq)?;
1522            match key.as_str() {
1523                "image_field" => image_field = Some(self.parse_string()?),
1524                "outputs" | "output_kinds" => output_kinds = self.parse_ai_string_list()?,
1525                "provider" => provider = Some(self.parse_string()?),
1526                "model" => model = Some(self.parse_string()?),
1527                other => {
1528                    return Err(ParseError::new(
1529                        format!(
1530                            "unsupported VISION policy option {other:?}; supported: image_field, outputs, provider, model"
1531                        ),
1532                        self.position(),
1533                    ));
1534                }
1535            }
1536            if !self.consume(&Token::Comma)? {
1537                break;
1538            }
1539        }
1540        self.expect(Token::RParen)?;
1541        let image_field = image_field.ok_or_else(|| {
1542            ParseError::new(
1543                "VISION policy requires image_field = '<col>'".to_string(),
1544                self.position(),
1545            )
1546        })?;
1547        if output_kinds.is_empty() {
1548            return Err(ParseError::new(
1549                "VISION policy requires outputs = ('<kind>', ...)".to_string(),
1550                self.position(),
1551            ));
1552        }
1553        let provider = provider.ok_or_else(|| {
1554            ParseError::new(
1555                "VISION policy requires provider = '<token>'".to_string(),
1556                self.position(),
1557            )
1558        })?;
1559        let model = model.ok_or_else(|| {
1560            ParseError::new(
1561                "VISION policy requires model = '<name>'".to_string(),
1562                self.position(),
1563            )
1564        })?;
1565        Ok(reddb_types::catalog::VisionPolicy {
1566            image_field,
1567            output_kinds,
1568            provider,
1569            model,
1570        })
1571    }
1572
1573    fn expect_ident_ci_ddl(&mut self, expected: &str) -> Result<(), ParseError> {
1574        if self.consume_ident_ci(expected)? {
1575            Ok(())
1576        } else {
1577            Err(ParseError::expected(
1578                vec![expected],
1579                self.peek(),
1580                self.position(),
1581            ))
1582        }
1583    }
1584
1585    fn parse_create_table_ttl_clause(&mut self) -> Result<Option<u64>, ParseError> {
1586        let option_name = self.expect_ident_or_keyword()?;
1587        if !option_name.eq_ignore_ascii_case("ttl") {
1588            return Err(ParseError::new(
1589                // F-05: `option_name` is caller-controlled identifier text.
1590                // Render via `{:?}` so embedded CR/LF/NUL/quotes are escaped
1591                // before the message reaches downstream serialization sinks.
1592                format!(
1593                    "unsupported CREATE TABLE option {option_name:?}; supported options: TTL <duration> [ms|s|m|h|d] (e.g. `WITH TTL 30 m`)"
1594                ),
1595                self.position(),
1596            ));
1597        }
1598
1599        let ttl_value = self.parse_float()?;
1600        let ttl_unit = match self.peek() {
1601            Token::Ident(unit) => {
1602                let unit = unit.clone();
1603                self.advance()?;
1604                unit
1605            }
1606            _ => "s".to_string(),
1607        };
1608
1609        let multiplier_ms = match ttl_unit.to_ascii_lowercase().as_str() {
1610            "ms" | "msec" | "millisecond" | "milliseconds" => 1.0,
1611            "s" | "sec" | "secs" | "second" | "seconds" => 1_000.0,
1612            "m" | "min" | "mins" | "minute" | "minutes" => 60_000.0,
1613            "h" | "hr" | "hrs" | "hour" | "hours" => 3_600_000.0,
1614            "d" | "day" | "days" => 86_400_000.0,
1615            other => {
1616                return Err(ParseError::new(
1617                    // F-05: render `other` via `{:?}` so caller-controlled
1618                    // bytes (CR / LF / NUL / quotes) are escaped before
1619                    // reaching downstream serialization sinks.
1620                    format!(
1621                        "unsupported TTL unit {other:?}; supported units: ms, s, m, h, d (e.g. `WITH TTL 30 m`)"
1622                    ),
1623                    self.position(),
1624                ));
1625            }
1626        };
1627
1628        if !ttl_value.is_finite() || ttl_value < 0.0 {
1629            return Err(ParseError::new(
1630                "TTL must be a finite, non-negative duration".to_string(),
1631                self.position(),
1632            ));
1633        }
1634
1635        let ttl_ms = ttl_value * multiplier_ms;
1636        if ttl_ms > u64::MAX as f64 {
1637            return Err(ParseError::new(
1638                "TTL duration is too large".to_string(),
1639                self.position(),
1640            ));
1641        }
1642        if ttl_ms.fract().abs() >= f64::EPSILON {
1643            return Err(ParseError::new(
1644                "TTL duration must resolve to a whole number of milliseconds".to_string(),
1645                self.position(),
1646            ));
1647        }
1648
1649        Ok(Some(ttl_ms as u64))
1650    }
1651
1652    /// Try to match IF NOT EXISTS sequence
1653    pub(crate) fn match_if_not_exists(&mut self) -> Result<bool, ParseError> {
1654        if self.check(&Token::If) {
1655            self.advance()?;
1656            self.expect(Token::Not)?;
1657            self.expect(Token::Exists)?;
1658            Ok(true)
1659        } else {
1660            Ok(false)
1661        }
1662    }
1663
1664    /// Try to match IF EXISTS sequence
1665    pub(crate) fn match_if_exists(&mut self) -> Result<bool, ParseError> {
1666        if self.check(&Token::If) {
1667            self.advance()?;
1668            self.expect(Token::Exists)?;
1669            Ok(true)
1670        } else {
1671            Ok(false)
1672        }
1673    }
1674
1675    /// Try to match NOT NULL sequence
1676    fn match_not_null(&mut self) -> Result<bool, ParseError> {
1677        if self.check(&Token::Not) {
1678            // Peek ahead - only consume if followed by NULL
1679            // We need to be careful: save state and try
1680            self.advance()?; // consume NOT
1681            if self.check(&Token::Null) {
1682                self.advance()?; // consume NULL
1683                Ok(true)
1684            } else {
1685                // This is tricky - NOT was consumed but next isn't NULL.
1686                // In column modifier context, NOT should only appear before NULL.
1687                // Return error for clarity.
1688                Err(ParseError::expected(
1689                    vec!["NULL (after NOT)"],
1690                    self.peek(),
1691                    self.position(),
1692                ))
1693            }
1694        } else {
1695            Ok(false)
1696        }
1697    }
1698
1699    /// Try to match PRIMARY KEY sequence
1700    fn match_primary_key(&mut self) -> Result<bool, ParseError> {
1701        if self.check(&Token::Primary) {
1702            self.advance()?;
1703            self.expect(Token::Key)?;
1704            Ok(true)
1705        } else {
1706            Ok(false)
1707        }
1708    }
1709}
1710
1711/// Decode a 64-char lowercase/uppercase hex string into a 32-byte array.
1712/// Returns a human-readable error message on length or character violations.
1713/// Used by `SIGNED_BY` clause parsing (issue #520) to surface bad pubkeys
1714/// at parse-time rather than downstream in the engine.
1715fn decode_hex_32(s: &str) -> Result<[u8; 32], String> {
1716    if s.len() != 64 {
1717        return Err(format!("expected 64 hex chars, got {}", s.len()));
1718    }
1719    let mut out = [0u8; 32];
1720    let bytes = s.as_bytes();
1721    for i in 0..32 {
1722        let hi = hex_nibble(bytes[i * 2])?;
1723        let lo = hex_nibble(bytes[i * 2 + 1])?;
1724        out[i] = (hi << 4) | lo;
1725    }
1726    Ok(out)
1727}
1728
1729fn hex_nibble(c: u8) -> Result<u8, String> {
1730    match c {
1731        b'0'..=b'9' => Ok(c - b'0'),
1732        b'a'..=b'f' => Ok(c - b'a' + 10),
1733        b'A'..=b'F' => Ok(c - b'A' + 10),
1734        _ => Err(format!("non-hex char: {:?}", c as char)),
1735    }
1736}
1737
1738#[cfg(test)]
1739mod tests {
1740    use super::*;
1741    use reddb_types::catalog::{AnalyticsOutput, CollectionModel, SubscriptionOperation};
1742
1743    fn parser(input: &str) -> Parser<'_> {
1744        Parser::new(input).unwrap_or_else(|err| panic!("failed to lex {input:?}: {err:?}"))
1745    }
1746
1747    #[test]
1748    fn parse_create_table_body_parenthesized_options_and_trailing_clauses() {
1749        let QueryExpr::CreateTable(table) = parser(
1750            "IF NOT EXISTS events (id INT, tenant_meta TEXT) \
1751             WITH (tenant_by = 'tenant_id', append_only = true, timestamps = false) \
1752             PARTITION BY HASH (id) TENANT BY (tenant_meta.tenant)",
1753        )
1754        .parse_create_table_body()
1755        .expect("create table body") else {
1756            panic!("Expected CreateTableQuery");
1757        };
1758
1759        assert_eq!(table.name, "events");
1760        assert!(table.if_not_exists);
1761        assert!(table.append_only);
1762        assert!(!table.timestamps);
1763        assert_eq!(table.tenant_by.as_deref(), Some("tenant_id"));
1764        assert_eq!(
1765            table
1766                .partition_by
1767                .as_ref()
1768                .map(|spec| (spec.kind, spec.column.as_str())),
1769            Some((PartitionKind::Hash, "id"))
1770        );
1771
1772        let err = parser("bad (id INT) WITH (tenant_by = 42)")
1773            .parse_create_table_body()
1774            .unwrap_err();
1775        assert!(
1776            err.to_string()
1777                .contains("WITH tenant_by expects a text literal"),
1778            "{err}"
1779        );
1780    }
1781
1782    #[test]
1783    fn parse_create_table_ai_policy_round_trips_all_modalities() {
1784        use reddb_types::catalog::{ModerateDegradedMode, ModerateRejectAction};
1785        let QueryExpr::CreateTable(table) = parser(
1786            "posts (id INT, title TEXT, body TEXT, photo TEXT) WITH ( \
1787               EMBED (fields = ('title', 'body'), provider = 'openai', model = 'text-embedding-3-small'), \
1788               MODERATE (fields = ('body'), provider = 'openai', model = 'omni-moderation-latest', sync = true, degraded = closed, on_reject = flag, hard_delete = true), \
1789               VISION (image_field = 'photo', outputs = ('caption', 'tags'), provider = 'openai', model = 'gpt-4o') \
1790             )",
1791        )
1792        .parse_create_table_body()
1793        .expect("create table body with ai policy") else {
1794            panic!("Expected CreateTableQuery");
1795        };
1796
1797        let policy = table.ai_policy.expect("ai policy present");
1798
1799        let embed = policy.embed.expect("embed block");
1800        assert_eq!(embed.fields, vec!["title".to_string(), "body".to_string()]);
1801        assert_eq!(embed.provider, "openai");
1802        assert_eq!(embed.model, "text-embedding-3-small");
1803
1804        let moderate = policy.moderate.expect("moderate block");
1805        assert_eq!(moderate.fields, vec!["body".to_string()]);
1806        assert!(moderate.sync_gate);
1807        assert_eq!(moderate.degraded_mode, ModerateDegradedMode::Closed);
1808        assert_eq!(moderate.reject_action, ModerateRejectAction::Flag);
1809        assert!(moderate.hard_delete_on_reject);
1810
1811        let vision = policy.vision.expect("vision block");
1812        assert_eq!(vision.image_field, "photo");
1813        assert_eq!(
1814            vision.output_kinds,
1815            vec!["caption".to_string(), "tags".to_string()]
1816        );
1817        assert_eq!(vision.model, "gpt-4o");
1818    }
1819
1820    #[test]
1821    fn parse_moderate_policy_aliases_and_error_branches() {
1822        use reddb_types::catalog::{ModerateDegradedMode, ModerateRejectAction};
1823        // Alias spellings hit the same match arms as the short forms.
1824        let QueryExpr::CreateTable(table) = parser(
1825            "t (id INT, body TEXT) WITH ( \
1826               MODERATE (fields = ('body'), provider = 'openai', model = 'm', \
1827                 sync_gate = true, degraded_mode = open, reject_action = redact, \
1828                 hard_delete_on_reject = true) \
1829             )",
1830        )
1831        .parse_create_table_body()
1832        .expect("alias spellings parse") else {
1833            panic!("Expected CreateTableQuery");
1834        };
1835        let moderate = table
1836            .ai_policy
1837            .expect("policy")
1838            .moderate
1839            .expect("moderate block");
1840        assert!(moderate.sync_gate);
1841        assert_eq!(moderate.degraded_mode, ModerateDegradedMode::Open);
1842        assert_eq!(moderate.reject_action, ModerateRejectAction::Redact);
1843        assert!(moderate.hard_delete_on_reject);
1844
1845        // Each invalid MODERATE option surfaces a descriptive parse error.
1846        for (sql, needle) in [
1847            (
1848                "t (id INT, body TEXT) WITH (MODERATE (fields = ('body'), provider = 'o', model = 'm', bogus = 1))",
1849                "unsupported MODERATE policy option",
1850            ),
1851            (
1852                "t (id INT, body TEXT) WITH (MODERATE (fields = ('body'), provider = 'o', model = 'm', degraded = sideways))",
1853                "unsupported MODERATE degraded mode",
1854            ),
1855            (
1856                "t (id INT, body TEXT) WITH (MODERATE (fields = ('body'), provider = 'o', model = 'm', on_reject = explode))",
1857                "unsupported MODERATE reject action",
1858            ),
1859            (
1860                "t (id INT, body TEXT) WITH (MODERATE (provider = 'o', model = 'm'))",
1861                "MODERATE policy requires fields",
1862            ),
1863        ] {
1864            let err = parser(sql)
1865                .parse_create_table_body()
1866                .expect_err("moderate policy error");
1867            assert!(format!("{err}").contains(needle), "got: {err}");
1868        }
1869    }
1870
1871    #[test]
1872    fn parse_embed_and_vision_policy_error_branches() {
1873        // EMBED + VISION unknown-option and missing-required-field arms each
1874        // surface a descriptive parse error.
1875        for (sql, needle) in [
1876            (
1877                "t (id INT, body TEXT) WITH (EMBED (fields = ('body'), provider = 'o', model = 'm', bogus = 1))",
1878                "unsupported EMBED policy option",
1879            ),
1880            (
1881                "t (id INT, body TEXT) WITH (EMBED (provider = 'o', model = 'm'))",
1882                "EMBED policy requires fields",
1883            ),
1884            (
1885                "t (id INT, body TEXT) WITH (EMBED (fields = ('body'), model = 'm'))",
1886                "EMBED policy requires provider",
1887            ),
1888            (
1889                "t (id INT, body TEXT) WITH (EMBED (fields = ('body'), provider = 'o'))",
1890                "EMBED policy requires model",
1891            ),
1892            (
1893                "t (id INT, photo TEXT) WITH (VISION (image_field = 'photo', provider = 'o', model = 'm', bogus = 1))",
1894                "unsupported VISION policy option",
1895            ),
1896            (
1897                "t (id INT, photo TEXT) WITH (VISION (provider = 'o', model = 'm'))",
1898                "VISION policy requires image_field",
1899            ),
1900        ] {
1901            let err = parser(sql)
1902                .parse_create_table_body()
1903                .expect_err("ai policy error");
1904            assert!(format!("{err}").contains(needle), "got: {err}");
1905        }
1906
1907        // The `output_kinds` alias for VISION parses like `outputs`.
1908        let QueryExpr::CreateTable(table) = parser(
1909            "t (id INT, photo TEXT) WITH (VISION (image_field = 'photo', \
1910               output_kinds = ('caption'), provider = 'o', model = 'm'))",
1911        )
1912        .parse_create_table_body()
1913        .expect("vision output_kinds alias") else {
1914            panic!("Expected CreateTableQuery");
1915        };
1916        let vision = table
1917            .ai_policy
1918            .expect("policy")
1919            .vision
1920            .expect("vision block");
1921        assert_eq!(vision.output_kinds, vec!["caption".to_string()]);
1922    }
1923
1924    #[test]
1925    fn parse_create_table_ai_policy_defaults_and_no_clause() {
1926        use reddb_types::catalog::{ModerateDegradedMode, ModerateRejectAction};
1927        // MODERATE with no sync/degraded/on_reject falls back to defaults.
1928        let QueryExpr::CreateTable(table) = parser(
1929            "msgs (id INT, body TEXT) WITH ( \
1930               MODERATE (fields = ('body'), provider = 'openai', model = 'omni-moderation-latest') \
1931             )",
1932        )
1933        .parse_create_table_body()
1934        .expect("create table body") else {
1935            panic!("Expected CreateTableQuery");
1936        };
1937        let moderate = table
1938            .ai_policy
1939            .expect("policy")
1940            .moderate
1941            .expect("moderate block");
1942        assert!(!moderate.sync_gate);
1943        assert_eq!(moderate.degraded_mode, ModerateDegradedMode::Open);
1944        assert_eq!(moderate.reject_action, ModerateRejectAction::Reject);
1945        assert!(!moderate.hard_delete_on_reject);
1946
1947        // No AI clause leaves the policy entirely absent.
1948        let QueryExpr::CreateTable(plain) = parser("plain (id INT)")
1949            .parse_create_table_body()
1950            .expect("create table body")
1951        else {
1952            panic!("Expected CreateTableQuery");
1953        };
1954        assert!(plain.ai_policy.is_none());
1955    }
1956
1957    #[test]
1958    fn parse_create_table_ai_policy_rejects_malformed_clauses() {
1959        // Missing provider.
1960        let err = parser("t (id INT) WITH (EMBED (fields = ('body'), model = 'm'))")
1961            .parse_create_table_body()
1962            .unwrap_err();
1963        assert!(
1964            err.to_string().contains("EMBED policy requires provider"),
1965            "{err}"
1966        );
1967
1968        // Unknown option key inside a clause.
1969        let err = parser(
1970            "t (id INT) WITH (VISION (image_field = 'p', outputs = ('caption'), provider = 'openai', model = 'm', bogus = 1))",
1971        )
1972        .parse_create_table_body()
1973        .unwrap_err();
1974        assert!(
1975            err.to_string().contains("unsupported VISION policy option"),
1976            "{err}"
1977        );
1978
1979        // Invalid moderation degraded mode.
1980        let err = parser(
1981            "t (id INT) WITH (MODERATE (fields = ('body'), provider = 'openai', model = 'm', degraded = maybe))",
1982        )
1983        .parse_create_table_body()
1984        .unwrap_err();
1985        assert!(
1986            err.to_string()
1987                .contains("unsupported MODERATE degraded mode"),
1988            "{err}"
1989        );
1990
1991        // Duplicate EMBED clause.
1992        let err = parser(
1993            "t (id INT) WITH (EMBED (fields = ('a'), provider = 'openai', model = 'm'), EMBED (fields = ('b'), provider = 'openai', model = 'm'))",
1994        )
1995        .parse_create_table_body()
1996        .unwrap_err();
1997        assert!(err.to_string().contains("duplicate EMBED clause"), "{err}");
1998    }
1999
2000    #[test]
2001    fn parse_keyed_bodies_cover_vault_analytics_and_dotted_drop_names() {
2002        let QueryExpr::CreateTable(vault) =
2003            parser("IF NOT EXISTS tenant.secrets WITH OWN MASTER KEY")
2004                .parse_create_keyed_body(CollectionModel::Vault)
2005                .expect("create vault")
2006        else {
2007            panic!("Expected CreateTableQuery");
2008        };
2009        assert_eq!(vault.collection_model, CollectionModel::Vault);
2010        assert_eq!(vault.name, "tenant.secrets");
2011        assert!(vault.if_not_exists);
2012        assert!(vault.vault_own_master_key);
2013
2014        let QueryExpr::CreateTable(graph) = parser(
2015            "g WITH ANALYTICS (centrality (using = pagerank, max_iterations = 12, tolerance = 0.001))",
2016        )
2017        .parse_create_keyed_body(CollectionModel::Graph)
2018        .expect("create graph")
2019        else {
2020            panic!("Expected CreateTableQuery");
2021        };
2022        assert_eq!(graph.analytics_config.len(), 1);
2023        let view = &graph.analytics_config[0];
2024        assert_eq!(view.output, AnalyticsOutput::Centrality);
2025        assert_eq!(view.algorithm.as_deref(), Some("pagerank"));
2026        assert_eq!(view.max_iterations, Some(12));
2027        assert_eq!(view.tolerance, Some(0.001));
2028
2029        let err = parser("g WITH OTHER")
2030            .parse_create_keyed_body(CollectionModel::Graph)
2031            .unwrap_err();
2032        assert!(err.to_string().contains("expected: ANALYTICS"), "{err}");
2033
2034        assert!(parser("CREATE KV cache WITH ANALYTICS (components)")
2035            .parse()
2036            .unwrap_err()
2037            .to_string()
2038            .contains("Unexpected token after query"));
2039
2040        let QueryExpr::DropKv(drop) = parser("IF EXISTS tenant.cache.*")
2041            .parse_drop_keyed_body(CollectionModel::Kv)
2042            .expect("drop kv")
2043        else {
2044            panic!("Expected DropKvQuery");
2045        };
2046        assert_eq!(drop.name, "tenant.cache.*");
2047        assert!(drop.if_exists);
2048        assert_eq!(drop.model, CollectionModel::Kv);
2049    }
2050
2051    #[test]
2052    fn parse_collection_signed_by_list_and_errors() {
2053        let pk_a = "aa".repeat(32);
2054        let pk_b = "BB".repeat(32);
2055        let QueryExpr::CreateCollection(collection) =
2056            parser(&format!("signed KIND graph SIGNED_BY ('{pk_a}', '{pk_b}')"))
2057                .parse_create_collection_body()
2058                .expect("create collection")
2059        else {
2060            panic!("Expected CreateCollectionQuery");
2061        };
2062        assert_eq!(collection.allowed_signers, vec![[0xaau8; 32], [0xBBu8; 32]]);
2063
2064        let err = parser("signed KIND graph SIGNED_BY (42)")
2065            .parse_create_collection_body()
2066            .unwrap_err();
2067        assert!(
2068            err.to_string()
2069                .contains("string literal (ed25519 pubkey hex)"),
2070            "{err}"
2071        );
2072
2073        let err = parser("signed KIND graph SIGNED_BY ('deadbeef')")
2074            .parse_create_collection_body()
2075            .unwrap_err();
2076        assert!(err.to_string().contains("expected 64 hex chars"), "{err}");
2077    }
2078
2079    #[test]
2080    fn parse_alter_operations_cover_subscriptions_partitions_tenancy_and_signers() {
2081        let pk = "11".repeat(32);
2082        let QueryExpr::AlterTable(alter) = parser(&format!(
2083            "ALTER COLLECTION audit \
2084             ADD SUBSCRIPTION pii TO audit_events REDACT (payload.ssn, *.secret) WHERE level = 'warn', \
2085             DROP SUBSCRIPTION pii, \
2086             ADD SIGNER '{pk}', \
2087             REVOKE SIGNER '{pk}', \
2088             ATTACH PARTITION audit_2026 FOR VALUES FROM (2026) TO (2027), \
2089             DETACH PARTITION audit_2026, \
2090             ENABLE EVENTS (INSERT, UPDATE) TO table_events ON ALL TENANTS, \
2091             DISABLE EVENTS, \
2092             ENABLE TENANCY ON (metadata.tenant), \
2093             DISABLE TENANCY, \
2094             SET APPEND_ONLY = true, \
2095             SET VERSIONED = false, \
2096             SET RETENTION 2 h, \
2097             UNSET RETENTION"
2098        ))
2099        .parse_alter_table_query()
2100        .expect("alter collection")
2101        else {
2102            panic!("Expected AlterTableQuery");
2103        };
2104
2105        assert_eq!(alter.name, "audit");
2106        assert_eq!(alter.operations.len(), 14);
2107        match &alter.operations[0] {
2108            AlterOperation::AddSubscription { name, descriptor } => {
2109                assert_eq!(name, "pii");
2110                assert_eq!(descriptor.target_queue, "audit_events");
2111                assert_eq!(descriptor.redact_fields, vec!["payload.ssn", "*.secret"]);
2112                assert_eq!(descriptor.where_filter.as_deref(), Some("LEVEL = 'warn'"));
2113            }
2114            other => panic!("expected AddSubscription, got {other:?}"),
2115        }
2116        assert!(matches!(
2117            &alter.operations[1],
2118            AlterOperation::DropSubscription { name } if name == "pii"
2119        ));
2120        assert!(matches!(
2121            &alter.operations[2],
2122            AlterOperation::AddSigner { pubkey } if *pubkey == [0x11; 32]
2123        ));
2124        assert!(matches!(
2125            &alter.operations[3],
2126            AlterOperation::RevokeSigner { pubkey } if *pubkey == [0x11; 32]
2127        ));
2128        assert!(matches!(
2129            &alter.operations[4],
2130            AlterOperation::AttachPartition { child, bound }
2131                if child == "audit_2026" && bound == "FROM ( 2026 ) TO ( 2027 )"
2132        ));
2133        assert!(matches!(
2134            &alter.operations[5],
2135            AlterOperation::DetachPartition { child } if child == "audit_2026"
2136        ));
2137        match &alter.operations[6] {
2138            AlterOperation::EnableEvents(descriptor) => {
2139                assert_eq!(
2140                    descriptor.ops_filter,
2141                    vec![SubscriptionOperation::Insert, SubscriptionOperation::Update]
2142                );
2143                assert_eq!(descriptor.target_queue, "table_events");
2144                assert!(descriptor.all_tenants);
2145            }
2146            other => panic!("expected EnableEvents, got {other:?}"),
2147        }
2148        assert!(matches!(
2149            &alter.operations[7],
2150            AlterOperation::DisableEvents
2151        ));
2152        assert!(matches!(
2153            &alter.operations[8],
2154            AlterOperation::EnableTenancy { column } if column == "METADATA.tenant"
2155        ));
2156        assert!(matches!(
2157            &alter.operations[9],
2158            AlterOperation::DisableTenancy
2159        ));
2160        assert!(matches!(
2161            &alter.operations[10],
2162            AlterOperation::SetAppendOnly(true)
2163        ));
2164        assert!(matches!(
2165            &alter.operations[11],
2166            AlterOperation::SetVersioned(false)
2167        ));
2168        assert!(matches!(
2169            &alter.operations[12],
2170            AlterOperation::SetRetention { duration_ms } if *duration_ms == 7_200_000
2171        ));
2172        assert!(matches!(
2173            &alter.operations[13],
2174            AlterOperation::UnsetRetention
2175        ));
2176    }
2177
2178    #[test]
2179    fn parse_alter_graph_analytics_keyword_errors() {
2180        let err = parser("ALTER GRAPH g ADD centrality")
2181            .parse_alter_graph_query()
2182            .unwrap_err();
2183        assert!(err.to_string().contains("expected: ANALYTICS"), "{err}");
2184
2185        let err = parser("ALTER GRAPH g DROP centrality")
2186            .parse_alter_graph_query()
2187            .unwrap_err();
2188        assert!(err.to_string().contains("expected: ANALYTICS"), "{err}");
2189    }
2190
2191    #[test]
2192    fn decode_hex_32_reports_length_and_character_errors() {
2193        assert_eq!(decode_hex_32(&"0f".repeat(32)).unwrap(), [0x0f; 32]);
2194        assert_eq!(
2195            decode_hex_32("deadbeef").unwrap_err(),
2196            "expected 64 hex chars, got 8"
2197        );
2198        assert!(decode_hex_32(&"gg".repeat(32))
2199            .unwrap_err()
2200            .contains("non-hex char"));
2201    }
2202}