1use crate::ast::{
2 AlterMetricQuery, AlterQueueQuery, AlterTableQuery, AlterUserStmt, ApplyMigrationQuery,
3 AskQuery, BinOp, CompareOp, ConfigCommand, CopyFormat, CopyFromQuery, CreateCollectionQuery,
4 CreateForeignTableQuery, CreateIndexQuery, CreateMetricQuery, CreateMigrationQuery,
5 CreatePolicyQuery, CreateQueueQuery, CreateSchemaQuery, CreateSequenceQuery, CreateServerQuery,
6 CreateSloQuery, CreateTableQuery, CreateTimeSeriesQuery, CreateTreeQuery, CreateUserStmt,
7 CreateVectorQuery, CreateViewQuery, DeleteQuery, DropCollectionQuery, DropDocumentQuery,
8 DropForeignTableQuery, DropGraphQuery, DropIndexQuery, DropKvQuery, DropPolicyQuery,
9 DropQueueQuery, DropSchemaQuery, DropSequenceQuery, DropServerQuery, DropTableQuery,
10 DropTimeSeriesQuery, DropTreeQuery, DropVectorQuery, DropViewQuery, EventsBackfillQuery,
11 ExplainAlterQuery, ExplainMigrationQuery, Expr, FieldRef, Filter, ForeignColumnDef, GrantStmt,
12 GraphCommand, GraphQuery, HybridQuery, InsertQuery, JoinQuery, KvCommand, MaintenanceCommand,
13 PathQuery, PolicyAction, ProbabilisticCommand, QueryExpr, QueueCommand, QueueSelectQuery,
14 RankOfQuery, RankRangeQuery, RefreshMaterializedViewQuery, RevokeStmt, RollbackMigrationQuery,
15 SearchCommand, Span, TableQuery, TreeCommand, TruncateQuery, TxnControl, UpdateQuery,
16 VectorQuery,
17};
18use crate::lexer::Token;
19use crate::parser::{ParseError, Parser, SafeTokenDisplay};
20use crate::sql_lowering::filter_to_expr;
21use reddb_types::catalog::CollectionModel;
22use reddb_types::types::Value;
23
24#[derive(Debug, Clone)]
29pub enum SqlStatement {
30 Query(SqlQuery),
31 Mutation(SqlMutation),
32 Schema(SqlSchemaCommand),
33 Admin(SqlAdminCommand),
34}
35
36#[derive(Debug, Clone)]
37#[allow(clippy::large_enum_variant)]
38pub enum FrontendStatement {
39 Sql(SqlStatement),
40 Graph(GraphQuery),
41 GraphCommand(GraphCommand),
42 Path(PathQuery),
43 Vector(VectorQuery),
44 Hybrid(HybridQuery),
45 Search(SearchCommand),
46 Ask(AskQuery),
47 QueueSelect(QueueSelectQuery),
48 QueueCommand(QueueCommand),
49 EventsBackfill(EventsBackfillQuery),
50 EventsBackfillStatus { collection: String },
51 TreeCommand(TreeCommand),
52 ProbabilisticCommand(ProbabilisticCommand),
53 KvCommand(KvCommand),
54 ConfigCommand(ConfigCommand),
55 Ranking(QueryExpr),
56}
57
58#[derive(Debug, Clone)]
59pub enum SqlCommand {
60 Select(TableQuery),
61 Join(JoinQuery),
62 Insert(InsertQuery),
63 Update(UpdateQuery),
64 Delete(DeleteQuery),
65 ExplainAlter(ExplainAlterQuery),
66 CreateTable(CreateTableQuery),
67 CreateCollection(CreateCollectionQuery),
68 CreateVector(CreateVectorQuery),
69 DropTable(DropTableQuery),
70 DropGraph(DropGraphQuery),
71 DropVector(DropVectorQuery),
72 DropDocument(DropDocumentQuery),
73 DropKv(DropKvQuery),
74 DropCollection(DropCollectionQuery),
75 Truncate(TruncateQuery),
76 AlterTable(AlterTableQuery),
77 CreateIndex(CreateIndexQuery),
78 DropIndex(DropIndexQuery),
79 CreateTimeSeries(CreateTimeSeriesQuery),
80 CreateMetric(CreateMetricQuery),
81 AlterMetric(AlterMetricQuery),
82 CreateSlo(CreateSloQuery),
83 DropTimeSeries(DropTimeSeriesQuery),
84 CreateQueue(CreateQueueQuery),
85 AlterQueue(AlterQueueQuery),
86 DropQueue(DropQueueQuery),
87 CreateTree(CreateTreeQuery),
88 DropTree(DropTreeQuery),
89 Probabilistic(ProbabilisticCommand),
90 SetConfig {
91 key: String,
92 value: Value,
93 },
94 ShowConfig {
95 prefix: Option<String>,
96 as_json: bool,
97 },
98 SetSecret {
99 key: String,
100 value: Value,
101 },
102 DeleteSecret {
103 key: String,
104 },
105 ShowSecrets {
106 prefix: Option<String>,
107 },
108 SetTenant(Option<String>),
109 ShowTenant,
110 TransactionControl(TxnControl),
111 Maintenance(MaintenanceCommand),
112 CreateSchema(CreateSchemaQuery),
113 DropSchema(DropSchemaQuery),
114 CreateSequence(CreateSequenceQuery),
115 DropSequence(DropSequenceQuery),
116 CopyFrom(CopyFromQuery),
117 CreateView(CreateViewQuery),
118 DropView(DropViewQuery),
119 RefreshMaterializedView(RefreshMaterializedViewQuery),
120 CreatePolicy(CreatePolicyQuery),
121 DropPolicy(DropPolicyQuery),
122 CreateServer(CreateServerQuery),
123 DropServer(DropServerQuery),
124 CreateForeignTable(CreateForeignTableQuery),
125 DropForeignTable(DropForeignTableQuery),
126 Grant(GrantStmt),
128 Revoke(RevokeStmt),
130 AlterUser(AlterUserStmt),
132 CreateUser(CreateUserStmt),
134 IamPolicy(QueryExpr),
139 CreateMigration(CreateMigrationQuery),
140 ApplyMigration(ApplyMigrationQuery),
141 RollbackMigration(RollbackMigrationQuery),
142 ExplainMigration(ExplainMigrationQuery),
143}
144
145fn analytics_v0_non_goal_create(token: &Token) -> Option<String> {
154 let ident = match token {
155 Token::Ident(s) => s,
156 _ => return None,
157 };
158 let upper = ident.to_ascii_uppercase();
159 let message = match upper.as_str() {
160 "ANALYTICS" => {
161 "CREATE ANALYTICS is not supported in Analytics v0 — \
162 use CREATE METRIC <dotted.path> for the metric-centric \
163 catalog (PRD #782 non-goal)"
164 }
165 "EVENT" => {
166 "CREATE EVENT is not supported in Analytics v0 — \
167 event-shaped data lives in ordinary TABLE/DOCUMENT \
168 collections, not a new storage model (PRD #782 non-goal)"
169 }
170 "COHORT" => {
171 "CREATE COHORT is not supported in Analytics v0 — \
172 cohort surfaces are deferred (PRD #782 non-goal)"
173 }
174 "FUNNEL" => {
175 "CREATE FUNNEL is not supported in Analytics v0 — \
176 funnel surfaces are deferred (PRD #782 non-goal)"
177 }
178 "SLA" => {
179 "CREATE SLA is not supported in Analytics v0 — \
180 SLA/legal/commercial contract modeling is post-MVP \
181 (PRD #782 non-goal)"
182 }
183 "ADAPTER" => {
184 "CREATE ADAPTER is not supported in Analytics v0 — \
185 Prometheus/Grafana/Snowplow/Google Analytics adapters \
186 are deferred (PRD #782 non-goal)"
187 }
188 _ => return None,
189 };
190 Some(message.to_string())
191}
192
193fn collection_model_filter(model: &str) -> Filter {
194 Filter::Compare {
195 field: FieldRef::column("", "model"),
196 op: CompareOp::Eq,
197 value: Value::Text(model.to_string().into()),
198 }
199}
200
201#[cfg(test)]
202mod tests {
203 use super::*;
204 use reddb_types::catalog::CollectionModel;
205
206 fn frontend(input: &str) -> FrontendStatement {
207 parse_frontend(input)
208 .unwrap_or_else(|err| panic!("failed to parse frontend {input:?}: {err:?}"))
209 }
210
211 fn expr(input: &str) -> QueryExpr {
212 frontend(input).into_query_expr()
213 }
214
215 fn sql_command(input: &str) -> SqlCommand {
216 sql_command_result(input)
217 .unwrap_or_else(|err| panic!("failed to parse SQL command {input:?}: {err:?}"))
218 }
219
220 fn sql_command_result(input: &str) -> Result<SqlCommand, ParseError> {
221 let mut parser = Parser::new(input)?;
222 parser.parse_sql_command()
223 }
224
225 fn assert_text(value: &Value, expected: &str) {
226 match value {
227 Value::Text(text) => assert_eq!(text.as_ref(), expected),
228 other => panic!("expected text {expected:?}, got {other:?}"),
229 }
230 }
231
232 #[test]
233 fn parse_frontend_routes_core_sql_statements() {
234 let FrontendStatement::Sql(SqlStatement::Query(SqlQuery::Select(query))) =
235 frontend("SELECT * FROM users")
236 else {
237 panic!("SELECT should route to SqlStatement::Query::Select");
238 };
239 assert_eq!(query.table, "users");
240
241 let QueryExpr::Insert(query) = expr("INSERT INTO users (id, name) VALUES (1, 'ada')")
242 else {
243 panic!("INSERT should lower through the SQL frontend");
244 };
245 assert_eq!(query.table, "users");
246 assert_eq!(query.columns, vec!["id", "name"]);
247 assert_eq!(query.values.len(), 1);
248
249 let QueryExpr::Update(query) = expr("UPDATE users SET name = 'ada' WHERE id = 1") else {
250 panic!("UPDATE should lower through the SQL frontend");
251 };
252 assert_eq!(query.table, "users");
253 assert_eq!(query.assignments[0].0, "name");
254
255 let QueryExpr::Delete(query) = expr("DELETE FROM users WHERE id = 1") else {
256 panic!("DELETE should lower through the SQL frontend");
257 };
258 assert_eq!(query.table, "users");
259 assert!(query.filter.is_some());
260
261 let QueryExpr::CreateTable(query) = expr("CREATE TABLE users (id INT, name TEXT)") else {
262 panic!("CREATE TABLE should lower through the SQL frontend");
263 };
264 assert_eq!(query.collection_model, CollectionModel::Table);
265 assert_eq!(query.name, "users");
266 assert_eq!(query.columns[0].name, "id");
267
268 let QueryExpr::DropTable(query) = expr("DROP TABLE IF EXISTS users") else {
269 panic!("DROP TABLE should lower through the SQL frontend");
270 };
271 assert_eq!(query.name, "users");
272 assert!(query.if_exists);
273 }
274
275 #[test]
276 fn parse_frontend_routes_admin_and_catalog_sql() {
277 let QueryExpr::Table(query) = expr("SHOW COLLECTIONS") else {
278 panic!("SHOW COLLECTIONS should become a red.collections table query");
279 };
280 assert_eq!(query.table, "red.collections");
281 assert!(query.filter.is_some());
282
283 let QueryExpr::Table(query) = expr("SHOW TABLES LIMIT 5") else {
284 panic!("SHOW TABLES should become a filtered red.collections table query");
285 };
286 assert_eq!(query.table, "red.collections");
287 assert_eq!(query.limit, Some(5));
288 assert!(query.filter.is_some());
289
290 assert!(matches!(
291 expr("SHOW CONFIG durability.mode"),
292 QueryExpr::ShowConfig { prefix: Some(prefix), as_json: false } if prefix == "durability.mode"
293 ));
294 assert!(matches!(
295 expr("SHOW CONFIG"),
296 QueryExpr::ShowConfig {
297 prefix: None,
298 as_json: false
299 }
300 ));
301 assert!(matches!(
302 expr("SHOW CONFIG runtime.result_cache AS JSON"),
303 QueryExpr::ShowConfig { prefix: Some(prefix), as_json: true } if prefix == "runtime.result_cache"
304 ));
305 assert!(matches!(
306 expr("SHOW CONFIG FORMAT JSON"),
307 QueryExpr::ShowConfig {
308 prefix: None,
309 as_json: true
310 }
311 ));
312
313 let QueryExpr::SetConfig { key, value } = expr("SET CONFIG durability.mode = 'sync'")
314 else {
315 panic!("SET CONFIG should stay on the SQL admin surface");
316 };
317 assert_eq!(key, "durability.mode");
318 assert_text(&value, "sync");
319
320 let QueryExpr::SetSecret { key, value } = expr("SET SECRET provider.api_key = 'sk_test'")
321 else {
322 panic!("SET SECRET should stay on the SQL admin surface");
323 };
324 assert_eq!(key, "provider.api_key");
325 assert_text(&value, "sk_test");
326 assert!(matches!(
327 expr("SET SECRET red.secrets.provider.api_key = 'sk_test'"),
328 QueryExpr::SetSecret { key, .. } if key == "red.secret.provider.api_key"
329 ));
330
331 assert!(matches!(
332 expr("DELETE SECRET provider.api_key"),
333 QueryExpr::DeleteSecret { key } if key == "provider.api_key"
334 ));
335 assert!(matches!(
336 expr("DELETE SECRET red.secrets.provider.api_key"),
337 QueryExpr::DeleteSecret { key } if key == "red.secret.provider.api_key"
338 ));
339 assert!(matches!(
340 expr("SHOW SECRETS provider"),
341 QueryExpr::ShowSecrets { prefix: Some(prefix) } if prefix == "provider"
342 ));
343 assert!(matches!(
344 expr("SHOW SECRETS red.secrets.provider"),
345 QueryExpr::ShowSecrets { prefix: Some(prefix) } if prefix == "red.secret.provider"
346 ));
347 assert!(matches!(
348 expr("SET TENANT 'acme'"),
349 QueryExpr::SetTenant(Some(tenant)) if tenant == "acme"
350 ));
351 assert!(matches!(expr("RESET TENANT"), QueryExpr::SetTenant(None)));
352 assert!(matches!(expr("SHOW TENANT"), QueryExpr::ShowTenant));
353 assert!(matches!(
354 expr("BEGIN ISOLATION LEVEL SNAPSHOT"),
355 QueryExpr::TransactionControl(TxnControl::Begin)
356 ));
357 assert!(matches!(
358 expr("ROLLBACK TO SAVEPOINT sp1"),
359 QueryExpr::TransactionControl(TxnControl::RollbackToSavepoint(name)) if name == "sp1"
360 ));
361 assert!(matches!(
362 expr("VACUUM FULL users"),
363 QueryExpr::MaintenanceCommand(MaintenanceCommand::Vacuum {
364 target: Some(target),
365 full: true,
366 }) if target == "users"
367 ));
368 }
369
370 #[test]
371 fn parse_frontend_routes_extended_schema_sql() {
372 assert!(matches!(
373 expr("CREATE SCHEMA IF NOT EXISTS app"),
374 QueryExpr::CreateSchema(CreateSchemaQuery {
375 name,
376 if_not_exists: true,
377 }) if name == "app"
378 ));
379 assert!(matches!(
380 expr("DROP SCHEMA IF EXISTS app CASCADE"),
381 QueryExpr::DropSchema(DropSchemaQuery {
382 name,
383 if_exists: true,
384 cascade: true,
385 }) if name == "app"
386 ));
387 assert!(matches!(
388 expr("CREATE SEQUENCE IF NOT EXISTS seq START WITH 10 INCREMENT BY 2"),
389 QueryExpr::CreateSequence(CreateSequenceQuery {
390 name,
391 if_not_exists: true,
392 start: 10,
393 increment: 2,
394 }) if name == "seq"
395 ));
396 assert!(matches!(
397 expr("DROP SEQUENCE IF EXISTS seq"),
398 QueryExpr::DropSequence(DropSequenceQuery {
399 name,
400 if_exists: true,
401 }) if name == "seq"
402 ));
403
404 let QueryExpr::CopyFrom(copy) = expr(
405 "COPY users FROM '/tmp/u.csv' WITH (FORMAT = csv, HEADER = true, DELIMITER = ';')",
406 ) else {
407 panic!("COPY should lower through SQL frontend");
408 };
409 assert_eq!(copy.table, "users");
410 assert_eq!(copy.path, "/tmp/u.csv");
411 assert_eq!(copy.format, CopyFormat::Csv);
412 assert_eq!(copy.delimiter, Some(';'));
413 assert!(copy.has_header);
414
415 let QueryExpr::CreateView(view) = expr(
416 "CREATE MATERIALIZED VIEW IF NOT EXISTS mv WITH RETENTION 1 h \
417 AS SELECT id FROM users REFRESH EVERY 5 s",
418 ) else {
419 panic!("CREATE MATERIALIZED VIEW should lower through SQL frontend");
420 };
421 assert_eq!(view.name, "mv");
422 assert!(view.materialized);
423 assert!(view.if_not_exists);
424 assert_eq!(view.retention_duration_ms, Some(3_600_000));
425 assert_eq!(view.refresh_every_ms, Some(5_000));
426 assert!(matches!(*view.query, QueryExpr::Table(_)));
427
428 assert!(matches!(
429 expr("DROP MATERIALIZED VIEW IF EXISTS mv"),
430 QueryExpr::DropView(DropViewQuery {
431 name,
432 materialized: true,
433 if_exists: true,
434 }) if name == "mv"
435 ));
436 assert!(matches!(
437 expr("REFRESH MATERIALIZED VIEW mv"),
438 QueryExpr::RefreshMaterializedView(RefreshMaterializedViewQuery { name }) if name == "mv"
439 ));
440 }
441
442 #[test]
443 fn parse_frontend_routes_fdw_policy_auth_and_migrations() {
444 let QueryExpr::CreateServer(server) = expr(
445 "CREATE SERVER IF NOT EXISTS csvsrv FOREIGN DATA WRAPPER csv OPTIONS (path '/data')",
446 ) else {
447 panic!("CREATE SERVER should lower through SQL frontend");
448 };
449 assert_eq!(server.name, "csvsrv");
450 assert_eq!(server.wrapper, "csv");
451 assert!(server.if_not_exists);
452 assert_eq!(
453 server.options,
454 vec![("path".to_string(), "/data".to_string())]
455 );
456
457 let QueryExpr::CreateForeignTable(table) = expr(
458 "CREATE FOREIGN TABLE IF NOT EXISTS ext_users \
459 (id INT, name TEXT) SERVER csvsrv OPTIONS (file 'users.csv')",
460 ) else {
461 panic!("CREATE FOREIGN TABLE should lower through SQL frontend");
462 };
463 assert_eq!(table.name, "ext_users");
464 assert_eq!(table.server, "csvsrv");
465 assert!(table.if_not_exists);
466 assert_eq!(table.columns.len(), 2);
467 assert!(!table.columns[0].not_null);
468
469 assert!(matches!(
470 expr("DROP SERVER IF EXISTS csvsrv CASCADE"),
471 QueryExpr::DropServer(DropServerQuery {
472 name,
473 if_exists: true,
474 cascade: true,
475 }) if name == "csvsrv"
476 ));
477 assert!(matches!(
478 expr("DROP FOREIGN TABLE IF EXISTS ext_users"),
479 QueryExpr::DropForeignTable(DropForeignTableQuery {
480 name,
481 if_exists: true,
482 }) if name == "ext_users"
483 ));
484
485 let QueryExpr::CreatePolicy(policy) = expr(
486 "CREATE POLICY readonly ON NODES OF mygraph FOR SELECT TO analytics USING (public = 1)",
487 ) else {
488 panic!("CREATE POLICY should lower through SQL frontend");
489 };
490 assert_eq!(policy.name, "readonly");
491 assert_eq!(policy.table, "mygraph");
492 assert_eq!(policy.action, Some(PolicyAction::Select));
493 assert_eq!(policy.role.as_deref(), Some("analytics"));
494 assert_eq!(policy.target_kind.as_ident(), "nodes");
495
496 assert!(matches!(
497 expr("DROP POLICY IF EXISTS readonly ON mygraph"),
498 QueryExpr::DropPolicy(DropPolicyQuery {
499 name,
500 table,
501 if_exists: true,
502 }) if name == "readonly" && table == "mygraph"
503 ));
504
505 assert!(matches!(
506 expr("GRANT SELECT ON TABLE public.users TO tenant1.alice"),
507 QueryExpr::Grant(grant)
508 if grant.actions == vec!["SELECT"]
509 && grant.objects[0].schema.as_deref() == Some("public")
510 ));
511 assert!(matches!(
512 expr("REVOKE GRANT OPTION FOR USAGE ON SCHEMA analytics FROM GROUP analysts"),
513 QueryExpr::Revoke(revoke) if revoke.grant_option_for && revoke.all == false
514 ));
515 assert!(matches!(
516 expr("ALTER USER bob ENABLE SET search_path TO 'public'"),
517 QueryExpr::AlterUser(user)
518 if user.username == "bob" && user.attributes.len() == 2
519 ));
520 assert!(matches!(
521 expr("CREATE USER tenant1.alice WITH PASSWORD 'pw' ROLE write"),
522 QueryExpr::CreateUser(user)
523 if user.tenant.as_deref() == Some("tenant1")
524 && user.username == "alice"
525 && user.password == "pw"
526 && user.role == "write"
527 ));
528
529 assert!(matches!(
530 expr("CREATE POLICY 'readonly' AS '{\"Statement\":[]}'"),
531 QueryExpr::CreateIamPolicy { id, json }
532 if id == "readonly" && json == "{\"Statement\":[]}"
533 ));
534 assert!(matches!(
535 expr("DROP POLICY 'readonly'"),
536 QueryExpr::DropIamPolicy { id } if id == "readonly"
537 ));
538
539 assert!(matches!(
540 expr("CREATE MIGRATION m2 DEPENDS ON m0 BATCH 10 ROWS AS CREATE TABLE accounts (id INT)"),
541 QueryExpr::CreateMigration(migration)
542 if migration.name == "m2"
543 && migration.depends_on == vec!["m0".to_string()]
544 && migration.batch_size == Some(10)
545 ));
546 assert!(matches!(
547 expr("APPLY MIGRATION * FOR TENANT tenant1"),
548 QueryExpr::ApplyMigration(apply)
549 if apply.for_tenant.as_deref() == Some("tenant1")
550 ));
551 assert!(matches!(
552 expr("ROLLBACK MIGRATION m2"),
553 QueryExpr::RollbackMigration(RollbackMigrationQuery { name }) if name == "m2"
554 ));
555 assert!(matches!(
556 expr("EXPLAIN MIGRATION m2"),
557 QueryExpr::ExplainMigration(ExplainMigrationQuery { name }) if name == "m2"
558 ));
559 }
560
561 #[test]
562 fn parse_sql_statement_covers_statement_category_wrapping() {
563 enum Expected {
564 Select,
565 Insert,
566 CreateSchema,
567 SetTenant,
568 }
569
570 let cases = [
571 ("SELECT * FROM users", Expected::Select),
572 ("INSERT INTO users (id) VALUES (1)", Expected::Insert),
573 ("CREATE SCHEMA app", Expected::CreateSchema),
574 ("SET TENANT 'acme'", Expected::SetTenant),
575 ];
576
577 for (input, expected) in cases {
578 let mut parser = Parser::new(input).expect("lexer");
579 let statement = parser
580 .parse_sql_statement()
581 .unwrap_or_else(|err| panic!("failed to parse {input:?}: {err:?}"));
582 let matched = match expected {
583 Expected::Select => matches!(statement, SqlStatement::Query(SqlQuery::Select(_))),
584 Expected::Insert => {
585 matches!(statement, SqlStatement::Mutation(SqlMutation::Insert(_)))
586 }
587 Expected::CreateSchema => matches!(
588 statement,
589 SqlStatement::Schema(SqlSchemaCommand::CreateSchema(_))
590 ),
591 Expected::SetTenant => {
592 matches!(
593 statement,
594 SqlStatement::Admin(SqlAdminCommand::SetTenant(_))
595 )
596 }
597 };
598 assert!(matched, "{input}");
599 }
600 }
601
602 #[test]
603 fn parse_frontend_routes_non_sql_frontends() {
604 let QueryExpr::KvCommand(KvCommand::Get {
605 model,
606 collection,
607 key,
608 }) = expr("KV GET settings.feature")
609 else {
610 panic!("KV GET should route to FrontendStatement::KvCommand");
611 };
612 assert_eq!(model, CollectionModel::Kv);
613 assert_eq!(collection, "settings");
614 assert_eq!(key, "feature");
615
616 let QueryExpr::ConfigCommand(ConfigCommand::Watch {
617 collection,
618 key,
619 prefix,
620 from_lsn,
621 }) = expr("WATCH CONFIG app PREFIX feature FROM LSN 7")
622 else {
623 panic!("WATCH CONFIG should route to FrontendStatement::ConfigCommand");
624 };
625 assert_eq!(collection, "app");
626 assert_eq!(key, "feature");
627 assert!(prefix);
628 assert_eq!(from_lsn, Some(7));
629
630 let QueryExpr::ConfigCommand(ConfigCommand::List {
631 collection,
632 prefix,
633 limit,
634 offset,
635 }) = expr("LIST CONFIG app PREFIX feature LIMIT 3 OFFSET 1")
636 else {
637 panic!("LIST CONFIG should route to FrontendStatement::ConfigCommand");
638 };
639 assert_eq!(collection, "app");
640 assert_eq!(prefix.as_deref(), Some("feature"));
641 assert_eq!(limit, Some(3));
642 assert_eq!(offset, 1);
643
644 let QueryExpr::KvCommand(KvCommand::List {
645 model,
646 collection,
647 prefix,
648 limit,
649 offset,
650 as_json,
651 }) = expr("KV LIST settings PREFIX 'feature.' LIMIT 10 OFFSET 2")
652 else {
653 panic!("KV LIST should route to FrontendStatement::KvCommand");
654 };
655 assert_eq!(model, CollectionModel::Kv);
656 assert_eq!(collection, "settings");
657 assert_eq!(prefix.as_deref(), Some("feature."));
658 assert_eq!(limit, Some(10));
659 assert_eq!(offset, 2);
660 assert!(!as_json);
661
662 let QueryExpr::KvCommand(KvCommand::List {
663 model,
664 collection,
665 prefix,
666 limit,
667 offset,
668 as_json,
669 }) = expr("LIST KV settings PREFIX feature LIMIT 10 OFFSET 2")
670 else {
671 panic!("LIST KV should route to FrontendStatement::KvCommand");
672 };
673 assert_eq!(model, CollectionModel::Kv);
674 assert_eq!(collection, "settings");
675 assert_eq!(prefix.as_deref(), Some("feature"));
676 assert_eq!(limit, Some(10));
677 assert_eq!(offset, 2);
678 assert!(!as_json);
679
680 let QueryExpr::KvCommand(KvCommand::List {
681 model,
682 collection,
683 prefix,
684 as_json,
685 ..
686 }) = expr("KV LIST settings PREFIX feature AS JSON")
687 else {
688 panic!("KV LIST AS JSON should route to FrontendStatement::KvCommand");
689 };
690 assert_eq!(model, CollectionModel::Kv);
691 assert_eq!(collection, "settings");
692 assert_eq!(prefix.as_deref(), Some("feature"));
693 assert!(as_json);
694
695 let QueryExpr::KvCommand(KvCommand::Watch {
696 model,
697 collection,
698 key,
699 prefix,
700 from_lsn,
701 }) = expr("WATCH sessions.user.* FROM LSN 3")
702 else {
703 panic!("bare WATCH should route to FrontendStatement::KvCommand");
704 };
705 assert_eq!(model, CollectionModel::Kv);
706 assert_eq!(collection, "sessions");
707 assert_eq!(key, "user");
708 assert!(prefix);
709 assert_eq!(from_lsn, Some(3));
710
711 let QueryExpr::KvCommand(KvCommand::Watch {
712 model,
713 collection,
714 key,
715 prefix,
716 from_lsn,
717 }) = expr("WATCH VAULT secrets PREFIX api FROM LSN 7")
718 else {
719 panic!("WATCH VAULT should route to FrontendStatement::KvCommand");
720 };
721 assert_eq!(model, CollectionModel::Vault);
722 assert_eq!(collection, "secrets");
723 assert_eq!(key, "api");
724 assert!(prefix);
725 assert_eq!(from_lsn, Some(7));
726
727 let QueryExpr::KvCommand(KvCommand::List {
728 model,
729 collection,
730 prefix,
731 limit,
732 offset,
733 as_json,
734 }) = expr("LIST VAULT secrets PREFIX api LIMIT 10 OFFSET 2")
735 else {
736 panic!("LIST VAULT should route to FrontendStatement::KvCommand");
737 };
738 assert_eq!(model, CollectionModel::Vault);
739 assert_eq!(collection, "secrets");
740 assert_eq!(prefix.as_deref(), Some("api"));
741 assert_eq!(limit, Some(10));
742 assert_eq!(offset, 2);
743 assert!(!as_json);
744
745 assert!(matches!(
746 expr("INVALIDATE CONFIG app feature_flag"),
747 QueryExpr::ConfigCommand(ConfigCommand::InvalidVolatileOperation {
748 operation,
749 collection,
750 key: Some(key),
751 }) if operation == "INVALIDATE" && collection == "app" && key == "feature_flag"
752 ));
753 assert!(matches!(
754 expr("INVALIDATE TAGS [user:42, org:7] FROM sessions"),
755 QueryExpr::KvCommand(KvCommand::InvalidateTags { collection, tags })
756 if collection == "sessions" && tags == vec!["user:42".to_string(), "org:7".to_string()]
757 ));
758
759 let QueryExpr::EventsBackfill(query) =
760 expr("EVENTS BACKFILL users WHERE status = 'active' TO audit LIMIT 10")
761 else {
762 panic!("EVENTS BACKFILL should route to FrontendStatement::EventsBackfill");
763 };
764 assert_eq!(query.collection, "users");
765 assert_eq!(query.where_filter.as_deref(), Some("status = 'active'"));
766 assert_eq!(query.target_queue, "audit");
767 assert_eq!(query.limit, Some(10));
768
769 let QueryExpr::Table(query) = expr("EVENTS STATUS users LIMIT 2") else {
770 panic!("EVENTS STATUS should route through the SQL select surface");
771 };
772 assert_eq!(query.table, "red.subscriptions");
773 assert_eq!(query.limit, Some(2));
774 assert!(query.filter.is_some());
775
776 assert!(matches!(
777 expr("EVENTS BACKFILL STATUS users"),
778 QueryExpr::EventsBackfillStatus { collection } if collection == "users"
779 ));
780 assert!(parse_frontend("LIST UNKNOWN").is_err());
781 assert!(parse_frontend("EVENTS UNKNOWN").is_err());
782 }
783
784 #[test]
785 fn parse_frontend_routes_ranking_reads() {
786 assert!(matches!(
787 expr("RANK OF 42 IN page_rank"),
788 QueryExpr::RankOf(RankOfQuery { ranking, entity_id })
789 if ranking == "page_rank" && entity_id == 42
790 ));
791 assert!(matches!(
792 expr("APPROX RANK OF 7 IN page_rank"),
793 QueryExpr::ApproxRankOf(RankOfQuery { ranking, entity_id })
794 if ranking == "page_rank" && entity_id == 7
795 ));
796 assert!(matches!(
797 expr("RANK RANGE 1 TO 3 IN page_rank"),
798 QueryExpr::RankRange(RankRangeQuery { ranking, lo, hi })
799 if ranking == "page_rank" && lo == 1 && hi == 3
800 ));
801 assert!(matches!(
802 expr("ZRANK page_rank 0"),
803 QueryExpr::RankOf(RankOfQuery { ranking, entity_id })
804 if ranking == "page_rank" && entity_id == 0
805 ));
806 assert!(matches!(
807 expr("ZRANGE page_rank 0 3 WITHSCORES"),
808 QueryExpr::RankRange(RankRangeQuery { ranking, lo, hi })
809 if ranking == "page_rank" && lo == 1 && hi == 4
810 ));
811 assert!(
812 parse_frontend("RANK RANGE 3 TO 1 IN page_rank").is_err(),
813 "rank range must reject reversed bounds"
814 );
815 }
816
817 #[test]
818 fn parse_frontend_covers_multimodel_command_routing() {
819 assert!(matches!(
820 expr("GRAPH CENTRALITY ALGORITHM pagerank LIMIT 5"),
821 QueryExpr::GraphCommand(GraphCommand::Centrality {
822 algorithm,
823 limit: Some(5),
824 ..
825 }) if algorithm == "pagerank"
826 ));
827 assert!(matches!(
828 expr("SEARCH TEXT 'login failure' COLLECTION incidents LIMIT 20 FUZZY"),
829 QueryExpr::SearchCommand(SearchCommand::Text {
830 query,
831 collection: Some(collection),
832 limit: 20,
833 fuzzy: true,
834 ..
835 }) if query == "login failure" && collection == "incidents"
836 ));
837 assert!(matches!(
838 expr("ASK 'why did login fail?' USING openai LIMIT 3"),
839 QueryExpr::Ask(query)
840 if query.question == "why did login fail?"
841 && query.provider.as_deref() == Some("openai")
842 && query.limit == Some(3)
843 ));
844 assert!(matches!(
845 expr("QUEUE LEN tasks"),
846 QueryExpr::QueueCommand(QueueCommand::Len { queue }) if queue == "tasks"
847 ));
848 assert!(matches!(
849 expr("TREE REBALANCE forest.org DRY RUN"),
850 QueryExpr::TreeCommand(TreeCommand::Rebalance {
851 collection,
852 tree_name,
853 dry_run: true,
854 }) if collection == "forest" && tree_name == "org"
855 ));
856 assert!(matches!(
857 expr("HLL COUNT visitors"),
858 QueryExpr::ProbabilisticCommand(ProbabilisticCommand::HllCount { names })
859 if names == vec!["visitors".to_string()]
860 ));
861 }
862
863 #[test]
864 fn sql_command_round_trips_multimodel_schema_variants() {
865 macro_rules! assert_command_round_trip {
866 ($input:expr, $pattern:pat) => {{
867 let command = sql_command($input);
868 assert!(matches!(command, $pattern), "unexpected command for {}", $input);
869
870 let statement = sql_command($input).into_statement();
871 let command = statement.into_command();
872 assert!(
873 matches!(command, $pattern),
874 "statement round trip changed command for {}",
875 $input
876 );
877
878 let expr = sql_command($input).into_query_expr();
879 assert!(
880 !matches!(expr, QueryExpr::Table(TableQuery { table, .. }) if table.is_empty()),
881 "lowering produced an empty table placeholder for {}",
882 $input
883 );
884 }};
885 }
886
887 assert_command_round_trip!(
888 "EXPLAIN ALTER FOR CREATE TABLE users (id INT) FORMAT JSON",
889 SqlCommand::ExplainAlter(_)
890 );
891 assert_command_round_trip!("CREATE TABLE users (id INT)", SqlCommand::CreateTable(_));
892 assert_command_round_trip!("DROP TABLE IF EXISTS users", SqlCommand::DropTable(_));
893 assert_command_round_trip!(
894 "ALTER TABLE users ADD COLUMN status TEXT",
895 SqlCommand::AlterTable(_)
896 );
897 assert_command_round_trip!(
898 "CREATE INDEX idx_email ON users (email) USING HASH",
899 SqlCommand::CreateIndex(_)
900 );
901 assert_command_round_trip!(
902 "DROP INDEX IF EXISTS idx_email ON users",
903 SqlCommand::DropIndex(_)
904 );
905 assert_command_round_trip!("CREATE GRAPH identity", SqlCommand::CreateTable(_));
906 assert_command_round_trip!("CREATE DOCUMENT docs", SqlCommand::CreateTable(_));
907 assert_command_round_trip!(
908 "CREATE VECTOR embeddings DIM 4",
909 SqlCommand::CreateVector(_)
910 );
911 assert_command_round_trip!(
912 "CREATE COLLECTION turbo KIND vector.turbo DIM 3",
913 SqlCommand::CreateCollection(_)
914 );
915 assert_command_round_trip!("CREATE KV settings", SqlCommand::CreateTable(_));
916 assert_command_round_trip!("CREATE CONFIG app", SqlCommand::CreateTable(_));
917 assert_command_round_trip!(
918 "CREATE VAULT secrets WITH OWN MASTER KEY",
919 SqlCommand::CreateTable(_)
920 );
921 assert_command_round_trip!(
922 "CREATE TIMESERIES metrics RETENTION 90 d",
923 SqlCommand::CreateTimeSeries(_)
924 );
925 assert_command_round_trip!(
926 "CREATE METRIC svc.latency TYPE gauge ROLE sli",
927 SqlCommand::CreateMetric(_)
928 );
929 assert_command_round_trip!(
930 "ALTER METRIC svc.latency SET ROLE internal",
931 SqlCommand::AlterMetric(_)
932 );
933 assert_command_round_trip!(
934 "CREATE SLO svc.availability ON svc.latency TARGET 99.9 WINDOW 5 m",
935 SqlCommand::CreateSlo(_)
936 );
937 assert_command_round_trip!(
938 "CREATE QUEUE tasks MAX_SIZE 100",
939 SqlCommand::CreateQueue(_)
940 );
941 assert_command_round_trip!(
942 "ALTER QUEUE tasks SET MODE FANOUT",
943 SqlCommand::AlterQueue(_)
944 );
945 assert_command_round_trip!(
946 "CREATE TREE org IN forest ROOT LABEL root MAX_CHILDREN 4",
947 SqlCommand::CreateTree(_)
948 );
949 assert_command_round_trip!(
950 "CREATE HLL visitors PRECISION 14",
951 SqlCommand::Probabilistic(_)
952 );
953 assert_command_round_trip!(
954 "CREATE SKETCH freqs WIDTH 512 DEPTH 3",
955 SqlCommand::Probabilistic(_)
956 );
957 assert_command_round_trip!(
958 "CREATE FILTER seen CAPACITY 1024",
959 SqlCommand::Probabilistic(_)
960 );
961 assert_command_round_trip!("COPY users FROM '/tmp/u.csv'", SqlCommand::CopyFrom(_));
962 assert_command_round_trip!(
963 "CREATE VIEW active_users AS SELECT * FROM users",
964 SqlCommand::CreateView(_)
965 );
966 assert_command_round_trip!("DROP VIEW active_users", SqlCommand::DropView(_));
967 assert_command_round_trip!(
968 "REFRESH MATERIALIZED VIEW active_users",
969 SqlCommand::RefreshMaterializedView(_)
970 );
971 assert_command_round_trip!(
972 "CREATE SERVER mycsv FOREIGN DATA WRAPPER csv OPTIONS (base_path '/data')",
973 SqlCommand::CreateServer(_)
974 );
975 assert_command_round_trip!(
976 "DROP SERVER IF EXISTS mycsv CASCADE",
977 SqlCommand::DropServer(_)
978 );
979 assert_command_round_trip!(
980 "CREATE FOREIGN TABLE ext_users (id INT, name TEXT) SERVER mycsv OPTIONS (path 'users.csv')",
981 SqlCommand::CreateForeignTable(_)
982 );
983 assert_command_round_trip!(
984 "DROP FOREIGN TABLE IF EXISTS ext_users",
985 SqlCommand::DropForeignTable(_)
986 );
987 }
988
989 #[test]
990 fn sql_command_round_trips_drop_truncate_and_maintenance_variants() {
991 macro_rules! assert_command_round_trip {
992 ($input:expr, $pattern:pat) => {{
993 let command = sql_command($input);
994 assert!(
995 matches!(command, $pattern),
996 "unexpected command for {}",
997 $input
998 );
999 let statement = sql_command($input).into_statement();
1000 assert!(
1001 matches!(statement.into_command(), $pattern),
1002 "statement round trip changed command for {}",
1003 $input
1004 );
1005 }};
1006 }
1007
1008 assert_command_round_trip!("DROP GRAPH IF EXISTS identity", SqlCommand::DropGraph(_));
1009 assert_command_round_trip!(
1010 "DROP VECTOR IF EXISTS embeddings",
1011 SqlCommand::DropVector(_)
1012 );
1013 assert_command_round_trip!("DROP DOCUMENT IF EXISTS docs", SqlCommand::DropDocument(_));
1014 assert_command_round_trip!("DROP KV IF EXISTS settings", SqlCommand::DropKv(_));
1015 assert_command_round_trip!("DROP CONFIG IF EXISTS app", SqlCommand::DropKv(_));
1016 assert_command_round_trip!("DROP VAULT IF EXISTS secrets", SqlCommand::DropKv(_));
1017 assert_command_round_trip!(
1018 "DROP COLLECTION IF EXISTS docs",
1019 SqlCommand::DropCollection(_)
1020 );
1021 assert_command_round_trip!(
1022 "DROP TIMESERIES IF EXISTS metrics",
1023 SqlCommand::DropTimeSeries(_)
1024 );
1025 assert_command_round_trip!(
1026 "DROP HYPERTABLE IF EXISTS metrics",
1027 SqlCommand::DropTimeSeries(_)
1028 );
1029 assert_command_round_trip!("DROP QUEUE IF EXISTS tasks", SqlCommand::DropQueue(_));
1030 assert_command_round_trip!("DROP TREE IF EXISTS org IN forest", SqlCommand::DropTree(_));
1031 assert_command_round_trip!("DROP HLL IF EXISTS visitors", SqlCommand::Probabilistic(_));
1032 assert_command_round_trip!("DROP SKETCH IF EXISTS freqs", SqlCommand::Probabilistic(_));
1033 assert_command_round_trip!("DROP FILTER IF EXISTS seen", SqlCommand::Probabilistic(_));
1034 assert_command_round_trip!(
1035 "TRUNCATE VECTOR IF EXISTS embeddings",
1036 SqlCommand::Truncate(_)
1037 );
1038 assert_command_round_trip!(
1039 "COMMIT WORK",
1040 SqlCommand::TransactionControl(TxnControl::Commit)
1041 );
1042 assert_command_round_trip!(
1043 "ROLLBACK",
1044 SqlCommand::TransactionControl(TxnControl::Rollback)
1045 );
1046 assert_command_round_trip!(
1047 "SAVEPOINT before_batch",
1048 SqlCommand::TransactionControl(TxnControl::Savepoint(_))
1049 );
1050 assert_command_round_trip!(
1051 "RELEASE SAVEPOINT before_batch",
1052 SqlCommand::TransactionControl(TxnControl::ReleaseSavepoint(_))
1053 );
1054 assert_command_round_trip!(
1055 "ANALYZE users",
1056 SqlCommand::Maintenance(MaintenanceCommand::Analyze { .. })
1057 );
1058 assert_command_round_trip!(
1059 "VACUUM",
1060 SqlCommand::Maintenance(MaintenanceCommand::Vacuum { .. })
1061 );
1062 }
1063
1064 #[test]
1065 fn parse_sql_command_covers_show_and_error_branches() {
1066 assert!(matches!(
1067 sql_command("SHOW CREATE TABLE public.users"),
1068 SqlCommand::Select(TableQuery { table, .. }) if table == "red.show_create"
1069 ));
1070 assert!(matches!(
1071 sql_command("SHOW COLLECTIONS INCLUDING INTERNAL LIMIT 2"),
1072 SqlCommand::Select(TableQuery { table, limit: Some(2), .. }) if table == "red.collections"
1073 ));
1074 assert!(matches!(
1075 sql_command("SHOW QUEUES INCLUDING INTERNAL"),
1076 SqlCommand::Select(TableQuery { table, filter: None, .. }) if table == "red.queues"
1077 ));
1078 assert!(matches!(
1079 sql_command("SHOW INDICES ON users"),
1080 SqlCommand::Select(TableQuery { table, filter: Some(_), .. }) if table == "red.show_indexes"
1081 ));
1082 assert!(matches!(
1083 sql_command("SHOW POLICIES ON users WHERE action = 'SELECT'"),
1084 SqlCommand::Select(TableQuery { table, filter: Some(_), .. }) if table == "red.policies"
1085 ));
1086 assert!(matches!(
1087 sql_command("SHOW STATS 'users' WHERE rows > 0"),
1088 SqlCommand::Select(TableQuery { table, filter: Some(_), .. }) if table == "red.stats"
1089 ));
1090 assert!(matches!(
1091 sql_command("SHOW SAMPLE users"),
1092 SqlCommand::Select(TableQuery { table, limit: Some(10), .. }) if table == "users"
1093 ));
1094 assert!(matches!(
1095 sql_command("DESC public.users"),
1096 SqlCommand::Select(TableQuery { table, filter: Some(_), .. }) if table == "red.describe"
1097 ));
1098 assert!(
1099 sql_command_result("CREATE VIEW v WITH RETENTION 1 h AS SELECT * FROM users").is_err()
1100 );
1101 assert!(sql_command_result("CREATE TABLE bad WITH ANALYTICS (centrality)").is_err());
1102 assert!(sql_command_result("BEGIN ISOLATION LEVEL SERIALIZABLE").is_err());
1103 assert!(sql_command_result("EVENTS BACKFILL STATUS users").is_err());
1104 }
1105
1106 #[test]
1107 fn parse_sql_command_covers_remaining_catalog_and_copy_shapes() {
1108 for input in [
1109 "SHOW VECTORS",
1110 "SHOW DOCUMENTS",
1111 "SHOW TIMESERIES",
1112 "SHOW GRAPHS",
1113 "SHOW CONFIGS",
1114 "SHOW VAULTS",
1115 "SHOW KV",
1116 "SHOW SCHEMA public.users",
1117 ] {
1118 assert!(
1119 matches!(sql_command(input), SqlCommand::Select(_)),
1120 "{input}"
1121 );
1122 }
1123
1124 for input in [
1125 "TRUNCATE TABLE users",
1126 "TRUNCATE GRAPH identity",
1127 "TRUNCATE DOCUMENT docs",
1128 "TRUNCATE TIMESERIES metrics",
1129 "TRUNCATE METRICS metrics",
1130 "TRUNCATE KV settings",
1131 "TRUNCATE QUEUE tasks",
1132 "TRUNCATE COLLECTION docs",
1133 ] {
1134 assert!(
1135 matches!(sql_command(input), SqlCommand::Truncate(_)),
1136 "{input}"
1137 );
1138 }
1139 assert!(sql_command_result("TRUNCATE UNKNOWN users").is_err());
1140
1141 let SqlCommand::CopyFrom(copy) = sql_command("COPY users FROM '/tmp/u.csv' WITH (HEADER)")
1142 else {
1143 panic!("expected COPY");
1144 };
1145 assert!(copy.has_header);
1146 assert_eq!(copy.delimiter, None);
1147
1148 let SqlCommand::CopyFrom(copy) =
1149 sql_command("COPY users FROM '/tmp/u.csv' WITH (HEADER = false)")
1150 else {
1151 panic!("expected COPY");
1152 };
1153 assert!(!copy.has_header);
1154
1155 let SqlCommand::CopyFrom(copy) =
1156 sql_command("COPY users FROM '/tmp/u.csv' DELIMITER '|' HEADER")
1157 else {
1158 panic!("expected COPY");
1159 };
1160 assert_eq!(copy.delimiter, Some('|'));
1161 assert!(copy.has_header);
1162 }
1163
1164 #[test]
1165 fn parse_sql_command_covers_remaining_ranking_and_event_errors() {
1166 assert!(matches!(
1167 expr("APPROXIMATE RANK OF 9 IN page_rank"),
1168 QueryExpr::ApproxRankOf(RankOfQuery { ranking, entity_id })
1169 if ranking == "page_rank" && entity_id == 9
1170 ));
1171 assert!(parse_frontend("APPROX OF 7 IN page_rank").is_err());
1172 assert!(parse_frontend("RANK 1 IN page_rank").is_err());
1173 assert!(parse_frontend("RANK RANGE 0 TO 3 IN page_rank").is_err());
1174 assert!(parse_frontend("ZRANK page_rank -1").is_err());
1175 assert!(parse_frontend("ZRANGE page_rank 3 1").is_err());
1176
1177 let QueryExpr::Table(query) = expr("EVENTS STATUS 'users' WHERE active = true") else {
1178 panic!("EVENTS STATUS should accept a quoted collection");
1179 };
1180 assert_eq!(query.table, "red.subscriptions");
1181 assert!(query.filter.is_some());
1182 assert!(query.where_expr.is_some());
1183
1184 let QueryExpr::EventsBackfill(query) = expr("EVENTS BACKFILL users TO audit") else {
1185 panic!("EVENTS BACKFILL should allow omitted filter and limit");
1186 };
1187 assert_eq!(query.collection, "users");
1188 assert_eq!(query.where_filter, None);
1189 assert_eq!(query.limit, None);
1190
1191 assert!(parse_frontend("EVENTS BACKFILL users WHERE TO audit").is_err());
1192 }
1193
1194 #[test]
1195 fn parse_sql_command_covers_analytics_non_goal_rejections() {
1196 for head in ["ANALYTICS", "EVENT", "COHORT", "FUNNEL", "SLA", "ADAPTER"] {
1197 let err = sql_command_result(&format!("CREATE {head} demo"))
1198 .expect_err("analytics v0 non-goal should be rejected");
1199 assert!(err.to_string().contains(&format!("CREATE {head}")), "{err}");
1200 }
1201 }
1202
1203 #[test]
1204 fn parse_sql_command_covers_transaction_isolation_edges() {
1205 for input in [
1206 "BEGIN ISOLATION LEVEL READ UNCOMMITTED",
1207 "BEGIN ISOLATION LEVEL READ COMMITTED",
1208 "BEGIN ISOLATION LEVEL REPEATABLE READ",
1209 "START TRANSACTION ISOLATION LEVEL SNAPSHOT",
1210 ] {
1211 assert!(
1212 matches!(
1213 sql_command(input),
1214 SqlCommand::TransactionControl(TxnControl::Begin)
1215 ),
1216 "{input}"
1217 );
1218 }
1219
1220 assert!(sql_command_result("BEGIN ISOLATION LEVEL READ").is_err());
1221 assert!(sql_command_result("BEGIN ISOLATION LEVEL REPEATABLE").is_err());
1222 assert!(sql_command_result("BEGIN ISOLATION LEVEL CHAOS").is_err());
1223 }
1224
1225 #[test]
1226 fn parse_sql_command_covers_iam_and_hypertable_dispatch_edges() {
1227 assert!(matches!(
1228 expr("CREATE HYPERTABLE metrics TIME_COLUMN ts CHUNK_INTERVAL '1d'"),
1229 QueryExpr::CreateTimeSeries(query)
1230 if query.name == "metrics" && query.hypertable.is_some()
1231 ));
1232 assert!(sql_command_result("CREATE OR TABLE bad (id INT)").is_err());
1233 assert!(sql_command_result("DROP MATERIALIZED TABLE bad").is_err());
1234
1235 assert!(matches!(
1236 expr("ATTACH POLICY 'readonly' TO USER tenant1.alice"),
1237 QueryExpr::AttachPolicy { policy_id, .. } if policy_id == "readonly"
1238 ));
1239 assert!(matches!(
1240 expr("DETACH POLICY 'readonly' FROM GROUP analysts"),
1241 QueryExpr::DetachPolicy { policy_id, .. } if policy_id == "readonly"
1242 ));
1243 assert!(matches!(
1244 expr("SHOW POLICIES FOR USER alice"),
1245 QueryExpr::ShowPolicies { filter: Some(_) }
1246 ));
1247 assert!(matches!(
1248 expr("SHOW EFFECTIVE PERMISSIONS FOR alice"),
1249 QueryExpr::ShowEffectivePermissions { resource: None, .. }
1250 ));
1251 assert!(matches!(
1252 expr("SIMULATE alice ACTION 'iam:PassRole' ON TABLE:public.orders"),
1253 QueryExpr::SimulatePolicy { action, .. } if action == "iam:PassRole"
1254 ));
1255 assert!(matches!(
1256 expr("LINT POLICY JSON '{\"Statement\":[]}'"),
1257 QueryExpr::LintPolicy { .. }
1258 ));
1259 assert!(matches!(
1260 expr("MIGRATE POLICY MODE TO 'policy_only' DRY RUN"),
1261 QueryExpr::MigratePolicyMode {
1262 target,
1263 dry_run: true,
1264 } if target == "policy_only"
1265 ));
1266 assert!(parse_frontend("MIGRATE OTHER").is_err());
1267 }
1268
1269 #[test]
1270 fn parse_frontend_rejects_trailing_tokens() {
1271 let err = parse_frontend("SET TENANT 'acme' junk")
1272 .expect_err("parse_frontend should reject trailing tokens");
1273 assert!(
1274 err.to_string().contains("Unexpected token after query"),
1275 "{err}"
1276 );
1277 }
1278}
1279
1280fn add_table_filter(query: &mut TableQuery, filter: Filter) {
1281 let combined = match query.filter.take() {
1282 Some(existing) => existing.and(filter),
1283 None => filter,
1284 };
1285 query.where_expr = Some(filter_to_expr(&combined));
1286 query.filter = Some(combined);
1287}
1288
1289fn parse_show_collections_by_model(
1290 parser: &mut Parser<'_>,
1291 model: &str,
1292) -> Result<TableQuery, ParseError> {
1293 let mut query = TableQuery::new("red.collections");
1294 parser.parse_table_clauses(&mut query)?;
1295 add_table_filter(&mut query, collection_model_filter(model));
1296 Ok(query)
1297}
1298
1299#[derive(Debug, Clone)]
1300#[allow(clippy::large_enum_variant)]
1301pub enum SqlQuery {
1302 Select(TableQuery),
1303 Join(JoinQuery),
1304}
1305
1306#[derive(Debug, Clone)]
1307pub enum SqlMutation {
1308 Insert(InsertQuery),
1309 Update(UpdateQuery),
1310 Delete(DeleteQuery),
1311}
1312
1313#[derive(Debug, Clone)]
1314pub enum SqlSchemaCommand {
1315 ExplainAlter(ExplainAlterQuery),
1316 CreateTable(CreateTableQuery),
1317 CreateCollection(CreateCollectionQuery),
1318 CreateVector(CreateVectorQuery),
1319 DropTable(DropTableQuery),
1320 DropGraph(DropGraphQuery),
1321 DropVector(DropVectorQuery),
1322 DropDocument(DropDocumentQuery),
1323 DropKv(DropKvQuery),
1324 DropCollection(DropCollectionQuery),
1325 Truncate(TruncateQuery),
1326 AlterTable(AlterTableQuery),
1327 CreateIndex(CreateIndexQuery),
1328 DropIndex(DropIndexQuery),
1329 CreateTimeSeries(CreateTimeSeriesQuery),
1330 CreateMetric(CreateMetricQuery),
1331 AlterMetric(AlterMetricQuery),
1332 CreateSlo(CreateSloQuery),
1333 DropTimeSeries(DropTimeSeriesQuery),
1334 CreateQueue(CreateQueueQuery),
1335 AlterQueue(AlterQueueQuery),
1336 DropQueue(DropQueueQuery),
1337 CreateTree(CreateTreeQuery),
1338 DropTree(DropTreeQuery),
1339 Probabilistic(ProbabilisticCommand),
1340 CreateSchema(CreateSchemaQuery),
1341 DropSchema(DropSchemaQuery),
1342 CreateSequence(CreateSequenceQuery),
1343 DropSequence(DropSequenceQuery),
1344 CopyFrom(CopyFromQuery),
1345 CreateView(CreateViewQuery),
1346 DropView(DropViewQuery),
1347 RefreshMaterializedView(RefreshMaterializedViewQuery),
1348 CreatePolicy(CreatePolicyQuery),
1349 DropPolicy(DropPolicyQuery),
1350 CreateServer(CreateServerQuery),
1351 DropServer(DropServerQuery),
1352 CreateForeignTable(CreateForeignTableQuery),
1353 DropForeignTable(DropForeignTableQuery),
1354 CreateMigration(CreateMigrationQuery),
1355 ApplyMigration(ApplyMigrationQuery),
1356 RollbackMigration(RollbackMigrationQuery),
1357 ExplainMigration(ExplainMigrationQuery),
1358}
1359
1360#[derive(Debug, Clone)]
1361#[allow(clippy::large_enum_variant)]
1362pub enum SqlAdminCommand {
1363 SetConfig {
1364 key: String,
1365 value: Value,
1366 },
1367 ShowConfig {
1368 prefix: Option<String>,
1369 as_json: bool,
1370 },
1371 SetSecret {
1372 key: String,
1373 value: Value,
1374 },
1375 DeleteSecret {
1376 key: String,
1377 },
1378 ShowSecrets {
1379 prefix: Option<String>,
1380 },
1381 SetTenant(Option<String>),
1382 ShowTenant,
1383 TransactionControl(TxnControl),
1384 Maintenance(MaintenanceCommand),
1385 Grant(GrantStmt),
1386 Revoke(RevokeStmt),
1387 AlterUser(AlterUserStmt),
1388 CreateUser(CreateUserStmt),
1389 IamPolicy(QueryExpr),
1390}
1391
1392impl SqlStatement {
1393 pub fn into_command(self) -> SqlCommand {
1394 match self {
1395 SqlStatement::Query(SqlQuery::Select(query)) => SqlCommand::Select(query),
1396 SqlStatement::Query(SqlQuery::Join(query)) => SqlCommand::Join(query),
1397 SqlStatement::Mutation(SqlMutation::Insert(query)) => SqlCommand::Insert(query),
1398 SqlStatement::Mutation(SqlMutation::Update(query)) => SqlCommand::Update(query),
1399 SqlStatement::Mutation(SqlMutation::Delete(query)) => SqlCommand::Delete(query),
1400 SqlStatement::Schema(SqlSchemaCommand::ExplainAlter(query)) => {
1401 SqlCommand::ExplainAlter(query)
1402 }
1403 SqlStatement::Schema(SqlSchemaCommand::CreateTable(query)) => {
1404 SqlCommand::CreateTable(query)
1405 }
1406 SqlStatement::Schema(SqlSchemaCommand::CreateCollection(query)) => {
1407 SqlCommand::CreateCollection(query)
1408 }
1409 SqlStatement::Schema(SqlSchemaCommand::CreateVector(query)) => {
1410 SqlCommand::CreateVector(query)
1411 }
1412 SqlStatement::Schema(SqlSchemaCommand::DropTable(query)) => {
1413 SqlCommand::DropTable(query)
1414 }
1415 SqlStatement::Schema(SqlSchemaCommand::DropGraph(query)) => {
1416 SqlCommand::DropGraph(query)
1417 }
1418 SqlStatement::Schema(SqlSchemaCommand::DropVector(query)) => {
1419 SqlCommand::DropVector(query)
1420 }
1421 SqlStatement::Schema(SqlSchemaCommand::DropDocument(query)) => {
1422 SqlCommand::DropDocument(query)
1423 }
1424 SqlStatement::Schema(SqlSchemaCommand::DropKv(query)) => SqlCommand::DropKv(query),
1425 SqlStatement::Schema(SqlSchemaCommand::DropCollection(query)) => {
1426 SqlCommand::DropCollection(query)
1427 }
1428 SqlStatement::Schema(SqlSchemaCommand::Truncate(query)) => SqlCommand::Truncate(query),
1429 SqlStatement::Schema(SqlSchemaCommand::AlterTable(query)) => {
1430 SqlCommand::AlterTable(query)
1431 }
1432 SqlStatement::Schema(SqlSchemaCommand::CreateIndex(query)) => {
1433 SqlCommand::CreateIndex(query)
1434 }
1435 SqlStatement::Schema(SqlSchemaCommand::DropIndex(query)) => {
1436 SqlCommand::DropIndex(query)
1437 }
1438 SqlStatement::Schema(SqlSchemaCommand::CreateTimeSeries(query)) => {
1439 SqlCommand::CreateTimeSeries(query)
1440 }
1441 SqlStatement::Schema(SqlSchemaCommand::CreateMetric(query)) => {
1442 SqlCommand::CreateMetric(query)
1443 }
1444 SqlStatement::Schema(SqlSchemaCommand::AlterMetric(query)) => {
1445 SqlCommand::AlterMetric(query)
1446 }
1447 SqlStatement::Schema(SqlSchemaCommand::CreateSlo(query)) => {
1448 SqlCommand::CreateSlo(query)
1449 }
1450 SqlStatement::Schema(SqlSchemaCommand::DropTimeSeries(query)) => {
1451 SqlCommand::DropTimeSeries(query)
1452 }
1453 SqlStatement::Schema(SqlSchemaCommand::CreateQueue(query)) => {
1454 SqlCommand::CreateQueue(query)
1455 }
1456 SqlStatement::Schema(SqlSchemaCommand::AlterQueue(query)) => {
1457 SqlCommand::AlterQueue(query)
1458 }
1459 SqlStatement::Schema(SqlSchemaCommand::DropQueue(query)) => {
1460 SqlCommand::DropQueue(query)
1461 }
1462 SqlStatement::Schema(SqlSchemaCommand::CreateTree(query)) => {
1463 SqlCommand::CreateTree(query)
1464 }
1465 SqlStatement::Schema(SqlSchemaCommand::DropTree(query)) => SqlCommand::DropTree(query),
1466 SqlStatement::Schema(SqlSchemaCommand::Probabilistic(command)) => {
1467 SqlCommand::Probabilistic(command)
1468 }
1469 SqlStatement::Admin(SqlAdminCommand::SetConfig { key, value }) => {
1470 SqlCommand::SetConfig { key, value }
1471 }
1472 SqlStatement::Admin(SqlAdminCommand::ShowConfig { prefix, as_json }) => {
1473 SqlCommand::ShowConfig { prefix, as_json }
1474 }
1475 SqlStatement::Admin(SqlAdminCommand::SetSecret { key, value }) => {
1476 SqlCommand::SetSecret { key, value }
1477 }
1478 SqlStatement::Admin(SqlAdminCommand::DeleteSecret { key }) => {
1479 SqlCommand::DeleteSecret { key }
1480 }
1481 SqlStatement::Admin(SqlAdminCommand::ShowSecrets { prefix }) => {
1482 SqlCommand::ShowSecrets { prefix }
1483 }
1484 SqlStatement::Admin(SqlAdminCommand::SetTenant(value)) => SqlCommand::SetTenant(value),
1485 SqlStatement::Admin(SqlAdminCommand::ShowTenant) => SqlCommand::ShowTenant,
1486 SqlStatement::Admin(SqlAdminCommand::TransactionControl(ctl)) => {
1487 SqlCommand::TransactionControl(ctl)
1488 }
1489 SqlStatement::Admin(SqlAdminCommand::Maintenance(cmd)) => SqlCommand::Maintenance(cmd),
1490 SqlStatement::Schema(SqlSchemaCommand::CreateSchema(q)) => SqlCommand::CreateSchema(q),
1491 SqlStatement::Schema(SqlSchemaCommand::DropSchema(q)) => SqlCommand::DropSchema(q),
1492 SqlStatement::Schema(SqlSchemaCommand::CreateSequence(q)) => {
1493 SqlCommand::CreateSequence(q)
1494 }
1495 SqlStatement::Schema(SqlSchemaCommand::DropSequence(q)) => SqlCommand::DropSequence(q),
1496 SqlStatement::Schema(SqlSchemaCommand::CopyFrom(q)) => SqlCommand::CopyFrom(q),
1497 SqlStatement::Schema(SqlSchemaCommand::CreateView(q)) => SqlCommand::CreateView(q),
1498 SqlStatement::Schema(SqlSchemaCommand::DropView(q)) => SqlCommand::DropView(q),
1499 SqlStatement::Schema(SqlSchemaCommand::RefreshMaterializedView(q)) => {
1500 SqlCommand::RefreshMaterializedView(q)
1501 }
1502 SqlStatement::Schema(SqlSchemaCommand::CreatePolicy(q)) => SqlCommand::CreatePolicy(q),
1503 SqlStatement::Schema(SqlSchemaCommand::DropPolicy(q)) => SqlCommand::DropPolicy(q),
1504 SqlStatement::Schema(SqlSchemaCommand::CreateServer(q)) => SqlCommand::CreateServer(q),
1505 SqlStatement::Schema(SqlSchemaCommand::DropServer(q)) => SqlCommand::DropServer(q),
1506 SqlStatement::Schema(SqlSchemaCommand::CreateForeignTable(q)) => {
1507 SqlCommand::CreateForeignTable(q)
1508 }
1509 SqlStatement::Schema(SqlSchemaCommand::DropForeignTable(q)) => {
1510 SqlCommand::DropForeignTable(q)
1511 }
1512 SqlStatement::Admin(SqlAdminCommand::Grant(s)) => SqlCommand::Grant(s),
1513 SqlStatement::Admin(SqlAdminCommand::Revoke(s)) => SqlCommand::Revoke(s),
1514 SqlStatement::Admin(SqlAdminCommand::AlterUser(s)) => SqlCommand::AlterUser(s),
1515 SqlStatement::Admin(SqlAdminCommand::CreateUser(s)) => SqlCommand::CreateUser(s),
1516 SqlStatement::Admin(SqlAdminCommand::IamPolicy(e)) => SqlCommand::IamPolicy(e),
1517 SqlStatement::Schema(SqlSchemaCommand::CreateMigration(q)) => {
1518 SqlCommand::CreateMigration(q)
1519 }
1520 SqlStatement::Schema(SqlSchemaCommand::ApplyMigration(q)) => {
1521 SqlCommand::ApplyMigration(q)
1522 }
1523 SqlStatement::Schema(SqlSchemaCommand::RollbackMigration(q)) => {
1524 SqlCommand::RollbackMigration(q)
1525 }
1526 SqlStatement::Schema(SqlSchemaCommand::ExplainMigration(q)) => {
1527 SqlCommand::ExplainMigration(q)
1528 }
1529 }
1530 }
1531
1532 pub fn into_query_expr(self) -> QueryExpr {
1533 self.into_command().into_query_expr()
1534 }
1535}
1536
1537impl FrontendStatement {
1538 pub fn into_query_expr(self) -> QueryExpr {
1539 match self {
1540 FrontendStatement::Sql(statement) => statement.into_query_expr(),
1541 FrontendStatement::Graph(query) => QueryExpr::Graph(query),
1542 FrontendStatement::GraphCommand(command) => QueryExpr::GraphCommand(command),
1543 FrontendStatement::Path(query) => QueryExpr::Path(query),
1544 FrontendStatement::Vector(query) => QueryExpr::Vector(query),
1545 FrontendStatement::Hybrid(query) => QueryExpr::Hybrid(query),
1546 FrontendStatement::Search(command) => QueryExpr::SearchCommand(command),
1547 FrontendStatement::Ask(query) => QueryExpr::Ask(query),
1548 FrontendStatement::QueueSelect(query) => QueryExpr::QueueSelect(query),
1549 FrontendStatement::QueueCommand(command) => QueryExpr::QueueCommand(command),
1550 FrontendStatement::EventsBackfill(query) => QueryExpr::EventsBackfill(query),
1551 FrontendStatement::EventsBackfillStatus { collection } => {
1552 QueryExpr::EventsBackfillStatus { collection }
1553 }
1554 FrontendStatement::TreeCommand(command) => QueryExpr::TreeCommand(command),
1555 FrontendStatement::ProbabilisticCommand(command) => {
1556 QueryExpr::ProbabilisticCommand(command)
1557 }
1558 FrontendStatement::KvCommand(command) => QueryExpr::KvCommand(command),
1559 FrontendStatement::ConfigCommand(command) => QueryExpr::ConfigCommand(command),
1560 FrontendStatement::Ranking(expr) => expr,
1561 }
1562 }
1563}
1564
1565pub fn parse_frontend(input: &str) -> Result<FrontendStatement, ParseError> {
1566 let mut parser = Parser::new(input)?;
1567 let statement = parser.parse_frontend_statement()?;
1568 if !parser.check(&Token::Eof) {
1569 return Err(ParseError::new(
1570 format!("Unexpected token after query: {:?}", parser.current.token),
1575 parser.position(),
1576 ));
1577 }
1578 Ok(statement)
1579}
1580
1581impl SqlCommand {
1582 pub fn into_query_expr(self) -> QueryExpr {
1583 match self {
1584 SqlCommand::Select(query) => QueryExpr::Table(query),
1585 SqlCommand::Join(query) => QueryExpr::Join(query),
1586 SqlCommand::Insert(query) => QueryExpr::Insert(query),
1587 SqlCommand::Update(query) => QueryExpr::Update(query),
1588 SqlCommand::Delete(query) => QueryExpr::Delete(query),
1589 SqlCommand::ExplainAlter(query) => QueryExpr::ExplainAlter(query),
1590 SqlCommand::CreateTable(query) => QueryExpr::CreateTable(query),
1591 SqlCommand::CreateCollection(query) => QueryExpr::CreateCollection(query),
1592 SqlCommand::CreateVector(query) => QueryExpr::CreateVector(query),
1593 SqlCommand::DropTable(query) => QueryExpr::DropTable(query),
1594 SqlCommand::DropGraph(query) => QueryExpr::DropGraph(query),
1595 SqlCommand::DropVector(query) => QueryExpr::DropVector(query),
1596 SqlCommand::DropDocument(query) => QueryExpr::DropDocument(query),
1597 SqlCommand::DropKv(query) => QueryExpr::DropKv(query),
1598 SqlCommand::DropCollection(query) => QueryExpr::DropCollection(query),
1599 SqlCommand::Truncate(query) => QueryExpr::Truncate(query),
1600 SqlCommand::AlterTable(query) => QueryExpr::AlterTable(query),
1601 SqlCommand::CreateIndex(query) => QueryExpr::CreateIndex(query),
1602 SqlCommand::DropIndex(query) => QueryExpr::DropIndex(query),
1603 SqlCommand::CreateTimeSeries(query) => QueryExpr::CreateTimeSeries(query),
1604 SqlCommand::CreateMetric(query) => QueryExpr::CreateMetric(query),
1605 SqlCommand::AlterMetric(query) => QueryExpr::AlterMetric(query),
1606 SqlCommand::CreateSlo(query) => QueryExpr::CreateSlo(query),
1607 SqlCommand::DropTimeSeries(query) => QueryExpr::DropTimeSeries(query),
1608 SqlCommand::CreateQueue(query) => QueryExpr::CreateQueue(query),
1609 SqlCommand::AlterQueue(query) => QueryExpr::AlterQueue(query),
1610 SqlCommand::DropQueue(query) => QueryExpr::DropQueue(query),
1611 SqlCommand::CreateTree(query) => QueryExpr::CreateTree(query),
1612 SqlCommand::DropTree(query) => QueryExpr::DropTree(query),
1613 SqlCommand::Probabilistic(command) => QueryExpr::ProbabilisticCommand(command),
1614 SqlCommand::SetConfig { key, value } => QueryExpr::SetConfig { key, value },
1615 SqlCommand::ShowConfig { prefix, as_json } => QueryExpr::ShowConfig { prefix, as_json },
1616 SqlCommand::SetSecret { key, value } => QueryExpr::SetSecret { key, value },
1617 SqlCommand::DeleteSecret { key } => QueryExpr::DeleteSecret { key },
1618 SqlCommand::ShowSecrets { prefix } => QueryExpr::ShowSecrets { prefix },
1619 SqlCommand::SetTenant(value) => QueryExpr::SetTenant(value),
1620 SqlCommand::ShowTenant => QueryExpr::ShowTenant,
1621 SqlCommand::TransactionControl(ctl) => QueryExpr::TransactionControl(ctl),
1622 SqlCommand::Maintenance(cmd) => QueryExpr::MaintenanceCommand(cmd),
1623 SqlCommand::CreateSchema(q) => QueryExpr::CreateSchema(q),
1624 SqlCommand::DropSchema(q) => QueryExpr::DropSchema(q),
1625 SqlCommand::CreateSequence(q) => QueryExpr::CreateSequence(q),
1626 SqlCommand::DropSequence(q) => QueryExpr::DropSequence(q),
1627 SqlCommand::CopyFrom(q) => QueryExpr::CopyFrom(q),
1628 SqlCommand::CreateView(q) => QueryExpr::CreateView(q),
1629 SqlCommand::DropView(q) => QueryExpr::DropView(q),
1630 SqlCommand::RefreshMaterializedView(q) => QueryExpr::RefreshMaterializedView(q),
1631 SqlCommand::CreatePolicy(q) => QueryExpr::CreatePolicy(q),
1632 SqlCommand::DropPolicy(q) => QueryExpr::DropPolicy(q),
1633 SqlCommand::CreateServer(q) => QueryExpr::CreateServer(q),
1634 SqlCommand::DropServer(q) => QueryExpr::DropServer(q),
1635 SqlCommand::CreateForeignTable(q) => QueryExpr::CreateForeignTable(q),
1636 SqlCommand::DropForeignTable(q) => QueryExpr::DropForeignTable(q),
1637 SqlCommand::Grant(s) => QueryExpr::Grant(s),
1638 SqlCommand::Revoke(s) => QueryExpr::Revoke(s),
1639 SqlCommand::AlterUser(s) => QueryExpr::AlterUser(s),
1640 SqlCommand::CreateUser(s) => QueryExpr::CreateUser(s),
1641 SqlCommand::IamPolicy(e) => e,
1642 SqlCommand::CreateMigration(q) => QueryExpr::CreateMigration(q),
1643 SqlCommand::ApplyMigration(q) => QueryExpr::ApplyMigration(q),
1644 SqlCommand::RollbackMigration(q) => QueryExpr::RollbackMigration(q),
1645 SqlCommand::ExplainMigration(q) => QueryExpr::ExplainMigration(q),
1646 }
1647 }
1648
1649 pub fn into_statement(self) -> SqlStatement {
1650 match self {
1651 SqlCommand::Select(query) => SqlStatement::Query(SqlQuery::Select(query)),
1652 SqlCommand::Join(query) => SqlStatement::Query(SqlQuery::Join(query)),
1653 SqlCommand::Insert(query) => SqlStatement::Mutation(SqlMutation::Insert(query)),
1654 SqlCommand::Update(query) => SqlStatement::Mutation(SqlMutation::Update(query)),
1655 SqlCommand::Delete(query) => SqlStatement::Mutation(SqlMutation::Delete(query)),
1656 SqlCommand::ExplainAlter(query) => {
1657 SqlStatement::Schema(SqlSchemaCommand::ExplainAlter(query))
1658 }
1659 SqlCommand::CreateTable(query) => {
1660 SqlStatement::Schema(SqlSchemaCommand::CreateTable(query))
1661 }
1662 SqlCommand::CreateCollection(query) => {
1663 SqlStatement::Schema(SqlSchemaCommand::CreateCollection(query))
1664 }
1665 SqlCommand::CreateVector(query) => {
1666 SqlStatement::Schema(SqlSchemaCommand::CreateVector(query))
1667 }
1668 SqlCommand::DropTable(query) => {
1669 SqlStatement::Schema(SqlSchemaCommand::DropTable(query))
1670 }
1671 SqlCommand::DropGraph(query) => {
1672 SqlStatement::Schema(SqlSchemaCommand::DropGraph(query))
1673 }
1674 SqlCommand::DropVector(query) => {
1675 SqlStatement::Schema(SqlSchemaCommand::DropVector(query))
1676 }
1677 SqlCommand::DropDocument(query) => {
1678 SqlStatement::Schema(SqlSchemaCommand::DropDocument(query))
1679 }
1680 SqlCommand::DropKv(query) => SqlStatement::Schema(SqlSchemaCommand::DropKv(query)),
1681 SqlCommand::DropCollection(query) => {
1682 SqlStatement::Schema(SqlSchemaCommand::DropCollection(query))
1683 }
1684 SqlCommand::Truncate(query) => SqlStatement::Schema(SqlSchemaCommand::Truncate(query)),
1685 SqlCommand::AlterTable(query) => {
1686 SqlStatement::Schema(SqlSchemaCommand::AlterTable(query))
1687 }
1688 SqlCommand::CreateIndex(query) => {
1689 SqlStatement::Schema(SqlSchemaCommand::CreateIndex(query))
1690 }
1691 SqlCommand::DropIndex(query) => {
1692 SqlStatement::Schema(SqlSchemaCommand::DropIndex(query))
1693 }
1694 SqlCommand::CreateTimeSeries(query) => {
1695 SqlStatement::Schema(SqlSchemaCommand::CreateTimeSeries(query))
1696 }
1697 SqlCommand::CreateMetric(query) => {
1698 SqlStatement::Schema(SqlSchemaCommand::CreateMetric(query))
1699 }
1700 SqlCommand::AlterMetric(query) => {
1701 SqlStatement::Schema(SqlSchemaCommand::AlterMetric(query))
1702 }
1703 SqlCommand::CreateSlo(query) => {
1704 SqlStatement::Schema(SqlSchemaCommand::CreateSlo(query))
1705 }
1706 SqlCommand::DropTimeSeries(query) => {
1707 SqlStatement::Schema(SqlSchemaCommand::DropTimeSeries(query))
1708 }
1709 SqlCommand::CreateQueue(query) => {
1710 SqlStatement::Schema(SqlSchemaCommand::CreateQueue(query))
1711 }
1712 SqlCommand::AlterQueue(query) => {
1713 SqlStatement::Schema(SqlSchemaCommand::AlterQueue(query))
1714 }
1715 SqlCommand::DropQueue(query) => {
1716 SqlStatement::Schema(SqlSchemaCommand::DropQueue(query))
1717 }
1718 SqlCommand::CreateTree(query) => {
1719 SqlStatement::Schema(SqlSchemaCommand::CreateTree(query))
1720 }
1721 SqlCommand::DropTree(query) => SqlStatement::Schema(SqlSchemaCommand::DropTree(query)),
1722 SqlCommand::Probabilistic(command) => {
1723 SqlStatement::Schema(SqlSchemaCommand::Probabilistic(command))
1724 }
1725 SqlCommand::SetConfig { key, value } => {
1726 SqlStatement::Admin(SqlAdminCommand::SetConfig { key, value })
1727 }
1728 SqlCommand::ShowConfig { prefix, as_json } => {
1729 SqlStatement::Admin(SqlAdminCommand::ShowConfig { prefix, as_json })
1730 }
1731 SqlCommand::SetSecret { key, value } => {
1732 SqlStatement::Admin(SqlAdminCommand::SetSecret { key, value })
1733 }
1734 SqlCommand::DeleteSecret { key } => {
1735 SqlStatement::Admin(SqlAdminCommand::DeleteSecret { key })
1736 }
1737 SqlCommand::ShowSecrets { prefix } => {
1738 SqlStatement::Admin(SqlAdminCommand::ShowSecrets { prefix })
1739 }
1740 SqlCommand::SetTenant(value) => SqlStatement::Admin(SqlAdminCommand::SetTenant(value)),
1741 SqlCommand::ShowTenant => SqlStatement::Admin(SqlAdminCommand::ShowTenant),
1742 SqlCommand::TransactionControl(ctl) => {
1743 SqlStatement::Admin(SqlAdminCommand::TransactionControl(ctl))
1744 }
1745 SqlCommand::Maintenance(cmd) => SqlStatement::Admin(SqlAdminCommand::Maintenance(cmd)),
1746 SqlCommand::CreateSchema(q) => SqlStatement::Schema(SqlSchemaCommand::CreateSchema(q)),
1747 SqlCommand::DropSchema(q) => SqlStatement::Schema(SqlSchemaCommand::DropSchema(q)),
1748 SqlCommand::CreateSequence(q) => {
1749 SqlStatement::Schema(SqlSchemaCommand::CreateSequence(q))
1750 }
1751 SqlCommand::DropSequence(q) => SqlStatement::Schema(SqlSchemaCommand::DropSequence(q)),
1752 SqlCommand::CopyFrom(q) => SqlStatement::Schema(SqlSchemaCommand::CopyFrom(q)),
1753 SqlCommand::CreateView(q) => SqlStatement::Schema(SqlSchemaCommand::CreateView(q)),
1754 SqlCommand::DropView(q) => SqlStatement::Schema(SqlSchemaCommand::DropView(q)),
1755 SqlCommand::RefreshMaterializedView(q) => {
1756 SqlStatement::Schema(SqlSchemaCommand::RefreshMaterializedView(q))
1757 }
1758 SqlCommand::CreatePolicy(q) => SqlStatement::Schema(SqlSchemaCommand::CreatePolicy(q)),
1759 SqlCommand::DropPolicy(q) => SqlStatement::Schema(SqlSchemaCommand::DropPolicy(q)),
1760 SqlCommand::CreateServer(q) => SqlStatement::Schema(SqlSchemaCommand::CreateServer(q)),
1761 SqlCommand::DropServer(q) => SqlStatement::Schema(SqlSchemaCommand::DropServer(q)),
1762 SqlCommand::CreateForeignTable(q) => {
1763 SqlStatement::Schema(SqlSchemaCommand::CreateForeignTable(q))
1764 }
1765 SqlCommand::DropForeignTable(q) => {
1766 SqlStatement::Schema(SqlSchemaCommand::DropForeignTable(q))
1767 }
1768 SqlCommand::Grant(s) => SqlStatement::Admin(SqlAdminCommand::Grant(s)),
1769 SqlCommand::Revoke(s) => SqlStatement::Admin(SqlAdminCommand::Revoke(s)),
1770 SqlCommand::AlterUser(s) => SqlStatement::Admin(SqlAdminCommand::AlterUser(s)),
1771 SqlCommand::CreateUser(s) => SqlStatement::Admin(SqlAdminCommand::CreateUser(s)),
1772 SqlCommand::IamPolicy(e) => SqlStatement::Admin(SqlAdminCommand::IamPolicy(e)),
1773 SqlCommand::CreateMigration(q) => {
1774 SqlStatement::Schema(SqlSchemaCommand::CreateMigration(q))
1775 }
1776 SqlCommand::ApplyMigration(q) => {
1777 SqlStatement::Schema(SqlSchemaCommand::ApplyMigration(q))
1778 }
1779 SqlCommand::RollbackMigration(q) => {
1780 SqlStatement::Schema(SqlSchemaCommand::RollbackMigration(q))
1781 }
1782 SqlCommand::ExplainMigration(q) => {
1783 SqlStatement::Schema(SqlSchemaCommand::ExplainMigration(q))
1784 }
1785 }
1786 }
1787}
1788
1789impl<'a> Parser<'a> {
1790 fn parse_events_command(&mut self) -> Result<QueryExpr, ParseError> {
1791 self.expect_ident()?; if self.consume_ident_ci("STATUS")? {
1793 let mut query = TableQuery::new("red.subscriptions");
1794 let collection = match self.peek().clone() {
1795 Token::Ident(name) => {
1796 self.advance()?;
1797 Some(name)
1798 }
1799 Token::String(name) => {
1800 self.advance()?;
1801 Some(name)
1802 }
1803 _ => None,
1804 };
1805 self.parse_table_clauses(&mut query)?;
1806 if let Some(collection) = collection {
1807 let filter = Filter::compare(
1808 FieldRef::column("red.subscriptions", "collection"),
1809 CompareOp::Eq,
1810 Value::text(collection),
1811 );
1812 let expr = filter_to_expr(&filter);
1813 query.where_expr = Some(match query.where_expr.take() {
1814 Some(existing) => Expr::binop(BinOp::And, existing, expr),
1815 None => expr,
1816 });
1817 query.filter = Some(match query.filter.take() {
1818 Some(existing) => existing.and(filter),
1819 None => filter,
1820 });
1821 }
1822 return Ok(QueryExpr::Table(query));
1823 }
1824
1825 if !self.consume_ident_ci("BACKFILL")? {
1826 return Err(ParseError::expected(
1827 vec!["BACKFILL", "STATUS"],
1828 self.peek(),
1829 self.position(),
1830 ));
1831 }
1832
1833 if self.consume_ident_ci("STATUS")? {
1834 let collection = self.expect_ident()?;
1835 return Ok(QueryExpr::EventsBackfillStatus { collection });
1836 }
1837
1838 let collection = self.expect_ident()?;
1839 let where_filter = if self.consume(&Token::Where)? {
1840 let mut parts = Vec::new();
1841 while !self.check(&Token::Eof) && !self.check(&Token::To) {
1842 parts.push(self.peek().to_string());
1843 self.advance()?;
1844 }
1845 if parts.is_empty() {
1846 return Err(ParseError::expected(
1847 vec!["predicate"],
1848 self.peek(),
1849 self.position(),
1850 ));
1851 }
1852 Some(parts.join(" "))
1853 } else {
1854 None
1855 };
1856
1857 self.expect(Token::To)?;
1858 let target_queue = self.expect_ident()?;
1859 let limit = if self.consume(&Token::Limit)? {
1860 Some(self.parse_positive_integer("LIMIT")? as u64)
1861 } else {
1862 None
1863 };
1864
1865 Ok(QueryExpr::EventsBackfill(EventsBackfillQuery {
1866 collection,
1867 where_filter,
1868 target_queue,
1869 limit,
1870 }))
1871 }
1872
1873 pub(crate) fn parse_fdw_options_clause(&mut self) -> Result<Vec<(String, String)>, ParseError> {
1878 if !self.consume(&Token::Options)? {
1879 return Ok(Vec::new());
1880 }
1881 self.expect(Token::LParen)?;
1882 let mut out: Vec<(String, String)> = Vec::new();
1883 loop {
1884 let was_ident = matches!(self.peek(), Token::Ident(_));
1889 let raw = self.expect_ident_or_keyword()?;
1890 let key = if was_ident {
1891 raw
1892 } else {
1893 raw.to_ascii_lowercase()
1894 };
1895 let value = self.parse_string()?;
1897 out.push((key, value));
1898 if !self.consume(&Token::Comma)? {
1899 break;
1900 }
1901 }
1902 self.expect(Token::RParen)?;
1903 Ok(out)
1904 }
1905
1906 pub fn parse_frontend_statement(&mut self) -> Result<FrontendStatement, ParseError> {
1908 self.enter_depth()?;
1909 let result = self.parse_frontend_statement_inner();
1910 self.exit_depth();
1911 result
1912 }
1913
1914 fn parse_frontend_statement_inner(&mut self) -> Result<FrontendStatement, ParseError> {
1915 match self.peek() {
1916 Token::Select => match self.parse_select_query()? {
1917 QueryExpr::Table(query) => Ok(FrontendStatement::Sql(SqlStatement::Query(
1918 SqlQuery::Select(query),
1919 ))),
1920 QueryExpr::Join(query) => Ok(FrontendStatement::Sql(SqlStatement::Query(
1921 SqlQuery::Join(query),
1922 ))),
1923 QueryExpr::QueueSelect(query) => Ok(FrontendStatement::QueueSelect(query)),
1924 other => Err(ParseError::new(
1925 format!("internal: SELECT produced unexpected query kind {other:?}"),
1926 self.position(),
1927 )),
1928 },
1929 Token::From
1930 | Token::Insert
1931 | Token::Update
1932 | Token::Truncate
1933 | Token::Create
1934 | Token::Drop
1935 | Token::Alter
1936 | Token::Set
1937 | Token::Begin
1938 | Token::Commit
1939 | Token::Rollback
1940 | Token::Savepoint
1941 | Token::Release
1942 | Token::Start
1943 | Token::Vacuum
1944 | Token::Analyze
1945 | Token::Copy
1946 | Token::Refresh => self.parse_sql_statement().map(FrontendStatement::Sql),
1947 Token::Explain => {
1948 if matches!(
1949 self.peek_next()?,
1950 Token::Ident(name) if name.eq_ignore_ascii_case("ASK")
1951 ) {
1952 match self.parse_explain_ask_query()? {
1953 QueryExpr::Ask(query) => Ok(FrontendStatement::Ask(query)),
1954 other => Err(ParseError::new(
1955 format!(
1956 "internal: EXPLAIN ASK produced unexpected query kind {other:?}"
1957 ),
1958 self.position(),
1959 )),
1960 }
1961 } else {
1962 self.parse_sql_statement().map(FrontendStatement::Sql)
1963 }
1964 }
1965 Token::Ident(name) if name.eq_ignore_ascii_case("SHOW") => {
1966 self.parse_sql_statement().map(FrontendStatement::Sql)
1967 }
1968 Token::Ident(name) if name.eq_ignore_ascii_case("RESET") => {
1969 self.parse_sql_statement().map(FrontendStatement::Sql)
1970 }
1971 Token::Ident(name)
1972 if name.eq_ignore_ascii_case("RANK")
1973 || name.eq_ignore_ascii_case("APPROX")
1974 || name.eq_ignore_ascii_case("APPROXIMATE")
1975 || name.eq_ignore_ascii_case("ZRANK")
1976 || name.eq_ignore_ascii_case("ZRANGE") =>
1977 {
1978 self.parse_ranking_read().map(FrontendStatement::Ranking)
1979 }
1980 Token::Desc => self.parse_sql_statement().map(FrontendStatement::Sql),
1981 Token::Ident(name)
1982 if name.eq_ignore_ascii_case("DESCRIBE") || name.eq_ignore_ascii_case("DESC") =>
1983 {
1984 self.parse_sql_statement().map(FrontendStatement::Sql)
1985 }
1986 Token::Ident(name)
1987 if name.eq_ignore_ascii_case("GRANT")
1988 || name.eq_ignore_ascii_case("REVOKE")
1989 || name.eq_ignore_ascii_case("SIMULATE")
1990 || name.eq_ignore_ascii_case("LINT")
1991 || name.eq_ignore_ascii_case("MIGRATE")
1992 || name.eq_ignore_ascii_case("APPLY") =>
1993 {
1994 self.parse_sql_statement().map(FrontendStatement::Sql)
1995 }
1996 Token::Ident(name) if name.eq_ignore_ascii_case("WATCH") => {
1997 self.advance()?;
1998 if matches!(
1999 self.peek(),
2000 Token::Ident(name) if name.eq_ignore_ascii_case("CONFIG")
2001 ) {
2002 match self.parse_config_watch_after_watch()? {
2003 QueryExpr::ConfigCommand(command) => {
2004 Ok(FrontendStatement::ConfigCommand(command))
2005 }
2006 other => Err(ParseError::new(
2007 format!(
2008 "internal: WATCH CONFIG produced unexpected query kind {other:?}"
2009 ),
2010 self.position(),
2011 )),
2012 }
2013 } else if matches!(
2014 self.peek(),
2015 Token::Ident(name) if name.eq_ignore_ascii_case("VAULT")
2016 ) {
2017 match self.parse_vault_watch_after_watch()? {
2018 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2019 other => Err(ParseError::new(
2020 format!(
2021 "internal: WATCH VAULT produced unexpected query kind {other:?}"
2022 ),
2023 self.position(),
2024 )),
2025 }
2026 } else {
2027 match self.parse_kv_watch(reddb_types::catalog::CollectionModel::Kv)? {
2028 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2029 other => Err(ParseError::new(
2030 format!("internal: WATCH produced unexpected query kind {other:?}"),
2031 self.position(),
2032 )),
2033 }
2034 }
2035 }
2036 Token::List => {
2037 self.advance()?;
2038 if matches!(
2039 self.peek(),
2040 Token::Ident(name) if name.eq_ignore_ascii_case("CONFIG")
2041 ) {
2042 match self.parse_config_list_after_list()? {
2043 QueryExpr::ConfigCommand(command) => {
2044 Ok(FrontendStatement::ConfigCommand(command))
2045 }
2046 other => Err(ParseError::new(
2047 format!(
2048 "internal: LIST CONFIG produced unexpected query kind {other:?}"
2049 ),
2050 self.position(),
2051 )),
2052 }
2053 } else if matches!(self.peek(), Token::Kv) {
2054 match self.parse_kv_list_after_list()? {
2055 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2056 other => Err(ParseError::new(
2057 format!("internal: LIST KV produced unexpected query kind {other:?}"),
2058 self.position(),
2059 )),
2060 }
2061 } else if matches!(
2062 self.peek(),
2063 Token::Ident(name) if name.eq_ignore_ascii_case("VAULT")
2064 ) {
2065 match self.parse_vault_list_after_list()? {
2066 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2067 other => Err(ParseError::new(
2068 format!(
2069 "internal: LIST VAULT produced unexpected query kind {other:?}"
2070 ),
2071 self.position(),
2072 )),
2073 }
2074 } else {
2075 Err(ParseError::expected(
2076 vec!["CONFIG", "KV", "VAULT"],
2077 self.peek(),
2078 self.position(),
2079 ))
2080 }
2081 }
2082 Token::Ident(name) if name.eq_ignore_ascii_case("LIST") => {
2083 self.advance()?;
2084 if matches!(
2085 self.peek(),
2086 Token::Ident(name) if name.eq_ignore_ascii_case("CONFIG")
2087 ) {
2088 match self.parse_config_list_after_list()? {
2089 QueryExpr::ConfigCommand(command) => {
2090 Ok(FrontendStatement::ConfigCommand(command))
2091 }
2092 other => Err(ParseError::new(
2093 format!(
2094 "internal: LIST CONFIG produced unexpected query kind {other:?}"
2095 ),
2096 self.position(),
2097 )),
2098 }
2099 } else if matches!(self.peek(), Token::Kv) {
2100 match self.parse_kv_list_after_list()? {
2101 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2102 other => Err(ParseError::new(
2103 format!("internal: LIST KV produced unexpected query kind {other:?}"),
2104 self.position(),
2105 )),
2106 }
2107 } else if matches!(
2108 self.peek(),
2109 Token::Ident(name) if name.eq_ignore_ascii_case("VAULT")
2110 ) {
2111 match self.parse_vault_list_after_list()? {
2112 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2113 other => Err(ParseError::new(
2114 format!(
2115 "internal: LIST VAULT produced unexpected query kind {other:?}"
2116 ),
2117 self.position(),
2118 )),
2119 }
2120 } else {
2121 Err(ParseError::expected(
2122 vec!["CONFIG", "KV", "VAULT"],
2123 self.peek(),
2124 self.position(),
2125 ))
2126 }
2127 }
2128 Token::Ident(name) if name.eq_ignore_ascii_case("INVALIDATE") => {
2129 if matches!(
2130 self.peek_next()?,
2131 Token::Ident(next) if next.eq_ignore_ascii_case("CONFIG")
2132 ) {
2133 match self.parse_config_command()? {
2134 QueryExpr::ConfigCommand(command) => {
2135 Ok(FrontendStatement::ConfigCommand(command))
2136 }
2137 other => Err(ParseError::new(
2138 format!("internal: CONFIG produced unexpected query kind {other:?}"),
2139 self.position(),
2140 )),
2141 }
2142 } else {
2143 self.advance()?;
2144 match self.parse_kv_invalidate_tags_after_invalidate()? {
2145 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2146 other => Err(ParseError::new(
2147 format!(
2148 "internal: INVALIDATE produced unexpected query kind {other:?}"
2149 ),
2150 self.position(),
2151 )),
2152 }
2153 }
2154 }
2155 Token::Attach | Token::Detach => self.parse_sql_statement().map(FrontendStatement::Sql),
2156 Token::Match => match self.parse_match_query()? {
2157 QueryExpr::Graph(query) => Ok(FrontendStatement::Graph(query)),
2158 other => Err(ParseError::new(
2159 format!("internal: MATCH produced unexpected query kind {other:?}"),
2160 self.position(),
2161 )),
2162 },
2163 Token::Path => match self.parse_path_query()? {
2164 QueryExpr::Path(query) => Ok(FrontendStatement::Path(query)),
2165 other => Err(ParseError::new(
2166 format!("internal: PATH produced unexpected query kind {other:?}"),
2167 self.position(),
2168 )),
2169 },
2170 Token::Vector => match self.parse_vector_query()? {
2171 QueryExpr::Vector(query) => Ok(FrontendStatement::Vector(query)),
2172 other => Err(ParseError::new(
2173 format!("internal: VECTOR produced unexpected query kind {other:?}"),
2174 self.position(),
2175 )),
2176 },
2177 Token::Hybrid => match self.parse_hybrid_query()? {
2178 QueryExpr::Hybrid(query) => Ok(FrontendStatement::Hybrid(query)),
2179 other => Err(ParseError::new(
2180 format!("internal: HYBRID produced unexpected query kind {other:?}"),
2181 self.position(),
2182 )),
2183 },
2184 Token::Graph => match self.parse_graph_command()? {
2185 QueryExpr::GraphCommand(command) => Ok(FrontendStatement::GraphCommand(command)),
2186 other => Err(ParseError::new(
2187 format!("internal: GRAPH produced unexpected query kind {other:?}"),
2188 self.position(),
2189 )),
2190 },
2191 Token::Search => match self.parse_search_command()? {
2192 QueryExpr::SearchCommand(command) => Ok(FrontendStatement::Search(command)),
2193 other => Err(ParseError::new(
2194 format!("internal: SEARCH produced unexpected query kind {other:?}"),
2195 self.position(),
2196 )),
2197 },
2198 Token::Ident(name) if name.eq_ignore_ascii_case("ASK") => {
2199 match self.parse_ask_query()? {
2200 QueryExpr::Ask(query) => Ok(FrontendStatement::Ask(query)),
2201 other => Err(ParseError::new(
2202 format!("internal: ASK produced unexpected query kind {other:?}"),
2203 self.position(),
2204 )),
2205 }
2206 }
2207 Token::Ident(name) if name.eq_ignore_ascii_case("UNSEAL") => {
2208 match self.parse_unseal_vault_command()? {
2209 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2210 other => Err(ParseError::new(
2211 format!("internal: UNSEAL VAULT produced unexpected query kind {other:?}"),
2212 self.position(),
2213 )),
2214 }
2215 }
2216 Token::Queue => match self.parse_queue_command()? {
2217 QueryExpr::QueueCommand(command) => Ok(FrontendStatement::QueueCommand(command)),
2218 other => Err(ParseError::new(
2219 format!("internal: QUEUE produced unexpected query kind {other:?}"),
2220 self.position(),
2221 )),
2222 },
2223 Token::Ident(name) if name.eq_ignore_ascii_case("EVENTS") => {
2224 match self.parse_events_command()? {
2225 QueryExpr::Table(query) => Ok(FrontendStatement::Sql(SqlStatement::Query(
2226 SqlQuery::Select(query),
2227 ))),
2228 QueryExpr::EventsBackfill(query) => {
2229 Ok(FrontendStatement::EventsBackfill(query))
2230 }
2231 QueryExpr::EventsBackfillStatus { collection } => {
2232 Ok(FrontendStatement::EventsBackfillStatus { collection })
2233 }
2234 other => Err(ParseError::new(
2235 format!("internal: EVENTS produced unexpected query kind {other:?}"),
2236 self.position(),
2237 )),
2238 }
2239 }
2240 Token::Kv => match self.parse_kv_command()? {
2241 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2242 other => Err(ParseError::new(
2243 format!("internal: KV produced unexpected query kind {other:?}"),
2244 self.position(),
2245 )),
2246 },
2247 Token::Delete => {
2248 if matches!(
2249 self.peek_next()?,
2250 Token::Ident(name) if name.eq_ignore_ascii_case("CONFIG")
2251 ) {
2252 match self.parse_config_command()? {
2253 QueryExpr::ConfigCommand(command) => {
2254 Ok(FrontendStatement::ConfigCommand(command))
2255 }
2256 other => Err(ParseError::new(
2257 format!("internal: CONFIG produced unexpected query kind {other:?}"),
2258 self.position(),
2259 )),
2260 }
2261 } else if matches!(
2262 self.peek_next()?,
2263 Token::Ident(name) if name.eq_ignore_ascii_case("VAULT")
2264 ) {
2265 match self.parse_vault_lifecycle_command()? {
2266 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2267 other => Err(ParseError::new(
2268 format!("internal: VAULT produced unexpected query kind {other:?}"),
2269 self.position(),
2270 )),
2271 }
2272 } else {
2273 self.parse_sql_statement().map(FrontendStatement::Sql)
2274 }
2275 }
2276 Token::Add => match self.parse_config_command()? {
2277 QueryExpr::ConfigCommand(command) => Ok(FrontendStatement::ConfigCommand(command)),
2278 other => Err(ParseError::new(
2279 format!("internal: CONFIG produced unexpected query kind {other:?}"),
2280 self.position(),
2281 )),
2282 },
2283 Token::Purge => match self.parse_vault_lifecycle_command()? {
2284 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2285 other => Err(ParseError::new(
2286 format!("internal: VAULT produced unexpected query kind {other:?}"),
2287 self.position(),
2288 )),
2289 },
2290 Token::Ident(name)
2291 if name.eq_ignore_ascii_case("PUT")
2292 || name.eq_ignore_ascii_case("GET")
2293 || name.eq_ignore_ascii_case("RESOLVE")
2294 || name.eq_ignore_ascii_case("ROTATE")
2295 || name.eq_ignore_ascii_case("HISTORY")
2296 || name.eq_ignore_ascii_case("PURGE")
2297 || name.eq_ignore_ascii_case("INCR")
2298 || name.eq_ignore_ascii_case("DECR")
2299 || name.eq_ignore_ascii_case("INVALIDATE") =>
2300 {
2301 if matches!(
2302 self.peek_next()?,
2303 Token::Ident(next) if next.eq_ignore_ascii_case("VAULT")
2304 ) {
2305 match self.parse_vault_lifecycle_command()? {
2306 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2307 other => Err(ParseError::new(
2308 format!("internal: VAULT produced unexpected query kind {other:?}"),
2309 self.position(),
2310 )),
2311 }
2312 } else {
2313 match self.parse_config_command()? {
2314 QueryExpr::ConfigCommand(command) => {
2315 Ok(FrontendStatement::ConfigCommand(command))
2316 }
2317 other => Err(ParseError::new(
2318 format!("internal: CONFIG produced unexpected query kind {other:?}"),
2319 self.position(),
2320 )),
2321 }
2322 }
2323 }
2324 Token::Ident(name) if name.eq_ignore_ascii_case("VAULT") => {
2325 match self.parse_vault_command()? {
2326 QueryExpr::KvCommand(command) => Ok(FrontendStatement::KvCommand(command)),
2327 other => Err(ParseError::new(
2328 format!("internal: VAULT produced unexpected query kind {other:?}"),
2329 self.position(),
2330 )),
2331 }
2332 }
2333 Token::Tree => match self.parse_tree_command()? {
2334 QueryExpr::TreeCommand(command) => Ok(FrontendStatement::TreeCommand(command)),
2335 other => Err(ParseError::new(
2336 format!("internal: TREE produced unexpected query kind {other:?}"),
2337 self.position(),
2338 )),
2339 },
2340 Token::Ident(name) if name.eq_ignore_ascii_case("HLL") => {
2341 match self.parse_hll_command()? {
2342 QueryExpr::ProbabilisticCommand(command) => {
2343 Ok(FrontendStatement::ProbabilisticCommand(command))
2344 }
2345 other => Err(ParseError::new(
2346 format!("internal: HLL produced unexpected query kind {other:?}"),
2347 self.position(),
2348 )),
2349 }
2350 }
2351 Token::Ident(name) if name.eq_ignore_ascii_case("SKETCH") => {
2352 match self.parse_sketch_command()? {
2353 QueryExpr::ProbabilisticCommand(command) => {
2354 Ok(FrontendStatement::ProbabilisticCommand(command))
2355 }
2356 other => Err(ParseError::new(
2357 format!("internal: SKETCH produced unexpected query kind {other:?}"),
2358 self.position(),
2359 )),
2360 }
2361 }
2362 Token::Ident(name) if name.eq_ignore_ascii_case("FILTER") => {
2363 match self.parse_filter_command()? {
2364 QueryExpr::ProbabilisticCommand(command) => {
2365 Ok(FrontendStatement::ProbabilisticCommand(command))
2366 }
2367 other => Err(ParseError::new(
2368 format!("internal: FILTER produced unexpected query kind {other:?}"),
2369 self.position(),
2370 )),
2371 }
2372 }
2373 Token::Ident(name) if name.eq_ignore_ascii_case("EVENTS") => self
2374 .parse_sql_command()
2375 .map(SqlCommand::into_statement)
2376 .map(FrontendStatement::Sql),
2377 other => Err(ParseError::expected(
2378 vec![
2379 "SELECT", "MATCH", "PATH", "FROM", "VECTOR", "HYBRID", "INSERT", "UPDATE",
2380 "DELETE", "TRUNCATE", "CREATE", "DROP", "ALTER", "GRAPH", "SEARCH", "ASK",
2381 "QUEUE", "EVENTS", "KV", "HLL", "TREE", "SKETCH", "FILTER", "SET", "SHOW",
2382 "RESET", "DESCRIBE", "DESC", "RANK", "ZRANK", "ZRANGE",
2383 ],
2384 other,
2385 self.position(),
2386 )),
2387 }
2388 }
2389
2390 fn parse_ranking_read(&mut self) -> Result<QueryExpr, ParseError> {
2391 let head = self.expect_ident()?;
2392 if head.eq_ignore_ascii_case("RANK") {
2393 return self.parse_rank_after_rank(false);
2394 }
2395 if head.eq_ignore_ascii_case("APPROX") || head.eq_ignore_ascii_case("APPROXIMATE") {
2396 if !self.consume_ident_ci("RANK")? {
2397 return Err(ParseError::expected(
2398 vec!["RANK"],
2399 self.peek(),
2400 self.position(),
2401 ));
2402 }
2403 return self.parse_rank_after_rank(true);
2404 }
2405 if head.eq_ignore_ascii_case("ZRANK") {
2406 return self.parse_zrank();
2407 }
2408 if head.eq_ignore_ascii_case("ZRANGE") {
2409 return self.parse_zrange();
2410 }
2411 Err(ParseError::expected(
2412 vec!["RANK", "APPROX RANK", "ZRANK", "ZRANGE"],
2413 self.peek(),
2414 self.position(),
2415 ))
2416 }
2417
2418 fn parse_rank_after_rank(&mut self, approximate: bool) -> Result<QueryExpr, ParseError> {
2419 if self.consume(&Token::Of)? {
2420 let entity_id = self.parse_u64_slot("rank entity id")?;
2421 self.expect(Token::In)?;
2422 let ranking = self.expect_ident()?;
2423 let query = RankOfQuery { ranking, entity_id };
2424 return Ok(if approximate {
2425 QueryExpr::ApproxRankOf(query)
2426 } else {
2427 QueryExpr::RankOf(query)
2428 });
2429 }
2430
2431 if !approximate && self.consume(&Token::Range)? {
2432 let lo = self.parse_positive_u64_slot("rank range lower bound")?;
2433 self.expect(Token::To)?;
2434 let hi = self.parse_positive_u64_slot("rank range upper bound")?;
2435 if hi < lo {
2436 return Err(ParseError::value_out_of_range(
2437 "rank range upper bound",
2438 "must be greater than or equal to the lower bound",
2439 self.position(),
2440 ));
2441 }
2442 self.expect(Token::In)?;
2443 let ranking = self.expect_ident()?;
2444 return Ok(QueryExpr::RankRange(RankRangeQuery { ranking, lo, hi }));
2445 }
2446
2447 Err(ParseError::expected(
2448 if approximate {
2449 vec!["OF"]
2450 } else {
2451 vec!["OF", "RANGE"]
2452 },
2453 self.peek(),
2454 self.position(),
2455 ))
2456 }
2457
2458 fn parse_zrank(&mut self) -> Result<QueryExpr, ParseError> {
2459 let ranking = self.expect_ident()?;
2460 let entity_id = self.parse_u64_slot("ZRANK entity id")?;
2461 Ok(QueryExpr::RankOf(RankOfQuery { ranking, entity_id }))
2462 }
2463
2464 fn parse_zrange(&mut self) -> Result<QueryExpr, ParseError> {
2465 let ranking = self.expect_ident()?;
2466 let start = self.parse_u64_slot("ZRANGE start")?;
2467 let stop = self.parse_u64_slot("ZRANGE stop")?;
2468 if stop < start {
2469 return Err(ParseError::value_out_of_range(
2470 "ZRANGE stop",
2471 "must be greater than or equal to start",
2472 self.position(),
2473 ));
2474 }
2475 let _with_scores = self.consume_ident_ci("WITHSCORES")?;
2476 Ok(QueryExpr::RankRange(RankRangeQuery {
2477 ranking,
2478 lo: start + 1,
2479 hi: stop + 1,
2480 }))
2481 }
2482
2483 fn parse_positive_u64_slot(&mut self, field: &'static str) -> Result<u64, ParseError> {
2484 let value = self.parse_u64_slot(field)?;
2485 if value == 0 {
2486 return Err(ParseError::value_out_of_range(
2487 field,
2488 "must be a positive integer",
2489 self.position(),
2490 ));
2491 }
2492 Ok(value)
2493 }
2494
2495 fn parse_u64_slot(&mut self, field: &'static str) -> Result<u64, ParseError> {
2496 let pos = self.position();
2497 if matches!(self.peek(), Token::Minus | Token::Dash) {
2498 return Err(ParseError::value_out_of_range(
2499 field,
2500 "must be an unsigned integer",
2501 pos,
2502 ));
2503 }
2504 let raw = self.parse_integer()?;
2505 u64::try_from(raw)
2506 .map_err(|_| ParseError::value_out_of_range(field, "must be an unsigned integer", pos))
2507 }
2508
2509 pub fn parse_sql_statement(&mut self) -> Result<SqlStatement, ParseError> {
2511 self.parse_sql_command().map(SqlCommand::into_statement)
2512 }
2513
2514 fn parse_dotted_admin_path(&mut self, lowercase: bool) -> Result<String, ParseError> {
2515 let mut path = self.expect_ident()?;
2516 while self.consume(&Token::Dot)? {
2517 let next = self.expect_ident_or_keyword()?;
2518 path = format!("{path}.{next}");
2519 }
2520 Ok(if lowercase {
2521 path.to_ascii_lowercase()
2522 } else {
2523 path
2524 })
2525 }
2526
2527 fn normalize_secret_admin_path(path: String) -> String {
2528 if let Some(rest) = path.strip_prefix("red.secrets.") {
2529 format!("red.secret.{rest}")
2530 } else if path == "red.secrets" {
2531 "red.secret".to_string()
2532 } else {
2533 path
2534 }
2535 }
2536
2537 #[inline(never)]
2546 fn parse_create_command(&mut self) -> Result<SqlCommand, ParseError> {
2547 let pos = self.position();
2548 self.advance()?;
2549
2550 let mut or_replace = false;
2554 if self.consume(&Token::Or)? || self.consume_ident_ci("OR")? {
2555 let _ = self.consume_ident_ci("REPLACE")?;
2556 or_replace = true;
2557 }
2558 let materialized = self.consume(&Token::Materialized)?;
2559 if self.check(&Token::View) {
2560 self.advance()?;
2561 let if_not_exists = self.match_if_not_exists()?;
2562 let name = self.expect_ident()?;
2563 let mut retention_duration_ms: Option<u64> = None;
2571 if self.check(&Token::With) {
2572 self.advance()?;
2573 if !self.consume(&Token::Retention)? && !self.consume_ident_ci("RETENTION")? {
2574 return Err(ParseError::expected(
2575 vec!["RETENTION"],
2576 self.peek(),
2577 self.position(),
2578 ));
2579 }
2580 if !materialized {
2581 return Err(ParseError::new(
2582 "WITH RETENTION is only valid on \
2583 CREATE MATERIALIZED VIEW"
2584 .to_string(),
2585 self.position(),
2586 ));
2587 }
2588 let value = self.parse_float()?;
2589 let unit_mult = self.parse_duration_unit()?;
2590 retention_duration_ms = Some((value * unit_mult).round() as u64);
2591 }
2592 if !self.consume(&Token::As)? && !self.consume_ident_ci("AS")? {
2595 return Err(ParseError::expected(
2596 vec!["AS"],
2597 self.peek(),
2598 self.position(),
2599 ));
2600 }
2601 let body = self.parse_sql_command()?.into_query_expr();
2604 let mut refresh_every_ms: Option<u64> = None;
2609 if self.check(&Token::Refresh) {
2610 if !materialized {
2611 return Err(ParseError::new(
2612 "REFRESH EVERY is only valid on \
2613 CREATE MATERIALIZED VIEW"
2614 .to_string(),
2615 self.position(),
2616 ));
2617 }
2618 self.advance()?;
2619 if !self.consume_ident_ci("EVERY")? {
2620 return Err(ParseError::expected(
2621 vec!["EVERY"],
2622 self.peek(),
2623 self.position(),
2624 ));
2625 }
2626 let value = self.parse_float()?;
2627 let unit_mult = self.parse_duration_unit()?;
2628 refresh_every_ms = Some((value * unit_mult).round() as u64);
2629 }
2630 return Ok(SqlCommand::CreateView(CreateViewQuery {
2631 name,
2632 query: Box::new(body),
2633 materialized,
2634 if_not_exists,
2635 or_replace,
2636 refresh_every_ms,
2637 retention_duration_ms,
2638 }));
2639 }
2640 if or_replace || materialized {
2643 return Err(ParseError::expected(
2644 vec!["VIEW"],
2645 self.peek(),
2646 self.position(),
2647 ));
2648 }
2649
2650 if matches!(self.peek(), Token::Ident(name) if name.eq_ignore_ascii_case("USER")) {
2651 let stmt = self.parse_create_user_statement()?;
2652 Ok(SqlCommand::CreateUser(stmt))
2653 } else if self.check(&Token::Index) || self.check(&Token::Unique) {
2654 match self.parse_create_index_query()? {
2655 QueryExpr::CreateIndex(query) => Ok(SqlCommand::CreateIndex(query)),
2656 other => Err(ParseError::new(
2657 format!("internal: CREATE INDEX produced unexpected kind {other:?}"),
2658 self.position(),
2659 )),
2660 }
2661 } else if self.check(&Token::Table) {
2662 self.expect(Token::Table)?;
2663 match self.parse_create_table_body()? {
2664 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2665 other => Err(ParseError::new(
2666 format!("internal: CREATE TABLE produced unexpected kind {other:?}"),
2667 self.position(),
2668 )),
2669 }
2670 } else if self.check(&Token::Graph) {
2671 self.advance()?;
2672 match self.parse_create_collection_model_body(CollectionModel::Graph)? {
2673 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2674 other => Err(ParseError::new(
2675 format!("internal: CREATE GRAPH produced unexpected kind {other:?}"),
2676 self.position(),
2677 )),
2678 }
2679 } else if self.check(&Token::Document) {
2680 self.advance()?;
2681 match self.parse_create_collection_model_body(CollectionModel::Document)? {
2682 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2683 other => Err(ParseError::new(
2684 format!("internal: CREATE DOCUMENT produced unexpected kind {other:?}"),
2685 self.position(),
2686 )),
2687 }
2688 } else if self.check(&Token::Vector) {
2689 self.advance()?;
2690 match self.parse_create_vector_body()? {
2691 QueryExpr::CreateVector(query) => Ok(SqlCommand::CreateVector(query)),
2692 other => Err(ParseError::new(
2693 format!("internal: CREATE VECTOR produced unexpected kind {other:?}"),
2694 self.position(),
2695 )),
2696 }
2697 } else if self.check(&Token::Collection) {
2698 self.advance()?;
2699 match self.parse_create_collection_body()? {
2700 QueryExpr::CreateCollection(query) => Ok(SqlCommand::CreateCollection(query)),
2701 other => Err(ParseError::new(
2702 format!("internal: CREATE COLLECTION produced unexpected kind {other:?}"),
2703 self.position(),
2704 )),
2705 }
2706 } else if self.check(&Token::Kv) {
2707 self.advance()?;
2708 match self.parse_create_keyed_body(CollectionModel::Kv)? {
2709 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2710 other => Err(ParseError::new(
2711 format!("internal: CREATE KV produced unexpected kind {other:?}"),
2712 self.position(),
2713 )),
2714 }
2715 } else if self.consume_ident_ci("CONFIG")? {
2716 match self.parse_create_keyed_body(CollectionModel::Config)? {
2717 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2718 other => Err(ParseError::new(
2719 format!("internal: CREATE CONFIG produced unexpected kind {other:?}"),
2720 self.position(),
2721 )),
2722 }
2723 } else if self.consume_ident_ci("VAULT")? {
2724 match self.parse_create_keyed_body(CollectionModel::Vault)? {
2725 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2726 other => Err(ParseError::new(
2727 format!("internal: CREATE VAULT produced unexpected kind {other:?}"),
2728 self.position(),
2729 )),
2730 }
2731 } else if self.check(&Token::Timeseries) {
2732 self.advance()?;
2733 match self.parse_create_timeseries_body()? {
2734 QueryExpr::CreateTimeSeries(query) => Ok(SqlCommand::CreateTimeSeries(query)),
2735 other => Err(ParseError::new(
2736 format!("internal: CREATE TIMESERIES produced unexpected kind {other:?}"),
2737 self.position(),
2738 )),
2739 }
2740 } else if self.check(&Token::Metric) {
2741 self.advance()?;
2742 match self.parse_create_metric_body()? {
2743 QueryExpr::CreateMetric(query) => Ok(SqlCommand::CreateMetric(query)),
2744 other => Err(ParseError::new(
2745 format!("internal: CREATE METRIC produced unexpected kind {other:?}"),
2746 self.position(),
2747 )),
2748 }
2749 } else if self.consume_ident_ci("METRICS")? {
2750 match self.parse_create_metrics_body()? {
2751 QueryExpr::CreateTable(query) => Ok(SqlCommand::CreateTable(query)),
2752 other => Err(ParseError::new(
2753 format!("internal: CREATE METRICS produced unexpected kind {other:?}"),
2754 self.position(),
2755 )),
2756 }
2757 } else if self.consume_ident_ci("SLO")? {
2758 match self.parse_create_slo_body()? {
2759 QueryExpr::CreateSlo(query) => Ok(SqlCommand::CreateSlo(query)),
2760 other => Err(ParseError::new(
2761 format!("internal: CREATE SLO produced unexpected kind {other:?}"),
2762 self.position(),
2763 )),
2764 }
2765 } else if matches!(self.peek(), Token::Ident(s) if s.eq_ignore_ascii_case("HYPERTABLE")) {
2766 self.advance()?;
2767 match self.parse_create_hypertable_body()? {
2768 QueryExpr::CreateTimeSeries(query) => Ok(SqlCommand::CreateTimeSeries(query)),
2769 other => Err(ParseError::new(
2770 format!("internal: CREATE HYPERTABLE produced unexpected kind {other:?}"),
2771 self.position(),
2772 )),
2773 }
2774 } else if self.check(&Token::Queue) {
2775 self.advance()?;
2776 match self.parse_create_queue_body()? {
2777 QueryExpr::CreateQueue(query) => Ok(SqlCommand::CreateQueue(query)),
2778 other => Err(ParseError::new(
2779 format!("internal: CREATE QUEUE produced unexpected kind {other:?}"),
2780 self.position(),
2781 )),
2782 }
2783 } else if self.check(&Token::Tree) {
2784 self.advance()?;
2785 match self.parse_create_tree_body()? {
2786 QueryExpr::CreateTree(query) => Ok(SqlCommand::CreateTree(query)),
2787 other => Err(ParseError::new(
2788 format!("internal: CREATE TREE produced unexpected kind {other:?}"),
2789 self.position(),
2790 )),
2791 }
2792 } else if matches!(self.peek(), Token::Ident(n) if
2793 n.eq_ignore_ascii_case("HLL") ||
2794 n.eq_ignore_ascii_case("SKETCH") ||
2795 n.eq_ignore_ascii_case("FILTER"))
2796 {
2797 match self.parse_create_probabilistic()? {
2798 QueryExpr::ProbabilisticCommand(command) => Ok(SqlCommand::Probabilistic(command)),
2799 other => Err(ParseError::new(
2800 format!("internal: CREATE probabilistic produced unexpected kind {other:?}"),
2801 self.position(),
2802 )),
2803 }
2804 } else if self.check(&Token::Schema) {
2805 self.advance()?;
2807 let if_not_exists = self.match_if_not_exists()?;
2808 let name = self.expect_ident()?;
2809 Ok(SqlCommand::CreateSchema(CreateSchemaQuery {
2810 name,
2811 if_not_exists,
2812 }))
2813 } else if self.check(&Token::Policy) {
2814 self.advance()?;
2819 if matches!(self.peek(), Token::String(_)) {
2820 let expr = self.parse_create_iam_policy_after_keywords()?;
2822 return Ok(SqlCommand::IamPolicy(expr));
2828 }
2829 let name = self.expect_ident()?;
2830 self.expect(Token::On)?;
2831
2832 let (target_kind, table) = {
2833 use crate::ast::PolicyTargetKind;
2834 let kw = match self.peek() {
2835 Token::Ident(s) => Some(s.to_ascii_uppercase()),
2836 _ => None,
2837 };
2838 let kind = kw.as_deref().and_then(|k| match k {
2839 "NODES" => Some(PolicyTargetKind::Nodes),
2840 "EDGES" => Some(PolicyTargetKind::Edges),
2841 "VECTORS" => Some(PolicyTargetKind::Vectors),
2842 "MESSAGES" => Some(PolicyTargetKind::Messages),
2843 "POINTS" => Some(PolicyTargetKind::Points),
2844 "DOCUMENTS" => Some(PolicyTargetKind::Documents),
2845 _ => None,
2846 });
2847 if let Some(k) = kind {
2848 self.advance()?;
2849 self.expect(Token::Of)?;
2850 let coll = self.expect_ident()?;
2851 (k, coll)
2852 } else {
2853 let coll = self.expect_ident()?;
2854 (PolicyTargetKind::Table, coll)
2855 }
2856 };
2857
2858 let action = if self.consume(&Token::For)? {
2859 let a = match self.peek() {
2860 Token::Select => {
2861 self.advance()?;
2862 Some(PolicyAction::Select)
2863 }
2864 Token::Insert => {
2865 self.advance()?;
2866 Some(PolicyAction::Insert)
2867 }
2868 Token::Update => {
2869 self.advance()?;
2870 Some(PolicyAction::Update)
2871 }
2872 Token::Delete => {
2873 self.advance()?;
2874 Some(PolicyAction::Delete)
2875 }
2876 Token::All => {
2877 self.advance()?;
2878 None
2879 }
2880 _ => None,
2881 };
2882 a
2883 } else {
2884 None
2885 };
2886
2887 let role = if self.consume(&Token::To)? {
2888 Some(self.expect_ident()?)
2889 } else {
2890 None
2891 };
2892
2893 self.expect(Token::Using)?;
2894 self.expect(Token::LParen)?;
2895 let filter = self.parse_filter()?;
2896 self.expect(Token::RParen)?;
2897
2898 Ok(SqlCommand::CreatePolicy(CreatePolicyQuery {
2899 name,
2900 table,
2901 action,
2902 role,
2903 using: Box::new(filter),
2904 target_kind,
2905 }))
2906 } else if self.check(&Token::Server) {
2907 self.advance()?;
2911 let if_not_exists = self.match_if_not_exists()?;
2912 let name = self.expect_ident()?;
2913 self.expect(Token::Foreign)?;
2914 self.expect(Token::Data)?;
2915 self.expect(Token::Wrapper)?;
2916 let wrapper = self.expect_ident()?;
2917 let options = self.parse_fdw_options_clause()?;
2918 Ok(SqlCommand::CreateServer(CreateServerQuery {
2919 name,
2920 wrapper,
2921 options,
2922 if_not_exists,
2923 }))
2924 } else if self.check(&Token::Foreign) {
2925 self.advance()?;
2929 self.expect(Token::Table)?;
2930 let if_not_exists = self.match_if_not_exists()?;
2931 let name = self.expect_ident()?;
2932 self.expect(Token::LParen)?;
2933 let mut columns = Vec::new();
2934 loop {
2935 let col_name = self.expect_ident()?;
2936 let data_type = self.expect_ident_or_keyword()?;
2937 let mut not_null = false;
2940 if matches!(self.peek(), Token::Ident(n) if n.eq_ignore_ascii_case("NOT")) {
2941 self.advance()?;
2942 if matches!(self.peek(), Token::Ident(n) if n.eq_ignore_ascii_case("NULL")) {
2943 self.advance()?;
2944 not_null = true;
2945 }
2946 }
2947 columns.push(ForeignColumnDef {
2948 name: col_name,
2949 data_type,
2950 not_null,
2951 });
2952 if !self.consume(&Token::Comma)? {
2953 break;
2954 }
2955 }
2956 self.expect(Token::RParen)?;
2957 self.expect(Token::Server)?;
2958 let server = self.expect_ident()?;
2959 let options = self.parse_fdw_options_clause()?;
2960 Ok(SqlCommand::CreateForeignTable(CreateForeignTableQuery {
2961 name,
2962 server,
2963 columns,
2964 options,
2965 if_not_exists,
2966 }))
2967 } else if self.check(&Token::Sequence) {
2968 self.advance()?;
2971 let if_not_exists = self.match_if_not_exists()?;
2972 let name = self.expect_ident()?;
2973 let mut start: i64 = 1;
2974 let mut increment: i64 = 1;
2975 loop {
2977 if self.consume(&Token::Start)? {
2978 let _ = self.consume(&Token::With)? || self.consume_ident_ci("WITH")?;
2980 start = self.parse_integer()?;
2981 } else if self.consume(&Token::Increment)? {
2982 let _ = self.consume(&Token::By)? || self.consume_ident_ci("BY")?;
2984 increment = self.parse_integer()?;
2985 } else {
2986 break;
2987 }
2988 }
2989 Ok(SqlCommand::CreateSequence(CreateSequenceQuery {
2990 name,
2991 if_not_exists,
2992 start,
2993 increment,
2994 }))
2995 } else if matches!(self.peek(), Token::Ident(n) if n.eq_ignore_ascii_case("MIGRATION")) {
2996 self.advance()?; match self.parse_create_migration_body()? {
2998 QueryExpr::CreateMigration(q) => Ok(SqlCommand::CreateMigration(q)),
2999 other => Err(ParseError::new(
3000 format!("internal: CREATE MIGRATION produced unexpected kind {other:?}"),
3001 self.position(),
3002 )),
3003 }
3004 } else if let Some(reason) = analytics_v0_non_goal_create(self.peek()) {
3005 Err(ParseError::new(reason, self.position()))
3013 } else if let Some(err) =
3014 ParseError::unsupported_recognized_token(self.peek(), self.position())
3015 {
3016 Err(err)
3017 } else {
3018 Err(ParseError::expected(
3019 vec![
3020 "TABLE",
3021 "GRAPH",
3022 "VECTOR",
3023 "DOCUMENT",
3024 "KV",
3025 "COLLECTION",
3026 "INDEX",
3027 "UNIQUE",
3028 "METRIC",
3029 "TIMESERIES",
3030 "QUEUE",
3031 "TREE",
3032 "HLL",
3033 "SKETCH",
3034 "FILTER",
3035 "SCHEMA",
3036 "SEQUENCE",
3037 "USER",
3038 "MIGRATION",
3039 ],
3040 self.peek(),
3041 pos,
3042 ))
3043 }
3044 }
3045
3046 pub fn parse_sql_command(&mut self) -> Result<SqlCommand, ParseError> {
3047 self.enter_depth()?;
3048 let result = self.parse_sql_command_inner();
3049 self.exit_depth();
3050 result
3051 }
3052
3053 fn parse_sql_command_inner(&mut self) -> Result<SqlCommand, ParseError> {
3054 match self.peek() {
3055 Token::Select => match self.parse_select_query()? {
3056 QueryExpr::Table(query) => Ok(SqlCommand::Select(query)),
3057 QueryExpr::Join(query) => Ok(SqlCommand::Join(query)),
3058 other => Err(ParseError::new(
3059 format!("internal: SELECT produced unexpected query kind {other:?}"),
3060 self.position(),
3061 )),
3062 },
3063 Token::From => match self.parse_from_query()? {
3064 QueryExpr::Table(query) => Ok(SqlCommand::Select(query)),
3065 QueryExpr::Join(query) => Ok(SqlCommand::Join(query)),
3066 other => Err(ParseError::new(
3067 format!("internal: FROM produced unexpected query kind {other:?}"),
3068 self.position(),
3069 )),
3070 },
3071 Token::Insert => match self.parse_insert_query()? {
3072 QueryExpr::Insert(query) => Ok(SqlCommand::Insert(query)),
3073 other => Err(ParseError::new(
3074 format!("internal: INSERT produced unexpected query kind {other:?}"),
3075 self.position(),
3076 )),
3077 },
3078 Token::Update => match self.parse_update_query()? {
3079 QueryExpr::Update(query) => Ok(SqlCommand::Update(query)),
3080 other => Err(ParseError::new(
3081 format!("internal: UPDATE produced unexpected query kind {other:?}"),
3082 self.position(),
3083 )),
3084 },
3085 Token::Delete => {
3086 if matches!(self.peek_next()?, Token::Ident(n) if n.eq_ignore_ascii_case("SECRET"))
3087 {
3088 self.advance()?; self.advance()?; let key =
3091 Self::normalize_secret_admin_path(self.parse_dotted_admin_path(true)?);
3092 Ok(SqlCommand::DeleteSecret { key })
3093 } else {
3094 match self.parse_delete_query()? {
3095 QueryExpr::Delete(query) => Ok(SqlCommand::Delete(query)),
3096 other => Err(ParseError::new(
3097 format!("internal: DELETE produced unexpected query kind {other:?}"),
3098 self.position(),
3099 )),
3100 }
3101 }
3102 }
3103 Token::Truncate => {
3104 self.advance()?;
3105 let model = if self.consume(&Token::Table)? {
3106 Some(CollectionModel::Table)
3107 } else if self.consume(&Token::Graph)? {
3108 Some(CollectionModel::Graph)
3109 } else if self.consume(&Token::Vector)? {
3110 Some(CollectionModel::Vector)
3111 } else if self.consume(&Token::Document)? {
3112 Some(CollectionModel::Document)
3113 } else if self.consume(&Token::Timeseries)? {
3114 Some(CollectionModel::TimeSeries)
3115 } else if self.consume_ident_ci("METRICS")? {
3116 Some(CollectionModel::Metrics)
3117 } else if self.consume(&Token::Kv)? {
3118 Some(CollectionModel::Kv)
3119 } else if self.consume(&Token::Queue)? {
3120 Some(CollectionModel::Queue)
3121 } else if self.consume(&Token::Collection)? {
3122 None
3123 } else {
3124 return Err(ParseError::expected(
3125 vec![
3126 "TABLE",
3127 "GRAPH",
3128 "VECTOR",
3129 "DOCUMENT",
3130 "TIMESERIES",
3131 "METRICS",
3132 "KV",
3133 "QUEUE",
3134 "COLLECTION",
3135 ],
3136 self.peek(),
3137 self.position(),
3138 ));
3139 };
3140 match self.parse_truncate_body(model)? {
3141 QueryExpr::Truncate(query) => Ok(SqlCommand::Truncate(query)),
3142 other => Err(ParseError::new(
3143 format!("internal: TRUNCATE produced unexpected kind {other:?}"),
3144 self.position(),
3145 )),
3146 }
3147 }
3148 Token::Explain => {
3149 if matches!(self.peek_next()?, Token::Ident(n) if n.eq_ignore_ascii_case("MIGRATION"))
3152 {
3153 self.advance()?; match self.parse_explain_migration_after_keyword()? {
3155 QueryExpr::ExplainMigration(q) => Ok(SqlCommand::ExplainMigration(q)),
3156 other => Err(ParseError::new(
3157 format!(
3158 "internal: EXPLAIN MIGRATION produced unexpected kind {other:?}"
3159 ),
3160 self.position(),
3161 )),
3162 }
3163 } else {
3164 match self.parse_explain_alter_query()? {
3165 QueryExpr::ExplainAlter(query) => Ok(SqlCommand::ExplainAlter(query)),
3166 other => Err(ParseError::new(
3167 format!("internal: EXPLAIN produced unexpected query kind {other:?}"),
3168 self.position(),
3169 )),
3170 }
3171 }
3172 }
3173 Token::Create => self.parse_create_command(),
3174 Token::Drop => {
3175 let pos = self.position();
3176 self.advance()?;
3177
3178 let materialized = self.consume(&Token::Materialized)?;
3180 if self.check(&Token::View) {
3181 self.advance()?;
3182 let if_exists = self.match_if_exists()?;
3183 let name = self.expect_ident()?;
3184 return Ok(SqlCommand::DropView(DropViewQuery {
3185 name,
3186 materialized,
3187 if_exists,
3188 }));
3189 }
3190 if materialized {
3191 return Err(ParseError::expected(
3192 vec!["VIEW"],
3193 self.peek(),
3194 self.position(),
3195 ));
3196 }
3197
3198 if self.check(&Token::Index) {
3199 match self.parse_drop_index_query()? {
3200 QueryExpr::DropIndex(query) => Ok(SqlCommand::DropIndex(query)),
3201 other => Err(ParseError::new(
3202 format!("internal: DROP INDEX produced unexpected kind {other:?}"),
3203 self.position(),
3204 )),
3205 }
3206 } else if self.check(&Token::Table) {
3207 self.expect(Token::Table)?;
3208 match self.parse_drop_table_body()? {
3209 QueryExpr::DropTable(query) => Ok(SqlCommand::DropTable(query)),
3210 other => Err(ParseError::new(
3211 format!("internal: DROP TABLE produced unexpected kind {other:?}"),
3212 self.position(),
3213 )),
3214 }
3215 } else if self.check(&Token::Graph) {
3216 self.advance()?;
3217 match self.parse_drop_graph_body()? {
3218 QueryExpr::DropGraph(query) => Ok(SqlCommand::DropGraph(query)),
3219 other => Err(ParseError::new(
3220 format!("internal: DROP GRAPH produced unexpected kind {other:?}"),
3221 self.position(),
3222 )),
3223 }
3224 } else if self.check(&Token::Vector) {
3225 self.advance()?;
3226 match self.parse_drop_vector_body()? {
3227 QueryExpr::DropVector(query) => Ok(SqlCommand::DropVector(query)),
3228 other => Err(ParseError::new(
3229 format!("internal: DROP VECTOR produced unexpected kind {other:?}"),
3230 self.position(),
3231 )),
3232 }
3233 } else if self.check(&Token::Document) {
3234 self.advance()?;
3235 match self.parse_drop_document_body()? {
3236 QueryExpr::DropDocument(query) => Ok(SqlCommand::DropDocument(query)),
3237 other => Err(ParseError::new(
3238 format!("internal: DROP DOCUMENT produced unexpected kind {other:?}"),
3239 self.position(),
3240 )),
3241 }
3242 } else if self.check(&Token::Kv) {
3243 self.advance()?;
3244 match self.parse_drop_kv_body()? {
3245 QueryExpr::DropKv(query) => Ok(SqlCommand::DropKv(query)),
3246 other => Err(ParseError::new(
3247 format!("internal: DROP KV produced unexpected kind {other:?}"),
3248 self.position(),
3249 )),
3250 }
3251 } else if self.consume_ident_ci("CONFIG")? {
3252 match self.parse_drop_keyed_body(CollectionModel::Config)? {
3253 QueryExpr::DropKv(query) => Ok(SqlCommand::DropKv(query)),
3254 other => Err(ParseError::new(
3255 format!("internal: DROP CONFIG produced unexpected kind {other:?}"),
3256 self.position(),
3257 )),
3258 }
3259 } else if self.consume_ident_ci("VAULT")? {
3260 match self.parse_drop_keyed_body(CollectionModel::Vault)? {
3261 QueryExpr::DropKv(query) => Ok(SqlCommand::DropKv(query)),
3262 other => Err(ParseError::new(
3263 format!("internal: DROP VAULT produced unexpected kind {other:?}"),
3264 self.position(),
3265 )),
3266 }
3267 } else if self.check(&Token::Collection) {
3268 self.advance()?;
3269 match self.parse_drop_collection_body()? {
3270 QueryExpr::DropCollection(query) => Ok(SqlCommand::DropCollection(query)),
3271 other => Err(ParseError::new(
3272 format!("internal: DROP COLLECTION produced unexpected kind {other:?}"),
3273 self.position(),
3274 )),
3275 }
3276 } else if self.check(&Token::Timeseries) {
3277 self.advance()?;
3278 match self.parse_drop_timeseries_body()? {
3279 QueryExpr::DropTimeSeries(query) => Ok(SqlCommand::DropTimeSeries(query)),
3280 other => Err(ParseError::new(
3281 format!("internal: DROP TIMESERIES produced unexpected kind {other:?}"),
3282 self.position(),
3283 )),
3284 }
3285 } else if self.consume_ident_ci("METRICS")? {
3286 match self.parse_drop_collection_model_body(Some(CollectionModel::Metrics))? {
3287 QueryExpr::DropCollection(query) => Ok(SqlCommand::DropCollection(query)),
3288 other => Err(ParseError::new(
3289 format!("internal: DROP METRICS produced unexpected kind {other:?}"),
3290 self.position(),
3291 )),
3292 }
3293 } else if matches!(self.peek(), Token::Ident(s) if s.eq_ignore_ascii_case("HYPERTABLE"))
3294 {
3295 self.advance()?;
3299 match self.parse_drop_timeseries_body()? {
3300 QueryExpr::DropTimeSeries(query) => Ok(SqlCommand::DropTimeSeries(query)),
3301 other => Err(ParseError::new(
3302 format!("internal: DROP HYPERTABLE produced unexpected kind {other:?}"),
3303 self.position(),
3304 )),
3305 }
3306 } else if self.check(&Token::Queue) {
3307 self.advance()?;
3308 match self.parse_drop_queue_body()? {
3309 QueryExpr::DropQueue(query) => Ok(SqlCommand::DropQueue(query)),
3310 other => Err(ParseError::new(
3311 format!("internal: DROP QUEUE produced unexpected kind {other:?}"),
3312 self.position(),
3313 )),
3314 }
3315 } else if self.check(&Token::Tree) {
3316 self.advance()?;
3317 match self.parse_drop_tree_body()? {
3318 QueryExpr::DropTree(query) => Ok(SqlCommand::DropTree(query)),
3319 other => Err(ParseError::new(
3320 format!("internal: DROP TREE produced unexpected kind {other:?}"),
3321 self.position(),
3322 )),
3323 }
3324 } else if matches!(self.peek(), Token::Ident(n) if
3325 n.eq_ignore_ascii_case("HLL") ||
3326 n.eq_ignore_ascii_case("SKETCH") ||
3327 n.eq_ignore_ascii_case("FILTER"))
3328 {
3329 match self.parse_drop_probabilistic()? {
3330 QueryExpr::ProbabilisticCommand(command) => {
3331 Ok(SqlCommand::Probabilistic(command))
3332 }
3333 other => Err(ParseError::new(
3334 format!(
3335 "internal: DROP probabilistic produced unexpected kind {other:?}"
3336 ),
3337 self.position(),
3338 )),
3339 }
3340 } else if self.check(&Token::Schema) {
3341 self.advance()?;
3343 let if_exists = self.match_if_exists()?;
3344 let name = self.expect_ident()?;
3345 let cascade = self.consume(&Token::Cascade)?;
3346 Ok(SqlCommand::DropSchema(DropSchemaQuery {
3347 name,
3348 if_exists,
3349 cascade,
3350 }))
3351 } else if self.check(&Token::Policy) {
3352 self.advance()?;
3356 if matches!(self.peek(), Token::String(_)) {
3357 let expr = self.parse_drop_iam_policy_after_keywords()?;
3358 return Ok(SqlCommand::IamPolicy(expr));
3359 }
3360 let if_exists = self.match_if_exists()?;
3361 let name = self.expect_ident()?;
3362 self.expect(Token::On)?;
3363 let table = self.expect_ident()?;
3364 Ok(SqlCommand::DropPolicy(DropPolicyQuery {
3365 name,
3366 table,
3367 if_exists,
3368 }))
3369 } else if self.check(&Token::Server) {
3370 self.advance()?;
3372 let if_exists = self.match_if_exists()?;
3373 let name = self.expect_ident()?;
3374 let cascade = self.consume(&Token::Cascade)?;
3375 Ok(SqlCommand::DropServer(DropServerQuery {
3376 name,
3377 if_exists,
3378 cascade,
3379 }))
3380 } else if self.check(&Token::Foreign) {
3381 self.advance()?;
3383 self.expect(Token::Table)?;
3384 let if_exists = self.match_if_exists()?;
3385 let name = self.expect_ident()?;
3386 Ok(SqlCommand::DropForeignTable(DropForeignTableQuery {
3387 name,
3388 if_exists,
3389 }))
3390 } else if self.check(&Token::Sequence) {
3391 self.advance()?;
3393 let if_exists = self.match_if_exists()?;
3394 let name = self.expect_ident()?;
3395 Ok(SqlCommand::DropSequence(DropSequenceQuery {
3396 name,
3397 if_exists,
3398 }))
3399 } else if let Some(err) =
3400 ParseError::unsupported_recognized_token(self.peek(), self.position())
3401 {
3402 Err(err)
3403 } else {
3404 Err(ParseError::expected(
3405 vec![
3406 "TABLE",
3407 "INDEX",
3408 "TIMESERIES",
3409 "QUEUE",
3410 "TREE",
3411 "HLL",
3412 "SKETCH",
3413 "FILTER",
3414 "SCHEMA",
3415 "SEQUENCE",
3416 ],
3417 self.peek(),
3418 pos,
3419 ))
3420 }
3421 }
3422 Token::Alter => {
3423 let next = self.peek_next()?.clone();
3428 if matches!(next, Token::Ident(ref s) if s.eq_ignore_ascii_case("USER")) {
3429 self.advance()?; let stmt = self.parse_alter_user_statement()?;
3431 Ok(SqlCommand::AlterUser(stmt))
3432 } else if matches!(next, Token::Queue) {
3433 self.advance()?; self.advance()?; match self.parse_alter_queue_body()? {
3436 QueryExpr::AlterQueue(query) => Ok(SqlCommand::AlterQueue(query)),
3437 other => Err(ParseError::new(
3438 format!("internal: ALTER QUEUE produced unexpected kind {other:?}"),
3439 self.position(),
3440 )),
3441 }
3442 } else if matches!(next, Token::Metric) {
3443 self.advance()?; self.advance()?; match self.parse_alter_metric_body()? {
3446 QueryExpr::AlterMetric(query) => Ok(SqlCommand::AlterMetric(query)),
3447 other => Err(ParseError::new(
3448 format!("internal: ALTER METRIC produced unexpected kind {other:?}"),
3449 self.position(),
3450 )),
3451 }
3452 } else if matches!(next, Token::Graph) {
3453 match self.parse_alter_graph_query()? {
3457 QueryExpr::AlterTable(query) => Ok(SqlCommand::AlterTable(query)),
3458 other => Err(ParseError::new(
3459 format!(
3460 "internal: ALTER GRAPH produced unexpected query kind {other:?}"
3461 ),
3462 self.position(),
3463 )),
3464 }
3465 } else if matches!(next, Token::Table)
3466 || matches!(next, Token::Collection)
3467 || matches!(next, Token::Ident(ref s) if s.eq_ignore_ascii_case("COLLECTION"))
3468 {
3469 match self.parse_alter_table_query()? {
3475 QueryExpr::AlterTable(query) => Ok(SqlCommand::AlterTable(query)),
3476 other => Err(ParseError::new(
3477 format!(
3478 "internal: ALTER TABLE produced unexpected query kind {other:?}"
3479 ),
3480 self.position(),
3481 )),
3482 }
3483 } else if let Some(err) =
3484 ParseError::unsupported_recognized_token(&next, self.position())
3485 {
3486 Err(err)
3487 } else {
3488 match self.parse_alter_table_query()? {
3489 QueryExpr::AlterTable(query) => Ok(SqlCommand::AlterTable(query)),
3490 other => Err(ParseError::new(
3491 format!("internal: ALTER produced unexpected query kind {other:?}"),
3492 self.position(),
3493 )),
3494 }
3495 }
3496 }
3497 Token::Ident(name) if name.eq_ignore_ascii_case("GRANT") => {
3498 let stmt = self.parse_grant_statement()?;
3499 Ok(SqlCommand::Grant(stmt))
3500 }
3501 Token::Ident(name) if name.eq_ignore_ascii_case("REVOKE") => {
3502 let stmt = self.parse_revoke_statement()?;
3503 Ok(SqlCommand::Revoke(stmt))
3504 }
3505 Token::Ident(name) if name.eq_ignore_ascii_case("EVENTS") => {
3506 self.advance()?;
3507 if self.consume_ident_ci("BACKFILL")? {
3508 return Err(ParseError::new(
3509 "EVENTS BACKFILL STATUS is not implemented; EVENTS BACKFILL runtime is available but durable progress tracking is not"
3510 .to_string(),
3511 self.position(),
3512 ));
3513 }
3514 if !self.consume_ident_ci("STATUS")? {
3515 return Err(ParseError::expected(
3516 vec!["STATUS"],
3517 self.peek(),
3518 self.position(),
3519 ));
3520 }
3521
3522 let mut query = TableQuery::new("red.subscriptions");
3523 let collection = match self.peek().clone() {
3524 Token::Ident(name) => {
3525 self.advance()?;
3526 Some(name)
3527 }
3528 Token::String(name) => {
3529 self.advance()?;
3530 Some(name)
3531 }
3532 _ => None,
3533 };
3534 self.parse_table_clauses(&mut query)?;
3535 if let Some(collection) = collection {
3536 let filter = Filter::compare(
3537 FieldRef::column("red.subscriptions", "collection"),
3538 CompareOp::Eq,
3539 Value::text(collection),
3540 );
3541 let expr = filter_to_expr(&filter);
3542 query.where_expr = Some(match query.where_expr.take() {
3543 Some(existing) => Expr::binop(BinOp::And, existing, expr),
3544 None => expr,
3545 });
3546 query.filter = Some(match query.filter.take() {
3547 Some(existing) => existing.and(filter),
3548 None => filter,
3549 });
3550 }
3551 Ok(SqlCommand::Select(query))
3552 }
3553 Token::Attach => {
3554 let expr = self.parse_attach_policy()?;
3555 Ok(SqlCommand::IamPolicy(expr))
3556 }
3557 Token::Detach => {
3558 let expr = self.parse_detach_policy()?;
3559 Ok(SqlCommand::IamPolicy(expr))
3560 }
3561 Token::Ident(name) if name.eq_ignore_ascii_case("SIMULATE") => {
3562 let expr = self.parse_simulate_policy()?;
3563 Ok(SqlCommand::IamPolicy(expr))
3564 }
3565 Token::Ident(name) if name.eq_ignore_ascii_case("LINT") => {
3566 let expr = self.parse_lint_policy()?;
3567 Ok(SqlCommand::IamPolicy(expr))
3568 }
3569 Token::Ident(name) if name.eq_ignore_ascii_case("MIGRATE") => {
3570 let next = self.peek_next()?.clone();
3574 let is_policy_mode = matches!(&next, Token::Policy)
3575 || matches!(&next, Token::Ident(name)
3576 if name.eq_ignore_ascii_case("POLICY"));
3577 if is_policy_mode {
3578 let expr = self.parse_migrate_policy_mode()?;
3579 return Ok(SqlCommand::IamPolicy(expr));
3580 }
3581 Err(ParseError::expected(
3582 vec!["POLICY"],
3583 self.peek(),
3584 self.position(),
3585 ))
3586 }
3587 Token::Set => {
3588 self.advance()?;
3589 if self.consume_ident_ci("CONFIG")? {
3590 let full_key = self.parse_dotted_admin_path(true)?;
3591 self.expect(Token::Eq)?;
3592 let value = self.parse_literal_value()?;
3593 Ok(SqlCommand::SetConfig {
3594 key: full_key,
3595 value,
3596 })
3597 } else if self.consume_ident_ci("SECRET")? {
3598 let key =
3599 Self::normalize_secret_admin_path(self.parse_dotted_admin_path(true)?);
3600 self.expect(Token::Eq)?;
3601 let value = self.parse_literal_value()?;
3602 Ok(SqlCommand::SetSecret { key, value })
3603 } else if self.consume_ident_ci("TENANT")? {
3604 let _ = self.consume(&Token::Eq)?;
3607 if self.consume_ident_ci("NULL")? {
3608 Ok(SqlCommand::SetTenant(None))
3609 } else {
3610 let value = self.parse_literal_value()?;
3611 match value {
3612 Value::Text(s) => Ok(SqlCommand::SetTenant(Some(s.to_string()))),
3613 Value::Null => Ok(SqlCommand::SetTenant(None)),
3614 other => Err(ParseError::new(
3615 format!("SET TENANT expects a text literal or NULL, got {other:?}"),
3616 self.position(),
3617 )),
3618 }
3619 }
3620 } else {
3621 Err(ParseError::expected(
3622 vec!["CONFIG", "SECRET", "TENANT"],
3623 self.peek(),
3624 self.position(),
3625 ))
3626 }
3627 }
3628 Token::Ident(name) if name.eq_ignore_ascii_case("APPLY") => {
3629 self.advance()?;
3630 match self.parse_apply_migration()? {
3631 QueryExpr::ApplyMigration(q) => Ok(SqlCommand::ApplyMigration(q)),
3632 other => Err(ParseError::new(
3633 format!("internal: APPLY MIGRATION produced unexpected kind {other:?}"),
3634 self.position(),
3635 )),
3636 }
3637 }
3638 Token::Ident(name) if name.eq_ignore_ascii_case("RESET") => {
3639 self.advance()?;
3641 if self.consume_ident_ci("TENANT")? {
3642 Ok(SqlCommand::SetTenant(None))
3643 } else {
3644 Err(ParseError::expected(
3645 vec!["TENANT"],
3646 self.peek(),
3647 self.position(),
3648 ))
3649 }
3650 }
3651 Token::Ident(name)
3652 if name.eq_ignore_ascii_case("DESCRIBE") || name.eq_ignore_ascii_case("DESC") =>
3653 {
3654 self.advance()?;
3655 let collection = self.parse_dotted_admin_path(false)?;
3656 let mut query = TableQuery::new("red.describe");
3657 query.filter = Some(Filter::compare(
3658 FieldRef::column("", "collection"),
3659 CompareOp::Eq,
3660 Value::text(collection),
3661 ));
3662 Ok(SqlCommand::Select(query))
3663 }
3664 Token::Desc => {
3665 self.advance()?;
3666 let collection = self.parse_dotted_admin_path(false)?;
3667 let mut query = TableQuery::new("red.describe");
3668 query.filter = Some(Filter::compare(
3669 FieldRef::column("", "collection"),
3670 CompareOp::Eq,
3671 Value::text(collection),
3672 ));
3673 Ok(SqlCommand::Select(query))
3674 }
3675 Token::Ident(name) if name.eq_ignore_ascii_case("SHOW") => {
3676 self.advance()?;
3677 if self.consume(&Token::Create)? || self.consume_ident_ci("CREATE")? {
3678 if !(self.consume(&Token::Table)? || self.consume_ident_ci("TABLE")?) {
3679 return Err(ParseError::expected(
3680 vec!["TABLE"],
3681 self.peek(),
3682 self.position(),
3683 ));
3684 }
3685 let collection = self.parse_dotted_admin_path(false)?;
3686 let mut query = TableQuery::new("red.show_create");
3687 query.filter = Some(Filter::compare(
3688 FieldRef::column("", "collection"),
3689 CompareOp::Eq,
3690 Value::text(collection),
3691 ));
3692 Ok(SqlCommand::Select(query))
3693 } else if self.consume_ident_ci("CONFIG")? {
3694 let prefix = if !(self.check(&Token::Eof)
3698 || self.check(&Token::As)
3699 || self.check(&Token::Format))
3700 {
3701 let first = self.expect_ident()?;
3702 let mut full = first;
3703 while self.consume(&Token::Dot)? {
3704 let next = self.expect_ident_or_keyword()?;
3705 full = format!("{full}.{next}");
3706 }
3707 Some(full.to_ascii_lowercase())
3710 } else {
3711 None
3712 };
3713 let as_json = if self.consume(&Token::As)? || self.consume(&Token::Format)? {
3714 if !self.consume(&Token::Json)? {
3715 return Err(ParseError::expected(
3716 vec!["JSON"],
3717 self.peek(),
3718 self.position(),
3719 ));
3720 }
3721 true
3722 } else {
3723 false
3724 };
3725 Ok(SqlCommand::ShowConfig { prefix, as_json })
3726 } else if self.consume_ident_ci("COLLECTIONS")? {
3727 let mut query = TableQuery::new("red.collections");
3728 let include_internal = if self.consume_ident_ci("INCLUDING")? {
3729 if !self.consume_ident_ci("INTERNAL")? {
3730 return Err(ParseError::expected(
3731 vec!["INTERNAL"],
3732 self.peek(),
3733 self.position(),
3734 ));
3735 }
3736 true
3737 } else {
3738 false
3739 };
3740 self.parse_table_clauses(&mut query)?;
3741 if !include_internal {
3742 let user_filter = query.filter.take();
3743 let hide_internal = crate::ast::Filter::Compare {
3744 field: FieldRef::column("", "internal"),
3745 op: CompareOp::Eq,
3746 value: Value::Boolean(false),
3747 };
3748 query.filter = Some(match user_filter {
3749 Some(filter) => filter.and(hide_internal),
3750 None => hide_internal,
3751 });
3752 }
3753 Ok(SqlCommand::Select(query))
3754 } else if self.consume_ident_ci("TABLES")? {
3755 Ok(SqlCommand::Select(parse_show_collections_by_model(
3756 self, "table",
3757 )?))
3758 } else if self.consume_ident_ci("QUEUES")? {
3759 let mut query = TableQuery::new("red.queues");
3767 let include_internal = if self.consume_ident_ci("INCLUDING")? {
3768 if !self.consume_ident_ci("INTERNAL")? {
3769 return Err(ParseError::expected(
3770 vec!["INTERNAL"],
3771 self.peek(),
3772 self.position(),
3773 ));
3774 }
3775 true
3776 } else {
3777 false
3778 };
3779 self.parse_table_clauses(&mut query)?;
3780 if !include_internal {
3781 let hide_internal = Filter::Compare {
3782 field: FieldRef::column("", "internal"),
3783 op: CompareOp::Eq,
3784 value: Value::Boolean(false),
3785 };
3786 add_table_filter(&mut query, hide_internal);
3787 }
3788 Ok(SqlCommand::Select(query))
3789 } else if self.consume(&Token::Vectors)? || self.consume_ident_ci("VECTORS")? {
3790 Ok(SqlCommand::Select(parse_show_collections_by_model(
3791 self, "vector",
3792 )?))
3793 } else if self.consume_ident_ci("DOCUMENTS")? {
3794 Ok(SqlCommand::Select(parse_show_collections_by_model(
3795 self, "document",
3796 )?))
3797 } else if self.consume(&Token::Timeseries)?
3798 || self.consume_ident_ci("TIMESERIES")?
3799 {
3800 Ok(SqlCommand::Select(parse_show_collections_by_model(
3801 self,
3802 "timeseries",
3803 )?))
3804 } else if self.consume_ident_ci("GRAPHS")? {
3805 Ok(SqlCommand::Select(parse_show_collections_by_model(
3806 self, "graph",
3807 )?))
3808 } else if self.consume_ident_ci("CONFIGS")? {
3809 Ok(SqlCommand::Select(parse_show_collections_by_model(
3810 self, "config",
3811 )?))
3812 } else if self.consume_ident_ci("VAULTS")? {
3813 Ok(SqlCommand::Select(parse_show_collections_by_model(
3814 self, "vault",
3815 )?))
3816 } else if self.consume(&Token::Kv)?
3817 || self.consume_ident_ci("KV")?
3818 || self.consume_ident_ci("KVS")?
3819 {
3820 Ok(SqlCommand::Select(parse_show_collections_by_model(
3821 self, "kv",
3822 )?))
3823 } else if self.consume(&Token::Schema)? || self.consume_ident_ci("SCHEMA")? {
3824 let collection = self.parse_dotted_admin_path(false)?;
3825 let mut query = TableQuery::new("red.columns");
3826 query.filter = Some(Filter::compare(
3827 FieldRef::column("", "collection"),
3828 CompareOp::Eq,
3829 Value::text(collection),
3830 ));
3831 Ok(SqlCommand::Select(query))
3832 } else if self.consume_ident_ci("INDICES")? || self.consume_ident_ci("INDEXES")? {
3833 let mut query = TableQuery::new("red.show_indexes");
3834 if self.consume(&Token::On)? {
3835 let collection = self.expect_ident_or_keyword()?;
3836 let filter = Filter::Compare {
3837 field: FieldRef::column("", "table"),
3838 op: CompareOp::Eq,
3839 value: Value::text(collection),
3840 };
3841 query.where_expr = Some(filter_to_expr(&filter));
3842 query.filter = Some(filter);
3843 }
3844 self.parse_table_clauses(&mut query)?;
3845 Ok(SqlCommand::Select(query))
3846 } else if self.consume_ident_ci("POLICIES")? {
3847 if self.consume(&Token::For)? || self.consume_ident_ci("FOR")? {
3848 let principal = self.parse_iam_principal_kind()?;
3849 return Ok(SqlCommand::IamPolicy(QueryExpr::ShowPolicies {
3850 filter: Some(principal),
3851 }));
3852 }
3853 let mut query = TableQuery::new("red.policies");
3854 let collection_filter =
3855 if self.consume(&Token::On)? || self.consume_ident_ci("ON")? {
3856 let collection = self.parse_dotted_admin_path(false)?;
3857 Some(Filter::Compare {
3858 field: FieldRef::TableColumn {
3859 table: String::new(),
3860 column: "collection".to_string(),
3861 },
3862 op: CompareOp::Eq,
3863 value: Value::text(collection),
3864 })
3865 } else {
3866 None
3867 };
3868 self.parse_table_clauses(&mut query)?;
3869 if let Some(collection_filter) = collection_filter {
3870 let combined = match query.filter.take() {
3871 Some(existing) => {
3872 Filter::And(Box::new(collection_filter), Box::new(existing))
3873 }
3874 None => collection_filter,
3875 };
3876 query.where_expr = Some(filter_to_expr(&combined));
3877 query.filter = Some(combined);
3878 }
3879 Ok(SqlCommand::Select(query))
3880 } else if self.consume_ident_ci("STATS")? {
3881 let mut query = TableQuery::new("red.stats");
3882 let collection = match self.peek().clone() {
3883 Token::Ident(name) => {
3884 self.advance()?;
3885 Some(name)
3886 }
3887 Token::String(name) => {
3888 self.advance()?;
3889 Some(name)
3890 }
3891 _ => None,
3892 };
3893 self.parse_table_clauses(&mut query)?;
3894 if let Some(collection) = collection {
3895 let filter = Filter::compare(
3896 FieldRef::column("red.stats", "collection"),
3897 CompareOp::Eq,
3898 Value::text(collection),
3899 );
3900 let expr = filter_to_expr(&filter);
3901 query.where_expr = Some(match query.where_expr.take() {
3902 Some(existing) => Expr::binop(BinOp::And, existing, expr),
3903 None => expr,
3904 });
3905 query.filter = Some(match query.filter.take() {
3906 Some(existing) => existing.and(filter),
3907 None => filter,
3908 });
3909 }
3910 Ok(SqlCommand::Select(query))
3911 } else if self.consume_ident_ci("SAMPLE")? {
3912 let mut query = TableQuery::new(&self.expect_ident()?);
3913 query.limit = if self.consume(&Token::Limit)? {
3914 Some(self.parse_integer()? as u64)
3915 } else {
3916 Some(10)
3917 };
3918 Ok(SqlCommand::Select(query))
3919 } else if self.consume_ident_ci("SECRET")? || self.consume_ident_ci("SECRETS")? {
3920 let prefix = if !self.check(&Token::Eof) {
3921 Some(Self::normalize_secret_admin_path(
3922 self.parse_dotted_admin_path(true)?,
3923 ))
3924 } else {
3925 None
3926 };
3927 Ok(SqlCommand::ShowSecrets { prefix })
3928 } else if self.consume_ident_ci("TENANT")? {
3929 Ok(SqlCommand::ShowTenant)
3930 } else if let Some(expr) = self.parse_show_iam_after_show()? {
3931 Ok(SqlCommand::IamPolicy(expr))
3932 } else {
3933 Err(ParseError::expected(
3934 vec![
3935 "CONFIG",
3936 "SECRET",
3937 "SECRETS",
3938 "COLLECTIONS",
3939 "TABLES",
3940 "QUEUES",
3941 "VECTORS",
3942 "DOCUMENTS",
3943 "TIMESERIES",
3944 "GRAPHS",
3945 "KV",
3946 "SCHEMA",
3947 "INDICES",
3948 "INDEXES",
3949 "SAMPLE",
3950 "POLICIES",
3951 "STATS",
3952 "TENANT",
3953 "EFFECTIVE",
3954 ],
3955 self.peek(),
3956 self.position(),
3957 ))
3958 }
3959 }
3960 Token::Begin | Token::Start => {
3972 self.advance()?;
3973 let _ = self.consume(&Token::Work)? || self.consume(&Token::Transaction)?;
3974 if self.consume_ident_ci("ISOLATION")? {
3976 self.expect(Token::Level)?;
3977 let mut parts: Vec<String> = Vec::new();
3981 if self.consume_ident_ci("READ")? {
3982 parts.push("READ".to_string());
3983 if self.consume_ident_ci("UNCOMMITTED")? {
3984 parts.push("UNCOMMITTED".to_string());
3985 } else if self.consume_ident_ci("COMMITTED")? {
3986 parts.push("COMMITTED".to_string());
3987 } else {
3988 return Err(ParseError::expected(
3989 vec!["UNCOMMITTED", "COMMITTED"],
3990 self.peek(),
3991 self.position(),
3992 ));
3993 }
3994 } else if self.consume_ident_ci("REPEATABLE")? {
3995 parts.push("REPEATABLE".to_string());
3996 if !self.consume_ident_ci("READ")? {
3997 return Err(ParseError::expected(
3998 vec!["READ"],
3999 self.peek(),
4000 self.position(),
4001 ));
4002 }
4003 parts.push("READ".to_string());
4004 } else if self.consume_ident_ci("SNAPSHOT")? {
4005 parts.push("SNAPSHOT".to_string());
4006 } else if self.consume_ident_ci("SERIALIZABLE")? {
4007 return Err(ParseError::new(
4008 "ISOLATION LEVEL SERIALIZABLE is not yet supported — reddb \
4009 currently provides SNAPSHOT ISOLATION (which PG calls \
4010 REPEATABLE READ). Use REPEATABLE READ / SNAPSHOT / \
4011 READ COMMITTED, or omit ISOLATION LEVEL for the default."
4012 .to_string(),
4013 self.position(),
4014 ));
4015 } else {
4016 return Err(ParseError::expected(
4017 vec!["READ", "REPEATABLE", "SNAPSHOT", "SERIALIZABLE"],
4018 self.peek(),
4019 self.position(),
4020 ));
4021 }
4022 let _ = parts;
4024 }
4025 Ok(SqlCommand::TransactionControl(TxnControl::Begin))
4026 }
4027 Token::Commit => {
4029 self.advance()?;
4030 let _ = self.consume(&Token::Work)? || self.consume(&Token::Transaction)?;
4031 Ok(SqlCommand::TransactionControl(TxnControl::Commit))
4032 }
4033 Token::Rollback => {
4036 self.advance()?;
4037 if matches!(self.peek(), Token::Ident(n) if n.eq_ignore_ascii_case("MIGRATION")) {
4038 match self.parse_rollback_migration_after_keyword()? {
4039 QueryExpr::RollbackMigration(q) => Ok(SqlCommand::RollbackMigration(q)),
4040 other => Err(ParseError::new(
4041 format!(
4042 "internal: ROLLBACK MIGRATION produced unexpected kind {other:?}"
4043 ),
4044 self.position(),
4045 )),
4046 }
4047 } else {
4048 let _ = self.consume(&Token::Work)? || self.consume(&Token::Transaction)?;
4049 if self.consume(&Token::To)? {
4050 let _ = self.consume(&Token::Savepoint)?;
4051 let name = self.expect_ident()?;
4052 Ok(SqlCommand::TransactionControl(
4053 TxnControl::RollbackToSavepoint(name),
4054 ))
4055 } else {
4056 Ok(SqlCommand::TransactionControl(TxnControl::Rollback))
4057 }
4058 }
4059 }
4060 Token::Savepoint => {
4062 self.advance()?;
4063 let name = self.expect_ident()?;
4064 Ok(SqlCommand::TransactionControl(TxnControl::Savepoint(name)))
4065 }
4066 Token::Release => {
4068 self.advance()?;
4069 let _ = self.consume(&Token::Savepoint)?;
4070 let name = self.expect_ident()?;
4071 Ok(SqlCommand::TransactionControl(
4072 TxnControl::ReleaseSavepoint(name),
4073 ))
4074 }
4075 Token::Vacuum => {
4077 self.advance()?;
4078 let full = self.consume(&Token::Full)?;
4079 let target = if self.check(&Token::Eof) {
4080 None
4081 } else {
4082 Some(self.expect_ident()?)
4083 };
4084 Ok(SqlCommand::Maintenance(MaintenanceCommand::Vacuum {
4085 target,
4086 full,
4087 }))
4088 }
4089 Token::Refresh => {
4091 self.advance()?;
4092 self.expect(Token::Materialized)?;
4093 self.expect(Token::View)?;
4094 let name = self.expect_ident()?;
4095 Ok(SqlCommand::RefreshMaterializedView(
4096 RefreshMaterializedViewQuery { name },
4097 ))
4098 }
4099 Token::Analyze => {
4101 self.advance()?;
4102 let target = if self.check(&Token::Eof) {
4103 None
4104 } else {
4105 Some(self.expect_ident()?)
4106 };
4107 Ok(SqlCommand::Maintenance(MaintenanceCommand::Analyze {
4108 target,
4109 }))
4110 }
4111 Token::Copy => {
4117 self.advance()?;
4118 let table = self.expect_ident()?;
4119 self.expect(Token::From)?;
4120 let path = self.parse_string()?;
4121
4122 let mut delimiter: Option<char> = None;
4123 let mut has_header = false;
4124 let format = CopyFormat::Csv;
4125
4126 if self.consume(&Token::With)? || self.consume_ident_ci("WITH")? {
4130 self.expect(Token::LParen)?;
4131 loop {
4132 if self.consume(&Token::Format)? || self.consume_ident_ci("FORMAT")? {
4133 let _ = self.consume(&Token::Eq)?;
4134 let _ = self.expect_ident()?;
4136 } else if self.consume(&Token::Header)? {
4137 let _ = self.consume(&Token::Eq)?;
4138 has_header = match self.peek().clone() {
4141 Token::True => {
4142 self.advance()?;
4143 true
4144 }
4145 Token::False => {
4146 self.advance()?;
4147 false
4148 }
4149 Token::Ident(ref n) if n.eq_ignore_ascii_case("true") => {
4150 self.advance()?;
4151 true
4152 }
4153 Token::Ident(ref n) if n.eq_ignore_ascii_case("false") => {
4154 self.advance()?;
4155 false
4156 }
4157 _ => true,
4158 };
4159 } else if self.consume(&Token::Delimiter)? {
4160 let _ = self.consume(&Token::Eq)?;
4161 let s = self.parse_string()?;
4162 delimiter = s.chars().next();
4163 } else {
4164 break;
4165 }
4166 if !self.consume(&Token::Comma)? {
4167 break;
4168 }
4169 }
4170 self.expect(Token::RParen)?;
4171 }
4172
4173 loop {
4175 if self.consume(&Token::Delimiter)? {
4176 let s = self.parse_string()?;
4177 delimiter = s.chars().next();
4178 } else if self.consume(&Token::Header)? {
4179 has_header = true;
4180 } else {
4181 break;
4182 }
4183 }
4184
4185 Ok(SqlCommand::CopyFrom(CopyFromQuery {
4186 table,
4187 path,
4188 format,
4189 delimiter,
4190 has_header,
4191 }))
4192 }
4193 other => Err(ParseError::expected(
4194 vec![
4195 "SELECT",
4196 "FROM",
4197 "INSERT",
4198 "UPDATE",
4199 "DELETE",
4200 "EXPLAIN",
4201 "CREATE",
4202 "DROP",
4203 "ALTER",
4204 "SET",
4205 "SHOW",
4206 "BEGIN",
4207 "COMMIT",
4208 "ROLLBACK",
4209 "SAVEPOINT",
4210 "RELEASE",
4211 "START",
4212 "VACUUM",
4213 "ANALYZE",
4214 "COPY",
4215 "REFRESH",
4216 "DESCRIBE",
4217 "DESC",
4218 ],
4219 other,
4220 self.position(),
4221 )),
4222 }
4223 }
4224}