1use super::*;
9use crate::catalog::CollectionModel;
10use crate::runtime::audit_log::{AuditAuthSource, AuditEvent, AuditFieldEscaper, Outcome};
11use crate::runtime::ddl::polymorphic_resolver;
12use reddb_engine::catalog::ddl::declared_column_contract_from_ddl;
13use reddb_rql::analyzer::analyze_create_table;
14use reddb_rql::ast::CreateColumnDef;
15use reddb_rql::ast::{
16 CreateVcsRefQuery, DropForkQuery, DropVcsRefQuery, ForkStoreQuery, VcsRefKind,
17};
18use std::collections::{BTreeSet, HashMap, HashSet};
19
20fn vault_master_key_ref(collection: &str) -> String {
21 format!("red.vault.{collection}.master_key")
22}
23
24impl RedDBRuntime {
25 pub fn execute_create_table(
31 &self,
32 raw_query: &str,
33 query: &CreateTableQuery,
34 ) -> RedDBResult<RuntimeQueryResult> {
35 if query.collection_model != CollectionModel::Table {
36 if query
37 .columns
38 .iter()
39 .any(|column| column.generated.is_some() || column.check.is_some())
40 {
41 return Err(RedDBError::Query(
42 "schema expressions currently require TABLE collections".to_string(),
43 ));
44 }
45 return self.execute_create_keyed_collection(raw_query, query);
46 }
47 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
48 if is_system_schema_name(&query.name) {
49 return Err(RedDBError::Query("system schema is read-only".to_string()));
50 }
51 let store = self.inner.db.store();
52 analyze_create_table(query).map_err(|err| RedDBError::Query(err.to_string()))?;
53 if let Some(policy) = &query.ai_policy {
57 validate_ai_policy_modalities(policy)?;
58 }
59 crate::reserved_fields::ensure_no_reserved_public_item_fields(
60 query.columns.iter().map(|column| column.name.as_str()),
61 &format!("table '{}'", query.name),
62 )?;
63 let exists = store.get_collection(&query.name).is_some();
65 if exists {
66 if query.if_not_exists {
67 return Ok(RuntimeQueryResult::ok_message(
68 raw_query.to_string(),
69 &format!("table '{}' already exists", query.name),
70 "create",
71 ));
72 }
73 return Err(RedDBError::Query(format!(
74 "table '{}' already exists",
75 query.name
76 )));
77 }
78
79 let contract = collection_contract_from_create_table(query)?;
82 crate::application::collection_contract_enforcer::validate_contract_expressions(&contract)?;
83 validate_event_subscriptions(self, &query.name, &contract.subscriptions)?;
84 store
86 .create_collection(&query.name)
87 .map_err(|err| RedDBError::Internal(err.to_string()))?;
88 for subscription in &contract.subscriptions {
89 ensure_event_target_queue(self, &subscription.target_queue)?;
90 }
91 if let Some(default_ttl_ms) = query.default_ttl_ms {
92 self.inner
93 .db
94 .set_collection_default_ttl_ms(&query.name, default_ttl_ms);
95 }
96 self.inner
97 .db
98 .save_collection_contract(contract)
99 .map_err(|err| RedDBError::Internal(err.to_string()))?;
100 if let Some(tenant_id) = crate::runtime::impl_core::current_tenant() {
101 store.set_config_tree(
102 &format!("red.collection_tenants.{}", query.name),
103 &crate::serde_json::Value::String(tenant_id),
104 );
105 }
106 self.inner
107 .db
108 .persist_metadata()
109 .map_err(|err| RedDBError::Internal(err.to_string()))?;
110 self.refresh_table_planner_stats(&query.name);
111 self.invalidate_result_cache();
112 let columns: Vec<String> = query.columns.iter().map(|col| col.name.clone()).collect();
115 self.schema_vocabulary_apply(
116 crate::runtime::schema_vocabulary::DdlEvent::CreateCollection {
117 collection: query.name.clone(),
118 columns,
119 type_tags: Vec::new(),
120 description: None,
121 },
122 );
123 if let Some(spec) = &query.partition_by {
130 let kind_str = match spec.kind {
131 reddb_rql::ast::PartitionKind::Range => "range",
132 reddb_rql::ast::PartitionKind::List => "list",
133 reddb_rql::ast::PartitionKind::Hash => "hash",
134 };
135 store.set_config_tree(
136 &format!("partition.{}.by", query.name),
137 &crate::serde_json::Value::String(kind_str.to_string()),
138 );
139 store.set_config_tree(
140 &format!("partition.{}.column", query.name),
141 &crate::serde_json::Value::String(spec.column.clone()),
142 );
143 }
144
145 if let Some(col) = &query.tenant_by {
156 store.set_config_tree(
157 &format!("tenant_tables.{}.column", query.name),
158 &crate::serde_json::Value::String(col.clone()),
159 );
160 self.register_tenant_table(&query.name, col);
161 }
162
163 let ttl_suffix = query
164 .default_ttl_ms
165 .map(|ttl_ms| format!(" with default TTL {}ms", ttl_ms))
166 .unwrap_or_default();
167
168 let tenant_suffix = query
169 .tenant_by
170 .as_ref()
171 .map(|col| format!(" (tenant-scoped by {col})"))
172 .unwrap_or_default();
173
174 Ok(RuntimeQueryResult::ok_message(
175 raw_query.to_string(),
176 &format!(
177 "table '{}' created{}{}",
178 query.name, ttl_suffix, tenant_suffix
179 ),
180 "create",
181 ))
182 }
183
184 fn execute_create_keyed_collection(
185 &self,
186 raw_query: &str,
187 query: &CreateTableQuery,
188 ) -> RedDBResult<RuntimeQueryResult> {
189 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
190 if is_system_schema_name(&query.name) {
191 return Err(RedDBError::Query("system schema is read-only".to_string()));
192 }
193 let store = self.inner.db.store();
194 let label = polymorphic_resolver::model_name(query.collection_model);
195 if store.get_collection(&query.name).is_some() {
196 if query.if_not_exists {
197 return Ok(RuntimeQueryResult::ok_message(
198 raw_query.to_string(),
199 &format!("{label} '{}' already exists", query.name),
200 "create",
201 ));
202 }
203 return Err(RedDBError::Query(format!(
204 "{label} '{}' already exists",
205 query.name
206 )));
207 }
208
209 if query.collection_model == CollectionModel::Vault {
210 self.reject_vault_transaction()?;
211 self.provision_vault_key_material(&query.name, query.vault_own_master_key)?;
212 }
213 store
214 .create_collection(&query.name)
215 .map_err(|err| RedDBError::Internal(err.to_string()))?;
216 if query.collection_model == CollectionModel::Vault {
217 let key_scope = if query.vault_own_master_key {
218 "own"
219 } else {
220 "cluster"
221 };
222 store.set_config_tree(
223 &format!("red.vault.{}.key_scope", query.name),
224 &crate::serde_json::Value::String(key_scope.to_string()),
225 );
226 store.set_config_tree(
227 &format!("red.vault.{}.status", query.name),
228 &crate::serde_json::Value::String("sealed".to_string()),
229 );
230 }
231 if query.collection_model == CollectionModel::Metrics {
232 for spec in &query.metrics_rollup_policies {
233 let policy = crate::storage::timeseries::retention::DownsamplePolicy::parse(spec)
234 .ok_or_else(|| {
235 RedDBError::Query(format!("invalid metrics rollup policy '{}'", spec))
236 })?;
237 if policy.source != "raw" {
238 return Err(RedDBError::Query(format!(
239 "invalid metrics rollup policy '{}': metrics v0 rollups must use raw as source",
240 spec
241 )));
242 }
243 if !matches!(
244 policy.aggregation.as_str(),
245 "avg" | "sum" | "min" | "max" | "count"
246 ) {
247 return Err(RedDBError::Query(format!(
248 "invalid metrics rollup policy '{}': supported aggregations are avg, sum, min, max, count",
249 spec
250 )));
251 }
252 }
253 if let Some(raw_retention_ms) = query.default_ttl_ms {
254 self.inner
255 .db
256 .set_collection_default_ttl_ms(&query.name, raw_retention_ms);
257 store.set_config_tree(
258 &format!("red.metrics.{}.raw_retention_ms", query.name),
259 &crate::serde_json::Value::Number(raw_retention_ms as f64),
260 );
261 }
262 let tenant_identity = query
263 .tenant_by
264 .clone()
265 .unwrap_or_else(|| "current_tenant".to_string());
266 store.set_config_tree(
267 &format!("red.metrics.{}.tenant_identity", query.name),
268 &crate::serde_json::Value::String(tenant_identity),
269 );
270 store.set_config_tree(
271 &format!("red.metrics.{}.namespace", query.name),
272 &crate::serde_json::Value::String("default".to_string()),
273 );
274 if !query.metrics_rollup_policies.is_empty() {
275 store.set_config_tree(
276 &format!("red.metrics.{}.rollup_policies", query.name),
277 &crate::serde_json::Value::Array(
278 query
279 .metrics_rollup_policies
280 .iter()
281 .cloned()
282 .map(crate::serde_json::Value::String)
283 .collect(),
284 ),
285 );
286 }
287 }
288 let contract = if query.collection_model == CollectionModel::Metrics {
289 metrics_collection_contract(query)
290 } else {
291 keyed_collection_contract(
292 &query.name,
293 query.collection_model,
294 query.analytics_config.clone(),
295 )
296 };
297 self.inner
298 .db
299 .save_collection_contract(contract)
300 .map_err(|err| RedDBError::Internal(err.to_string()))?;
301 if let Some(tenant_id) = crate::runtime::impl_core::current_tenant() {
302 store.set_config_tree(
303 &format!("red.collection_tenants.{}", query.name),
304 &crate::serde_json::Value::String(tenant_id),
305 );
306 }
307 self.inner
308 .db
309 .persist_metadata()
310 .map_err(|err| RedDBError::Internal(err.to_string()))?;
311 self.invalidate_result_cache();
312
313 Ok(RuntimeQueryResult::ok_message(
314 raw_query.to_string(),
315 &format!("{label} '{}' created", query.name),
316 "create",
317 ))
318 }
319
320 pub fn ensure_system_graph_with_analytics(
329 &self,
330 name: &str,
331 outputs: &[crate::catalog::AnalyticsOutput],
332 ) -> RedDBResult<()> {
333 let store = self.inner.db.store();
334 if store.get_collection(name).is_some() {
335 return Ok(());
336 }
337 let analytics_config = outputs
338 .iter()
339 .map(|output| crate::catalog::AnalyticsViewDescriptor {
340 output: *output,
341 algorithm: None,
342 resolution: None,
343 max_iterations: None,
344 tolerance: None,
345 })
346 .collect();
347 store
348 .create_collection(name)
349 .map_err(|err| RedDBError::Internal(err.to_string()))?;
350 let contract = keyed_collection_contract(
351 name,
352 crate::catalog::CollectionModel::Graph,
353 analytics_config,
354 );
355 self.inner
356 .db
357 .save_collection_contract(contract)
358 .map_err(|err| RedDBError::Internal(err.to_string()))?;
359 self.inner
360 .db
361 .persist_metadata()
362 .map_err(|err| RedDBError::Internal(err.to_string()))?;
363 self.invalidate_result_cache_process_only();
364 Ok(())
365 }
366
367 pub fn execute_create_collection(
368 &self,
369 raw_query: &str,
370 query: &CreateCollectionQuery,
371 ) -> RedDBResult<RuntimeQueryResult> {
372 let model = match query.kind.as_str() {
373 "graph" => CollectionModel::Graph,
374 "document" => CollectionModel::Document,
375 "metrics" => CollectionModel::Metrics,
376 "vector.turbo" => {
377 let dimension = query.vector_dimension.ok_or_else(|| {
378 RedDBError::Query(
379 "CREATE COLLECTION KIND vector.turbo requires DIM".to_string(),
380 )
381 })?;
382 let create = CreateVectorQuery {
383 name: query.name.clone(),
384 dimension,
385 metric: query
386 .vector_metric
387 .unwrap_or(crate::storage::engine::distance::DistanceMetric::Cosine),
388 if_not_exists: query.if_not_exists,
389 };
390 let result = self.execute_create_vector(raw_query, &create)?;
391 let store = self.inner.db.store();
398 crate::runtime::vector_turbo_kind::mark_as_turbo(&store, &query.name);
399 self.inner
400 .db
401 .persist_metadata()
402 .map_err(|err| RedDBError::Internal(err.to_string()))?;
403 let _ = self.inner.db.turbo_state(&query.name);
408 return Ok(result);
409 }
410 "blockchain" => CollectionModel::Table,
415 other => {
416 return Err(RedDBError::Query(format!(
417 "NOT_YET_SUPPORTED: CREATE COLLECTION KIND {other} is not implemented"
418 )));
419 }
420 };
421 let create = CreateTableQuery {
422 collection_model: model,
423 name: query.name.clone(),
424 columns: Vec::new(),
425 unique_constraints: Vec::new(),
426 if_not_exists: query.if_not_exists,
427 default_ttl_ms: None,
428 metrics_rollup_policies: Vec::new(),
429 context_index_fields: Vec::new(),
430 context_index_enabled: false,
431 timestamps: false,
432 partition_by: None,
433 tenant_by: None,
434 append_only: false,
435 subscriptions: Vec::new(),
436 analytics_config: Vec::new(),
437 vault_own_master_key: false,
438 ai_policy: None,
439 };
440 let result = self.execute_create_table(raw_query, &create)?;
441 if query.kind == "blockchain" {
442 self.install_blockchain_kind(&query.name)?;
443 }
444 if !query.allowed_signers.is_empty() {
449 let actor = crate::runtime::impl_core::current_user_projected()
450 .unwrap_or_else(|| "@system/create-collection".to_string());
451 crate::runtime::signed_writes_kind::try_install(
452 &self.inner.db.store(),
453 &query.name,
454 &query.allowed_signers,
455 &actor,
456 )?;
457 }
458 Ok(result)
459 }
460
461 fn install_blockchain_kind(&self, name: &str) -> RedDBResult<()> {
465 use crate::runtime::blockchain_kind;
466 use crate::storage::unified::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
467 use std::sync::Arc;
468
469 let store = self.inner.db.store();
470 blockchain_kind::try_mark_as_chain(&store, name)?;
471
472 let existing_tip = blockchain_kind::try_chain_tip(&store, name)?;
473 if existing_tip.height.is_some() {
474 return Ok(());
475 }
476
477 let fields = blockchain_kind::genesis_fields(blockchain_kind::now_ms());
478 let named: std::collections::HashMap<String, reddb_types::Value> =
479 fields.into_iter().collect();
480 let entity = UnifiedEntity::new(
481 EntityId::new(0),
482 EntityKind::TableRow {
483 table: Arc::from(name),
484 row_id: 0,
485 },
486 EntityData::Row(RowData {
487 columns: Vec::new(),
488 named: Some(named),
489 schema: None,
490 }),
491 );
492 store
493 .insert_auto(name, entity)
494 .map_err(|err| RedDBError::Internal(err.to_string()))?;
495 if let Some(tip) = blockchain_kind::try_chain_tip_full(&store, name)? {
498 self.inner
499 .chain_tip_cache
500 .lock()
501 .insert(name.to_string(), tip);
502 }
503 Ok(())
504 }
505
506 pub fn execute_create_vector(
507 &self,
508 raw_query: &str,
509 query: &CreateVectorQuery,
510 ) -> RedDBResult<RuntimeQueryResult> {
511 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
512 if is_system_schema_name(&query.name) {
513 return Err(RedDBError::Query("system schema is read-only".to_string()));
514 }
515 let store = self.inner.db.store();
516 if store.get_collection(&query.name).is_some() {
517 if query.if_not_exists {
518 return Ok(RuntimeQueryResult::ok_message(
519 raw_query.to_string(),
520 &format!("vector '{}' already exists", query.name),
521 "create",
522 ));
523 }
524 return Err(RedDBError::Query(format!(
525 "vector '{}' already exists",
526 query.name
527 )));
528 }
529
530 store
531 .create_collection(&query.name)
532 .map_err(|err| RedDBError::Internal(err.to_string()))?;
533 self.inner
534 .db
535 .save_collection_contract(vector_collection_contract(query))
536 .map_err(|err| RedDBError::Internal(err.to_string()))?;
537 if let Some(tenant_id) = crate::runtime::impl_core::current_tenant() {
538 store.set_config_tree(
539 &format!("red.collection_tenants.{}", query.name),
540 &crate::serde_json::Value::String(tenant_id),
541 );
542 }
543 crate::runtime::vector_turbo_kind::mark_as_turbo(&store, &query.name);
552 self.inner
553 .db
554 .persist_metadata()
555 .map_err(|err| RedDBError::Internal(err.to_string()))?;
556 let _ = self.inner.db.turbo_state(&query.name);
561 self.invalidate_result_cache();
562
563 Ok(RuntimeQueryResult::ok_message(
564 raw_query.to_string(),
565 &format!("vector '{}' created", query.name),
566 "create",
567 ))
568 }
569
570 fn provision_vault_key_material(
571 &self,
572 collection: &str,
573 own_master_key: bool,
574 ) -> RedDBResult<()> {
575 let auth_store = self.inner.auth_store.read().clone().ok_or_else(|| {
576 RedDBError::Query("CREATE VAULT requires an enabled, unsealed vault".to_string())
577 })?;
578 if !auth_store.is_vault_backed() {
579 return Err(RedDBError::Query(
580 "CREATE VAULT requires an enabled, unsealed vault".to_string(),
581 ));
582 }
583
584 auth_store
585 .ensure_vault_secret_key_result()
586 .map_err(|err| RedDBError::Query(err.to_string()))?;
587
588 if own_master_key
589 && auth_store
590 .vault_kv_get(&vault_master_key_ref(collection))
591 .is_none()
592 {
593 let key = crate::auth::store::random_bytes(32);
594 auth_store
595 .vault_kv_try_set(vault_master_key_ref(collection), hex::encode(key))
596 .map_err(|err| RedDBError::Query(err.to_string()))?;
597 }
598
599 Ok(())
600 }
601
602 pub fn execute_drop_table(
606 &self,
607 raw_query: &str,
608 query: &DropTableQuery,
609 ) -> RedDBResult<RuntimeQueryResult> {
610 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
611 let store = self.inner.db.store();
612
613 if is_system_schema_name(&query.name) {
614 return Err(RedDBError::Query("system schema is read-only".to_string()));
615 }
616
617 let exists = store.get_collection(&query.name).is_some();
618 if !exists {
619 if query.if_exists {
620 return Ok(RuntimeQueryResult::ok_message(
621 raw_query.to_string(),
622 &format!("table '{}' does not exist", query.name),
623 "drop",
624 ));
625 }
626 return Err(RedDBError::NotFound(format!(
627 "table '{}' not found",
628 query.name
629 )));
630 }
631 let actual =
632 polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
633 polymorphic_resolver::ensure_model_match(CollectionModel::Table, actual)?;
634
635 let final_count = store
638 .get_collection(&query.name)
639 .map(|manager| manager.query_all(|_| true).len() as u64)
640 .unwrap_or(0);
641 crate::runtime::mutation::emit_collection_dropped_event_for_collection(
642 self,
643 &query.name,
644 final_count,
645 )?;
646
647 let topology_lock = self.inner.index_store.collection_topology_lock(&query.name);
648 let _topology_guard = topology_lock.write();
649 let orphaned_indices: Vec<String> = self
650 .inner
651 .index_store
652 .list_indices(&query.name)
653 .into_iter()
654 .map(|index| index.name)
655 .collect();
656 for name in &orphaned_indices {
657 self.inner.index_store.drop_index(name, &query.name);
658 }
659
660 store
661 .drop_collection(&query.name)
662 .map_err(|err| RedDBError::Internal(err.to_string()))?;
663 self.inner.db.invalidate_vector_index(&query.name);
664 self.inner.db.clear_collection_default_ttl_ms(&query.name);
665 self.inner
666 .db
667 .remove_collection_contract(&query.name)
668 .map_err(|err| RedDBError::Internal(err.to_string()))?;
669 self.clear_table_planner_stats(&query.name);
670 self.invalidate_result_cache();
671 if let Some(store) = self.inner.auth_store.read().clone() {
675 store.invalidate_visible_collections_cache();
676 }
677 self.schema_vocabulary_apply(
686 crate::runtime::schema_vocabulary::DdlEvent::DropCollection {
687 collection: query.name.clone(),
688 },
689 );
690
691 Ok(RuntimeQueryResult::ok_message(
692 raw_query.to_string(),
693 &format!("table '{}' dropped", query.name),
694 "drop",
695 ))
696 }
697
698 pub fn execute_drop_graph(
699 &self,
700 raw_query: &str,
701 query: &DropGraphQuery,
702 ) -> RedDBResult<RuntimeQueryResult> {
703 self.execute_drop_typed_collection(
704 raw_query,
705 &query.name,
706 query.if_exists,
707 CollectionModel::Graph,
708 "graph",
709 )
710 }
711
712 pub fn execute_drop_vector(
713 &self,
714 raw_query: &str,
715 query: &DropVectorQuery,
716 ) -> RedDBResult<RuntimeQueryResult> {
717 self.execute_drop_typed_collection(
718 raw_query,
719 &query.name,
720 query.if_exists,
721 CollectionModel::Vector,
722 "vector",
723 )
724 }
725
726 pub fn execute_drop_document(
727 &self,
728 raw_query: &str,
729 query: &DropDocumentQuery,
730 ) -> RedDBResult<RuntimeQueryResult> {
731 self.execute_drop_typed_collection(
732 raw_query,
733 &query.name,
734 query.if_exists,
735 CollectionModel::Document,
736 "document",
737 )
738 }
739
740 pub fn execute_drop_kv(
741 &self,
742 raw_query: &str,
743 query: &DropKvQuery,
744 ) -> RedDBResult<RuntimeQueryResult> {
745 let label = polymorphic_resolver::model_name(query.model);
746 self.execute_drop_typed_collection(
747 raw_query,
748 &query.name,
749 query.if_exists,
750 query.model,
751 label,
752 )
753 }
754
755 pub fn execute_drop_collection(
756 &self,
757 raw_query: &str,
758 query: &DropCollectionQuery,
759 ) -> RedDBResult<RuntimeQueryResult> {
760 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
761 if is_system_schema_name(&query.name) {
762 return Err(RedDBError::Query("system schema is read-only".to_string()));
763 }
764 let store = self.inner.db.store();
765 if store.get_collection(&query.name).is_none() {
766 if query.if_exists {
767 return Ok(RuntimeQueryResult::ok_message(
768 raw_query.to_string(),
769 &format!("collection '{}' does not exist", query.name),
770 "drop",
771 ));
772 }
773 return Err(RedDBError::NotFound(format!(
774 "collection '{}' not found",
775 query.name
776 )));
777 }
778
779 let actual =
780 polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
781 if let Some(expected) = query.model {
782 polymorphic_resolver::ensure_model_match(expected, actual)?;
783 }
784
785 match actual {
786 CollectionModel::Table => self.execute_drop_table(
787 raw_query,
788 &DropTableQuery {
789 name: query.name.clone(),
790 if_exists: query.if_exists,
791 },
792 ),
793 CollectionModel::TimeSeries => self.execute_drop_timeseries(
794 raw_query,
795 &DropTimeSeriesQuery {
796 name: query.name.clone(),
797 if_exists: query.if_exists,
798 },
799 ),
800 CollectionModel::Queue => self.execute_drop_queue(
801 raw_query,
802 &DropQueueQuery {
803 name: query.name.clone(),
804 if_exists: query.if_exists,
805 },
806 ),
807 CollectionModel::Graph => self.execute_drop_graph(
808 raw_query,
809 &DropGraphQuery {
810 name: query.name.clone(),
811 if_exists: query.if_exists,
812 },
813 ),
814 CollectionModel::Vector => self.execute_drop_vector(
815 raw_query,
816 &DropVectorQuery {
817 name: query.name.clone(),
818 if_exists: query.if_exists,
819 },
820 ),
821 CollectionModel::Document => self.execute_drop_document(
822 raw_query,
823 &DropDocumentQuery {
824 name: query.name.clone(),
825 if_exists: query.if_exists,
826 },
827 ),
828 CollectionModel::Kv => self.execute_drop_kv(
829 raw_query,
830 &DropKvQuery {
831 name: query.name.clone(),
832 if_exists: query.if_exists,
833 model: CollectionModel::Kv,
834 },
835 ),
836 CollectionModel::Config => self.execute_drop_kv(
837 raw_query,
838 &DropKvQuery {
839 name: query.name.clone(),
840 if_exists: query.if_exists,
841 model: CollectionModel::Config,
842 },
843 ),
844 CollectionModel::Vault => self.execute_drop_kv(
845 raw_query,
846 &DropKvQuery {
847 name: query.name.clone(),
848 if_exists: query.if_exists,
849 model: CollectionModel::Vault,
850 },
851 ),
852 CollectionModel::Hll => self.execute_probabilistic_command(
853 raw_query,
854 &ProbabilisticCommand::DropHll {
855 name: query.name.clone(),
856 if_exists: query.if_exists,
857 },
858 ),
859 CollectionModel::Sketch => self.execute_probabilistic_command(
860 raw_query,
861 &ProbabilisticCommand::DropSketch {
862 name: query.name.clone(),
863 if_exists: query.if_exists,
864 },
865 ),
866 CollectionModel::Filter => self.execute_probabilistic_command(
867 raw_query,
868 &ProbabilisticCommand::DropFilter {
869 name: query.name.clone(),
870 if_exists: query.if_exists,
871 },
872 ),
873 CollectionModel::Metrics => self.execute_drop_typed_collection(
874 raw_query,
875 &query.name,
876 query.if_exists,
877 CollectionModel::Metrics,
878 "metrics",
879 ),
880 CollectionModel::Mixed => self.execute_drop_typed_collection(
881 raw_query,
882 &query.name,
883 query.if_exists,
884 CollectionModel::Mixed,
885 "collection",
886 ),
887 }
888 }
889
890 fn require_graph_for_analytics(&self, name: &str) -> RedDBResult<()> {
896 let is_graph = self
897 .inner
898 .db
899 .collection_contract(name)
900 .map(|c| c.declared_model == crate::catalog::CollectionModel::Graph)
901 .unwrap_or(false);
902 if !is_graph {
903 return Err(RedDBError::Query(format!(
904 "ALTER GRAPH ... ANALYTICS: '{name}' is not a graph collection"
905 )));
906 }
907 Ok(())
908 }
909
910 pub fn execute_alter_table(
916 &self,
917 raw_query: &str,
918 query: &AlterTableQuery,
919 ) -> RedDBResult<RuntimeQueryResult> {
920 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
921 let expression_contract =
922 self.db()
923 .collection_contract(&query.name)
924 .is_some_and(|contract| {
925 crate::application::collection_contract_enforcer::has_contract_expressions(
926 &contract,
927 )
928 });
929 if query.operations.iter().any(|operation| match operation {
930 AlterOperation::AddColumn(column) => {
931 expression_contract || column.generated.is_some() || column.check.is_some()
932 }
933 AlterOperation::DropColumn(_) | AlterOperation::RenameColumn { .. } => {
934 expression_contract
935 }
936 _ => false,
937 }) {
938 return Err(RedDBError::Query("ALTER of expression schemas requires validated backfill; create a new collection and import explicitly".to_string()));
939 }
940
941 let store = self.inner.db.store();
942
943 if store.get_collection(&query.name).is_none() {
945 return Err(RedDBError::NotFound(format!(
946 "table '{}' not found",
947 query.name
948 )));
949 }
950
951 let mut messages = Vec::new();
952
953 let fields_added: Vec<String> = query
955 .operations
956 .iter()
957 .filter_map(|op| {
958 if let AlterOperation::AddColumn(col) = op {
959 Some(col.name.clone())
960 } else {
961 None
962 }
963 })
964 .collect();
965 let fields_removed: Vec<String> = query
966 .operations
967 .iter()
968 .filter_map(|op| {
969 if let AlterOperation::DropColumn(name) = op {
970 Some(name.clone())
971 } else {
972 None
973 }
974 })
975 .collect();
976
977 for op in &query.operations {
978 match op {
979 AlterOperation::AddColumn(col) => {
980 messages.push(format!("column '{}' added", col.name));
982 }
983 AlterOperation::DropColumn(name) => {
984 messages.push(format!("column '{}' dropped", name));
985 }
986 AlterOperation::RenameColumn { from, to } => {
987 messages.push(format!("column '{}' renamed to '{}'", from, to));
988 }
989 AlterOperation::AttachPartition { child, bound } => {
990 store.set_config_tree(
994 &format!("partition.{}.children.{}", query.name, child),
995 &crate::serde_json::Value::String(bound.clone()),
996 );
997 messages.push(format!(
998 "partition '{child}' attached to '{}' ({bound})",
999 query.name
1000 ));
1001 }
1002 AlterOperation::DetachPartition { child } => {
1003 store.set_config_tree(
1004 &format!("partition.{}.children.{}", query.name, child),
1005 &crate::serde_json::Value::Null,
1006 );
1007 messages.push(format!(
1008 "partition '{child}' detached from '{}'",
1009 query.name
1010 ));
1011 }
1012 AlterOperation::EnableRowLevelSecurity => {
1013 self.inner
1014 .rls_enabled_tables
1015 .write()
1016 .insert(query.name.clone());
1017 store.set_config_tree(
1019 &format!("rls.enabled.{}", query.name),
1020 &crate::serde_json::Value::Bool(true),
1021 );
1022 self.invalidate_plan_cache();
1023 messages.push(format!("row level security enabled on '{}'", query.name));
1024 self.invalidate_result_cache();
1025 }
1026 AlterOperation::DisableRowLevelSecurity => {
1027 self.inner.rls_enabled_tables.write().remove(&query.name);
1028 store.set_config_tree(
1029 &format!("rls.enabled.{}", query.name),
1030 &crate::serde_json::Value::Null,
1031 );
1032 self.invalidate_plan_cache();
1033 messages.push(format!("row level security disabled on '{}'", query.name));
1034 self.invalidate_result_cache();
1035 }
1036 AlterOperation::EnableTenancy { column } => {
1038 store.set_config_tree(
1039 &format!("tenant_tables.{}.column", query.name),
1040 &crate::serde_json::Value::String(column.clone()),
1041 );
1042 self.register_tenant_table(&query.name, column);
1043 self.invalidate_plan_cache();
1044 messages.push(format!(
1045 "tenancy enabled on '{}' by column '{column}'",
1046 query.name
1047 ));
1048 }
1049 AlterOperation::DisableTenancy => {
1050 store.set_config_tree(
1051 &format!("tenant_tables.{}.column", query.name),
1052 &crate::serde_json::Value::Null,
1053 );
1054 self.unregister_tenant_table(&query.name);
1055 self.invalidate_plan_cache();
1056 messages.push(format!("tenancy disabled on '{}'", query.name));
1057 }
1058 AlterOperation::SetAppendOnly(on) => {
1059 messages.push(format!(
1064 "append_only {} on '{}'",
1065 if *on { "enabled" } else { "disabled" },
1066 query.name
1067 ));
1068 }
1069 AlterOperation::SetVersioned(on) => {
1070 self.vcs_set_versioned(&query.name, *on)?;
1077 messages.push(format!(
1078 "versioned {} on '{}'",
1079 if *on { "enabled" } else { "disabled" },
1080 query.name
1081 ));
1082 }
1083 AlterOperation::EnableEvents(subscription) => {
1084 let mut subscription = subscription.clone();
1085 subscription.source = query.name.clone();
1086 validate_event_subscriptions(
1087 self,
1088 &query.name,
1089 std::slice::from_ref(&subscription),
1090 )?;
1091 ensure_event_target_queue(self, &subscription.target_queue)?;
1092 messages.push(format!(
1093 "events enabled on '{}' to '{}'",
1094 query.name, subscription.target_queue
1095 ));
1096 }
1097 AlterOperation::DisableEvents => {
1098 messages.push(format!("events disabled on '{}'", query.name));
1099 }
1100 AlterOperation::AddSubscription { name, descriptor } => {
1101 let mut sub = descriptor.clone();
1102 sub.name = name.clone();
1103 sub.source = query.name.clone();
1104 validate_event_subscriptions(self, &query.name, std::slice::from_ref(&sub))?;
1105 ensure_event_target_queue(self, &sub.target_queue)?;
1106 messages.push(format!(
1107 "subscription '{}' added on '{}' to '{}'",
1108 name, query.name, sub.target_queue
1109 ));
1110 }
1111 AlterOperation::DropSubscription { name } => {
1112 messages.push(format!(
1113 "subscription '{}' dropped on '{}'",
1114 name, query.name
1115 ));
1116 }
1117 AlterOperation::AddSigner { pubkey } => {
1118 if !crate::runtime::signed_writes_kind::try_is_signed(&store, &query.name)? {
1125 return Err(RedDBError::Query(format!(
1126 "ALTER COLLECTION ADD SIGNER: '{}' has no signer registry; \
1127 recreate it with CREATE COLLECTION ... SIGNED_BY (...)",
1128 query.name
1129 )));
1130 }
1131 let actor = crate::runtime::impl_core::current_user_projected()
1132 .unwrap_or_else(|| "@system/alter".to_string());
1133 let changed = crate::runtime::signed_writes_kind::try_add_signer(
1134 &store,
1135 &query.name,
1136 *pubkey,
1137 &actor,
1138 )?;
1139 messages.push(format!(
1140 "signer {} on '{}'",
1141 if changed { "added" } else { "already present" },
1142 query.name
1143 ));
1144 }
1145 AlterOperation::RevokeSigner { pubkey } => {
1146 if !crate::runtime::signed_writes_kind::try_is_signed(&store, &query.name)? {
1147 return Err(RedDBError::Query(format!(
1148 "ALTER COLLECTION REVOKE SIGNER: '{}' has no signer registry",
1149 query.name
1150 )));
1151 }
1152 let actor = crate::runtime::impl_core::current_user_projected()
1153 .unwrap_or_else(|| "@system/alter".to_string());
1154 let changed = crate::runtime::signed_writes_kind::try_revoke_signer(
1155 &store,
1156 &query.name,
1157 pubkey,
1158 &actor,
1159 )?;
1160 messages.push(format!(
1161 "signer {} on '{}'",
1162 if changed {
1163 "revoked"
1164 } else {
1165 "already revoked"
1166 },
1167 query.name
1168 ));
1169 }
1170 AlterOperation::SetRetention { duration_ms } => {
1171 let existing = self.inner.db.collection_contract(&query.name);
1177 let has_ts_column = existing
1178 .as_ref()
1179 .map(retention_timestamp_column_exists)
1180 .unwrap_or(false);
1181 if !has_ts_column {
1182 return Err(RedDBError::Query(format!(
1183 "ALTER COLLECTION SET RETENTION: '{}' has no timestamp \
1184 column — declare a TIMESTAMP/TIMESTAMPMS/DATETIME column \
1185 or enable WITH timestamps = true before setting a \
1186 retention policy",
1187 query.name
1188 )));
1189 }
1190 messages.push(format!(
1191 "retention set to {duration_ms} ms on '{}'",
1192 query.name
1193 ));
1194 }
1195 AlterOperation::UnsetRetention => {
1196 messages.push(format!("retention cleared on '{}'", query.name));
1197 }
1198 AlterOperation::AddAnalytics(views) => {
1199 self.require_graph_for_analytics(&query.name)?;
1204 let existing = self.inner.db.collection_contract(&query.name);
1205 let enabled: std::collections::BTreeSet<crate::catalog::AnalyticsOutput> =
1206 existing
1207 .as_ref()
1208 .map(|c| c.analytics_config.iter().map(|v| v.output).collect())
1209 .unwrap_or_default();
1210 for view in views {
1211 if enabled.contains(&view.output) {
1212 messages.push(format!(
1215 "analytics '{}' already enabled on '{}'",
1216 view.output.as_str(),
1217 query.name
1218 ));
1219 } else {
1220 messages.push(format!(
1221 "analytics '{}' enabled on '{}'",
1222 view.output.as_str(),
1223 query.name
1224 ));
1225 }
1226 }
1227 }
1228 AlterOperation::DropAnalytics(output) => {
1229 self.require_graph_for_analytics(&query.name)?;
1230 let enabled = self
1231 .inner
1232 .db
1233 .collection_contract(&query.name)
1234 .map(|c| c.analytics_config.iter().any(|v| v.output == *output))
1235 .unwrap_or(false);
1236 if !enabled {
1237 return Err(RedDBError::Query(format!(
1240 "ALTER GRAPH DROP ANALYTICS: analytics output '{}' is not enabled on graph '{}'",
1241 output.as_str(),
1242 query.name
1243 )));
1244 }
1245 messages.push(format!(
1246 "analytics '{}' disabled on '{}'",
1247 output.as_str(),
1248 query.name
1249 ));
1250 }
1251 }
1252 }
1253
1254 let mut contract = self
1255 .inner
1256 .db
1257 .collection_contract(&query.name)
1258 .unwrap_or_else(|| default_collection_contract_for_existing_table(&query.name));
1259 apply_alter_operations_to_contract(&mut contract, &query.operations);
1260 contract.version = contract.version.saturating_add(1);
1261 contract.updated_at_unix_ms = current_unix_ms();
1262 self.inner
1263 .db
1264 .save_collection_contract(contract)
1265 .map_err(|err| RedDBError::Internal(err.to_string()))?;
1266 if !fields_added.is_empty() || !fields_removed.is_empty() {
1270 let sub_names: Vec<String> = self
1271 .inner
1272 .db
1273 .collection_contract(&query.name)
1274 .map(|c| {
1275 c.subscriptions
1276 .iter()
1277 .filter(|s| s.enabled)
1278 .map(|s| s.name.clone())
1279 .collect()
1280 })
1281 .unwrap_or_default();
1282 if !sub_names.is_empty() {
1283 crate::telemetry::operator_event::OperatorEvent::SubscriptionSchemaChange {
1284 collection: query.name.clone(),
1285 subscription_names: sub_names.join(", "),
1286 fields_added: fields_added.join(", "),
1287 fields_removed: fields_removed.join(", "),
1288 lsn: self.cdc_current_lsn(),
1289 }
1290 .emit_global();
1291 }
1292 }
1293
1294 self.clear_table_planner_stats(&query.name);
1295 self.invalidate_result_cache();
1296 let post_alter_columns: Vec<String> = self
1301 .inner
1302 .db
1303 .collection_contract(&query.name)
1304 .map(|contract| {
1305 contract
1306 .declared_columns
1307 .iter()
1308 .map(|col| col.name.clone())
1309 .collect()
1310 })
1311 .unwrap_or_default();
1312 self.schema_vocabulary_apply(
1313 crate::runtime::schema_vocabulary::DdlEvent::AlterCollection {
1314 collection: query.name.clone(),
1315 columns: post_alter_columns,
1316 type_tags: Vec::new(),
1317 description: None,
1318 },
1319 );
1320
1321 let message = if messages.is_empty() {
1322 format!("table '{}' altered (no operations)", query.name)
1323 } else {
1324 format!("table '{}' altered: {}", query.name, messages.join(", "))
1325 };
1326
1327 Ok(RuntimeQueryResult::ok_message(
1328 raw_query.to_string(),
1329 &message,
1330 "alter",
1331 ))
1332 }
1333
1334 pub fn execute_create_vcs_ref(
1335 &self,
1336 raw_query: &str,
1337 query: &CreateVcsRefQuery,
1338 ) -> RedDBResult<RuntimeQueryResult> {
1339 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1340 let connection_id = crate::runtime::impl_core::current_connection_id();
1341 match query.kind {
1342 VcsRefKind::Branch => {
1343 self.vcs_branch_create(crate::application::vcs::CreateBranchInput {
1344 name: query.name.clone(),
1345 from: query.target.clone(),
1346 connection_id,
1347 })?;
1348 }
1349 VcsRefKind::Tag => {
1350 let target = match &query.target {
1351 Some(target) => target.clone(),
1352 None => self
1353 .vcs_status(crate::application::vcs::StatusInput { connection_id })?
1354 .head_commit
1355 .filter(|head| !head.is_empty())
1356 .ok_or_else(|| {
1357 RedDBError::InvalidConfig(
1358 "cannot create tag: HEAD has no commits".to_string(),
1359 )
1360 })?,
1361 };
1362 self.vcs_tag_create(crate::application::vcs::CreateTagInput {
1363 name: query.name.clone(),
1364 target,
1365 annotation: None,
1366 })?;
1367 }
1368 }
1369 let kind = match query.kind {
1370 VcsRefKind::Branch => "branch",
1371 VcsRefKind::Tag => "tag",
1372 };
1373 Ok(RuntimeQueryResult::ok_message(
1374 raw_query.to_string(),
1375 &format!("{kind} '{}' created", query.name),
1376 "create",
1377 ))
1378 }
1379
1380 pub fn execute_drop_vcs_ref(
1381 &self,
1382 raw_query: &str,
1383 query: &DropVcsRefQuery,
1384 ) -> RedDBResult<RuntimeQueryResult> {
1385 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1386 match query.kind {
1387 VcsRefKind::Branch => self.vcs_branch_delete(&query.name)?,
1388 VcsRefKind::Tag => self.vcs_tag_delete(&query.name)?,
1389 }
1390 let kind = match query.kind {
1391 VcsRefKind::Branch => "branch",
1392 VcsRefKind::Tag => "tag",
1393 };
1394 Ok(RuntimeQueryResult::ok_message(
1395 raw_query.to_string(),
1396 &format!("{kind} '{}' dropped", query.name),
1397 "drop",
1398 ))
1399 }
1400
1401 pub fn execute_fork_store(
1402 &self,
1403 raw_query: &str,
1404 query: &ForkStoreQuery,
1405 ) -> RedDBResult<RuntimeQueryResult> {
1406 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1407 let fork = self.fork_store_at_lsn(&query.name, query.at_lsn)?;
1408 let path = self
1409 .fork_store_path(&query.name)?
1410 .ok_or_else(|| RedDBError::Internal("created fork path is missing".into()))?;
1411 Ok(RuntimeQueryResult::ok_message(
1412 raw_query.to_string(),
1413 &format!(
1414 "store fork '{}' created at LSN {}; database {}",
1415 fork.name,
1416 fork.fork_lsn,
1417 path.display()
1418 ),
1419 "fork_store",
1420 ))
1421 }
1422
1423 pub fn execute_drop_fork(
1424 &self,
1425 raw_query: &str,
1426 query: &DropForkQuery,
1427 ) -> RedDBResult<RuntimeQueryResult> {
1428 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1429 if !self.drop_fork_store(&query.name)? && !query.if_exists {
1430 return Err(RedDBError::NotFound(format!("store fork '{}'", query.name)));
1431 }
1432 Ok(RuntimeQueryResult::ok_message(
1433 raw_query.to_string(),
1434 &format!("store fork '{}' dropped", query.name),
1435 "drop_fork",
1436 ))
1437 }
1438
1439 pub fn execute_promote_fork(
1440 &self,
1441 raw_query: &str,
1442 query: &PromoteForkQuery,
1443 ) -> RedDBResult<RuntimeQueryResult> {
1444 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1445 let path = self.promote_fork_store(&query.name)?;
1446 Ok(RuntimeQueryResult::ok_message(raw_query.to_string(),
1447 &format!("store fork '{}' promoted; reopen the server with database {}; the previous database is preserved", query.name, path.display()), "promote_fork"))
1448 }
1449
1450 pub fn execute_explain_alter(
1457 &self,
1458 raw_query: &str,
1459 query: &ExplainAlterQuery,
1460 ) -> RedDBResult<RuntimeQueryResult> {
1461 let analyzed = analyze_create_table(&query.target)
1465 .map_err(|err| RedDBError::Query(err.to_string()))?;
1466
1467 let current_contract = self.inner.db.collection_contract(&query.target.name);
1468
1469 let current_unique = current_contract
1472 .as_ref()
1473 .map(|contract| {
1474 contract
1475 .table_def
1476 .iter()
1477 .flat_map(|table| &table.constraints)
1478 .filter(|constraint| {
1479 constraint.constraint_type == reddb_types::ConstraintType::Unique
1480 && !contract.declared_columns.iter().any(|column| {
1481 column.unique
1482 && constraint.columns == [column.name.clone()]
1483 && constraint.name == format!("uniq_{}", column.name)
1484 })
1485 })
1486 .map(|constraint| (constraint.name.clone(), constraint.columns.clone()))
1487 .collect::<BTreeSet<_>>()
1488 })
1489 .unwrap_or_default();
1490 let target_unique = analyzed
1491 .unique_constraints
1492 .into_iter()
1493 .map(|constraint| (constraint.name, constraint.columns))
1494 .collect::<BTreeSet<_>>();
1495 if current_unique != target_unique {
1496 return Err(RedDBError::Query("NOT_YET_SUPPORTED: EXPLAIN ALTER cannot plan changes to table-level UNIQUE constraints".to_string()));
1497 }
1498
1499 let current_columns: Vec<crate::physical::DeclaredColumnContract> = current_contract
1500 .as_ref()
1501 .map(|c| c.declared_columns.clone())
1502 .unwrap_or_default();
1503
1504 let diff = super::schema_diff::compute_column_diff(
1505 &query.target.name,
1506 ¤t_columns,
1507 &query.target.columns,
1508 );
1509
1510 let rendered = match query.format {
1511 ExplainFormat::Sql => super::schema_diff::format_as_sql(&diff),
1512 ExplainFormat::Json => super::schema_diff::format_as_json(&diff),
1513 };
1514
1515 let format_label = match query.format {
1516 ExplainFormat::Sql => "sql",
1517 ExplainFormat::Json => "json",
1518 };
1519
1520 let columns = vec![
1521 "table".to_string(),
1522 "format".to_string(),
1523 "diff".to_string(),
1524 ];
1525 let row = vec![
1526 ("table".to_string(), Value::text(query.target.name.clone())),
1527 ("format".to_string(), Value::text(format_label.to_string())),
1528 ("diff".to_string(), Value::text(rendered)),
1529 ];
1530
1531 Ok(RuntimeQueryResult::ok_records(
1532 raw_query.to_string(),
1533 columns,
1534 vec![row],
1535 "explain",
1536 ))
1537 }
1538
1539 pub fn execute_create_index(
1544 &self,
1545 raw_query: &str,
1546 query: &CreateIndexQuery,
1547 ) -> RedDBResult<RuntimeQueryResult> {
1548 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1549 let topology_lock = self
1550 .inner
1551 .index_store
1552 .collection_topology_lock(&query.table);
1553 let _topology_guard = topology_lock.write();
1554 let store = self.inner.db.store();
1555
1556 let manager = store
1558 .get_collection(&query.table)
1559 .ok_or_else(|| RedDBError::NotFound(format!("table '{}' not found", query.table)))?;
1560
1561 if query.if_not_exists
1562 && self
1563 .inner
1564 .index_store
1565 .list_indices(&query.table)
1566 .iter()
1567 .any(|index| index.name == query.name)
1568 {
1569 return Ok(RuntimeQueryResult::ok_message(
1570 raw_query.to_string(),
1571 &format!(
1572 "index '{}' already exists on '{}'; skipped",
1573 query.name, query.table
1574 ),
1575 "create",
1576 ));
1577 }
1578
1579 let method_kind = match query.method {
1580 IndexMethod::Hash => super::index_store::IndexMethodKind::Hash,
1581 IndexMethod::BTree => super::index_store::IndexMethodKind::BTree,
1582 IndexMethod::Bitmap => super::index_store::IndexMethodKind::Bitmap,
1583 IndexMethod::Spatial => super::index_store::IndexMethodKind::default_spatial(),
1586 IndexMethod::H3 { resolution } => {
1587 super::index_store::IndexMethodKind::H3 { resolution }
1588 }
1589 };
1590
1591 let entities = manager.query_all(|_| true);
1601 let mut entity_fields: Vec<(crate::storage::unified::EntityId, Vec<(String, Value)>)> =
1602 entities
1603 .iter()
1604 .map(|e| {
1605 let fields = match &e.data {
1606 crate::storage::EntityData::Row(row) => {
1607 if let Some(ref named) = row.named {
1608 named.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
1609 } else if let Some(ref schema) = row.schema {
1610 schema
1614 .iter()
1615 .zip(row.columns.iter())
1616 .map(|(k, v)| (k.clone(), v.clone()))
1617 .collect()
1618 } else {
1619 Vec::new()
1620 }
1621 }
1622 crate::storage::EntityData::Node(node) => node
1623 .properties
1624 .iter()
1625 .map(|(k, v)| (k.clone(), v.clone()))
1626 .collect(),
1627 crate::storage::EntityData::TimeSeries(ts) => ts
1628 .fields
1629 .iter()
1630 .map(|(k, v)| (k.clone(), v.clone()))
1631 .collect(),
1632 _ => Vec::new(),
1633 };
1634 (e.id, fields)
1635 })
1636 .collect();
1637
1638 if query.unique
1639 && matches!(
1640 method_kind,
1641 super::index_store::IndexMethodKind::Hash
1642 | super::index_store::IndexMethodKind::BTree
1643 )
1644 {
1645 let snapshots = self.snapshot_manager();
1650 let own_xids = self.own_transaction_xids();
1651 let mut live = entities.iter().map(|entity| {
1652 snapshots.row_reserves_unique_key(entity.xmin, entity.xmax, &own_xids)
1653 });
1654 entity_fields.retain(|_| live.next().unwrap_or(false));
1655 if let Some((earlier, later)) =
1656 super::index_store::first_duplicate_unique_key(&query.columns, &entity_fields)
1657 {
1658 return Err(RedDBError::Query(format!(
1659 "cannot create unique index '{}' on '{}': rows {} and {} share the same ({}) key",
1660 query.name,
1661 query.table,
1662 earlier.raw(),
1663 later.raw(),
1664 query.columns.join(", ")
1665 )));
1666 }
1667 }
1668
1669 let index_growth_rows = entity_fields
1670 .iter()
1671 .map(|(_, fields)| fields.clone())
1672 .collect::<Vec<_>>();
1673 let _reservation = self.admit_non_evictable_growth(
1674 crate::storage::memory_pools::MemoryPool::IndexMemory,
1675 &format!("create index {}", query.name),
1676 crate::runtime::memory_admission::estimate_index_growth(
1677 &index_growth_rows,
1678 &query.columns,
1679 ),
1680 )?;
1681
1682 let indexed_count = self
1684 .inner
1685 .index_store
1686 .create_index(
1687 &query.name,
1688 &query.table,
1689 &query.columns,
1690 method_kind,
1691 query.unique,
1692 &entity_fields,
1693 )
1694 .map_err(RedDBError::Internal)?;
1695
1696 let analyzed = crate::storage::query::planner::stats_catalog::analyze_entity_fields(
1697 &query.table,
1698 &entity_fields,
1699 );
1700 crate::storage::query::planner::stats_catalog::persist_table_stats(&store, &analyzed);
1701 self.invalidate_plan_cache();
1702
1703 self.inner
1705 .index_store
1706 .register(super::index_store::RegisteredIndex {
1707 name: query.name.clone(),
1708 collection: query.table.clone(),
1709 columns: query.columns.clone(),
1710 method: method_kind,
1711 unique: query.unique,
1712 });
1713 self.persist_runtime_index_descriptor(super::index_store::RegisteredIndex {
1714 name: query.name.clone(),
1715 collection: query.table.clone(),
1716 columns: query.columns.clone(),
1717 method: method_kind,
1718 unique: query.unique,
1719 })?;
1720 self.schema_vocabulary_apply(crate::runtime::schema_vocabulary::DdlEvent::CreateIndex {
1724 collection: query.table.clone(),
1725 index: query.name.clone(),
1726 columns: query.columns.clone(),
1727 });
1728
1729 let method_str = format!("{}", query.method);
1730 let unique_str = if query.unique { "unique " } else { "" };
1731 let cols = query.columns.join(", ");
1732 let total_count = entity_fields.len();
1733 let coverage_detail =
1734 if matches!(method_kind, super::index_store::IndexMethodKind::H3 { .. })
1735 && total_count > 0
1736 && indexed_count == 0
1737 {
1738 format!(
1739 "{} of {} entities indexed — no indexable geo value in '{}'; expected {}",
1740 indexed_count,
1741 total_count,
1742 cols,
1743 crate::geo::RECOGNIZED_GEO_SHAPES
1744 )
1745 } else if matches!(method_kind, super::index_store::IndexMethodKind::H3 { .. }) {
1746 format!("{} of {} entities indexed", indexed_count, total_count)
1747 } else {
1748 format!("{} entities indexed", indexed_count)
1749 };
1750
1751 Ok(RuntimeQueryResult::ok_message(
1752 raw_query.to_string(),
1753 &format!(
1754 "{}index '{}' created on '{}' ({}) using {} ({})",
1755 unique_str, query.name, query.table, cols, method_str, coverage_detail
1756 ),
1757 "create",
1758 ))
1759 }
1760
1761 pub fn execute_drop_index(
1765 &self,
1766 raw_query: &str,
1767 query: &DropIndexQuery,
1768 ) -> RedDBResult<RuntimeQueryResult> {
1769 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1770 let topology_lock = self
1771 .inner
1772 .index_store
1773 .collection_topology_lock(&query.table);
1774 let _topology_guard = topology_lock.write();
1775 let store = self.inner.db.store();
1776
1777 if store.get_collection(&query.table).is_none() {
1779 if query.if_exists {
1780 return Ok(RuntimeQueryResult::ok_message(
1781 raw_query.to_string(),
1782 &format!("table '{}' does not exist", query.table),
1783 "drop",
1784 ));
1785 }
1786 return Err(RedDBError::NotFound(format!(
1787 "table '{}' not found",
1788 query.table
1789 )));
1790 }
1791
1792 self.inner.index_store.drop_index(&query.name, &query.table);
1794 self.persist_runtime_index_drop(&query.table, &query.name)?;
1795 self.invalidate_plan_cache();
1796 self.schema_vocabulary_apply(crate::runtime::schema_vocabulary::DdlEvent::DropIndex {
1798 collection: query.table.clone(),
1799 index: query.name.clone(),
1800 });
1801
1802 Ok(RuntimeQueryResult::ok_message(
1803 raw_query.to_string(),
1804 &format!("index '{}' dropped from '{}'", query.name, query.table),
1805 "drop",
1806 ))
1807 }
1808
1809 fn execute_drop_typed_collection(
1810 &self,
1811 raw_query: &str,
1812 name: &str,
1813 if_exists: bool,
1814 expected_model: CollectionModel,
1815 label: &str,
1816 ) -> RedDBResult<RuntimeQueryResult> {
1817 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1818 if is_system_schema_name(name) {
1819 return Err(RedDBError::Query("system schema is read-only".to_string()));
1820 }
1821 let store = self.inner.db.store();
1822 if store.get_collection(name).is_none() {
1823 if if_exists {
1824 return Ok(RuntimeQueryResult::ok_message(
1825 raw_query.to_string(),
1826 &format!("{label} '{name}' does not exist"),
1827 "drop",
1828 ));
1829 }
1830 return Err(RedDBError::NotFound(format!("{label} '{name}' not found")));
1831 }
1832
1833 let actual = self
1834 .inner
1835 .db
1836 .collection_contract(name)
1837 .map(|contract| contract.declared_model)
1838 .map(Ok)
1839 .unwrap_or_else(|| {
1840 polymorphic_resolver::resolve(name, &self.inner.db.catalog_model_snapshot())
1841 })?;
1842 polymorphic_resolver::ensure_model_match(expected_model, actual)?;
1843 self.drop_collection_storage(raw_query, name, label)
1844 }
1845
1846 pub fn execute_truncate(
1847 &self,
1848 raw_query: &str,
1849 query: &TruncateQuery,
1850 ) -> RedDBResult<RuntimeQueryResult> {
1851 self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1852 if is_system_schema_name(&query.name) {
1853 return Err(RedDBError::Query("system schema is read-only".to_string()));
1854 }
1855
1856 let label = query
1857 .model
1858 .map(polymorphic_resolver::model_name)
1859 .unwrap_or("collection");
1860 let store = self.inner.db.store();
1861 if store.get_collection(&query.name).is_none() {
1862 if query.if_exists {
1863 return Ok(RuntimeQueryResult::ok_message(
1864 raw_query.to_string(),
1865 &format!("{label} '{}' does not exist", query.name),
1866 "truncate",
1867 ));
1868 }
1869 return Err(RedDBError::NotFound(format!(
1870 "{label} '{}' not found",
1871 query.name
1872 )));
1873 }
1874
1875 let actual =
1876 polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
1877 if let Some(expected) = query.model {
1878 polymorphic_resolver::ensure_model_match(expected, actual)?;
1879 }
1880
1881 if actual == CollectionModel::Queue {
1882 return self.execute_queue_command(
1883 raw_query,
1884 &QueueCommand::Purge {
1885 queue: query.name.clone(),
1886 },
1887 );
1888 }
1889
1890 let affected = self.truncate_collection_entities(&query.name)?;
1892 crate::runtime::mutation::emit_truncate_event_for_collection(self, &query.name, affected)?;
1894 self.inner.db.invalidate_vector_index(&query.name);
1895 self.clear_table_planner_stats(&query.name);
1896 self.invalidate_result_cache();
1897
1898 Ok(RuntimeQueryResult::ok_message(
1899 raw_query.to_string(),
1900 &format!(
1901 "{affected} entities truncated from {label} '{}'",
1902 query.name
1903 ),
1904 "truncate",
1905 ))
1906 }
1907
1908 fn truncate_collection_entities(&self, name: &str) -> RedDBResult<u64> {
1909 let _indexed_mutation = self.index_store_ref().begin_indexed_mutation();
1910 let store = self.inner.db.store();
1911 let Some(manager) = store.get_collection(name) else {
1912 return Ok(0);
1913 };
1914 let entities = manager.query_all(|_| true);
1915 if entities.is_empty() {
1916 return Ok(0);
1917 }
1918
1919 for entity in &entities {
1920 let fields = entity_index_fields(&entity.data);
1921 self.inner
1922 .index_store
1923 .index_entity_delete(name, entity.id, &fields)
1924 .map_err(RedDBError::Internal)?;
1925 }
1926
1927 let ids = entities.iter().map(|entity| entity.id).collect::<Vec<_>>();
1928 let deleted_ids = store
1929 .delete_batch(name, &ids)
1930 .map_err(|err| RedDBError::Internal(err.to_string()))?;
1931 for id in &deleted_ids {
1932 store.context_index().remove_entity(*id);
1933 }
1934 Ok(deleted_ids.len() as u64)
1935 }
1936
1937 fn drop_collection_storage(
1938 &self,
1939 raw_query: &str,
1940 name: &str,
1941 label: &str,
1942 ) -> RedDBResult<RuntimeQueryResult> {
1943 let store = self.inner.db.store();
1944
1945 let final_count = store
1948 .get_collection(name)
1949 .map(|manager| manager.query_all(|_| true).len() as u64)
1950 .unwrap_or(0);
1951 crate::runtime::mutation::emit_collection_dropped_event_for_collection(
1952 self,
1953 name,
1954 final_count,
1955 )?;
1956
1957 let topology_lock = self.inner.index_store.collection_topology_lock(name);
1958 let _topology_guard = topology_lock.write();
1959 let orphaned_indices: Vec<String> = self
1960 .inner
1961 .index_store
1962 .list_indices(name)
1963 .into_iter()
1964 .map(|index| index.name)
1965 .collect();
1966 for index_name in &orphaned_indices {
1967 self.inner.index_store.drop_index(index_name, name);
1968 }
1969
1970 store
1971 .drop_collection(name)
1972 .map_err(|err| RedDBError::Internal(err.to_string()))?;
1973 self.inner.db.invalidate_vector_index(name);
1974 self.inner.db.clear_collection_default_ttl_ms(name);
1975 self.inner
1976 .db
1977 .remove_collection_contract(name)
1978 .map_err(|err| RedDBError::Internal(err.to_string()))?;
1979 self.clear_table_planner_stats(name);
1980 self.invalidate_result_cache();
1981 if let Some(store) = self.inner.auth_store.read().clone() {
1982 store.invalidate_visible_collections_cache();
1983 }
1984 self.schema_vocabulary_apply(
1988 crate::runtime::schema_vocabulary::DdlEvent::DropCollection {
1989 collection: name.to_string(),
1990 },
1991 );
1992
1993 Ok(RuntimeQueryResult::ok_message(
1994 raw_query.to_string(),
1995 &format!("{label} '{name}' dropped"),
1996 "drop",
1997 ))
1998 }
1999}
2000
2001pub(crate) fn is_system_schema_name(name: &str) -> bool {
2002 name == "red" || name.starts_with("red.") || name.starts_with("__red_schema_")
2003}
2004
2005fn entity_index_fields(data: &EntityData) -> Vec<(String, Value)> {
2006 match data {
2007 EntityData::Row(row) => {
2008 if let Some(ref named) = row.named {
2009 named
2010 .iter()
2011 .map(|(key, value)| (key.clone(), value.clone()))
2012 .collect()
2013 } else if let Some(ref schema) = row.schema {
2014 schema
2015 .iter()
2016 .zip(row.columns.iter())
2017 .map(|(key, value)| (key.clone(), value.clone()))
2018 .collect()
2019 } else {
2020 Vec::new()
2021 }
2022 }
2023 EntityData::Node(node) => node
2024 .properties
2025 .iter()
2026 .map(|(key, value)| (key.clone(), value.clone()))
2027 .collect(),
2028 _ => Vec::new(),
2029 }
2030}
2031
2032fn validate_ai_policy_modalities(policy: &crate::catalog::AiPolicy) -> RedDBResult<()> {
2038 use crate::runtime::ai::provider_capabilities::{Modality, Registry};
2039 let registry = Registry::new();
2040 let check = |provider: &str, model: &str, modality: Modality| -> RedDBResult<()> {
2041 registry
2042 .validate_policy_modality(provider, model, modality)
2043 .map_err(|err| RedDBError::Query(err.to_string()))
2044 };
2045 if let Some(embed) = &policy.embed {
2046 check(&embed.provider, &embed.model, Modality::Embed)?;
2047 }
2048 if let Some(moderate) = &policy.moderate {
2049 check(&moderate.provider, &moderate.model, Modality::Moderate)?;
2050 }
2051 if let Some(vision) = &policy.vision {
2052 check(&vision.provider, &vision.model, Modality::Vision)?;
2053 }
2054 Ok(())
2055}
2056
2057fn collection_contract_from_create_table(
2058 query: &CreateTableQuery,
2059) -> RedDBResult<crate::physical::CollectionContract> {
2060 reddb_engine::catalog::ddl::collection_contract_from_create_table(query, current_unix_ms())
2061 .map_err(RedDBError::Query)
2062}
2063
2064fn default_collection_contract_for_existing_table(
2065 name: &str,
2066) -> crate::physical::CollectionContract {
2067 let now = current_unix_ms();
2068 crate::physical::CollectionContract {
2069 name: name.to_string(),
2070 declared_model: crate::catalog::CollectionModel::Table,
2071 schema_mode: crate::catalog::SchemaMode::SemiStructured,
2072 origin: crate::physical::ContractOrigin::Explicit,
2073 version: 0,
2074 created_at_unix_ms: now,
2075 updated_at_unix_ms: now,
2076 default_ttl_ms: None,
2077 vector_dimension: None,
2078 vector_metric: None,
2079 context_index_fields: Vec::new(),
2080 declared_columns: Vec::new(),
2081 table_def: Some(reddb_types::TableDef::new(name.to_string())),
2082 timestamps_enabled: false,
2083 context_index_enabled: false,
2084 metrics_raw_retention_ms: None,
2085 metrics_rollup_policies: Vec::new(),
2086 metrics_tenant_identity: None,
2087 metrics_namespace: None,
2088 append_only: false,
2089 subscriptions: Vec::new(),
2090 analytics_config: Vec::new(),
2091 session_key: None,
2092 session_gap_ms: None,
2093 retention_duration_ms: None,
2094 analytical_storage: None,
2095
2096 ai_policy: None,
2097 }
2098}
2099
2100fn keyed_collection_contract(
2101 name: &str,
2102 model: crate::catalog::CollectionModel,
2103 analytics_config: Vec<crate::catalog::AnalyticsViewDescriptor>,
2104) -> crate::physical::CollectionContract {
2105 let now = current_unix_ms();
2106 crate::physical::CollectionContract {
2107 name: name.to_string(),
2108 declared_model: model,
2109 schema_mode: crate::catalog::SchemaMode::Dynamic,
2110 origin: crate::physical::ContractOrigin::Explicit,
2111 version: 1,
2112 created_at_unix_ms: now,
2113 updated_at_unix_ms: now,
2114 default_ttl_ms: None,
2115 vector_dimension: None,
2116 vector_metric: None,
2117 context_index_fields: Vec::new(),
2118 declared_columns: Vec::new(),
2119 table_def: None,
2120 timestamps_enabled: false,
2121 context_index_enabled: false,
2122 metrics_raw_retention_ms: None,
2123 metrics_rollup_policies: Vec::new(),
2124 metrics_tenant_identity: None,
2125 metrics_namespace: None,
2126 append_only: false,
2127 subscriptions: Vec::new(),
2128 analytics_config,
2129 session_key: None,
2130 session_gap_ms: None,
2131 retention_duration_ms: None,
2132 analytical_storage: None,
2133
2134 ai_policy: None,
2135 }
2136}
2137
2138fn metrics_collection_contract(query: &CreateTableQuery) -> crate::physical::CollectionContract {
2139 let now = current_unix_ms();
2140 crate::physical::CollectionContract {
2141 name: query.name.clone(),
2142 declared_model: crate::catalog::CollectionModel::Metrics,
2143 schema_mode: crate::catalog::SchemaMode::SemiStructured,
2144 origin: crate::physical::ContractOrigin::Explicit,
2145 version: 1,
2146 created_at_unix_ms: now,
2147 updated_at_unix_ms: now,
2148 default_ttl_ms: query.default_ttl_ms,
2149 vector_dimension: None,
2150 vector_metric: None,
2151 context_index_fields: Vec::new(),
2152 declared_columns: Vec::new(),
2153 table_def: None,
2154 timestamps_enabled: false,
2155 context_index_enabled: false,
2156 metrics_raw_retention_ms: query.default_ttl_ms,
2157 metrics_rollup_policies: query.metrics_rollup_policies.clone(),
2158 metrics_tenant_identity: Some(
2159 query
2160 .tenant_by
2161 .clone()
2162 .unwrap_or_else(|| "current_tenant".to_string()),
2163 ),
2164 metrics_namespace: Some("default".to_string()),
2165 append_only: true,
2166 subscriptions: Vec::new(),
2167 analytics_config: Vec::new(),
2168 session_key: None,
2169 session_gap_ms: None,
2170 retention_duration_ms: None,
2171 analytical_storage: None,
2172
2173 ai_policy: None,
2174 }
2175}
2176
2177fn vector_collection_contract(query: &CreateVectorQuery) -> crate::physical::CollectionContract {
2178 let now = current_unix_ms();
2179 crate::physical::CollectionContract {
2180 name: query.name.clone(),
2181 declared_model: crate::catalog::CollectionModel::Vector,
2182 schema_mode: crate::catalog::SchemaMode::Dynamic,
2183 origin: crate::physical::ContractOrigin::Explicit,
2184 version: 1,
2185 created_at_unix_ms: now,
2186 updated_at_unix_ms: now,
2187 default_ttl_ms: None,
2188 vector_dimension: Some(query.dimension),
2189 vector_metric: Some(query.metric),
2190 context_index_fields: Vec::new(),
2191 declared_columns: Vec::new(),
2192 table_def: None,
2193 timestamps_enabled: false,
2194 context_index_enabled: false,
2195 metrics_raw_retention_ms: None,
2196 metrics_rollup_policies: Vec::new(),
2197 metrics_tenant_identity: None,
2198 metrics_namespace: None,
2199 append_only: false,
2200 subscriptions: Vec::new(),
2201 analytics_config: Vec::new(),
2202 session_key: None,
2203 session_gap_ms: None,
2204 retention_duration_ms: None,
2205 analytical_storage: None,
2206
2207 ai_policy: None,
2208 }
2209}
2210
2211fn apply_alter_operations_to_contract(
2212 contract: &mut crate::physical::CollectionContract,
2213 operations: &[AlterOperation],
2214) {
2215 if contract.table_def.is_none() {
2216 contract.table_def = Some(reddb_types::TableDef::new(contract.name.clone()));
2217 }
2218 for operation in operations {
2219 match operation {
2220 AlterOperation::AddColumn(column) => {
2221 if !contract
2222 .declared_columns
2223 .iter()
2224 .any(|existing| existing.name == column.name)
2225 {
2226 contract
2227 .declared_columns
2228 .push(declared_column_contract_from_ddl(column));
2229 }
2230 if let Some(table_def) = contract.table_def.as_mut() {
2231 if table_def.get_column(&column.name).is_none() {
2232 if let Ok(column_def) = column_def_from_ddl(column) {
2233 if column.primary_key {
2234 table_def.primary_key.push(column.name.clone());
2235 table_def.constraints.push(
2236 reddb_types::Constraint::new(
2237 format!("pk_{}", column.name),
2238 reddb_types::ConstraintType::PrimaryKey,
2239 )
2240 .on_columns(vec![column.name.clone()]),
2241 );
2242 }
2243 if column.unique {
2244 table_def.constraints.push(
2245 reddb_types::Constraint::new(
2246 format!("uniq_{}", column.name),
2247 reddb_types::ConstraintType::Unique,
2248 )
2249 .on_columns(vec![column.name.clone()]),
2250 );
2251 }
2252 if column.not_null {
2253 table_def.constraints.push(
2254 reddb_types::Constraint::new(
2255 format!("not_null_{}", column.name),
2256 reddb_types::ConstraintType::NotNull,
2257 )
2258 .on_columns(vec![column.name.clone()]),
2259 );
2260 }
2261 table_def.columns.push(column_def);
2262 }
2263 }
2264 }
2265 }
2266 AlterOperation::DropColumn(name) => {
2267 contract
2268 .declared_columns
2269 .retain(|column| column.name != *name);
2270 if let Some(table_def) = contract.table_def.as_mut() {
2271 if let Some(index) = table_def.column_index(name) {
2272 table_def.columns.remove(index);
2273 }
2274 table_def.primary_key.retain(|column| column != name);
2275 table_def.constraints.retain(|constraint| {
2276 !constraint.columns.iter().any(|column| column == name)
2277 });
2278 table_def
2279 .indexes
2280 .retain(|index| !index.columns.iter().any(|column| column == name));
2281 }
2282 }
2283 AlterOperation::RenameColumn { from, to } => {
2284 if contract
2285 .declared_columns
2286 .iter()
2287 .any(|column| column.name == *to)
2288 {
2289 continue;
2290 }
2291 if let Some(column) = contract
2292 .declared_columns
2293 .iter_mut()
2294 .find(|column| column.name == *from)
2295 {
2296 column.name = to.clone();
2297 }
2298 if let Some(table_def) = contract.table_def.as_mut() {
2299 if let Some(column) = table_def
2300 .columns
2301 .iter_mut()
2302 .find(|column| column.name == *from)
2303 {
2304 column.name = to.clone();
2305 }
2306 for primary_key in &mut table_def.primary_key {
2307 if *primary_key == *from {
2308 *primary_key = to.clone();
2309 }
2310 }
2311 for constraint in &mut table_def.constraints {
2312 for column in &mut constraint.columns {
2313 if *column == *from {
2314 *column = to.clone();
2315 }
2316 }
2317 if let Some(ref_columns) = constraint.ref_columns.as_mut() {
2318 for column in ref_columns {
2319 if *column == *from {
2320 *column = to.clone();
2321 }
2322 }
2323 }
2324 }
2325 for index in &mut table_def.indexes {
2326 for column in &mut index.columns {
2327 if *column == *from {
2328 *column = to.clone();
2329 }
2330 }
2331 }
2332 }
2333 }
2334 AlterOperation::AttachPartition { .. } | AlterOperation::DetachPartition { .. } => {}
2337 AlterOperation::EnableRowLevelSecurity | AlterOperation::DisableRowLevelSecurity => {}
2341 AlterOperation::EnableTenancy { .. } | AlterOperation::DisableTenancy => {}
2344 AlterOperation::SetAppendOnly(on) => {
2345 contract.append_only = *on;
2346 }
2347 AlterOperation::SetVersioned(_) => {}
2350 AlterOperation::EnableEvents(subscription) => {
2351 let mut subscription = subscription.clone();
2352 subscription.source = contract.name.clone();
2353 subscription.enabled = true;
2354 if let Some(existing) = contract
2355 .subscriptions
2356 .iter_mut()
2357 .find(|existing| existing.target_queue == subscription.target_queue)
2358 {
2359 *existing = subscription;
2360 } else {
2361 contract.subscriptions.push(subscription);
2362 }
2363 }
2364 AlterOperation::DisableEvents => {
2365 for subscription in &mut contract.subscriptions {
2366 subscription.enabled = false;
2367 }
2368 }
2369 AlterOperation::AddSubscription { name, descriptor } => {
2370 let mut sub = descriptor.clone();
2371 sub.name = name.clone();
2372 sub.source = contract.name.clone();
2373 sub.enabled = true;
2374 if let Some(existing) = contract.subscriptions.iter_mut().find(|s| s.name == *name)
2375 {
2376 *existing = sub;
2377 } else {
2378 contract.subscriptions.push(sub);
2379 }
2380 }
2381 AlterOperation::DropSubscription { name } => {
2382 contract.subscriptions.retain(|s| s.name != *name);
2383 }
2384 AlterOperation::AddSigner { .. } | AlterOperation::RevokeSigner { .. } => {}
2389 AlterOperation::SetRetention { duration_ms } => {
2390 contract.retention_duration_ms = Some(*duration_ms);
2391 }
2392 AlterOperation::UnsetRetention => {
2393 contract.retention_duration_ms = None;
2394 }
2395 AlterOperation::AddAnalytics(views) => {
2399 for view in views {
2400 if !contract
2404 .analytics_config
2405 .iter()
2406 .any(|existing| existing.output == view.output)
2407 {
2408 contract.analytics_config.push(view.clone());
2409 }
2410 }
2411 }
2412 AlterOperation::DropAnalytics(output) => {
2413 contract
2414 .analytics_config
2415 .retain(|view| view.output != *output);
2416 }
2417 }
2418 }
2419}
2420
2421pub(crate) fn retention_timestamp_column_exists(
2426 contract: &crate::physical::CollectionContract,
2427) -> bool {
2428 if contract.timestamps_enabled {
2429 return true;
2430 }
2431 if matches!(
2432 contract.declared_model,
2433 crate::catalog::CollectionModel::TimeSeries | crate::catalog::CollectionModel::Metrics
2434 ) {
2435 return true;
2439 }
2440 contract
2441 .declared_columns
2442 .iter()
2443 .any(|column| is_temporal_data_type(&column.data_type))
2444}
2445
2446fn is_temporal_data_type(data_type: &str) -> bool {
2447 let upper = data_type.to_ascii_uppercase();
2448 matches!(
2449 upper.as_str(),
2450 "TIMESTAMP" | "TIMESTAMPMS" | "TIMESTAMP_MS" | "DATETIME" | "DATE"
2451 )
2452}
2453
2454fn validate_event_subscriptions(
2455 runtime: &RedDBRuntime,
2456 source: &str,
2457 subscriptions: &[crate::catalog::SubscriptionDescriptor],
2458) -> RedDBResult<()> {
2459 for subscription in subscriptions
2460 .iter()
2461 .filter(|subscription| subscription.enabled)
2462 {
2463 if subscription.all_tenants && crate::runtime::impl_core::current_tenant().is_some() {
2464 return Err(RedDBError::Query(
2465 "cross-tenant subscription requires cluster-admin capability (events:cluster_subscribe)".to_string(),
2466 ));
2467 }
2468 validate_subscription_auth(runtime, source, subscription)?;
2469 if subscription.target_queue == source
2470 || subscription_would_create_cycle(
2471 &runtime.inner.db,
2472 source,
2473 &subscription.target_queue,
2474 )
2475 {
2476 return Err(RedDBError::Query(
2477 "subscription would create cycle".to_string(),
2478 ));
2479 }
2480 audit_subscription_redact_gap(runtime, source, subscription);
2481 }
2482 Ok(())
2483}
2484
2485fn validate_subscription_auth(
2486 runtime: &RedDBRuntime,
2487 source: &str,
2488 subscription: &crate::catalog::SubscriptionDescriptor,
2489) -> RedDBResult<()> {
2490 let auth_store = match runtime.inner.auth_store.read().clone() {
2491 Some(store) => store,
2492 None => return Ok(()),
2493 };
2494 let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
2495 Some(identity) => identity,
2496 None => return Ok(()),
2497 };
2498 let tenant = crate::runtime::impl_core::current_tenant();
2499 let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
2500
2501 if auth_store.iam_authorization_enabled() {
2502 let ctx = crate::auth::policies::EvalContext {
2503 principal_tenant: tenant.clone(),
2504 current_tenant: tenant.clone(),
2505 peer_ip: None,
2506 mfa_present: false,
2507 now_ms: crate::auth::now_ms(),
2508 principal_is_admin_role: role == crate::auth::Role::Admin,
2509 principal_is_platform_scoped: principal.tenant.is_none(),
2510 };
2511 let mut source_resource = crate::auth::policies::ResourceRef::new("table", source);
2512 if let Some(t) = tenant.as_deref() {
2513 source_resource = source_resource.with_tenant(t.to_string());
2514 }
2515 if !auth_store.check_policy_authz_with_role(
2516 &principal,
2517 "select",
2518 &source_resource,
2519 &ctx,
2520 role,
2521 ) {
2522 return Err(RedDBError::Query(format!(
2523 "permission denied: principal=`{}` action=`select` resource=`{}:{}` denied by IAM policy",
2524 principal, source_resource.kind, source_resource.name
2525 )));
2526 }
2527
2528 let mut target_resource =
2529 crate::auth::policies::ResourceRef::new("queue", subscription.target_queue.clone());
2530 if let Some(t) = tenant.as_deref() {
2531 target_resource = target_resource.with_tenant(t.to_string());
2532 }
2533 if !auth_store.check_policy_authz_with_role(
2534 &principal,
2535 "write",
2536 &target_resource,
2537 &ctx,
2538 role,
2539 ) {
2540 return Err(RedDBError::Query(format!(
2541 "permission denied: principal=`{}` action=`write` resource=`{}:{}` denied by IAM policy",
2542 principal, target_resource.kind, target_resource.name
2543 )));
2544 }
2545 return Ok(());
2546 }
2547
2548 let ctx = crate::auth::privileges::AuthzContext {
2549 principal: &username,
2550 effective_role: role,
2551 tenant: tenant.as_deref(),
2552 };
2553 auth_store
2554 .check_grant(
2555 &ctx,
2556 crate::auth::privileges::Action::Select,
2557 &crate::auth::privileges::Resource::table_from_name(source),
2558 )
2559 .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))?;
2560 auth_store
2561 .check_grant(
2562 &ctx,
2563 crate::auth::privileges::Action::Insert,
2564 &crate::auth::privileges::Resource::table_from_name(&subscription.target_queue),
2565 )
2566 .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))?;
2567 Ok(())
2568}
2569
2570fn audit_subscription_redact_gap(
2571 runtime: &RedDBRuntime,
2572 source: &str,
2573 subscription: &crate::catalog::SubscriptionDescriptor,
2574) {
2575 let auth_store = match runtime.inner.auth_store.read().clone() {
2576 Some(store) if store.iam_authorization_enabled() => store,
2577 _ => return,
2578 };
2579 let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
2580 Some(identity) => identity,
2581 None => return,
2582 };
2583 let tenant = crate::runtime::impl_core::current_tenant();
2584 let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
2585 let missing = subscription_redact_gap_columns(&auth_store, &principal, source, subscription);
2586 if missing.is_empty() {
2587 return;
2588 }
2589
2590 let columns = missing.into_iter().collect::<Vec<_>>().join(", ");
2591 tracing::warn!(
2592 target: "reddb::operator",
2593 "subscription_redact_gap: source={} target_queue={} columns=[{}]",
2594 source,
2595 subscription.target_queue,
2596 columns
2597 );
2598 let mut event = AuditEvent::builder("subscription_redact_gap")
2599 .principal(username)
2600 .source(AuditAuthSource::System)
2601 .resource(format!(
2602 "subscription:{}->{}",
2603 source, subscription.target_queue
2604 ))
2605 .outcome(Outcome::Success)
2606 .field(AuditFieldEscaper::field("source", source))
2607 .field(AuditFieldEscaper::field(
2608 "target_queue",
2609 subscription.target_queue.clone(),
2610 ))
2611 .field(AuditFieldEscaper::field(
2612 "subscription",
2613 subscription.name.clone(),
2614 ))
2615 .field(AuditFieldEscaper::field("columns", columns))
2616 .field(AuditFieldEscaper::field("role", role.as_str()));
2617 if let Some(t) = tenant {
2618 event = event.tenant(t);
2619 }
2620 runtime.inner.audit_log.record_event(event.build());
2621}
2622
2623fn subscription_redact_gap_columns(
2624 auth_store: &crate::auth::store::AuthStore,
2625 principal: &crate::auth::UserId,
2626 source: &str,
2627 subscription: &crate::catalog::SubscriptionDescriptor,
2628) -> BTreeSet<String> {
2629 let redacted: HashSet<String> = subscription
2630 .redact_fields
2631 .iter()
2632 .map(|field| field.to_ascii_lowercase())
2633 .collect();
2634 auth_store
2635 .effective_policies(principal)
2636 .iter()
2637 .flat_map(|policy| policy.statements.iter())
2638 .filter(|statement| statement.effect == crate::auth::policies::Effect::Deny)
2639 .filter(|statement| statement.actions.iter().any(action_pattern_matches_select))
2640 .flat_map(|statement| statement.resources.iter())
2641 .filter_map(|resource| denied_column_for_source(resource, source))
2642 .filter(|column| !redact_covers_column(&redacted, source, column))
2643 .collect()
2644}
2645
2646fn action_pattern_matches_select(pattern: &crate::auth::policies::ActionPattern) -> bool {
2647 match pattern {
2648 crate::auth::policies::ActionPattern::Wildcard => true,
2649 crate::auth::policies::ActionPattern::Exact(action) => action == "select",
2650 crate::auth::policies::ActionPattern::Prefix(prefix) => {
2651 "select".len() > prefix.len() + 1
2652 && "select".starts_with(prefix)
2653 && "select".as_bytes()[prefix.len()] == b':'
2654 }
2655 }
2656}
2657
2658fn denied_column_for_source(
2659 resource: &crate::auth::policies::ResourcePattern,
2660 source: &str,
2661) -> Option<String> {
2662 let crate::auth::policies::ResourcePattern::Exact { kind, name } = resource else {
2663 return None;
2664 };
2665 if kind != "column" {
2666 return None;
2667 }
2668 let column = crate::auth::ColumnRef::parse_resource_name(name).ok()?;
2669 (column.table_resource_name() == source).then_some(column.column)
2670}
2671
2672fn redact_covers_column(redacted: &HashSet<String>, source: &str, column: &str) -> bool {
2673 let column = column.to_ascii_lowercase();
2674 let qualified = format!("{}.{}", source.to_ascii_lowercase(), column);
2675 redacted.contains("*") || redacted.contains(&column) || redacted.contains(&qualified)
2676}
2677
2678fn subscription_would_create_cycle(
2679 db: &crate::storage::unified::devx::RedDB,
2680 source: &str,
2681 target: &str,
2682) -> bool {
2683 let mut graph: HashMap<String, Vec<String>> = HashMap::new();
2684 for contract in db.collection_contracts() {
2685 for subscription in contract
2686 .subscriptions
2687 .into_iter()
2688 .filter(|subscription| subscription.enabled)
2689 {
2690 graph
2691 .entry(subscription.source)
2692 .or_default()
2693 .push(subscription.target_queue);
2694 }
2695 }
2696 graph
2697 .entry(source.to_string())
2698 .or_default()
2699 .push(target.to_string());
2700
2701 let mut stack = vec![target.to_string()];
2702 let mut seen = HashSet::new();
2703 while let Some(node) = stack.pop() {
2704 if node == source {
2705 return true;
2706 }
2707 if !seen.insert(node.clone()) {
2708 continue;
2709 }
2710 if let Some(next) = graph.get(&node) {
2711 stack.extend(next.iter().cloned());
2712 }
2713 }
2714 false
2715}
2716
2717pub(crate) fn ensure_event_target_queue_pub(
2718 runtime: &RedDBRuntime,
2719 queue: &str,
2720) -> RedDBResult<()> {
2721 ensure_event_target_queue(runtime, queue)
2722}
2723
2724fn ensure_event_target_queue(runtime: &RedDBRuntime, queue: &str) -> RedDBResult<()> {
2725 let store = runtime.inner.db.store();
2726 if store.get_collection(queue).is_some() {
2727 return Ok(());
2728 }
2729 store
2730 .create_collection(queue)
2731 .map_err(|err| RedDBError::Internal(err.to_string()))?;
2732 crate::runtime::queue_storage_index::prepare_queue(runtime, queue)?;
2733 runtime
2734 .inner
2735 .db
2736 .save_collection_contract(event_queue_collection_contract(queue))
2737 .map_err(|err| RedDBError::Internal(err.to_string()))?;
2738 store.set_config_tree(
2739 &format!("queue.{queue}.mode"),
2740 &crate::serde_json::Value::String("fanout".to_string()),
2741 );
2742 Ok(())
2743}
2744
2745fn event_queue_collection_contract(queue: &str) -> crate::physical::CollectionContract {
2746 let now = current_unix_ms();
2747 crate::physical::CollectionContract {
2748 name: queue.to_string(),
2749 declared_model: crate::catalog::CollectionModel::Queue,
2750 schema_mode: crate::catalog::SchemaMode::Dynamic,
2751 origin: crate::physical::ContractOrigin::Implicit,
2752 version: 1,
2753 created_at_unix_ms: now,
2754 updated_at_unix_ms: now,
2755 default_ttl_ms: None,
2756 vector_dimension: None,
2757 vector_metric: None,
2758 context_index_fields: Vec::new(),
2759 declared_columns: Vec::new(),
2760 table_def: None,
2761 timestamps_enabled: false,
2762 context_index_enabled: false,
2763 metrics_raw_retention_ms: None,
2764 metrics_rollup_policies: Vec::new(),
2765 metrics_tenant_identity: None,
2766 metrics_namespace: None,
2767 append_only: true,
2768 subscriptions: Vec::new(),
2769 analytics_config: Vec::new(),
2770 session_key: None,
2771 session_gap_ms: None,
2772 retention_duration_ms: None,
2773 analytical_storage: None,
2774
2775 ai_policy: None,
2776 }
2777}
2778
2779fn column_def_from_ddl(column: &CreateColumnDef) -> RedDBResult<reddb_types::ColumnDef> {
2780 reddb_engine::catalog::ddl::column_def_from_ddl(column).map_err(RedDBError::Query)
2781}
2782
2783fn current_unix_ms() -> u128 {
2784 std::time::SystemTime::now()
2785 .duration_since(std::time::UNIX_EPOCH)
2786 .unwrap_or_default()
2787 .as_millis()
2788}
2789
2790#[cfg(test)]
2791mod tests {
2792 use crate::auth::policies::{ActionPattern, Effect, Policy, ResourcePattern, Statement};
2793 use crate::auth::store::{AuthStore, PrincipalRef};
2794 use crate::auth::UserId;
2795 use crate::auth::{AuthConfig, Role};
2796 use crate::runtime::impl_core::{clear_current_auth_identity, set_current_auth_identity};
2797 use crate::{RedDBOptions, RedDBRuntime};
2798 use reddb_types::Value;
2799 use std::sync::Arc;
2800
2801 fn make_allow_policy(id: &str, action: &str, collection: &str) -> Policy {
2802 Policy {
2803 id: id.to_string(),
2804 version: 1,
2805 tenant: None,
2806 created_at: 0,
2807 updated_at: 0,
2808 statements: vec![Statement {
2809 sid: None,
2810 effect: Effect::Allow,
2811 actions: vec![ActionPattern::Exact(action.to_string())],
2812 resources: vec![ResourcePattern::Exact {
2813 kind: "collection".to_string(),
2814 name: collection.to_string(),
2815 }],
2816 condition: None,
2817 }],
2818 }
2819 }
2820
2821 fn wire_auth_store(rt: &RedDBRuntime) -> Arc<AuthStore> {
2822 let store = Arc::new(AuthStore::new(AuthConfig::default()));
2823 *rt.inner.auth_store.write() = Some(store.clone());
2824 store
2825 }
2826
2827 #[test]
2828 fn drop_denied_without_iam_policy() {
2829 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2830 rt.execute_query("CREATE TABLE foo (id INT)").unwrap();
2831 let store = wire_auth_store(&rt);
2832 let select_only = Policy {
2834 id: "select-only".to_string(),
2835 version: 1,
2836 tenant: None,
2837 created_at: 0,
2838 updated_at: 0,
2839 statements: vec![Statement {
2840 sid: None,
2841 effect: Effect::Allow,
2842 actions: vec![ActionPattern::Exact("select".to_string())],
2843 resources: vec![ResourcePattern::Wildcard],
2844 condition: None,
2845 }],
2846 };
2847 store.put_policy_internal(select_only).unwrap();
2848 let alice = UserId::from_parts(None, "alice");
2849 store
2850 .attach_policy(PrincipalRef::User(alice), "select-only")
2851 .unwrap();
2852 set_current_auth_identity("alice".to_string(), Role::Write);
2853 let err = rt.execute_query("DROP TABLE foo").unwrap_err();
2854 clear_current_auth_identity();
2855 assert!(
2856 format!("{err}").contains("denied by IAM policy"),
2857 "got: {err}"
2858 );
2859 }
2860
2861 #[test]
2862 fn drop_allowed_with_explicit_iam_policy() {
2863 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2864 rt.execute_query("CREATE TABLE bar (id INT)").unwrap();
2865 let store = wire_auth_store(&rt);
2866 let policy = make_allow_policy("allow-drop-bar", "drop", "bar");
2867 store.put_policy_internal(policy).unwrap();
2868 let bob = UserId::from_parts(None, "bob");
2869 store
2870 .attach_policy(PrincipalRef::User(bob), "allow-drop-bar")
2871 .unwrap();
2872 set_current_auth_identity("bob".to_string(), Role::Write);
2873 rt.execute_query("DROP TABLE bar").unwrap();
2874 clear_current_auth_identity();
2875 }
2876
2877 #[test]
2878 fn drop_allowed_with_wildcard_iam_policy() {
2879 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2880 rt.execute_query("CREATE TABLE baz (id INT)").unwrap();
2881 let store = wire_auth_store(&rt);
2882 let policy = Policy {
2883 id: "allow-drop-all".to_string(),
2884 version: 1,
2885 tenant: None,
2886 created_at: 0,
2887 updated_at: 0,
2888 statements: vec![Statement {
2889 sid: None,
2890 effect: Effect::Allow,
2891 actions: vec![ActionPattern::Exact("drop".to_string())],
2892 resources: vec![ResourcePattern::Wildcard],
2893 condition: None,
2894 }],
2895 };
2896 store.put_policy_internal(policy).unwrap();
2897 let carl = UserId::from_parts(None, "carl");
2898 store
2899 .attach_policy(PrincipalRef::User(carl), "allow-drop-all")
2900 .unwrap();
2901 set_current_auth_identity("carl".to_string(), Role::Write);
2902 rt.execute_query("DROP TABLE baz").unwrap();
2903 clear_current_auth_identity();
2904 }
2905
2906 #[test]
2907 fn truncate_denied_without_iam_policy() {
2908 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2909 rt.execute_query("CREATE TABLE qux (id INT)").unwrap();
2910 let store = wire_auth_store(&rt);
2911 store
2918 .set_enforcement_mode(crate::auth::enforcement_mode::PolicyEnforcementMode::PolicyOnly);
2919 let select_only = Policy {
2921 id: "select-only-2".to_string(),
2922 version: 1,
2923 tenant: None,
2924 created_at: 0,
2925 updated_at: 0,
2926 statements: vec![Statement {
2927 sid: None,
2928 effect: Effect::Allow,
2929 actions: vec![ActionPattern::Exact("select".to_string())],
2930 resources: vec![ResourcePattern::Wildcard],
2931 condition: None,
2932 }],
2933 };
2934 store.put_policy_internal(select_only).unwrap();
2935 let dana = UserId::from_parts(None, "dana");
2936 store
2937 .attach_policy(PrincipalRef::User(dana), "select-only-2")
2938 .unwrap();
2939 set_current_auth_identity("dana".to_string(), Role::Write);
2940 let err = rt.execute_query("TRUNCATE TABLE qux").unwrap_err();
2941 clear_current_auth_identity();
2942 assert!(
2943 format!("{err}").contains("denied by IAM policy"),
2944 "got: {err}"
2945 );
2946 }
2947
2948 #[test]
2949 fn truncate_table_clears_rows_and_preserves_schema_and_indexes() {
2950 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2951 rt.execute_query("CREATE TABLE users (id INT, name TEXT)")
2952 .unwrap();
2953 rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'ana'), (2, 'bob')")
2954 .unwrap();
2955 rt.execute_query("CREATE INDEX idx_users_id ON users (id) USING HASH")
2956 .unwrap();
2957
2958 let truncated = rt.execute_query("TRUNCATE TABLE users").unwrap();
2959 assert_eq!(truncated.statement_type, "truncate");
2960 assert_eq!(truncated.affected_rows, 0);
2961
2962 let empty = rt.execute_query("SELECT id FROM users").unwrap();
2963 assert!(empty.result.records.is_empty());
2964
2965 rt.execute_query("INSERT INTO users (id, name) VALUES (3, 'cy')")
2966 .unwrap();
2967 let selected = rt
2968 .execute_query("SELECT name FROM users WHERE id = 3")
2969 .unwrap();
2970 let name = selected.result.records[0].get("name").unwrap();
2971 assert_eq!(name, &Value::text("cy"));
2972 assert!(rt.db().collection_contract("users").is_some());
2973 assert!(rt
2974 .inner
2975 .index_store
2976 .list_indices("users")
2977 .iter()
2978 .any(|index| index.name == "idx_users_id"));
2979 }
2980
2981 #[test]
2982 fn truncate_collection_is_polymorphic_and_typed_mismatch_fails() {
2983 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2984 rt.execute_query("CREATE QUEUE tasks").unwrap();
2985 rt.execute_query("QUEUE PUSH tasks {'job':'a'}").unwrap();
2986
2987 let err = rt.execute_query("TRUNCATE TABLE tasks").unwrap_err();
2988 assert!(format!("{err}").contains("model mismatch: expected table, got queue"));
2989
2990 rt.execute_query("TRUNCATE COLLECTION tasks").unwrap();
2991 let len = rt.execute_query("QUEUE LEN tasks").unwrap();
2992 assert_eq!(
2993 len.result.records[0].get("len"),
2994 Some(&Value::UnsignedInteger(0))
2995 );
2996 }
2997
2998 #[test]
2999 fn truncate_system_schema_is_read_only() {
3000 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3001 let err = rt
3002 .execute_query("TRUNCATE COLLECTION red.collections")
3003 .unwrap_err();
3004 assert!(format!("{err}").contains("system schema is read-only"));
3005 }
3006
3007 fn queue_payloads(rt: &RedDBRuntime, queue: &str) -> Vec<crate::json::Value> {
3010 let result = rt
3011 .execute_query(&format!("QUEUE PEEK {queue} 100"))
3012 .expect("peek queue");
3013 result
3014 .result
3015 .records
3016 .iter()
3017 .map(
3018 |record| match record.get("payload").expect("payload column") {
3019 Value::Json(bytes) => crate::json::from_slice(bytes).expect("json payload"),
3020 other => panic!("expected JSON queue payload, got {other:?}"),
3021 },
3022 )
3023 .collect()
3024 }
3025
3026 #[test]
3029 fn truncate_event_enabled_table_emits_single_truncate_event() {
3030 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3031 rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3032 .unwrap();
3033 rt.execute_query(
3034 "INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob'), (3, 'carol')",
3035 )
3036 .unwrap();
3037
3038 rt.execute_query("QUEUE POP users_events COUNT 10").unwrap();
3040
3041 rt.execute_query("TRUNCATE TABLE users").unwrap();
3042
3043 let events = queue_payloads(&rt, "users_events");
3044 assert_eq!(
3046 events.len(),
3047 1,
3048 "expected 1 truncate event, got {}",
3049 events.len()
3050 );
3051 let ev = events[0].as_object().expect("event is object");
3052 assert_eq!(
3053 ev.get("op").and_then(crate::json::Value::as_str),
3054 Some("truncate")
3055 );
3056 assert_eq!(
3057 ev.get("collection").and_then(crate::json::Value::as_str),
3058 Some("users")
3059 );
3060 assert_eq!(
3061 ev.get("entities_count")
3062 .and_then(crate::json::Value::as_u64),
3063 Some(3)
3064 );
3065 assert!(ev.get("ts").and_then(crate::json::Value::as_u64).is_some());
3066 assert!(ev.get("lsn").and_then(crate::json::Value::as_u64).is_some());
3067 assert!(ev
3068 .get("event_id")
3069 .and_then(crate::json::Value::as_str)
3070 .is_some_and(|s| !s.is_empty()));
3071 }
3072
3073 #[test]
3075 fn truncate_no_events_collection_emits_nothing() {
3076 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3077 rt.execute_query("CREATE TABLE plain (id INT, val TEXT)")
3078 .unwrap();
3079 rt.execute_query("INSERT INTO plain (id, val) VALUES (1, 'a'), (2, 'b')")
3080 .unwrap();
3081 rt.execute_query("TRUNCATE TABLE plain").unwrap();
3083 let rows = rt.execute_query("SELECT id FROM plain").unwrap();
3085 assert!(rows.result.records.is_empty());
3086 }
3087
3088 #[test]
3092 fn drop_event_enabled_table_emits_single_collection_dropped_event() {
3093 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3094 rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3095 .unwrap();
3096 rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob')")
3097 .unwrap();
3098
3099 rt.execute_query("QUEUE POP users_events COUNT 10").unwrap();
3101
3102 rt.execute_query("DROP TABLE users").unwrap();
3103
3104 let events = queue_payloads(&rt, "users_events");
3106 assert_eq!(
3107 events.len(),
3108 1,
3109 "expected 1 collection_dropped event, got {}",
3110 events.len()
3111 );
3112 let ev = events[0].as_object().expect("event is object");
3113 assert_eq!(
3114 ev.get("op").and_then(crate::json::Value::as_str),
3115 Some("collection_dropped")
3116 );
3117 assert_eq!(
3118 ev.get("collection").and_then(crate::json::Value::as_str),
3119 Some("users")
3120 );
3121 assert_eq!(
3122 ev.get("final_entities_count")
3123 .and_then(crate::json::Value::as_u64),
3124 Some(2)
3125 );
3126 assert!(ev.get("ts").and_then(crate::json::Value::as_u64).is_some());
3127 assert!(ev.get("lsn").and_then(crate::json::Value::as_u64).is_some());
3128 assert!(ev
3129 .get("event_id")
3130 .and_then(crate::json::Value::as_str)
3131 .is_some_and(|s| !s.is_empty()));
3132
3133 let err = rt.execute_query("SELECT id FROM users").unwrap_err();
3135 assert!(
3136 format!("{err}").contains("users"),
3137 "expected not-found error"
3138 );
3139 }
3140
3141 #[test]
3144 fn drop_no_events_collection_emits_nothing() {
3145 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3146 rt.execute_query("CREATE TABLE plain (id INT, val TEXT)")
3147 .unwrap();
3148 rt.execute_query("INSERT INTO plain (id, val) VALUES (1, 'a')")
3149 .unwrap();
3150 rt.execute_query("DROP TABLE plain").unwrap();
3151 let err = rt.execute_query("SELECT id FROM plain").unwrap_err();
3153 assert!(format!("{err}").contains("plain"));
3154 }
3155
3156 #[test]
3160 fn ops_filter_insert_only_ignores_update_and_delete() {
3161 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3162 rt.execute_query(
3163 "CREATE TABLE items (id INT, val TEXT) WITH EVENTS (INSERT) TO items_events",
3164 )
3165 .unwrap();
3166 rt.execute_query("INSERT INTO items (id, val) VALUES (1, 'a')")
3167 .unwrap();
3168 rt.execute_query("UPDATE items SET val = 'b' WHERE id = 1")
3169 .unwrap();
3170 rt.execute_query("DELETE FROM items WHERE id = 1").unwrap();
3171
3172 let events = queue_payloads(&rt, "items_events");
3173 assert_eq!(
3175 events.len(),
3176 1,
3177 "expected 1 insert event, got {}",
3178 events.len()
3179 );
3180 assert_eq!(
3181 events[0]
3182 .as_object()
3183 .unwrap()
3184 .get("op")
3185 .and_then(crate::json::Value::as_str),
3186 Some("insert")
3187 );
3188 }
3189
3190 #[test]
3192 fn where_filter_skips_rows_that_do_not_match() {
3193 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3194 rt.execute_query(
3195 "CREATE TABLE users (id INT, status TEXT) WITH EVENTS WHERE status = 'active' TO users_events",
3196 )
3197 .unwrap();
3198
3199 rt.execute_query("INSERT INTO users (id, status) VALUES (1, 'active')")
3201 .unwrap();
3202 rt.execute_query("INSERT INTO users (id, status) VALUES (2, 'inactive')")
3204 .unwrap();
3205
3206 let events = queue_payloads(&rt, "users_events");
3207 assert_eq!(
3208 events.len(),
3209 1,
3210 "expected 1 event (only active), got {}",
3211 events.len()
3212 );
3213 let ev = events[0].as_object().unwrap();
3214 assert_eq!(
3215 ev.get("op").and_then(crate::json::Value::as_str),
3216 Some("insert")
3217 );
3218 let after = ev.get("after").unwrap().as_object().unwrap();
3219 assert_eq!(
3220 after.get("status").and_then(crate::json::Value::as_str),
3221 Some("active")
3222 );
3223 }
3224
3225 #[test]
3227 fn ops_filter_and_where_filter_combined() {
3228 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3229 rt.execute_query(
3230 "CREATE TABLE items (id INT, status TEXT) WITH EVENTS (INSERT, UPDATE) WHERE status = 'active' TO items_events",
3231 )
3232 .unwrap();
3233
3234 rt.execute_query("INSERT INTO items (id, status) VALUES (1, 'active')")
3236 .unwrap();
3237 rt.execute_query("INSERT INTO items (id, status) VALUES (2, 'inactive')")
3239 .unwrap();
3240 rt.execute_query("UPDATE items SET status = 'inactive' WHERE id = 1")
3242 .unwrap();
3243 rt.execute_query("DELETE FROM items WHERE id = 2").unwrap();
3245
3246 let events = queue_payloads(&rt, "items_events");
3247 assert_eq!(
3249 events.len(),
3250 1,
3251 "expected 1 event, got {}: {events:?}",
3252 events.len()
3253 );
3254 assert_eq!(
3255 events[0]
3256 .as_object()
3257 .unwrap()
3258 .get("op")
3259 .and_then(crate::json::Value::as_str),
3260 Some("insert")
3261 );
3262 }
3263
3264 #[test]
3266 fn where_filter_on_delete_checks_before_state() {
3267 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3268 rt.execute_query(
3269 "CREATE TABLE users (id INT, status TEXT) WITH EVENTS (DELETE) WHERE status = 'active' TO users_events",
3270 )
3271 .unwrap();
3272
3273 rt.execute_query("INSERT INTO users (id, status) VALUES (1, 'active'), (2, 'inactive')")
3274 .unwrap();
3275
3276 rt.execute_query("DELETE FROM users WHERE id = 1").unwrap();
3278 rt.execute_query("DELETE FROM users WHERE id = 2").unwrap();
3280
3281 let events = queue_payloads(&rt, "users_events");
3282 assert_eq!(
3283 events.len(),
3284 1,
3285 "expected 1 delete event, got {}",
3286 events.len()
3287 );
3288 let ev = events[0].as_object().unwrap();
3289 assert_eq!(
3290 ev.get("op").and_then(crate::json::Value::as_str),
3291 Some("delete")
3292 );
3293 }
3294
3295 #[test]
3299 fn alter_add_column_on_event_enabled_table_succeeds() {
3300 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3301 rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3302 .unwrap();
3303 rt.execute_query("ALTER TABLE users ADD COLUMN phone TEXT")
3305 .unwrap();
3306 let contract = rt.db().collection_contract("users").unwrap();
3308 assert!(
3309 contract.declared_columns.iter().any(|c| c.name == "phone"),
3310 "phone column should be in contract"
3311 );
3312 assert!(
3314 contract.subscriptions.iter().any(|s| s.enabled),
3315 "subscription should remain enabled"
3316 );
3317 }
3318
3319 #[test]
3322 fn alter_drop_column_and_rls_on_event_enabled_table_succeeds() {
3323 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3324 rt.execute_query(
3325 "CREATE TABLE items (id INT, secret TEXT, status TEXT) WITH EVENTS TO items_events",
3326 )
3327 .unwrap();
3328 rt.execute_query("ALTER TABLE items DROP COLUMN secret")
3330 .unwrap();
3331 let contract = rt.db().collection_contract("items").unwrap();
3332 assert!(
3333 !contract.declared_columns.iter().any(|c| c.name == "secret"),
3334 "secret column should be removed"
3335 );
3336 rt.execute_query("ALTER TABLE items ENABLE ROW LEVEL SECURITY")
3338 .unwrap();
3339 assert!(
3341 contract.subscriptions.iter().any(|s| s.enabled),
3342 "subscription should remain enabled"
3343 );
3344 }
3345
3346 #[test]
3352 fn create_vector_marks_collection_as_turbo_baseline() {
3353 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3354 rt.execute_query("CREATE VECTOR embeddings DIM 4").unwrap();
3355 let store = rt.db().store();
3356 assert!(
3357 crate::runtime::vector_turbo_kind::is_turbo(&store, "embeddings"),
3358 "new vector collections must be turbo-marked baseline"
3359 );
3360 assert!(
3361 rt.db().turbo_state("embeddings").is_some(),
3362 "turbo_state must materialise after CREATE VECTOR"
3363 );
3364 }
3365
3366 #[test]
3370 fn create_table_does_not_mark_turbo() {
3371 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3372 rt.execute_query("CREATE TABLE plain (id INT)").unwrap();
3373 let store = rt.db().store();
3374 assert!(
3375 !crate::runtime::vector_turbo_kind::is_turbo(&store, "plain"),
3376 "non-vector collections must not gain the turbo marker"
3377 );
3378 assert!(rt.db().turbo_state("plain").is_none());
3379 }
3380
3381 #[test]
3384 fn create_collection_kind_vector_turbo_still_marked() {
3385 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3386 rt.execute_query("CREATE COLLECTION turbo_v KIND vector.turbo DIM 4")
3387 .unwrap();
3388 let store = rt.db().store();
3389 assert!(crate::runtime::vector_turbo_kind::is_turbo(
3390 &store, "turbo_v"
3391 ));
3392 assert!(rt.db().turbo_state("turbo_v").is_some());
3393 }
3394
3395 #[test]
3398 fn create_table_persists_and_introspects_ai_policy() {
3399 use crate::catalog::{ModerateDegradedMode, ModerateRejectAction};
3400 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3401 rt.execute_query(
3402 "CREATE TABLE posts (id INT, title TEXT, body TEXT, photo TEXT) WITH ( \
3403 EMBED (fields = ('title', 'body'), provider = 'openai', model = 'text-embedding-3-small'), \
3404 MODERATE (fields = ('body'), provider = 'openai', model = 'omni-moderation-latest', sync = true, degraded = closed, on_reject = flag), \
3405 VISION (image_field = 'photo', outputs = ('caption'), provider = 'openai', model = 'gpt-4o') \
3406 )",
3407 )
3408 .expect("create table with ai policy");
3409
3410 let contracts = rt.db().collection_contracts();
3411 let contract = contracts
3412 .iter()
3413 .find(|c| c.name == "posts")
3414 .expect("contract persisted");
3415 let policy = contract.ai_policy.as_ref().expect("ai policy persisted");
3416
3417 let embed = policy.embed.as_ref().expect("embed");
3418 assert_eq!(embed.fields, vec!["title".to_string(), "body".to_string()]);
3419 assert_eq!(embed.model, "text-embedding-3-small");
3420
3421 let moderate = policy.moderate.as_ref().expect("moderate");
3422 assert!(moderate.sync_gate);
3423 assert_eq!(moderate.degraded_mode, ModerateDegradedMode::Closed);
3424 assert_eq!(moderate.reject_action, ModerateRejectAction::Flag);
3425
3426 let vision = policy.vision.as_ref().expect("vision");
3427 assert_eq!(vision.image_field, "photo");
3428 assert_eq!(vision.model, "gpt-4o");
3429 }
3430
3431 #[test]
3435 fn create_table_rejects_ai_policy_incapable_provider() {
3436 let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3437 let err = rt
3438 .execute_query(
3439 "CREATE TABLE bad (id INT, body TEXT) WITH ( \
3440 EMBED (fields = ('body'), provider = 'anthropic', model = 'claude-3-5-sonnet') \
3441 )",
3442 )
3443 .unwrap_err();
3444 assert!(
3445 err.to_string()
3446 .contains("cannot serve the 'embed' modality"),
3447 "{err}"
3448 );
3449 assert!(
3450 rt.db().store().get_collection("bad").is_none(),
3451 "rejected DDL must not create the collection"
3452 );
3453 }
3454}