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