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 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 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 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 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 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 if tenant_by.is_none() && self.consume_ident_ci("TENANT")? {
293 self.expect(Token::By)?;
294 self.expect(Token::LParen)?;
295 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 let pubkey = self.parse_single_signer_hex()?;
912 Ok(AlterOperation::AddSigner { pubkey })
913 } else {
914 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 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 let sub_name = self.expect_ident_or_keyword()?;
934 Ok(AlterOperation::DropSubscription { name: sub_name })
935 } else {
936 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 let _ = self.consume(&Token::Column)?; 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 self.expect(Token::Partition)?;
951 let child = self.expect_ident()?;
952 self.expect(Token::For)?;
953 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 self.expect(Token::Partition)?;
968 let child = self.expect_ident()?;
969 Ok(AlterOperation::DetachPartition { child })
970 } else if self.consume(&Token::Enable)? {
971 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 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 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 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 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 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 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 if self.consume_ident_ci("REQUIRES")? {
1122 self.consume_ident_ci("CAPABILITY")?;
1123 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 fn match_not_null(&mut self) -> Result<bool, ParseError> {
1677 if self.check(&Token::Not) {
1678 self.advance()?; if self.check(&Token::Null) {
1682 self.advance()?; Ok(true)
1684 } else {
1685 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 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
1711fn 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 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 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 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 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 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 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 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 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 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 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}