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