reddb_server/runtime/
impl_ddl.rs

1//! DDL execution: CREATE TABLE, DROP TABLE, ALTER TABLE via SQL AST
2//!
3//! Translates DDL statements into collection-level operations on the
4//! underlying `UnifiedStore`.  RedDB uses a flexible schema-on-read
5//! model, so column definitions are advisory metadata rather than
6//! rigid constraints.
7
8use 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    /// Execute CREATE TABLE
26    ///
27    /// Creates a new collection in the store.  Column definitions are
28    /// recorded for introspection but do not enforce rigid schema
29    /// constraints.
30    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        // Reject an AI policy that wires a provider/model to a modality it
54        // cannot serve — fail at DDL time, before the collection exists,
55        // using the capability matrix from #1269.
56        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        // Check if the collection already exists.
64        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        // Build and validate the contract before mutating storage so invalid
80        // SQL types / duplicate columns do not leave partial side effects.
81        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        // Create the collection.
85        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        // Issue #120 — feed the create into the schema-vocabulary
113        // reverse index so AskPipeline (#121) sees this collection.
114        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        // Partition metadata (Phase 2.2 PG parity).
124        //
125        // When the CREATE TABLE carries a `PARTITION BY RANGE|LIST|HASH (col)`
126        // clause, stamp the partition config into `red_config` under
127        // `partition.{table}.{by,column}`. Children are registered separately
128        // via `ALTER TABLE parent ATTACH PARTITION child ...`.
129        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        // Table-scoped multi-tenancy (Phase 2.5.4).
146        //
147        // `CREATE TABLE t (...) TENANT BY (col)` declaration:
148        //   1. Persists the `tenant_tables.{table}.column` marker so
149        //      INSERTs can auto-fill and future opens re-hydrate.
150        //   2. Registers the table in the in-memory `tenant_tables`
151        //      HashMap used by the DML auto-fill path.
152        //   3. Installs an implicit RLS policy equivalent to
153        //      `USING (col = CURRENT_TENANT())` across all actions.
154        //   4. Flips `rls_enabled_tables` on so the policy applies.
155        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    /// Bootstrap an internal graph collection declared `WITH ANALYTICS`
321    /// (issue #803). The SQL DDL path refuses `red.*` names via
322    /// [`is_system_schema_name`], so built-ins like `red.topology.cluster`
323    /// cannot be created through `CREATE GRAPH`; this method drives the same
324    /// contract-build / save / persist path directly, bypassing only that
325    /// guard. Idempotent: a no-op when the collection already exists (its
326    /// analytics contract is recovered from the WAL on restart), so it is safe
327    /// to call unconditionally on every boot.
328    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                // Issue #693 — durable kind marker that distinguishes
392                // this collection from the legacy `vector` path on
393                // every restart. Runtime state (TurboQuantIndex +
394                // TurboExtent) is lazily materialised by
395                // `RedDB::turbo_state` on first INSERT/SEARCH so the
396                // marker is the only thing that needs to survive.
397                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                // Eagerly construct the state so the TurboExtent (if
404                // any) is reserved at creation time rather than on
405                // first INSERT. Errors are swallowed: the lazy path in
406                // `turbo_state` will retry on the next access.
407                let _ = self.inner.db.turbo_state(&query.name);
408                return Ok(result);
409            }
410            // KIND blockchain — issue #523 foundation: stored on top of a
411            // Table-shaped collection. The `chain` marker + reserved-column
412            // discipline make the difference. Schema validation, conflict
413            // retries, and `verify_chain` come in later iterations.
414            "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        // Issue #522 — wire `SIGNED_BY (...)` into the runtime. The parser
445        // already produces a validated 32-byte pubkey list; installing
446        // the registry stamps the per-collection signer set into
447        // `red_config` so the INSERT path can load it cheaply.
448        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    /// Stamp `red.collection.{name}.kind = "chain"` and append the genesis
462    /// row. Idempotent against `IF NOT EXISTS`: if the collection already
463    /// has a row at height 0 we leave it.
464    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        // #524: prime the in-memory tip cache so the chain-tip endpoint and
496        // subsequent INSERTs don't have to scan the collection to find genesis.
497        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        // Slice G (#675) — register `vector.turbo` as the built-in
544        // baseline for every newly-created vector collection. The
545        // marker is the durable signal `RedDB::turbo_state` reads to
546        // materialise the TurboQuantIndex + TurboExtent on first
547        // INSERT/SEARCH. Pre-existing vector collections were created
548        // without the marker and stay on the brute-force fallback path
549        // until they are explicitly migrated — there is no implicit
550        // backfill in this slice.
551        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        // Eagerly materialise the per-collection turbo state so the
557        // TurboExtent (when the store is paged) is reserved at CREATE
558        // time rather than on the first INSERT. Failures are swallowed
559        // — the lazy path in `turbo_state` retries on next access.
560        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    /// Execute DROP TABLE
603    ///
604    /// Drops the collection and all its data from the store.
605    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        // Emit 1 collection_dropped event before storage is wiped.
636        // Queue is preserved; subscription is removed with the contract below.
637        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        // Issue #119: a dropped collection vanishes from every
672        // (tenant, role)'s visible-collections set. Auth is optional in
673        // embedded mode so guard the call.
674        if let Some(store) = self.inner.auth_store.read().clone() {
675            store.invalidate_visible_collections_cache();
676        }
677        // The native registry and targeted contract metadata publication above
678        // are the DROP commit boundary; do not turn this into a full database
679        // fingerprint/checkpoint.
680        // Issue #120 — drop both the collection entries *and* every
681        // index entry that was scoped to this collection.  Dropping the
682        // collection wipes columns + collection-name + type-tags +
683        // index hits in one pass via `purge_collection_entries`, so
684        // the explicit `DropIndex` calls would be redundant.
685        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    /// Issue #801 — guard `ALTER GRAPH ... ADD|DROP ANALYTICS` so analytics
891    /// lifecycle mutations only land on a collection declared as a graph. The
892    /// `<graph>.<output>` resolver only consults `analytics_config` on
893    /// graph-model contracts, so allowing the op on any other model would write
894    /// dead config the reader can never surface.
895    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    /// Execute ALTER TABLE
911    ///
912    /// In RedDB's schema-on-read model, ALTER TABLE operations are advisory.
913    /// ADD COLUMN records the schema intent, DROP COLUMN removes it, and
914    /// RENAME COLUMN is a metadata rename.  Existing data is not rewritten.
915    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        // Verify the table exists.
944        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        // Collect column-level changes upfront for schema-change event emission below.
954        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                    // Schema-on-read: column will be available on next insert.
981                    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                    // Persist child → parent binding in red_config so the
991                    // future planner-side pruner can enumerate children and
992                    // evaluate their bounds.
993                    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                    // Persist flag so RLS survives restart via red_config.
1018                    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                // Phase 2.5.4: retrofit tenancy onto an existing table.
1037                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                    // Contract is the single source of truth for the
1060                    // UPDATE/DELETE parse-time guard. The flag lands
1061                    // below via `apply_alter_operations_to_contract`;
1062                    // here we only publish the human-readable message.
1063                    messages.push(format!(
1064                        "append_only {} on '{}'",
1065                        if *on { "enabled" } else { "disabled" },
1066                        query.name
1067                    ));
1068                }
1069                AlterOperation::SetVersioned(on) => {
1070                    // Opt the collection into (or out of) Git-for-Data.
1071                    // Persists a row in red_vcs_settings; next AS OF /
1072                    // merge / diff against this table honours the
1073                    // flag. Retroactive: existing row versions whose
1074                    // xmins are still pinned by commits become
1075                    // reachable via AS OF COMMIT immediately.
1076                    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                    // Issue #522 — admin-gated registry mutation. The
1119                    // standard DDL `check_write` above gates by role; we
1120                    // additionally verify the collection actually has a
1121                    // signed-writes registry installed so this isn't a
1122                    // covert way to retrofit one (use `CREATE COLLECTION
1123                    // ... SIGNED_BY (...)` for that).
1124                    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                    // Issue #580 — validate that the collection has a
1172                    // timestamp column the lazy-on-scan filter can key
1173                    // off. Without one there is no anchor for "older
1174                    // than now - duration"; reject at ALTER time so the
1175                    // policy can never silently hide all rows.
1176                    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                    // Issue #801 — analytics lifecycle is graph-only. Reject the
1200                    // op on a non-graph collection so a misrouted ALTER fails
1201                    // loudly instead of silently writing analytics_config onto a
1202                    // model whose reader never resolves `<name>.<output>`.
1203                    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                            // Idempotent: adding an already-enabled output is a
1213                            // no-op — no error, no duplicate state.
1214                            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                        // Dropping an output that was never enabled is a clean,
1238                        // explicit error (AC4) rather than a silent no-op.
1239                        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        // Issue #301 — emit OperatorEvent when column schema changes on a
1267        // collection that has active event subscriptions, so operators know
1268        // downstream consumers may see a different payload shape.
1269        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        // Issue #120 — refresh the schema-vocabulary entries from the
1297        // post-ALTER contract. Drop+recreate inside the index keeps
1298        // the invalidation guarantee complete (no stale columns from
1299        // before an ALTER ... DROP COLUMN).
1300        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        Err(RedDBError::InvalidOperation(
1408            super::impl_primary_replica_file::EXTERNAL_FORK_DISABLED.into(),
1409        ))
1410    }
1411
1412    pub fn execute_drop_fork(
1413        &self,
1414        _raw_query: &str,
1415        _query: &DropForkQuery,
1416    ) -> RedDBResult<RuntimeQueryResult> {
1417        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1418        Err(RedDBError::InvalidOperation(
1419            super::impl_primary_replica_file::EXTERNAL_FORK_DISABLED.into(),
1420        ))
1421    }
1422
1423    pub fn execute_promote_fork(
1424        &self,
1425        _raw_query: &str,
1426        _query: &PromoteForkQuery,
1427    ) -> RedDBResult<RuntimeQueryResult> {
1428        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1429        Err(RedDBError::InvalidOperation(
1430            super::impl_primary_replica_file::EXTERNAL_FORK_DISABLED.into(),
1431        ))
1432    }
1433
1434    /// Execute EXPLAIN ALTER FOR CREATE TABLE
1435    ///
1436    /// Pure read: computes the schema diff between the target table's
1437    /// current `CollectionContract` and the embedded `CREATE TABLE` body,
1438    /// and returns it as SQL `ALTER TABLE` text (default) or structured
1439    /// JSON. Never mutates storage.
1440    pub fn execute_explain_alter(
1441        &self,
1442        raw_query: &str,
1443        query: &ExplainAlterQuery,
1444    ) -> RedDBResult<RuntimeQueryResult> {
1445        // Validate the target CREATE TABLE body so syntactically valid
1446        // but semantically broken targets (bad SQL types, duplicate
1447        // columns) are caught here rather than inside the diff engine.
1448        let analyzed = analyze_create_table(&query.target)
1449            .map_err(|err| RedDBError::Query(err.to_string()))?;
1450
1451        let current_contract = self.inner.db.collection_contract(&query.target.name);
1452
1453        // The column-diff planner cannot emit ADD/DROP CONSTRAINT. Do not
1454        // silently report a complete migration when table-level UNIQUE changes.
1455        let current_unique = current_contract
1456            .as_ref()
1457            .map(|contract| {
1458                contract
1459                    .table_def
1460                    .iter()
1461                    .flat_map(|table| &table.constraints)
1462                    .filter(|constraint| {
1463                        constraint.constraint_type == reddb_types::ConstraintType::Unique
1464                            && !contract.declared_columns.iter().any(|column| {
1465                                column.unique
1466                                    && constraint.columns == [column.name.clone()]
1467                                    && constraint.name == format!("uniq_{}", column.name)
1468                            })
1469                    })
1470                    .map(|constraint| (constraint.name.clone(), constraint.columns.clone()))
1471                    .collect::<BTreeSet<_>>()
1472            })
1473            .unwrap_or_default();
1474        let target_unique = analyzed
1475            .unique_constraints
1476            .into_iter()
1477            .map(|constraint| (constraint.name, constraint.columns))
1478            .collect::<BTreeSet<_>>();
1479        if current_unique != target_unique {
1480            return Err(RedDBError::Query("NOT_YET_SUPPORTED: EXPLAIN ALTER cannot plan changes to table-level UNIQUE constraints".to_string()));
1481        }
1482
1483        let current_columns: Vec<crate::physical::DeclaredColumnContract> = current_contract
1484            .as_ref()
1485            .map(|c| c.declared_columns.clone())
1486            .unwrap_or_default();
1487
1488        let diff = super::schema_diff::compute_column_diff(
1489            &query.target.name,
1490            &current_columns,
1491            &query.target.columns,
1492        );
1493
1494        let rendered = match query.format {
1495            ExplainFormat::Sql => super::schema_diff::format_as_sql(&diff),
1496            ExplainFormat::Json => super::schema_diff::format_as_json(&diff),
1497        };
1498
1499        let format_label = match query.format {
1500            ExplainFormat::Sql => "sql",
1501            ExplainFormat::Json => "json",
1502        };
1503
1504        let columns = vec![
1505            "table".to_string(),
1506            "format".to_string(),
1507            "diff".to_string(),
1508        ];
1509        let row = vec![
1510            ("table".to_string(), Value::text(query.target.name.clone())),
1511            ("format".to_string(), Value::text(format_label.to_string())),
1512            ("diff".to_string(), Value::text(rendered)),
1513        ];
1514
1515        Ok(RuntimeQueryResult::ok_records(
1516            raw_query.to_string(),
1517            columns,
1518            vec![row],
1519            "explain",
1520        ))
1521    }
1522
1523    /// Execute CREATE INDEX
1524    ///
1525    /// Registers a new index on a collection, builds it from existing data,
1526    /// and makes it available to the query executor for O(1) lookups.
1527    pub fn execute_create_index(
1528        &self,
1529        raw_query: &str,
1530        query: &CreateIndexQuery,
1531    ) -> RedDBResult<RuntimeQueryResult> {
1532        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1533        let topology_lock = self
1534            .inner
1535            .index_store
1536            .collection_topology_lock(&query.table);
1537        let _topology_guard = topology_lock.write();
1538        let store = self.inner.db.store();
1539
1540        // Verify the table exists
1541        let manager = store
1542            .get_collection(&query.table)
1543            .ok_or_else(|| RedDBError::NotFound(format!("table '{}' not found", query.table)))?;
1544
1545        if query.if_not_exists
1546            && self
1547                .inner
1548                .index_store
1549                .list_indices(&query.table)
1550                .iter()
1551                .any(|index| index.name == query.name)
1552        {
1553            return Ok(RuntimeQueryResult::ok_message(
1554                raw_query.to_string(),
1555                &format!(
1556                    "index '{}' already exists on '{}'; skipped",
1557                    query.name, query.table
1558                ),
1559                "create",
1560            ));
1561        }
1562
1563        let method_kind = match query.method {
1564            IndexMethod::Hash => super::index_store::IndexMethodKind::Hash,
1565            IndexMethod::BTree => super::index_store::IndexMethodKind::BTree,
1566            IndexMethod::Bitmap => super::index_store::IndexMethodKind::Bitmap,
1567            // Generic spatial request → engine default spatial backend
1568            // (disk-resident H3 as of PRD #1574 slice 4, #1578).
1569            IndexMethod::Spatial => super::index_store::IndexMethodKind::default_spatial(),
1570            IndexMethod::H3 { resolution } => {
1571                super::index_store::IndexMethodKind::H3 { resolution }
1572            }
1573        };
1574
1575        // Extract fields from existing entities for indexing. Row
1576        // entities may arrive in either the "named" HashMap layout
1577        // (gRPC `BulkInsertBinary` path) OR the columnar layout
1578        // (HTTP `POST /collections/X/bulk/rows` fast path, which uses
1579        // `schema: Some(Arc<Vec<String>>)` + `columns: Vec<Value>` and
1580        // leaves `named == None`). Prior to this commit the columnar
1581        // branch returned an empty field list, so `CREATE INDEX` built
1582        // a zero-entity index over HTTP-inserted data even though the
1583        // data was queryable via `SELECT`.
1584        let entities = manager.query_all(|_| true);
1585        let mut entity_fields: Vec<(crate::storage::unified::EntityId, Vec<(String, Value)>)> =
1586            entities
1587                .iter()
1588                .map(|e| {
1589                    let fields = match &e.data {
1590                        crate::storage::EntityData::Row(row) => {
1591                            if let Some(ref named) = row.named {
1592                                named.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
1593                            } else if let Some(ref schema) = row.schema {
1594                                // Columnar layout — pair each column
1595                                // with its positional name from the
1596                                // shared schema Arc.
1597                                schema
1598                                    .iter()
1599                                    .zip(row.columns.iter())
1600                                    .map(|(k, v)| (k.clone(), v.clone()))
1601                                    .collect()
1602                            } else {
1603                                Vec::new()
1604                            }
1605                        }
1606                        crate::storage::EntityData::Node(node) => node
1607                            .properties
1608                            .iter()
1609                            .map(|(k, v)| (k.clone(), v.clone()))
1610                            .collect(),
1611                        crate::storage::EntityData::TimeSeries(ts) => ts
1612                            .fields
1613                            .iter()
1614                            .map(|(k, v)| (k.clone(), v.clone()))
1615                            .collect(),
1616                        _ => Vec::new(),
1617                    };
1618                    (e.id, fields)
1619                })
1620                .collect();
1621
1622        if query.unique
1623            && matches!(
1624                method_kind,
1625                super::index_store::IndexMethodKind::Hash
1626                    | super::index_store::IndexMethodKind::BTree
1627            )
1628        {
1629            // Only live rows reserve a key: the dead versions of an updated or
1630            // deleted row share its key but are not duplicates. This is DDL
1631            // only — reopening must still rebuild an index over data that an
1632            // older release let duplicate.
1633            let snapshots = self.snapshot_manager();
1634            let own_xids = self.own_transaction_xids();
1635            let mut live = entities.iter().map(|entity| {
1636                snapshots.row_reserves_unique_key(entity.xmin, entity.xmax, &own_xids)
1637            });
1638            entity_fields.retain(|_| live.next().unwrap_or(false));
1639            if let Some((earlier, later)) =
1640                super::index_store::first_duplicate_unique_key(&query.columns, &entity_fields)
1641            {
1642                return Err(RedDBError::Query(format!(
1643                    "cannot create unique index '{}' on '{}': rows {} and {} share the same ({}) key",
1644                    query.name,
1645                    query.table,
1646                    earlier.raw(),
1647                    later.raw(),
1648                    query.columns.join(", ")
1649                )));
1650            }
1651        }
1652
1653        let index_growth_rows = entity_fields
1654            .iter()
1655            .map(|(_, fields)| fields.clone())
1656            .collect::<Vec<_>>();
1657        let _reservation = self.admit_non_evictable_growth(
1658            crate::storage::memory_pools::MemoryPool::IndexMemory,
1659            &format!("create index {}", query.name),
1660            crate::runtime::memory_admission::estimate_index_growth(
1661                &index_growth_rows,
1662                &query.columns,
1663            ),
1664        )?;
1665
1666        // Build the index
1667        let indexed_count = self
1668            .inner
1669            .index_store
1670            .create_index(
1671                &query.name,
1672                &query.table,
1673                &query.columns,
1674                method_kind,
1675                query.unique,
1676                &entity_fields,
1677            )
1678            .map_err(RedDBError::Internal)?;
1679
1680        let analyzed = crate::storage::query::planner::stats_catalog::analyze_entity_fields(
1681            &query.table,
1682            &entity_fields,
1683        );
1684        crate::storage::query::planner::stats_catalog::persist_table_stats(&store, &analyzed);
1685        self.invalidate_plan_cache();
1686
1687        // Register metadata
1688        self.inner
1689            .index_store
1690            .register(super::index_store::RegisteredIndex {
1691                name: query.name.clone(),
1692                collection: query.table.clone(),
1693                columns: query.columns.clone(),
1694                method: method_kind,
1695                unique: query.unique,
1696            });
1697        self.persist_runtime_index_descriptor(super::index_store::RegisteredIndex {
1698            name: query.name.clone(),
1699            collection: query.table.clone(),
1700            columns: query.columns.clone(),
1701            method: method_kind,
1702            unique: query.unique,
1703        })?;
1704        // Issue #120 — surface the index name + indexed columns in
1705        // the schema-vocabulary so AskPipeline (#121) can resolve
1706        // "the email index" back to its collection.
1707        self.schema_vocabulary_apply(crate::runtime::schema_vocabulary::DdlEvent::CreateIndex {
1708            collection: query.table.clone(),
1709            index: query.name.clone(),
1710            columns: query.columns.clone(),
1711        });
1712
1713        let method_str = format!("{}", query.method);
1714        let unique_str = if query.unique { "unique " } else { "" };
1715        let cols = query.columns.join(", ");
1716        let total_count = entity_fields.len();
1717        let coverage_detail =
1718            if matches!(method_kind, super::index_store::IndexMethodKind::H3 { .. })
1719                && total_count > 0
1720                && indexed_count == 0
1721            {
1722                format!(
1723                    "{} of {} entities indexed — no indexable geo value in '{}'; expected {}",
1724                    indexed_count,
1725                    total_count,
1726                    cols,
1727                    crate::geo::RECOGNIZED_GEO_SHAPES
1728                )
1729            } else if matches!(method_kind, super::index_store::IndexMethodKind::H3 { .. }) {
1730                format!("{} of {} entities indexed", indexed_count, total_count)
1731            } else {
1732                format!("{} entities indexed", indexed_count)
1733            };
1734
1735        Ok(RuntimeQueryResult::ok_message(
1736            raw_query.to_string(),
1737            &format!(
1738                "{}index '{}' created on '{}' ({}) using {} ({})",
1739                unique_str, query.name, query.table, cols, method_str, coverage_detail
1740            ),
1741            "create",
1742        ))
1743    }
1744
1745    /// Execute DROP INDEX
1746    ///
1747    /// Removes an index from a collection.
1748    pub fn execute_drop_index(
1749        &self,
1750        raw_query: &str,
1751        query: &DropIndexQuery,
1752    ) -> RedDBResult<RuntimeQueryResult> {
1753        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1754        let topology_lock = self
1755            .inner
1756            .index_store
1757            .collection_topology_lock(&query.table);
1758        let _topology_guard = topology_lock.write();
1759        let store = self.inner.db.store();
1760
1761        // Verify the table exists
1762        if store.get_collection(&query.table).is_none() {
1763            if query.if_exists {
1764                return Ok(RuntimeQueryResult::ok_message(
1765                    raw_query.to_string(),
1766                    &format!("table '{}' does not exist", query.table),
1767                    "drop",
1768                ));
1769            }
1770            return Err(RedDBError::NotFound(format!(
1771                "table '{}' not found",
1772                query.table
1773            )));
1774        }
1775
1776        // Remove from IndexStore
1777        self.inner.index_store.drop_index(&query.name, &query.table);
1778        self.persist_runtime_index_drop(&query.table, &query.name)?;
1779        self.invalidate_plan_cache();
1780        // Issue #120 — keep the schema-vocabulary index entry in sync.
1781        self.schema_vocabulary_apply(crate::runtime::schema_vocabulary::DdlEvent::DropIndex {
1782            collection: query.table.clone(),
1783            index: query.name.clone(),
1784        });
1785
1786        Ok(RuntimeQueryResult::ok_message(
1787            raw_query.to_string(),
1788            &format!("index '{}' dropped from '{}'", query.name, query.table),
1789            "drop",
1790        ))
1791    }
1792
1793    fn execute_drop_typed_collection(
1794        &self,
1795        raw_query: &str,
1796        name: &str,
1797        if_exists: bool,
1798        expected_model: CollectionModel,
1799        label: &str,
1800    ) -> RedDBResult<RuntimeQueryResult> {
1801        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1802        if is_system_schema_name(name) {
1803            return Err(RedDBError::Query("system schema is read-only".to_string()));
1804        }
1805        let store = self.inner.db.store();
1806        if store.get_collection(name).is_none() {
1807            if if_exists {
1808                return Ok(RuntimeQueryResult::ok_message(
1809                    raw_query.to_string(),
1810                    &format!("{label} '{name}' does not exist"),
1811                    "drop",
1812                ));
1813            }
1814            return Err(RedDBError::NotFound(format!("{label} '{name}' not found")));
1815        }
1816
1817        let actual = self
1818            .inner
1819            .db
1820            .collection_contract(name)
1821            .map(|contract| contract.declared_model)
1822            .map(Ok)
1823            .unwrap_or_else(|| {
1824                polymorphic_resolver::resolve(name, &self.inner.db.catalog_model_snapshot())
1825            })?;
1826        polymorphic_resolver::ensure_model_match(expected_model, actual)?;
1827        self.drop_collection_storage(raw_query, name, label)
1828    }
1829
1830    pub fn execute_truncate(
1831        &self,
1832        raw_query: &str,
1833        query: &TruncateQuery,
1834    ) -> RedDBResult<RuntimeQueryResult> {
1835        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1836        if is_system_schema_name(&query.name) {
1837            return Err(RedDBError::Query("system schema is read-only".to_string()));
1838        }
1839
1840        let label = query
1841            .model
1842            .map(polymorphic_resolver::model_name)
1843            .unwrap_or("collection");
1844        let store = self.inner.db.store();
1845        if store.get_collection(&query.name).is_none() {
1846            if query.if_exists {
1847                return Ok(RuntimeQueryResult::ok_message(
1848                    raw_query.to_string(),
1849                    &format!("{label} '{}' does not exist", query.name),
1850                    "truncate",
1851                ));
1852            }
1853            return Err(RedDBError::NotFound(format!(
1854                "{label} '{}' not found",
1855                query.name
1856            )));
1857        }
1858
1859        let actual =
1860            polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
1861        if let Some(expected) = query.model {
1862            polymorphic_resolver::ensure_model_match(expected, actual)?;
1863        }
1864
1865        if actual == CollectionModel::Queue {
1866            return self.execute_queue_command(
1867                raw_query,
1868                &QueueCommand::Purge {
1869                    queue: query.name.clone(),
1870                },
1871            );
1872        }
1873
1874        // Count before wiping so we can emit the aggregated truncate event.
1875        let affected = self.truncate_collection_entities(&query.name)?;
1876        // Emit 1 truncate event (not N delete events) for event-enabled collections.
1877        crate::runtime::mutation::emit_truncate_event_for_collection(self, &query.name, affected)?;
1878        self.inner.db.invalidate_vector_index(&query.name);
1879        self.clear_table_planner_stats(&query.name);
1880        self.invalidate_result_cache();
1881
1882        Ok(RuntimeQueryResult::ok_message(
1883            raw_query.to_string(),
1884            &format!(
1885                "{affected} entities truncated from {label} '{}'",
1886                query.name
1887            ),
1888            "truncate",
1889        ))
1890    }
1891
1892    fn truncate_collection_entities(&self, name: &str) -> RedDBResult<u64> {
1893        let _indexed_mutation = self.index_store_ref().begin_indexed_mutation();
1894        let store = self.inner.db.store();
1895        let Some(manager) = store.get_collection(name) else {
1896            return Ok(0);
1897        };
1898        let entities = manager.query_all(|_| true);
1899        if entities.is_empty() {
1900            return Ok(0);
1901        }
1902
1903        for entity in &entities {
1904            let fields = entity_index_fields(&entity.data);
1905            self.inner
1906                .index_store
1907                .index_entity_delete(name, entity.id, &fields)
1908                .map_err(RedDBError::Internal)?;
1909        }
1910
1911        let ids = entities.iter().map(|entity| entity.id).collect::<Vec<_>>();
1912        let deleted_ids = store
1913            .delete_batch(name, &ids)
1914            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1915        for id in &deleted_ids {
1916            store.context_index().remove_entity(*id);
1917        }
1918        Ok(deleted_ids.len() as u64)
1919    }
1920
1921    fn drop_collection_storage(
1922        &self,
1923        raw_query: &str,
1924        name: &str,
1925        label: &str,
1926    ) -> RedDBResult<RuntimeQueryResult> {
1927        let store = self.inner.db.store();
1928
1929        // Emit 1 collection_dropped event before storage is wiped.
1930        // Queue is preserved; subscription is removed with the contract below.
1931        let final_count = store
1932            .get_collection(name)
1933            .map(|manager| manager.query_all(|_| true).len() as u64)
1934            .unwrap_or(0);
1935        crate::runtime::mutation::emit_collection_dropped_event_for_collection(
1936            self,
1937            name,
1938            final_count,
1939        )?;
1940
1941        let topology_lock = self.inner.index_store.collection_topology_lock(name);
1942        let _topology_guard = topology_lock.write();
1943        let orphaned_indices: Vec<String> = self
1944            .inner
1945            .index_store
1946            .list_indices(name)
1947            .into_iter()
1948            .map(|index| index.name)
1949            .collect();
1950        for index_name in &orphaned_indices {
1951            self.inner.index_store.drop_index(index_name, name);
1952        }
1953
1954        store
1955            .drop_collection(name)
1956            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1957        self.inner.db.invalidate_vector_index(name);
1958        self.inner.db.clear_collection_default_ttl_ms(name);
1959        self.inner
1960            .db
1961            .remove_collection_contract(name)
1962            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1963        self.clear_table_planner_stats(name);
1964        self.invalidate_result_cache();
1965        if let Some(store) = self.inner.auth_store.read().clone() {
1966            store.invalidate_visible_collections_cache();
1967        }
1968        // The native registry and targeted contract metadata publication above
1969        // are the DROP commit boundary; do not turn this into a full database
1970        // fingerprint/checkpoint.
1971        self.schema_vocabulary_apply(
1972            crate::runtime::schema_vocabulary::DdlEvent::DropCollection {
1973                collection: name.to_string(),
1974            },
1975        );
1976
1977        Ok(RuntimeQueryResult::ok_message(
1978            raw_query.to_string(),
1979            &format!("{label} '{name}' dropped"),
1980            "drop",
1981        ))
1982    }
1983}
1984
1985pub(crate) fn is_system_schema_name(name: &str) -> bool {
1986    name == "red" || name.starts_with("red.") || name.starts_with("__red_schema_")
1987}
1988
1989fn entity_index_fields(data: &EntityData) -> Vec<(String, Value)> {
1990    match data {
1991        EntityData::Row(row) => {
1992            if let Some(ref named) = row.named {
1993                named
1994                    .iter()
1995                    .map(|(key, value)| (key.clone(), value.clone()))
1996                    .collect()
1997            } else if let Some(ref schema) = row.schema {
1998                schema
1999                    .iter()
2000                    .zip(row.columns.iter())
2001                    .map(|(key, value)| (key.clone(), value.clone()))
2002                    .collect()
2003            } else {
2004                Vec::new()
2005            }
2006        }
2007        EntityData::Node(node) => node
2008            .properties
2009            .iter()
2010            .map(|(key, value)| (key.clone(), value.clone()))
2011            .collect(),
2012        _ => Vec::new(),
2013    }
2014}
2015
2016/// DDL-time gate for a declared AI policy (issue #1271). Each declared
2017/// modality block is checked against the provider capability matrix
2018/// (#1269): a provider/model that cannot serve the requested modality is
2019/// rejected before the collection is created. Uses the built-in registry
2020/// (no per-deployment overrides at DDL time).
2021fn validate_ai_policy_modalities(policy: &crate::catalog::AiPolicy) -> RedDBResult<()> {
2022    use crate::runtime::ai::provider_capabilities::{Modality, Registry};
2023    let registry = Registry::new();
2024    let check = |provider: &str, model: &str, modality: Modality| -> RedDBResult<()> {
2025        registry
2026            .validate_policy_modality(provider, model, modality)
2027            .map_err(|err| RedDBError::Query(err.to_string()))
2028    };
2029    if let Some(embed) = &policy.embed {
2030        check(&embed.provider, &embed.model, Modality::Embed)?;
2031    }
2032    if let Some(moderate) = &policy.moderate {
2033        check(&moderate.provider, &moderate.model, Modality::Moderate)?;
2034    }
2035    if let Some(vision) = &policy.vision {
2036        check(&vision.provider, &vision.model, Modality::Vision)?;
2037    }
2038    Ok(())
2039}
2040
2041fn collection_contract_from_create_table(
2042    query: &CreateTableQuery,
2043) -> RedDBResult<crate::physical::CollectionContract> {
2044    reddb_engine::catalog::ddl::collection_contract_from_create_table(query, current_unix_ms())
2045        .map_err(RedDBError::Query)
2046}
2047
2048fn default_collection_contract_for_existing_table(
2049    name: &str,
2050) -> crate::physical::CollectionContract {
2051    let now = current_unix_ms();
2052    crate::physical::CollectionContract {
2053        name: name.to_string(),
2054        declared_model: crate::catalog::CollectionModel::Table,
2055        schema_mode: crate::catalog::SchemaMode::SemiStructured,
2056        origin: crate::physical::ContractOrigin::Explicit,
2057        version: 0,
2058        created_at_unix_ms: now,
2059        updated_at_unix_ms: now,
2060        default_ttl_ms: None,
2061        vector_dimension: None,
2062        vector_metric: None,
2063        context_index_fields: Vec::new(),
2064        declared_columns: Vec::new(),
2065        table_def: Some(reddb_types::TableDef::new(name.to_string())),
2066        timestamps_enabled: false,
2067        context_index_enabled: false,
2068        metrics_raw_retention_ms: None,
2069        metrics_rollup_policies: Vec::new(),
2070        metrics_tenant_identity: None,
2071        metrics_namespace: None,
2072        append_only: false,
2073        subscriptions: Vec::new(),
2074        analytics_config: Vec::new(),
2075        session_key: None,
2076        session_gap_ms: None,
2077        retention_duration_ms: None,
2078        analytical_storage: None,
2079
2080        ai_policy: None,
2081    }
2082}
2083
2084fn keyed_collection_contract(
2085    name: &str,
2086    model: crate::catalog::CollectionModel,
2087    analytics_config: Vec<crate::catalog::AnalyticsViewDescriptor>,
2088) -> crate::physical::CollectionContract {
2089    let now = current_unix_ms();
2090    crate::physical::CollectionContract {
2091        name: name.to_string(),
2092        declared_model: model,
2093        schema_mode: crate::catalog::SchemaMode::Dynamic,
2094        origin: crate::physical::ContractOrigin::Explicit,
2095        version: 1,
2096        created_at_unix_ms: now,
2097        updated_at_unix_ms: now,
2098        default_ttl_ms: None,
2099        vector_dimension: None,
2100        vector_metric: None,
2101        context_index_fields: Vec::new(),
2102        declared_columns: Vec::new(),
2103        table_def: None,
2104        timestamps_enabled: false,
2105        context_index_enabled: false,
2106        metrics_raw_retention_ms: None,
2107        metrics_rollup_policies: Vec::new(),
2108        metrics_tenant_identity: None,
2109        metrics_namespace: None,
2110        append_only: false,
2111        subscriptions: Vec::new(),
2112        analytics_config,
2113        session_key: None,
2114        session_gap_ms: None,
2115        retention_duration_ms: None,
2116        analytical_storage: None,
2117
2118        ai_policy: None,
2119    }
2120}
2121
2122fn metrics_collection_contract(query: &CreateTableQuery) -> crate::physical::CollectionContract {
2123    let now = current_unix_ms();
2124    crate::physical::CollectionContract {
2125        name: query.name.clone(),
2126        declared_model: crate::catalog::CollectionModel::Metrics,
2127        schema_mode: crate::catalog::SchemaMode::SemiStructured,
2128        origin: crate::physical::ContractOrigin::Explicit,
2129        version: 1,
2130        created_at_unix_ms: now,
2131        updated_at_unix_ms: now,
2132        default_ttl_ms: query.default_ttl_ms,
2133        vector_dimension: None,
2134        vector_metric: None,
2135        context_index_fields: Vec::new(),
2136        declared_columns: Vec::new(),
2137        table_def: None,
2138        timestamps_enabled: false,
2139        context_index_enabled: false,
2140        metrics_raw_retention_ms: query.default_ttl_ms,
2141        metrics_rollup_policies: query.metrics_rollup_policies.clone(),
2142        metrics_tenant_identity: Some(
2143            query
2144                .tenant_by
2145                .clone()
2146                .unwrap_or_else(|| "current_tenant".to_string()),
2147        ),
2148        metrics_namespace: Some("default".to_string()),
2149        append_only: true,
2150        subscriptions: Vec::new(),
2151        analytics_config: Vec::new(),
2152        session_key: None,
2153        session_gap_ms: None,
2154        retention_duration_ms: None,
2155        analytical_storage: None,
2156
2157        ai_policy: None,
2158    }
2159}
2160
2161fn vector_collection_contract(query: &CreateVectorQuery) -> crate::physical::CollectionContract {
2162    let now = current_unix_ms();
2163    crate::physical::CollectionContract {
2164        name: query.name.clone(),
2165        declared_model: crate::catalog::CollectionModel::Vector,
2166        schema_mode: crate::catalog::SchemaMode::Dynamic,
2167        origin: crate::physical::ContractOrigin::Explicit,
2168        version: 1,
2169        created_at_unix_ms: now,
2170        updated_at_unix_ms: now,
2171        default_ttl_ms: None,
2172        vector_dimension: Some(query.dimension),
2173        vector_metric: Some(query.metric),
2174        context_index_fields: Vec::new(),
2175        declared_columns: Vec::new(),
2176        table_def: None,
2177        timestamps_enabled: false,
2178        context_index_enabled: false,
2179        metrics_raw_retention_ms: None,
2180        metrics_rollup_policies: Vec::new(),
2181        metrics_tenant_identity: None,
2182        metrics_namespace: None,
2183        append_only: false,
2184        subscriptions: Vec::new(),
2185        analytics_config: Vec::new(),
2186        session_key: None,
2187        session_gap_ms: None,
2188        retention_duration_ms: None,
2189        analytical_storage: None,
2190
2191        ai_policy: None,
2192    }
2193}
2194
2195fn apply_alter_operations_to_contract(
2196    contract: &mut crate::physical::CollectionContract,
2197    operations: &[AlterOperation],
2198) {
2199    if contract.table_def.is_none() {
2200        contract.table_def = Some(reddb_types::TableDef::new(contract.name.clone()));
2201    }
2202    for operation in operations {
2203        match operation {
2204            AlterOperation::AddColumn(column) => {
2205                if !contract
2206                    .declared_columns
2207                    .iter()
2208                    .any(|existing| existing.name == column.name)
2209                {
2210                    contract
2211                        .declared_columns
2212                        .push(declared_column_contract_from_ddl(column));
2213                }
2214                if let Some(table_def) = contract.table_def.as_mut() {
2215                    if table_def.get_column(&column.name).is_none() {
2216                        if let Ok(column_def) = column_def_from_ddl(column) {
2217                            if column.primary_key {
2218                                table_def.primary_key.push(column.name.clone());
2219                                table_def.constraints.push(
2220                                    reddb_types::Constraint::new(
2221                                        format!("pk_{}", column.name),
2222                                        reddb_types::ConstraintType::PrimaryKey,
2223                                    )
2224                                    .on_columns(vec![column.name.clone()]),
2225                                );
2226                            }
2227                            if column.unique {
2228                                table_def.constraints.push(
2229                                    reddb_types::Constraint::new(
2230                                        format!("uniq_{}", column.name),
2231                                        reddb_types::ConstraintType::Unique,
2232                                    )
2233                                    .on_columns(vec![column.name.clone()]),
2234                                );
2235                            }
2236                            if column.not_null {
2237                                table_def.constraints.push(
2238                                    reddb_types::Constraint::new(
2239                                        format!("not_null_{}", column.name),
2240                                        reddb_types::ConstraintType::NotNull,
2241                                    )
2242                                    .on_columns(vec![column.name.clone()]),
2243                                );
2244                            }
2245                            table_def.columns.push(column_def);
2246                        }
2247                    }
2248                }
2249            }
2250            AlterOperation::DropColumn(name) => {
2251                contract
2252                    .declared_columns
2253                    .retain(|column| column.name != *name);
2254                if let Some(table_def) = contract.table_def.as_mut() {
2255                    if let Some(index) = table_def.column_index(name) {
2256                        table_def.columns.remove(index);
2257                    }
2258                    table_def.primary_key.retain(|column| column != name);
2259                    table_def.constraints.retain(|constraint| {
2260                        !constraint.columns.iter().any(|column| column == name)
2261                    });
2262                    table_def
2263                        .indexes
2264                        .retain(|index| !index.columns.iter().any(|column| column == name));
2265                }
2266            }
2267            AlterOperation::RenameColumn { from, to } => {
2268                if contract
2269                    .declared_columns
2270                    .iter()
2271                    .any(|column| column.name == *to)
2272                {
2273                    continue;
2274                }
2275                if let Some(column) = contract
2276                    .declared_columns
2277                    .iter_mut()
2278                    .find(|column| column.name == *from)
2279                {
2280                    column.name = to.clone();
2281                }
2282                if let Some(table_def) = contract.table_def.as_mut() {
2283                    if let Some(column) = table_def
2284                        .columns
2285                        .iter_mut()
2286                        .find(|column| column.name == *from)
2287                    {
2288                        column.name = to.clone();
2289                    }
2290                    for primary_key in &mut table_def.primary_key {
2291                        if *primary_key == *from {
2292                            *primary_key = to.clone();
2293                        }
2294                    }
2295                    for constraint in &mut table_def.constraints {
2296                        for column in &mut constraint.columns {
2297                            if *column == *from {
2298                                *column = to.clone();
2299                            }
2300                        }
2301                        if let Some(ref_columns) = constraint.ref_columns.as_mut() {
2302                            for column in ref_columns {
2303                                if *column == *from {
2304                                    *column = to.clone();
2305                                }
2306                            }
2307                        }
2308                    }
2309                    for index in &mut table_def.indexes {
2310                        for column in &mut index.columns {
2311                            if *column == *from {
2312                                *column = to.clone();
2313                            }
2314                        }
2315                    }
2316                }
2317            }
2318            // Partition ops don't touch the column contract — metadata is
2319            // persisted separately via `red_config.partition.*`.
2320            AlterOperation::AttachPartition { .. } | AlterOperation::DetachPartition { .. } => {}
2321            // RLS toggles don't touch the column contract — flag is persisted
2322            // separately via `red_config.rls.enabled.{table}` and enforced
2323            // through the in-memory `rls_enabled_tables` set.
2324            AlterOperation::EnableRowLevelSecurity | AlterOperation::DisableRowLevelSecurity => {}
2325            // Phase 2.5.4: tenancy toggles persist via `red_config.tenant_tables.*`
2326            // and are enforced through `tenant_tables` + RLS auto-policy.
2327            AlterOperation::EnableTenancy { .. } | AlterOperation::DisableTenancy => {}
2328            AlterOperation::SetAppendOnly(on) => {
2329                contract.append_only = *on;
2330            }
2331            // VCS opt-in is persisted to red_vcs_settings by the
2332            // executor, not the contract — nothing to do here.
2333            AlterOperation::SetVersioned(_) => {}
2334            AlterOperation::EnableEvents(subscription) => {
2335                let mut subscription = subscription.clone();
2336                subscription.source = contract.name.clone();
2337                subscription.enabled = true;
2338                if let Some(existing) = contract
2339                    .subscriptions
2340                    .iter_mut()
2341                    .find(|existing| existing.target_queue == subscription.target_queue)
2342                {
2343                    *existing = subscription;
2344                } else {
2345                    contract.subscriptions.push(subscription);
2346                }
2347            }
2348            AlterOperation::DisableEvents => {
2349                for subscription in &mut contract.subscriptions {
2350                    subscription.enabled = false;
2351                }
2352            }
2353            AlterOperation::AddSubscription { name, descriptor } => {
2354                let mut sub = descriptor.clone();
2355                sub.name = name.clone();
2356                sub.source = contract.name.clone();
2357                sub.enabled = true;
2358                if let Some(existing) = contract.subscriptions.iter_mut().find(|s| s.name == *name)
2359                {
2360                    *existing = sub;
2361                } else {
2362                    contract.subscriptions.push(sub);
2363                }
2364            }
2365            AlterOperation::DropSubscription { name } => {
2366                contract.subscriptions.retain(|s| s.name != *name);
2367            }
2368            // Signer registry mutations live in `red_config` outside the
2369            // contract surface — the executor applied them directly via
2370            // `signed_writes_kind::{add,revoke}_signer`. Nothing to fold
2371            // into the column-shaped contract.
2372            AlterOperation::AddSigner { .. } | AlterOperation::RevokeSigner { .. } => {}
2373            AlterOperation::SetRetention { duration_ms } => {
2374                contract.retention_duration_ms = Some(*duration_ms);
2375            }
2376            AlterOperation::UnsetRetention => {
2377                contract.retention_duration_ms = None;
2378            }
2379            // Issue #801 — fold analytics lifecycle into the WAL-backed
2380            // `analytics_config` so the change is durable and the
2381            // `<graph>.<output>` resolver picks it up on the next read.
2382            AlterOperation::AddAnalytics(views) => {
2383                for view in views {
2384                    // Idempotent: skip outputs already enabled so a repeated
2385                    // ADD never duplicates state. The first declaration wins —
2386                    // re-declaring options requires DROP then ADD.
2387                    if !contract
2388                        .analytics_config
2389                        .iter()
2390                        .any(|existing| existing.output == view.output)
2391                    {
2392                        contract.analytics_config.push(view.clone());
2393                    }
2394                }
2395            }
2396            AlterOperation::DropAnalytics(output) => {
2397                contract
2398                    .analytics_config
2399                    .retain(|view| view.output != *output);
2400            }
2401        }
2402    }
2403}
2404
2405/// Issue #580 — returns true if the contract carries at least one
2406/// column the retention filter can use as a timestamp anchor:
2407/// either `WITH timestamps = true` (auto `created_at` / `updated_at`)
2408/// or a user-declared column with a temporal data_type.
2409pub(crate) fn retention_timestamp_column_exists(
2410    contract: &crate::physical::CollectionContract,
2411) -> bool {
2412    if contract.timestamps_enabled {
2413        return true;
2414    }
2415    if matches!(
2416        contract.declared_model,
2417        crate::catalog::CollectionModel::TimeSeries | crate::catalog::CollectionModel::Metrics
2418    ) {
2419        // Time-series and metrics collections carry an intrinsic
2420        // timestamp axis on every row even without a declared column;
2421        // retention has a natural anchor.
2422        return true;
2423    }
2424    contract
2425        .declared_columns
2426        .iter()
2427        .any(|column| is_temporal_data_type(&column.data_type))
2428}
2429
2430fn is_temporal_data_type(data_type: &str) -> bool {
2431    let upper = data_type.to_ascii_uppercase();
2432    matches!(
2433        upper.as_str(),
2434        "TIMESTAMP" | "TIMESTAMPMS" | "TIMESTAMP_MS" | "DATETIME" | "DATE"
2435    )
2436}
2437
2438fn validate_event_subscriptions(
2439    runtime: &RedDBRuntime,
2440    source: &str,
2441    subscriptions: &[crate::catalog::SubscriptionDescriptor],
2442) -> RedDBResult<()> {
2443    for subscription in subscriptions
2444        .iter()
2445        .filter(|subscription| subscription.enabled)
2446    {
2447        if subscription.all_tenants && crate::runtime::impl_core::current_tenant().is_some() {
2448            return Err(RedDBError::Query(
2449                "cross-tenant subscription requires cluster-admin capability (events:cluster_subscribe)".to_string(),
2450            ));
2451        }
2452        validate_subscription_auth(runtime, source, subscription)?;
2453        if subscription.target_queue == source
2454            || subscription_would_create_cycle(
2455                &runtime.inner.db,
2456                source,
2457                &subscription.target_queue,
2458            )
2459        {
2460            return Err(RedDBError::Query(
2461                "subscription would create cycle".to_string(),
2462            ));
2463        }
2464        audit_subscription_redact_gap(runtime, source, subscription);
2465    }
2466    Ok(())
2467}
2468
2469fn validate_subscription_auth(
2470    runtime: &RedDBRuntime,
2471    source: &str,
2472    subscription: &crate::catalog::SubscriptionDescriptor,
2473) -> RedDBResult<()> {
2474    let auth_store = match runtime.inner.auth_store.read().clone() {
2475        Some(store) => store,
2476        None => return Ok(()),
2477    };
2478    let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
2479        Some(identity) => identity,
2480        None => return Ok(()),
2481    };
2482    let tenant = crate::runtime::impl_core::current_tenant();
2483    let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
2484
2485    if auth_store.iam_authorization_enabled() {
2486        let ctx = crate::auth::policies::EvalContext {
2487            principal_tenant: tenant.clone(),
2488            current_tenant: tenant.clone(),
2489            peer_ip: None,
2490            mfa_present: false,
2491            now_ms: crate::auth::now_ms(),
2492            principal_is_admin_role: role == crate::auth::Role::Admin,
2493            principal_is_platform_scoped: principal.tenant.is_none(),
2494        };
2495        let mut source_resource = crate::auth::policies::ResourceRef::new("table", source);
2496        if let Some(t) = tenant.as_deref() {
2497            source_resource = source_resource.with_tenant(t.to_string());
2498        }
2499        if !auth_store.check_policy_authz_with_role(
2500            &principal,
2501            "select",
2502            &source_resource,
2503            &ctx,
2504            role,
2505        ) {
2506            return Err(RedDBError::Query(format!(
2507                "permission denied: principal=`{}` action=`select` resource=`{}:{}` denied by IAM policy",
2508                principal, source_resource.kind, source_resource.name
2509            )));
2510        }
2511
2512        let mut target_resource =
2513            crate::auth::policies::ResourceRef::new("queue", subscription.target_queue.clone());
2514        if let Some(t) = tenant.as_deref() {
2515            target_resource = target_resource.with_tenant(t.to_string());
2516        }
2517        if !auth_store.check_policy_authz_with_role(
2518            &principal,
2519            "write",
2520            &target_resource,
2521            &ctx,
2522            role,
2523        ) {
2524            return Err(RedDBError::Query(format!(
2525                "permission denied: principal=`{}` action=`write` resource=`{}:{}` denied by IAM policy",
2526                principal, target_resource.kind, target_resource.name
2527            )));
2528        }
2529        return Ok(());
2530    }
2531
2532    let ctx = crate::auth::privileges::AuthzContext {
2533        principal: &username,
2534        effective_role: role,
2535        tenant: tenant.as_deref(),
2536    };
2537    auth_store
2538        .check_grant(
2539            &ctx,
2540            crate::auth::privileges::Action::Select,
2541            &crate::auth::privileges::Resource::table_from_name(source),
2542        )
2543        .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))?;
2544    auth_store
2545        .check_grant(
2546            &ctx,
2547            crate::auth::privileges::Action::Insert,
2548            &crate::auth::privileges::Resource::table_from_name(&subscription.target_queue),
2549        )
2550        .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))?;
2551    Ok(())
2552}
2553
2554fn audit_subscription_redact_gap(
2555    runtime: &RedDBRuntime,
2556    source: &str,
2557    subscription: &crate::catalog::SubscriptionDescriptor,
2558) {
2559    let auth_store = match runtime.inner.auth_store.read().clone() {
2560        Some(store) if store.iam_authorization_enabled() => store,
2561        _ => return,
2562    };
2563    let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
2564        Some(identity) => identity,
2565        None => return,
2566    };
2567    let tenant = crate::runtime::impl_core::current_tenant();
2568    let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
2569    let missing = subscription_redact_gap_columns(&auth_store, &principal, source, subscription);
2570    if missing.is_empty() {
2571        return;
2572    }
2573
2574    let columns = missing.into_iter().collect::<Vec<_>>().join(", ");
2575    tracing::warn!(
2576        target: "reddb::operator",
2577        "subscription_redact_gap: source={} target_queue={} columns=[{}]",
2578        source,
2579        subscription.target_queue,
2580        columns
2581    );
2582    let mut event = AuditEvent::builder("subscription_redact_gap")
2583        .principal(username)
2584        .source(AuditAuthSource::System)
2585        .resource(format!(
2586            "subscription:{}->{}",
2587            source, subscription.target_queue
2588        ))
2589        .outcome(Outcome::Success)
2590        .field(AuditFieldEscaper::field("source", source))
2591        .field(AuditFieldEscaper::field(
2592            "target_queue",
2593            subscription.target_queue.clone(),
2594        ))
2595        .field(AuditFieldEscaper::field(
2596            "subscription",
2597            subscription.name.clone(),
2598        ))
2599        .field(AuditFieldEscaper::field("columns", columns))
2600        .field(AuditFieldEscaper::field("role", role.as_str()));
2601    if let Some(t) = tenant {
2602        event = event.tenant(t);
2603    }
2604    runtime.inner.audit_log.record_event(event.build());
2605}
2606
2607fn subscription_redact_gap_columns(
2608    auth_store: &crate::auth::store::AuthStore,
2609    principal: &crate::auth::UserId,
2610    source: &str,
2611    subscription: &crate::catalog::SubscriptionDescriptor,
2612) -> BTreeSet<String> {
2613    let redacted: HashSet<String> = subscription
2614        .redact_fields
2615        .iter()
2616        .map(|field| field.to_ascii_lowercase())
2617        .collect();
2618    auth_store
2619        .effective_policies(principal)
2620        .iter()
2621        .flat_map(|policy| policy.statements.iter())
2622        .filter(|statement| statement.effect == crate::auth::policies::Effect::Deny)
2623        .filter(|statement| statement.actions.iter().any(action_pattern_matches_select))
2624        .flat_map(|statement| statement.resources.iter())
2625        .filter_map(|resource| denied_column_for_source(resource, source))
2626        .filter(|column| !redact_covers_column(&redacted, source, column))
2627        .collect()
2628}
2629
2630fn action_pattern_matches_select(pattern: &crate::auth::policies::ActionPattern) -> bool {
2631    match pattern {
2632        crate::auth::policies::ActionPattern::Wildcard => true,
2633        crate::auth::policies::ActionPattern::Exact(action) => action == "select",
2634        crate::auth::policies::ActionPattern::Prefix(prefix) => {
2635            "select".len() > prefix.len() + 1
2636                && "select".starts_with(prefix)
2637                && "select".as_bytes()[prefix.len()] == b':'
2638        }
2639    }
2640}
2641
2642fn denied_column_for_source(
2643    resource: &crate::auth::policies::ResourcePattern,
2644    source: &str,
2645) -> Option<String> {
2646    let crate::auth::policies::ResourcePattern::Exact { kind, name } = resource else {
2647        return None;
2648    };
2649    if kind != "column" {
2650        return None;
2651    }
2652    let column = crate::auth::ColumnRef::parse_resource_name(name).ok()?;
2653    (column.table_resource_name() == source).then_some(column.column)
2654}
2655
2656fn redact_covers_column(redacted: &HashSet<String>, source: &str, column: &str) -> bool {
2657    let column = column.to_ascii_lowercase();
2658    let qualified = format!("{}.{}", source.to_ascii_lowercase(), column);
2659    redacted.contains("*") || redacted.contains(&column) || redacted.contains(&qualified)
2660}
2661
2662fn subscription_would_create_cycle(
2663    db: &crate::storage::unified::devx::RedDB,
2664    source: &str,
2665    target: &str,
2666) -> bool {
2667    let mut graph: HashMap<String, Vec<String>> = HashMap::new();
2668    for contract in db.collection_contracts() {
2669        for subscription in contract
2670            .subscriptions
2671            .into_iter()
2672            .filter(|subscription| subscription.enabled)
2673        {
2674            graph
2675                .entry(subscription.source)
2676                .or_default()
2677                .push(subscription.target_queue);
2678        }
2679    }
2680    graph
2681        .entry(source.to_string())
2682        .or_default()
2683        .push(target.to_string());
2684
2685    let mut stack = vec![target.to_string()];
2686    let mut seen = HashSet::new();
2687    while let Some(node) = stack.pop() {
2688        if node == source {
2689            return true;
2690        }
2691        if !seen.insert(node.clone()) {
2692            continue;
2693        }
2694        if let Some(next) = graph.get(&node) {
2695            stack.extend(next.iter().cloned());
2696        }
2697    }
2698    false
2699}
2700
2701pub(crate) fn ensure_event_target_queue_pub(
2702    runtime: &RedDBRuntime,
2703    queue: &str,
2704) -> RedDBResult<()> {
2705    ensure_event_target_queue(runtime, queue)
2706}
2707
2708fn ensure_event_target_queue(runtime: &RedDBRuntime, queue: &str) -> RedDBResult<()> {
2709    let store = runtime.inner.db.store();
2710    if store.get_collection(queue).is_some() {
2711        return Ok(());
2712    }
2713    store
2714        .create_collection(queue)
2715        .map_err(|err| RedDBError::Internal(err.to_string()))?;
2716    crate::runtime::queue_storage_index::prepare_queue(runtime, queue)?;
2717    runtime
2718        .inner
2719        .db
2720        .save_collection_contract(event_queue_collection_contract(queue))
2721        .map_err(|err| RedDBError::Internal(err.to_string()))?;
2722    store.set_config_tree(
2723        &format!("queue.{queue}.mode"),
2724        &crate::serde_json::Value::String("fanout".to_string()),
2725    );
2726    Ok(())
2727}
2728
2729fn event_queue_collection_contract(queue: &str) -> crate::physical::CollectionContract {
2730    let now = current_unix_ms();
2731    crate::physical::CollectionContract {
2732        name: queue.to_string(),
2733        declared_model: crate::catalog::CollectionModel::Queue,
2734        schema_mode: crate::catalog::SchemaMode::Dynamic,
2735        origin: crate::physical::ContractOrigin::Implicit,
2736        version: 1,
2737        created_at_unix_ms: now,
2738        updated_at_unix_ms: now,
2739        default_ttl_ms: None,
2740        vector_dimension: None,
2741        vector_metric: None,
2742        context_index_fields: Vec::new(),
2743        declared_columns: Vec::new(),
2744        table_def: None,
2745        timestamps_enabled: false,
2746        context_index_enabled: false,
2747        metrics_raw_retention_ms: None,
2748        metrics_rollup_policies: Vec::new(),
2749        metrics_tenant_identity: None,
2750        metrics_namespace: None,
2751        append_only: true,
2752        subscriptions: Vec::new(),
2753        analytics_config: Vec::new(),
2754        session_key: None,
2755        session_gap_ms: None,
2756        retention_duration_ms: None,
2757        analytical_storage: None,
2758
2759        ai_policy: None,
2760    }
2761}
2762
2763fn column_def_from_ddl(column: &CreateColumnDef) -> RedDBResult<reddb_types::ColumnDef> {
2764    reddb_engine::catalog::ddl::column_def_from_ddl(column).map_err(RedDBError::Query)
2765}
2766
2767fn current_unix_ms() -> u128 {
2768    std::time::SystemTime::now()
2769        .duration_since(std::time::UNIX_EPOCH)
2770        .unwrap_or_default()
2771        .as_millis()
2772}
2773
2774#[cfg(test)]
2775mod tests {
2776    use crate::auth::policies::{ActionPattern, Effect, Policy, ResourcePattern, Statement};
2777    use crate::auth::store::{AuthStore, PrincipalRef};
2778    use crate::auth::UserId;
2779    use crate::auth::{AuthConfig, Role};
2780    use crate::runtime::impl_core::{clear_current_auth_identity, set_current_auth_identity};
2781    use crate::{RedDBOptions, RedDBRuntime};
2782    use reddb_types::Value;
2783    use std::sync::Arc;
2784
2785    fn make_allow_policy(id: &str, action: &str, collection: &str) -> Policy {
2786        Policy {
2787            id: id.to_string(),
2788            version: 1,
2789            tenant: None,
2790            created_at: 0,
2791            updated_at: 0,
2792            statements: vec![Statement {
2793                sid: None,
2794                effect: Effect::Allow,
2795                actions: vec![ActionPattern::Exact(action.to_string())],
2796                resources: vec![ResourcePattern::Exact {
2797                    kind: "collection".to_string(),
2798                    name: collection.to_string(),
2799                }],
2800                condition: None,
2801            }],
2802        }
2803    }
2804
2805    fn wire_auth_store(rt: &RedDBRuntime) -> Arc<AuthStore> {
2806        let store = Arc::new(AuthStore::new(AuthConfig::default()));
2807        *rt.inner.auth_store.write() = Some(store.clone());
2808        store
2809    }
2810
2811    #[test]
2812    fn drop_denied_without_iam_policy() {
2813        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2814        rt.execute_query("CREATE TABLE foo (id INT)").unwrap();
2815        let store = wire_auth_store(&rt);
2816        // Put a select-only policy so IAM mode activates, but give alice no drop policy.
2817        let select_only = Policy {
2818            id: "select-only".to_string(),
2819            version: 1,
2820            tenant: None,
2821            created_at: 0,
2822            updated_at: 0,
2823            statements: vec![Statement {
2824                sid: None,
2825                effect: Effect::Allow,
2826                actions: vec![ActionPattern::Exact("select".to_string())],
2827                resources: vec![ResourcePattern::Wildcard],
2828                condition: None,
2829            }],
2830        };
2831        store.put_policy_internal(select_only).unwrap();
2832        let alice = UserId::from_parts(None, "alice");
2833        store
2834            .attach_policy(PrincipalRef::User(alice), "select-only")
2835            .unwrap();
2836        set_current_auth_identity("alice".to_string(), Role::Write);
2837        let err = rt.execute_query("DROP TABLE foo").unwrap_err();
2838        clear_current_auth_identity();
2839        assert!(
2840            format!("{err}").contains("denied by IAM policy"),
2841            "got: {err}"
2842        );
2843    }
2844
2845    #[test]
2846    fn drop_allowed_with_explicit_iam_policy() {
2847        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2848        rt.execute_query("CREATE TABLE bar (id INT)").unwrap();
2849        let store = wire_auth_store(&rt);
2850        let policy = make_allow_policy("allow-drop-bar", "drop", "bar");
2851        store.put_policy_internal(policy).unwrap();
2852        let bob = UserId::from_parts(None, "bob");
2853        store
2854            .attach_policy(PrincipalRef::User(bob), "allow-drop-bar")
2855            .unwrap();
2856        set_current_auth_identity("bob".to_string(), Role::Write);
2857        rt.execute_query("DROP TABLE bar").unwrap();
2858        clear_current_auth_identity();
2859    }
2860
2861    #[test]
2862    fn drop_allowed_with_wildcard_iam_policy() {
2863        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2864        rt.execute_query("CREATE TABLE baz (id INT)").unwrap();
2865        let store = wire_auth_store(&rt);
2866        let policy = Policy {
2867            id: "allow-drop-all".to_string(),
2868            version: 1,
2869            tenant: None,
2870            created_at: 0,
2871            updated_at: 0,
2872            statements: vec![Statement {
2873                sid: None,
2874                effect: Effect::Allow,
2875                actions: vec![ActionPattern::Exact("drop".to_string())],
2876                resources: vec![ResourcePattern::Wildcard],
2877                condition: None,
2878            }],
2879        };
2880        store.put_policy_internal(policy).unwrap();
2881        let carl = UserId::from_parts(None, "carl");
2882        store
2883            .attach_policy(PrincipalRef::User(carl), "allow-drop-all")
2884            .unwrap();
2885        set_current_auth_identity("carl".to_string(), Role::Write);
2886        rt.execute_query("DROP TABLE baz").unwrap();
2887        clear_current_auth_identity();
2888    }
2889
2890    #[test]
2891    fn truncate_denied_without_iam_policy() {
2892        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2893        rt.execute_query("CREATE TABLE qux (id INT)").unwrap();
2894        let store = wire_auth_store(&rt);
2895        // Acceptance #2 (#712 / S5A): in `policy_only` mode a
2896        // principal with no matching policy is denied even if their
2897        // role would have permitted the action. The pre-#712 default
2898        // of "no policy → deny" only holds under `policy_only`, so
2899        // pin the mode explicitly here so the assertion holds
2900        // regardless of the construction-time default.
2901        store
2902            .set_enforcement_mode(crate::auth::enforcement_mode::PolicyEnforcementMode::PolicyOnly);
2903        // A policy exists (IAM active) but gives no truncate right.
2904        let select_only = Policy {
2905            id: "select-only-2".to_string(),
2906            version: 1,
2907            tenant: None,
2908            created_at: 0,
2909            updated_at: 0,
2910            statements: vec![Statement {
2911                sid: None,
2912                effect: Effect::Allow,
2913                actions: vec![ActionPattern::Exact("select".to_string())],
2914                resources: vec![ResourcePattern::Wildcard],
2915                condition: None,
2916            }],
2917        };
2918        store.put_policy_internal(select_only).unwrap();
2919        let dana = UserId::from_parts(None, "dana");
2920        store
2921            .attach_policy(PrincipalRef::User(dana), "select-only-2")
2922            .unwrap();
2923        set_current_auth_identity("dana".to_string(), Role::Write);
2924        let err = rt.execute_query("TRUNCATE TABLE qux").unwrap_err();
2925        clear_current_auth_identity();
2926        assert!(
2927            format!("{err}").contains("denied by IAM policy"),
2928            "got: {err}"
2929        );
2930    }
2931
2932    #[test]
2933    fn truncate_table_clears_rows_and_preserves_schema_and_indexes() {
2934        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2935        rt.execute_query("CREATE TABLE users (id INT, name TEXT)")
2936            .unwrap();
2937        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'ana'), (2, 'bob')")
2938            .unwrap();
2939        rt.execute_query("CREATE INDEX idx_users_id ON users (id) USING HASH")
2940            .unwrap();
2941
2942        let truncated = rt.execute_query("TRUNCATE TABLE users").unwrap();
2943        assert_eq!(truncated.statement_type, "truncate");
2944        assert_eq!(truncated.affected_rows, 0);
2945
2946        let empty = rt.execute_query("SELECT id FROM users").unwrap();
2947        assert!(empty.result.records.is_empty());
2948
2949        rt.execute_query("INSERT INTO users (id, name) VALUES (3, 'cy')")
2950            .unwrap();
2951        let selected = rt
2952            .execute_query("SELECT name FROM users WHERE id = 3")
2953            .unwrap();
2954        let name = selected.result.records[0].get("name").unwrap();
2955        assert_eq!(name, &Value::text("cy"));
2956        assert!(rt.db().collection_contract("users").is_some());
2957        assert!(rt
2958            .inner
2959            .index_store
2960            .list_indices("users")
2961            .iter()
2962            .any(|index| index.name == "idx_users_id"));
2963    }
2964
2965    #[test]
2966    fn truncate_collection_is_polymorphic_and_typed_mismatch_fails() {
2967        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2968        rt.execute_query("CREATE QUEUE tasks").unwrap();
2969        rt.execute_query("QUEUE PUSH tasks {'job':'a'}").unwrap();
2970
2971        let err = rt.execute_query("TRUNCATE TABLE tasks").unwrap_err();
2972        assert!(format!("{err}").contains("model mismatch: expected table, got queue"));
2973
2974        rt.execute_query("TRUNCATE COLLECTION tasks").unwrap();
2975        let len = rt.execute_query("QUEUE LEN tasks").unwrap();
2976        assert_eq!(
2977            len.result.records[0].get("len"),
2978            Some(&Value::UnsignedInteger(0))
2979        );
2980    }
2981
2982    #[test]
2983    fn truncate_system_schema_is_read_only() {
2984        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
2985        let err = rt
2986            .execute_query("TRUNCATE COLLECTION red.collections")
2987            .unwrap_err();
2988        assert!(format!("{err}").contains("system schema is read-only"));
2989    }
2990
2991    // ── #302 / #310: TRUNCATE / DROP single-event semantics ────────────────
2992
2993    fn queue_payloads(rt: &RedDBRuntime, queue: &str) -> Vec<crate::json::Value> {
2994        let result = rt
2995            .execute_query(&format!("QUEUE PEEK {queue} 100"))
2996            .expect("peek queue");
2997        result
2998            .result
2999            .records
3000            .iter()
3001            .map(
3002                |record| match record.get("payload").expect("payload column") {
3003                    Value::Json(bytes) => crate::json::from_slice(bytes).expect("json payload"),
3004                    other => panic!("expected JSON queue payload, got {other:?}"),
3005                },
3006            )
3007            .collect()
3008    }
3009
3010    /// `TRUNCATE users` on an event-enabled collection emits exactly 1
3011    /// `truncate` event, not one delete event per row.
3012    #[test]
3013    fn truncate_event_enabled_table_emits_single_truncate_event() {
3014        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3015        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3016            .unwrap();
3017        rt.execute_query(
3018            "INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob'), (3, 'carol')",
3019        )
3020        .unwrap();
3021
3022        // Drain the 3 insert events so we start clean.
3023        rt.execute_query("QUEUE POP users_events COUNT 10").unwrap();
3024
3025        rt.execute_query("TRUNCATE TABLE users").unwrap();
3026
3027        let events = queue_payloads(&rt, "users_events");
3028        // Must be exactly 1 truncate event, not 3 delete events.
3029        assert_eq!(
3030            events.len(),
3031            1,
3032            "expected 1 truncate event, got {}",
3033            events.len()
3034        );
3035        let ev = events[0].as_object().expect("event is object");
3036        assert_eq!(
3037            ev.get("op").and_then(crate::json::Value::as_str),
3038            Some("truncate")
3039        );
3040        assert_eq!(
3041            ev.get("collection").and_then(crate::json::Value::as_str),
3042            Some("users")
3043        );
3044        assert_eq!(
3045            ev.get("entities_count")
3046                .and_then(crate::json::Value::as_u64),
3047            Some(3)
3048        );
3049        assert!(ev.get("ts").and_then(crate::json::Value::as_u64).is_some());
3050        assert!(ev.get("lsn").and_then(crate::json::Value::as_u64).is_some());
3051        assert!(ev
3052            .get("event_id")
3053            .and_then(crate::json::Value::as_str)
3054            .is_some_and(|s| !s.is_empty()));
3055    }
3056
3057    /// `TRUNCATE users` on a collection without event subscription emits no events.
3058    #[test]
3059    fn truncate_no_events_collection_emits_nothing() {
3060        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3061        rt.execute_query("CREATE TABLE plain (id INT, val TEXT)")
3062            .unwrap();
3063        rt.execute_query("INSERT INTO plain (id, val) VALUES (1, 'a'), (2, 'b')")
3064            .unwrap();
3065        // No EVENTS subscription — truncate must work without touching any queue.
3066        rt.execute_query("TRUNCATE TABLE plain").unwrap();
3067        // No crash, no queue to check. Just verify truncation happened.
3068        let rows = rt.execute_query("SELECT id FROM plain").unwrap();
3069        assert!(rows.result.records.is_empty());
3070    }
3071
3072    /// `DROP TABLE users` on an event-enabled collection emits exactly 1
3073    /// `collection_dropped` event. The subscription is removed from the
3074    /// source contract but the target queue is preserved for consumer drain.
3075    #[test]
3076    fn drop_event_enabled_table_emits_single_collection_dropped_event() {
3077        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3078        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3079            .unwrap();
3080        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob')")
3081            .unwrap();
3082
3083        // Drain insert events so we start clean.
3084        rt.execute_query("QUEUE POP users_events COUNT 10").unwrap();
3085
3086        rt.execute_query("DROP TABLE users").unwrap();
3087
3088        // Queue must still exist with 1 collection_dropped event.
3089        let events = queue_payloads(&rt, "users_events");
3090        assert_eq!(
3091            events.len(),
3092            1,
3093            "expected 1 collection_dropped event, got {}",
3094            events.len()
3095        );
3096        let ev = events[0].as_object().expect("event is object");
3097        assert_eq!(
3098            ev.get("op").and_then(crate::json::Value::as_str),
3099            Some("collection_dropped")
3100        );
3101        assert_eq!(
3102            ev.get("collection").and_then(crate::json::Value::as_str),
3103            Some("users")
3104        );
3105        assert_eq!(
3106            ev.get("final_entities_count")
3107                .and_then(crate::json::Value::as_u64),
3108            Some(2)
3109        );
3110        assert!(ev.get("ts").and_then(crate::json::Value::as_u64).is_some());
3111        assert!(ev.get("lsn").and_then(crate::json::Value::as_u64).is_some());
3112        assert!(ev
3113            .get("event_id")
3114            .and_then(crate::json::Value::as_str)
3115            .is_some_and(|s| !s.is_empty()));
3116
3117        // Source collection is gone.
3118        let err = rt.execute_query("SELECT id FROM users").unwrap_err();
3119        assert!(
3120            format!("{err}").contains("users"),
3121            "expected not-found error"
3122        );
3123    }
3124
3125    /// `DROP TABLE users` on a collection without event subscription works
3126    /// normally with no event emitted.
3127    #[test]
3128    fn drop_no_events_collection_emits_nothing() {
3129        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3130        rt.execute_query("CREATE TABLE plain (id INT, val TEXT)")
3131            .unwrap();
3132        rt.execute_query("INSERT INTO plain (id, val) VALUES (1, 'a')")
3133            .unwrap();
3134        rt.execute_query("DROP TABLE plain").unwrap();
3135        // No crash and collection is gone.
3136        let err = rt.execute_query("SELECT id FROM plain").unwrap_err();
3137        assert!(format!("{err}").contains("plain"));
3138    }
3139
3140    // ── #297: ops_filter + WHERE filter ────────────────────────────────────
3141
3142    /// `WITH EVENTS (INSERT)` — UPDATE and DELETE events must NOT be emitted.
3143    #[test]
3144    fn ops_filter_insert_only_ignores_update_and_delete() {
3145        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3146        rt.execute_query(
3147            "CREATE TABLE items (id INT, val TEXT) WITH EVENTS (INSERT) TO items_events",
3148        )
3149        .unwrap();
3150        rt.execute_query("INSERT INTO items (id, val) VALUES (1, 'a')")
3151            .unwrap();
3152        rt.execute_query("UPDATE items SET val = 'b' WHERE id = 1")
3153            .unwrap();
3154        rt.execute_query("DELETE FROM items WHERE id = 1").unwrap();
3155
3156        let events = queue_payloads(&rt, "items_events");
3157        // Only the INSERT should have fired.
3158        assert_eq!(
3159            events.len(),
3160            1,
3161            "expected 1 insert event, got {}",
3162            events.len()
3163        );
3164        assert_eq!(
3165            events[0]
3166                .as_object()
3167                .unwrap()
3168                .get("op")
3169                .and_then(crate::json::Value::as_str),
3170            Some("insert")
3171        );
3172    }
3173
3174    /// `WITH EVENTS WHERE status = 'active'` — only rows matching the predicate generate events.
3175    #[test]
3176    fn where_filter_skips_rows_that_do_not_match() {
3177        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3178        rt.execute_query(
3179            "CREATE TABLE users (id INT, status TEXT) WITH EVENTS WHERE status = 'active' TO users_events",
3180        )
3181        .unwrap();
3182
3183        // This row should generate an event.
3184        rt.execute_query("INSERT INTO users (id, status) VALUES (1, 'active')")
3185            .unwrap();
3186        // This row should NOT generate an event.
3187        rt.execute_query("INSERT INTO users (id, status) VALUES (2, 'inactive')")
3188            .unwrap();
3189
3190        let events = queue_payloads(&rt, "users_events");
3191        assert_eq!(
3192            events.len(),
3193            1,
3194            "expected 1 event (only active), got {}",
3195            events.len()
3196        );
3197        let ev = events[0].as_object().unwrap();
3198        assert_eq!(
3199            ev.get("op").and_then(crate::json::Value::as_str),
3200            Some("insert")
3201        );
3202        let after = ev.get("after").unwrap().as_object().unwrap();
3203        assert_eq!(
3204            after.get("status").and_then(crate::json::Value::as_str),
3205            Some("active")
3206        );
3207    }
3208
3209    /// `WITH EVENTS (INSERT, UPDATE) WHERE status = 'active'` — combination functional.
3210    #[test]
3211    fn ops_filter_and_where_filter_combined() {
3212        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3213        rt.execute_query(
3214            "CREATE TABLE items (id INT, status TEXT) WITH EVENTS (INSERT, UPDATE) WHERE status = 'active' TO items_events",
3215        )
3216        .unwrap();
3217
3218        // INSERT active → event
3219        rt.execute_query("INSERT INTO items (id, status) VALUES (1, 'active')")
3220            .unwrap();
3221        // INSERT inactive → no event
3222        rt.execute_query("INSERT INTO items (id, status) VALUES (2, 'inactive')")
3223            .unwrap();
3224        // UPDATE row 1 to inactive → after = inactive, no event
3225        rt.execute_query("UPDATE items SET status = 'inactive' WHERE id = 1")
3226            .unwrap();
3227        // DELETE → ops_filter excludes it
3228        rt.execute_query("DELETE FROM items WHERE id = 2").unwrap();
3229
3230        let events = queue_payloads(&rt, "items_events");
3231        // Only the first INSERT (active) fires; UPDATE result is inactive so skipped; DELETE excluded by ops_filter.
3232        assert_eq!(
3233            events.len(),
3234            1,
3235            "expected 1 event, got {}: {events:?}",
3236            events.len()
3237        );
3238        assert_eq!(
3239            events[0]
3240                .as_object()
3241                .unwrap()
3242                .get("op")
3243                .and_then(crate::json::Value::as_str),
3244            Some("insert")
3245        );
3246    }
3247
3248    /// WHERE filter on DELETE events — the before-state (pre-image) is evaluated.
3249    #[test]
3250    fn where_filter_on_delete_checks_before_state() {
3251        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3252        rt.execute_query(
3253            "CREATE TABLE users (id INT, status TEXT) WITH EVENTS (DELETE) WHERE status = 'active' TO users_events",
3254        )
3255        .unwrap();
3256
3257        rt.execute_query("INSERT INTO users (id, status) VALUES (1, 'active'), (2, 'inactive')")
3258            .unwrap();
3259
3260        // Delete active row → event (before-state was active)
3261        rt.execute_query("DELETE FROM users WHERE id = 1").unwrap();
3262        // Delete inactive row → no event (before-state was inactive)
3263        rt.execute_query("DELETE FROM users WHERE id = 2").unwrap();
3264
3265        let events = queue_payloads(&rt, "users_events");
3266        assert_eq!(
3267            events.len(),
3268            1,
3269            "expected 1 delete event, got {}",
3270            events.len()
3271        );
3272        let ev = events[0].as_object().unwrap();
3273        assert_eq!(
3274            ev.get("op").and_then(crate::json::Value::as_str),
3275            Some("delete")
3276        );
3277    }
3278
3279    // ── #301: schema evolution OperatorEvent on ALTER ───────────────────────
3280
3281    /// ADD COLUMN on event-enabled table must succeed (OperatorEvent is best-effort).
3282    #[test]
3283    fn alter_add_column_on_event_enabled_table_succeeds() {
3284        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3285        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3286            .unwrap();
3287        // Must not error — OperatorEvent emission is best-effort (no global sink in tests).
3288        rt.execute_query("ALTER TABLE users ADD COLUMN phone TEXT")
3289            .unwrap();
3290        // The column is now in the contract.
3291        let contract = rt.db().collection_contract("users").unwrap();
3292        assert!(
3293            contract.declared_columns.iter().any(|c| c.name == "phone"),
3294            "phone column should be in contract"
3295        );
3296        // Subscription still enabled after the alter.
3297        assert!(
3298            contract.subscriptions.iter().any(|s| s.enabled),
3299            "subscription should remain enabled"
3300        );
3301    }
3302
3303    /// DROP COLUMN on event-enabled table must succeed; non-column ALTERs
3304    /// (like ENABLE ROW LEVEL SECURITY) must also succeed without emitting.
3305    #[test]
3306    fn alter_drop_column_and_rls_on_event_enabled_table_succeeds() {
3307        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3308        rt.execute_query(
3309            "CREATE TABLE items (id INT, secret TEXT, status TEXT) WITH EVENTS TO items_events",
3310        )
3311        .unwrap();
3312        // DROP COLUMN — schema change event path exercises, must not error.
3313        rt.execute_query("ALTER TABLE items DROP COLUMN secret")
3314            .unwrap();
3315        let contract = rt.db().collection_contract("items").unwrap();
3316        assert!(
3317            !contract.declared_columns.iter().any(|c| c.name == "secret"),
3318            "secret column should be removed"
3319        );
3320        // ENABLE RLS — non-column op, no schema-change event (coverage).
3321        rt.execute_query("ALTER TABLE items ENABLE ROW LEVEL SECURITY")
3322            .unwrap();
3323        // Collection and subscription still intact.
3324        assert!(
3325            contract.subscriptions.iter().any(|s| s.enabled),
3326            "subscription should remain enabled"
3327        );
3328    }
3329
3330    /// Slice G (#675) — every newly-created `CollectionModel::Vector`
3331    /// collection is marked `vector.turbo` and gains a materialised
3332    /// `TurboCollectionState`. The marker is what the SEARCH/INSERT
3333    /// hot paths read to branch between TurboQuant and the legacy
3334    /// brute-force fallback.
3335    #[test]
3336    fn create_vector_marks_collection_as_turbo_baseline() {
3337        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3338        rt.execute_query("CREATE VECTOR embeddings DIM 4").unwrap();
3339        let store = rt.db().store();
3340        assert!(
3341            crate::runtime::vector_turbo_kind::is_turbo(&store, "embeddings"),
3342            "new vector collections must be turbo-marked baseline"
3343        );
3344        assert!(
3345            rt.db().turbo_state("embeddings").is_some(),
3346            "turbo_state must materialise after CREATE VECTOR"
3347        );
3348    }
3349
3350    /// Non-vector collections must NOT be turbo-marked. Guards against
3351    /// the always-baseline change accidentally leaking the marker onto
3352    /// CREATE TABLE / CREATE DOCUMENT / etc.
3353    #[test]
3354    fn create_table_does_not_mark_turbo() {
3355        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3356        rt.execute_query("CREATE TABLE plain (id INT)").unwrap();
3357        let store = rt.db().store();
3358        assert!(
3359            !crate::runtime::vector_turbo_kind::is_turbo(&store, "plain"),
3360            "non-vector collections must not gain the turbo marker"
3361        );
3362        assert!(rt.db().turbo_state("plain").is_none());
3363    }
3364
3365    /// `CREATE COLLECTION KIND vector.turbo` continues to mark the
3366    /// collection (idempotent with the new always-baseline path).
3367    #[test]
3368    fn create_collection_kind_vector_turbo_still_marked() {
3369        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3370        rt.execute_query("CREATE COLLECTION turbo_v KIND vector.turbo DIM 4")
3371            .unwrap();
3372        let store = rt.db().store();
3373        assert!(crate::runtime::vector_turbo_kind::is_turbo(
3374            &store, "turbo_v"
3375        ));
3376        assert!(rt.db().turbo_state("turbo_v").is_some());
3377    }
3378
3379    /// Issue #1271: a declared AI policy is persisted in the catalog and
3380    /// reads back identical via the collection contract (introspection).
3381    #[test]
3382    fn create_table_persists_and_introspects_ai_policy() {
3383        use crate::catalog::{ModerateDegradedMode, ModerateRejectAction};
3384        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3385        rt.execute_query(
3386            "CREATE TABLE posts (id INT, title TEXT, body TEXT, photo TEXT) WITH ( \
3387               EMBED (fields = ('title', 'body'), provider = 'openai', model = 'text-embedding-3-small'), \
3388               MODERATE (fields = ('body'), provider = 'openai', model = 'omni-moderation-latest', sync = true, degraded = closed, on_reject = flag), \
3389               VISION (image_field = 'photo', outputs = ('caption'), provider = 'openai', model = 'gpt-4o') \
3390             )",
3391        )
3392        .expect("create table with ai policy");
3393
3394        let contracts = rt.db().collection_contracts();
3395        let contract = contracts
3396            .iter()
3397            .find(|c| c.name == "posts")
3398            .expect("contract persisted");
3399        let policy = contract.ai_policy.as_ref().expect("ai policy persisted");
3400
3401        let embed = policy.embed.as_ref().expect("embed");
3402        assert_eq!(embed.fields, vec!["title".to_string(), "body".to_string()]);
3403        assert_eq!(embed.model, "text-embedding-3-small");
3404
3405        let moderate = policy.moderate.as_ref().expect("moderate");
3406        assert!(moderate.sync_gate);
3407        assert_eq!(moderate.degraded_mode, ModerateDegradedMode::Closed);
3408        assert_eq!(moderate.reject_action, ModerateRejectAction::Flag);
3409
3410        let vision = policy.vision.as_ref().expect("vision");
3411        assert_eq!(vision.image_field, "photo");
3412        assert_eq!(vision.model, "gpt-4o");
3413    }
3414
3415    /// Issue #1271 / #1269: a policy wiring a provider to a modality it
3416    /// cannot serve is rejected at DDL time, and the collection is not
3417    /// created. Anthropic has no embeddings product in the matrix.
3418    #[test]
3419    fn create_table_rejects_ai_policy_incapable_provider() {
3420        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3421        let err = rt
3422            .execute_query(
3423                "CREATE TABLE bad (id INT, body TEXT) WITH ( \
3424                   EMBED (fields = ('body'), provider = 'anthropic', model = 'claude-3-5-sonnet') \
3425                 )",
3426            )
3427            .unwrap_err();
3428        assert!(
3429            err.to_string()
3430                .contains("cannot serve the 'embed' modality"),
3431            "{err}"
3432        );
3433        assert!(
3434            rt.db().store().get_collection("bad").is_none(),
3435            "rejected DDL must not create the collection"
3436        );
3437    }
3438}
reddb_server/runtime/impl_ddl.rs

reddb_server/runtime/
impl_ddl.rs
