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