1use std::sync::Arc;
27
28use super::persistent_value::persistent_value_expression;
29use crate::utils::dummy_token_clone;
30use radixdb_catalog::ObjectId;
31use radixdb_core::{
32 DataType, Error, ForeignKeyAction, ForeignKeyConstraint, Result, Row, Schema, SchemaBuilder,
33 SchemaColumn, SchemaConstraintKind, Value, ValueSet,
34};
35use radixdb_sql::ast::*;
36use radixdb_storage::traits::{
37 Engine, PendingIndexDefinition, PendingIndexDrop, PendingIndexRename, QueryResult,
38 SchemaPhysicalTransition, Table, Transaction,
39};
40
41fn walk_persistent_expression_mut(
42 expression: &mut Expression,
43 visitor: &mut impl FnMut(&mut Expression) -> Result<()>,
44) -> Result<()> {
45 visitor(expression)?;
46 match expression {
47 Expression::Prefix(value) => walk_persistent_expression_mut(&mut value.right, visitor)?,
48 Expression::Infix(value) => {
49 walk_persistent_expression_mut(&mut value.left, visitor)?;
50 walk_persistent_expression_mut(&mut value.right, visitor)?;
51 }
52 Expression::List(value) => {
53 for expression in &mut value.elements {
54 walk_persistent_expression_mut(expression, visitor)?;
55 }
56 }
57 Expression::Distinct(value) => walk_persistent_expression_mut(&mut value.expr, visitor)?,
58 Expression::In(value) => {
59 walk_persistent_expression_mut(&mut value.left, visitor)?;
60 walk_persistent_expression_mut(&mut value.right, visitor)?;
61 }
62 Expression::InHashSet(value) => walk_persistent_expression_mut(&mut value.column, visitor)?,
63 Expression::Between(value) => {
64 walk_persistent_expression_mut(&mut value.expr, visitor)?;
65 walk_persistent_expression_mut(&mut value.lower, visitor)?;
66 walk_persistent_expression_mut(&mut value.upper, visitor)?;
67 }
68 Expression::Like(value) => {
69 walk_persistent_expression_mut(&mut value.left, visitor)?;
70 walk_persistent_expression_mut(&mut value.pattern, visitor)?;
71 if let Some(escape) = &mut value.escape {
72 walk_persistent_expression_mut(escape, visitor)?;
73 }
74 }
75 Expression::ExpressionList(value) => {
76 for expression in &mut value.expressions {
77 walk_persistent_expression_mut(expression, visitor)?;
78 }
79 }
80 Expression::Case(value) => {
81 if let Some(expression) = &mut value.value {
82 walk_persistent_expression_mut(expression, visitor)?;
83 }
84 for clause in &mut value.when_clauses {
85 walk_persistent_expression_mut(&mut clause.condition, visitor)?;
86 walk_persistent_expression_mut(&mut clause.then_result, visitor)?;
87 }
88 if let Some(expression) = &mut value.else_value {
89 walk_persistent_expression_mut(expression, visitor)?;
90 }
91 }
92 Expression::Cast(value) => walk_persistent_expression_mut(&mut value.expr, visitor)?,
93 Expression::FunctionCall(value) => {
94 for argument in &mut value.arguments {
95 walk_persistent_expression_mut(argument, visitor)?;
96 }
97 for order in &mut value.order_by {
98 walk_persistent_expression_mut(&mut order.expression, visitor)?;
99 }
100 if let Some(filter) = &mut value.filter {
101 walk_persistent_expression_mut(filter, visitor)?;
102 }
103 }
104 Expression::Aliased(value) => {
105 walk_persistent_expression_mut(&mut value.expression, visitor)?
106 }
107 Expression::Window(value) => {
108 for argument in &mut value.function.arguments {
109 walk_persistent_expression_mut(argument, visitor)?;
110 }
111 for order in &mut value.function.order_by {
112 walk_persistent_expression_mut(&mut order.expression, visitor)?;
113 }
114 if let Some(filter) = &mut value.function.filter {
115 walk_persistent_expression_mut(filter, visitor)?;
116 }
117 for partition in &mut value.partition_by {
118 walk_persistent_expression_mut(partition, visitor)?;
119 }
120 for order in &mut value.order_by {
121 walk_persistent_expression_mut(&mut order.expression, visitor)?;
122 }
123 }
124 Expression::Exists(_)
125 | Expression::AllAny(_)
126 | Expression::ScalarSubquery(_)
127 | Expression::TableSource(_)
128 | Expression::JoinSource(_)
129 | Expression::SubquerySource(_)
130 | Expression::ValuesSource(_)
131 | Expression::CteReference(_)
132 | Expression::FunctionTableSource(_) => {
133 return Err(Error::NotSupported(
134 "subqueries and table sources are not supported in persistent schema expressions"
135 .to_string(),
136 ));
137 }
138 Expression::Identifier(_)
139 | Expression::QualifiedIdentifier(_)
140 | Expression::IntegerLiteral(_)
141 | Expression::FloatLiteral(_)
142 | Expression::StringLiteral(_)
143 | Expression::BooleanLiteral(_)
144 | Expression::NullLiteral(_)
145 | Expression::IntervalLiteral(_)
146 | Expression::BoundValue(_)
147 | Expression::Parameter(_)
148 | Expression::Star(_)
149 | Expression::QualifiedStar(_)
150 | Expression::Default(_) => {}
151 }
152 Ok(())
153}
154
155fn materialize_persistent_expression(
156 expression: &Expression,
157 ctx: &ExecutionContext,
158) -> Result<Expression> {
159 let mut expression = expression.clone();
160 walk_persistent_expression_mut(&mut expression, &mut |expression| {
161 let (token, value) = match expression {
162 Expression::Parameter(parameter) => {
163 let value = if parameter.name.starts_with(':') {
164 let name = ¶meter.name[1..];
165 ctx.get_named_param(name).cloned().ok_or_else(|| {
166 Error::invalid_argument(format!("missing named parameter :{name}"))
167 })?
168 } else {
169 ctx.get_param(parameter.index).cloned().ok_or_else(|| {
170 Error::invalid_argument(format!(
171 "missing positional parameter ${}",
172 parameter.index
173 ))
174 })?
175 };
176 (parameter.token.clone(), value)
177 }
178 Expression::BoundValue(value) => {
179 return persistent_value_expression(value, &dummy_token_clone()).map(|bound| {
180 *expression = bound;
181 });
182 }
183 _ => return Ok(()),
184 };
185 *expression = persistent_value_expression(&value, &token)?;
186 Ok(())
187 })?;
188 Ok(expression)
189}
190
191fn bind_persistent_expression(expression: &Expression, ctx: &ExecutionContext) -> Result<String> {
192 Ok(materialize_persistent_expression(expression, ctx)?.to_string())
193}
194
195fn bind_index_option_value(expression: &Expression, ctx: &ExecutionContext) -> Result<String> {
196 let expression = materialize_persistent_expression(expression, ctx)?;
197 if let Expression::Identifier(identifier) = &expression {
198 return Ok(identifier.value_lower.to_string());
199 }
200 let value = ExpressionEval::compile(&expression, &[])?
201 .with_context(ctx)
202 .eval_slice(&Row::new())?;
203 match value {
204 Value::Integer(value) => Ok(value.to_string()),
205 Value::Float(value) if value.is_finite() => Ok(value.to_string()),
206 Value::Text(value) => Ok(value.to_string()),
207 Value::Boolean(value) => Ok(value.to_string()),
208 other => Err(Error::invalid_argument(format!(
209 "index option must be a finite scalar value, got {other:?}"
210 ))),
211 }
212}
213
214fn materialize_catalog_create_table(
215 statement: &CreateTableStatement,
216 ctx: &ExecutionContext,
217) -> Result<CreateTableStatement> {
218 let mut statement = statement.clone();
219 for column in &mut statement.columns {
220 for constraint in &mut column.constraints {
221 match constraint {
222 ColumnConstraint::Default(expression) | ColumnConstraint::Check(expression) => {
223 *expression = materialize_persistent_expression(expression, ctx)?;
224 }
225 _ => {}
226 }
227 }
228 }
229 for constraint in &mut statement.table_constraints {
230 if let TableConstraint::Check(expression) = constraint {
231 **expression = materialize_persistent_expression(expression, ctx)?;
232 }
233 }
234 Ok(statement)
235}
236
237fn materialize_catalog_create_index(
238 statement: &CreateIndexStatement,
239 ctx: &ExecutionContext,
240) -> Result<CreateIndexStatement> {
241 let mut statement = statement.clone();
242 for (_, expression) in &mut statement.options {
243 *expression = materialize_persistent_expression(expression, ctx)?;
244 }
245 if let Some(predicate) = &mut statement.where_clause {
246 **predicate = materialize_persistent_expression(predicate, ctx)?;
247 }
248 Ok(statement)
249}
250
251fn generated_catalog_index(table_name: &str, index_name: &str, columns: &[String]) -> Statement {
252 Statement::CreateIndex(CreateIndexStatement {
253 token: dummy_token_clone(),
254 index_name: Identifier::new(dummy_token_clone(), index_name.to_owned()),
255 table_name: Identifier::new(dummy_token_clone(), table_name.to_owned()),
256 columns: columns
257 .iter()
258 .map(|column| Identifier::new(dummy_token_clone(), column.clone()))
259 .collect(),
260 is_unique: false,
261 if_not_exists: false,
262 index_method: None,
263 options: Vec::new(),
264 where_clause: None,
265 operator_class: None,
266 })
267}
268
269fn stage_generated_foreign_key_indexes(
270 catalog: &mut crate::catalog::DdlTransaction,
271 table_name: &str,
272 columns: &[String],
273 actor: ObjectId,
274 current_database: Option<&str>,
275) -> Result<()> {
276 for column in columns {
277 let index_name = format!("fk_{table_name}_{column}");
278 catalog.stage_statement_as(
279 generated_catalog_index(table_name, &index_name, std::slice::from_ref(column)),
280 actor,
281 current_database,
282 )?;
283 }
284 Ok(())
285}
286
287#[allow(clippy::too_many_arguments)]
292fn validate_fk_reference(
293 engine: &dyn Engine,
294 transaction_parent: Option<&dyn Table>,
295 schema_builder: &SchemaBuilder,
296 fk_col_name: &str,
297 fk_col_display: &str,
298 ref_table_lower: &str,
299 ref_table_display: &str,
300 current_table_lower: &str,
301 ref_col_opt: Option<&str>,
302 on_delete: ForeignKeyAction,
303 on_update: ForeignKeyAction,
304) -> Result<ForeignKeyConstraint> {
305 let fk_col_idx = schema_builder.column_index(fk_col_name).ok_or_else(|| {
306 Error::internal(format!(
307 "foreign key column '{}' not found in table definition",
308 fk_col_display
309 ))
310 })?;
311 let fk_col_def = schema_builder
312 .column_definition(fk_col_idx)
313 .ok_or_else(|| {
314 Error::internal(format!(
315 "foreign key column '{}' has no declared type",
316 fk_col_display
317 ))
318 })?;
319
320 if ref_table_lower == current_table_lower {
321 let (ref_col_idx, ref_col_def) = if let Some(ref_col_name) = ref_col_opt {
322 let ref_col_idx = schema_builder.column_index(ref_col_name).ok_or_else(|| {
323 Error::InvalidArgument(format!(
324 "foreign key references non-existent column '{}' in table '{}'",
325 ref_col_name, ref_table_display
326 ))
327 })?;
328 let ref_col_def = schema_builder
329 .column_definition(ref_col_idx)
330 .ok_or_else(|| {
331 Error::internal("self-referencing FK column disappeared during schema binding")
332 })?;
333 (ref_col_idx, ref_col_def)
334 } else {
335 schema_builder.primary_key_column().ok_or_else(|| {
336 Error::InvalidArgument(format!(
337 "table '{}' has no single primary key for FK reference default",
338 ref_table_display
339 ))
340 })?
341 };
342 if !ref_col_def.primary_key {
343 return Err(Error::InvalidArgument(format!(
344 "self-referencing foreign key on '{}' must reference the table PRIMARY KEY; '{}' is not PRIMARY KEY",
345 fk_col_display, ref_col_def.name
346 )));
347 }
348 if !fk_col_def.has_same_declared_type(ref_col_def) {
349 return Err(Error::InvalidArgument(format!(
350 "foreign key column '{}' has type {}, but referenced column '{}.{}' has type {}",
351 fk_col_display,
352 fk_col_def.formatted_data_type(),
353 ref_table_display,
354 ref_col_def.name,
355 ref_col_def.formatted_data_type()
356 )));
357 }
358 if (matches!(on_delete, ForeignKeyAction::SetNull)
359 || matches!(on_update, ForeignKeyAction::SetNull))
360 && !schema_builder.is_column_nullable(fk_col_idx)
361 {
362 return Err(Error::InvalidArgument(format!(
363 "foreign key column '{}' has SET NULL action but is NOT NULL",
364 fk_col_display
365 )));
366 }
367 return Ok(ForeignKeyConstraint {
368 column_index: fk_col_idx,
369 column_name: fk_col_name.to_string(),
370 referenced_table: current_table_lower.to_string(),
371 referenced_column: schema_builder
372 .column_definition(ref_col_idx)
373 .expect("bound self-reference column")
374 .name_lower
375 .clone(),
376 on_delete,
377 on_update,
378 });
379 }
380
381 if transaction_parent.is_none() && !engine.table_exists(ref_table_lower)? {
383 return Err(Error::internal(format!(
384 "foreign key on column '{}' references non-existent table '{}'",
385 fk_col_display, ref_table_display
386 )));
387 }
388
389 let parent_schema = if let Some(parent) = transaction_parent {
390 parent.schema().clone()
391 } else {
392 engine.get_table_schema(ref_table_lower)?.as_ref().clone()
393 };
394
395 let ref_col_name = if let Some(rc) = ref_col_opt {
397 rc.to_string()
398 } else {
399 let pk_indices = parent_schema.primary_key_indices();
400 if pk_indices.len() == 1 {
401 parent_schema.columns[pk_indices[0]].name.to_lowercase()
402 } else {
403 return Err(Error::internal(format!(
404 "table '{}' has no primary key for FK reference default",
405 ref_table_display
406 )));
407 }
408 };
409
410 let (ref_col_idx, ref_col_def) = parent_schema.find_column(&ref_col_name).ok_or_else(|| {
412 Error::internal(format!(
413 "foreign key references non-existent column '{}' in table '{}'",
414 ref_col_name, ref_table_display
415 ))
416 })?;
417
418 if !ref_col_def.primary_key {
419 let has_unique = if let Some(parent) = transaction_parent {
420 parent.get_indexes().iter().any(|idx| {
421 idx.is_unique()
422 && idx.partial_predicate().is_none()
423 && idx.column_ids().len() == 1
424 && idx.column_ids()[0] as usize == ref_col_idx
425 })
426 } else {
427 engine
428 .get_all_indexes(ref_table_lower)
429 .map(|indexes| {
430 indexes.iter().any(|idx| {
431 idx.is_unique()
432 && idx.partial_predicate().is_none()
433 && idx.column_ids().len() == 1
434 && idx.column_ids()[0] as usize == ref_col_idx
435 })
436 })
437 .unwrap_or(false)
438 };
439
440 if !has_unique {
441 return Err(Error::internal(format!(
442 "foreign key on '{}' references column '{}' in '{}' which is neither PRIMARY KEY nor UNIQUE",
443 fk_col_display, ref_col_name, ref_table_display
444 )));
445 }
446 }
447
448 if !fk_col_def.has_same_declared_type(ref_col_def) {
449 return Err(Error::InvalidArgument(format!(
450 "foreign key column '{}' has type {}, but referenced column '{}.{}' has type {}",
451 fk_col_display,
452 fk_col_def.formatted_data_type(),
453 ref_table_display,
454 ref_col_name,
455 ref_col_def.formatted_data_type()
456 )));
457 }
458
459 if (matches!(on_delete, ForeignKeyAction::SetNull)
461 || matches!(on_update, ForeignKeyAction::SetNull))
462 && !schema_builder.is_column_nullable(fk_col_idx)
463 {
464 return Err(Error::internal(format!(
465 "foreign key column '{}' has ON {} SET NULL but is NOT NULL",
466 fk_col_display,
467 if matches!(on_delete, ForeignKeyAction::SetNull) {
468 "DELETE"
469 } else {
470 "UPDATE"
471 }
472 )));
473 }
474
475 Ok(ForeignKeyConstraint {
476 column_index: fk_col_idx,
477 column_name: fk_col_name.to_string(),
478 referenced_table: ref_table_lower.to_string(),
479 referenced_column: ref_col_name,
480 on_delete,
481 on_update,
482 })
483}
484
485fn catalog_create_table_statement(schema: &Schema) -> Result<Statement> {
486 fn quote_identifier(value: &str) -> String {
487 format!("\"{}\"", value.replace('"', "\"\""))
488 }
489
490 let columns = schema
491 .columns
492 .iter()
493 .map(|column| {
494 format!(
495 "{} {}{}",
496 quote_identifier(&column.name),
497 column.formatted_data_type(),
498 if column.nullable { "" } else { " NOT NULL" }
499 )
500 })
501 .collect::<Vec<_>>()
502 .join(", ");
503 let sql = format!(
504 "CREATE TABLE {} ({columns})",
505 quote_identifier(&schema.table_name)
506 );
507 let mut statements =
508 radixdb_sql::parse_sql(&sql).map_err(|error| Error::Parse(error.to_string()))?;
509 if statements.len() != 1 || !matches!(statements.first(), Some(Statement::CreateTable(_))) {
510 return Err(Error::internal(
511 "CTAS schema did not produce one catalog CREATE TABLE statement",
512 ));
513 }
514 Ok(statements.pop().expect("single catalog statement exists"))
515}
516
517use crate::context::{
518 invalidate_in_subquery_cache_for_table, invalidate_scalar_subquery_cache_for_table,
519 invalidate_semi_join_cache_for_table, ExecutionContext,
520};
521use crate::expression::ExpressionEval;
522use crate::mutation::host::MutationHost;
523use crate::mutation::validation::{
524 compile_table_check_constraints, validate_resulting_row_constraints,
525};
526use crate::result::ExecResult;
527
528#[doc(hidden)]
529pub trait DdlExecutorExt: MutationHost {
530 fn execute_create_schema(
531 &self,
532 stmt: &CreateSchemaStatement,
533 ctx: &ExecutionContext,
534 ) -> Result<Box<dyn QueryResult>> {
535 self.execute_security_catalog_statement(Statement::CreateSchema(stmt.clone()), ctx)
536 }
537
538 fn execute_create_principal(
539 &self,
540 stmt: &CreatePrincipalStatement,
541 ctx: &ExecutionContext,
542 ) -> Result<Box<dyn QueryResult>> {
543 self.execute_security_catalog_statement(Statement::CreatePrincipal(stmt.clone()), ctx)
544 }
545
546 fn execute_create_role(
547 &self,
548 stmt: &CreateRoleStatement,
549 ctx: &ExecutionContext,
550 ) -> Result<Box<dyn QueryResult>> {
551 self.execute_security_catalog_statement(Statement::CreateRole(stmt.clone()), ctx)
552 }
553
554 fn execute_alter_security_subject(
555 &self,
556 stmt: &AlterSecuritySubjectStatement,
557 ctx: &ExecutionContext,
558 ) -> Result<Box<dyn QueryResult>> {
559 self.execute_security_catalog_statement(Statement::AlterSecuritySubject(stmt.clone()), ctx)
560 }
561
562 fn execute_drop_security_subject(
563 &self,
564 stmt: &DropSecuritySubjectStatement,
565 ctx: &ExecutionContext,
566 ) -> Result<Box<dyn QueryResult>> {
567 self.execute_security_catalog_statement(Statement::DropSecuritySubject(stmt.clone()), ctx)
568 }
569
570 fn execute_grant(
571 &self,
572 stmt: &GrantStatement,
573 ctx: &ExecutionContext,
574 ) -> Result<Box<dyn QueryResult>> {
575 self.execute_security_catalog_statement(Statement::Grant(Box::new(stmt.clone())), ctx)
576 }
577
578 fn execute_revoke(
579 &self,
580 stmt: &RevokeStatement,
581 ctx: &ExecutionContext,
582 ) -> Result<Box<dyn QueryResult>> {
583 self.execute_security_catalog_statement(Statement::Revoke(Box::new(stmt.clone())), ctx)
584 }
585
586 fn execute_alter_owner(
587 &self,
588 stmt: &AlterOwnerStatement,
589 ctx: &ExecutionContext,
590 ) -> Result<Box<dyn QueryResult>> {
591 self.execute_security_catalog_statement(Statement::AlterOwner(Box::new(stmt.clone())), ctx)
592 }
593
594 fn execute_drop_routine(
595 &self,
596 stmt: &DropRoutineStatement,
597 ctx: &ExecutionContext,
598 ) -> Result<Box<dyn QueryResult>> {
599 self.execute_security_catalog_statement(Statement::DropRoutine(Box::new(stmt.clone())), ctx)
600 }
601
602 fn execute_drop_trigger(
603 &self,
604 stmt: &DropTriggerStatement,
605 ctx: &ExecutionContext,
606 ) -> Result<Box<dyn QueryResult>> {
607 self.execute_security_catalog_statement(Statement::DropTrigger(Box::new(stmt.clone())), ctx)
608 }
609
610 fn execute_drop_job(
611 &self,
612 stmt: &DropJobStatement,
613 ctx: &ExecutionContext,
614 ) -> Result<Box<dyn QueryResult>> {
615 self.execute_security_catalog_statement(Statement::DropJob(Box::new(stmt.clone())), ctx)
616 }
617
618 fn execute_alter_job(
619 &self,
620 stmt: &AlterJobStatement,
621 ctx: &ExecutionContext,
622 ) -> Result<Box<dyn QueryResult>> {
623 self.execute_security_catalog_statement(Statement::AlterJob(Box::new(stmt.clone())), ctx)
624 }
625
626 fn execute_security_catalog_statement(
627 &self,
628 statement: Statement,
629 ctx: &ExecutionContext,
630 ) -> Result<Box<dyn QueryResult>> {
631 let mutation = self.mutation_stage_catalog_statement_as(
632 statement,
633 ctx.effective_principal_id(),
634 ctx.current_database(),
635 )?;
636 if let Some(mutation) = mutation {
637 let mut transaction = self.mutation_engine().begin_transaction()?;
638 transaction.stage_catalog_mutation(mutation)?;
639 transaction.commit()?;
640 }
641 self.mutation_invalidate_authorization_caches();
642 Ok(Box::new(ExecResult::empty()))
643 }
644
645 fn execute_create_routine(
652 &self,
653 stmt: &CreateRoutineStatement,
654 ctx: &ExecutionContext,
655 ) -> Result<Box<dyn QueryResult>> {
656 if stmt.native.is_some() {
657 return self.execute_create_native_function(stmt, ctx);
658 }
659 let active_catalog = {
660 let active = self.mutation_active_transaction().lock().unwrap();
661 active.as_ref().map(|state| {
662 (
663 state.catalog.clone(),
664 state.catalog.working_generation_shared(),
665 )
666 })
667 };
668
669 if let Some((mut catalog, pinned_working)) = active_catalog {
670 self.compile_and_stage_routine(stmt, &mut catalog, ctx.effective_principal_id())?;
671 let mut active = self.mutation_active_transaction().lock().unwrap();
672 let state = active.as_mut().ok_or_else(|| {
673 Error::internal("explicit transaction disappeared during routine compilation")
674 })?;
675 if !state.catalog.shares_working_generation(&pinned_working) {
676 return Err(Error::InvalidArgument(
677 "transaction-private catalog changed during routine compilation; retry the statement"
678 .to_string(),
679 ));
680 }
681 state.catalog = catalog;
682 return Ok(Box::new(ExecResult::empty()));
683 }
684
685 let generation = self.mutation_engine().pin_catalog()?;
686 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
687 self.compile_and_stage_routine(stmt, &mut catalog, ctx.effective_principal_id())?;
688 if let Some(mutation) = catalog.pending_mutation()? {
689 let mut transaction = self.mutation_engine().begin_transaction()?;
690 transaction.stage_catalog_mutation(mutation)?;
691 transaction.commit()?;
692 }
693 Ok(Box::new(ExecResult::empty()))
694 }
695
696 fn execute_create_native_function(
697 &self,
698 stmt: &CreateRoutineStatement,
699 ctx: &ExecutionContext,
700 ) -> Result<Box<dyn QueryResult>> {
701 let statement = Statement::CreateRoutine(Box::new(stmt.clone()));
702 let active_catalog = {
703 let active = self.mutation_active_transaction().lock().unwrap();
704 active.as_ref().map(|state| {
705 (
706 state.catalog.clone(),
707 state.catalog.working_generation_shared(),
708 )
709 })
710 };
711 if let Some((mut catalog, pinned_working)) = active_catalog {
712 catalog.stage_statement_as(
713 statement,
714 ctx.effective_principal_id(),
715 ctx.current_database(),
716 )?;
717 let mut active = self.mutation_active_transaction().lock().unwrap();
718 let state = active.as_mut().ok_or_else(|| {
719 Error::internal("explicit transaction disappeared during native function binding")
720 })?;
721 if !state.catalog.shares_working_generation(&pinned_working) {
722 return Err(Error::InvalidArgument(
723 "transaction-private catalog changed during native function binding; retry the statement"
724 .to_owned(),
725 ));
726 }
727 state.catalog = catalog;
728 return Ok(Box::new(ExecResult::empty()));
729 }
730
731 let generation = self.mutation_engine().pin_catalog()?;
732 let mut catalog = crate::catalog::DdlTransaction::begin_shared_with_plugin_registry(
733 generation,
734 Arc::clone(self.mutation_plugin_registry()),
735 );
736 catalog.stage_statement_as(
737 statement,
738 ctx.effective_principal_id(),
739 ctx.current_database(),
740 )?;
741 if let Some(mutation) = catalog.pending_mutation()? {
742 let mut transaction = self.mutation_engine().begin_transaction()?;
743 transaction.stage_catalog_mutation(mutation)?;
744 transaction.commit()?;
745 }
746 Ok(Box::new(ExecResult::empty()))
747 }
748
749 fn compile_and_stage_routine(
750 &self,
751 stmt: &CreateRoutineStatement,
752 catalog: &mut crate::catalog::DdlTransaction,
753 actor: ObjectId,
754 ) -> Result<()> {
755 let (identity, search_path) = catalog.prepare_routine_compile(stmt)?;
756 let object_id = identity.object_id;
757 let dependencies = self.mutation_compile_routine(
758 stmt,
759 catalog.working_generation(),
760 identity,
761 search_path,
762 )?;
763 catalog.stage_compiled_routine_as(stmt, object_id, dependencies, actor)
764 }
765
766 fn execute_create_trigger(
767 &self,
768 stmt: &CreateTriggerStatement,
769 ctx: &ExecutionContext,
770 ) -> Result<Box<dyn QueryResult>> {
771 let active_catalog = {
772 let active = self.mutation_active_transaction().lock().unwrap();
773 active.as_ref().map(|state| {
774 (
775 state.catalog.clone(),
776 state.catalog.working_generation_shared(),
777 )
778 })
779 };
780 if let Some((mut catalog, pinned_working)) = active_catalog {
781 self.mutation_validate_trigger(stmt, catalog.working_generation())?;
782 catalog.stage_statement_as(
783 Statement::CreateTrigger(Box::new(stmt.clone())),
784 ctx.effective_principal_id(),
785 ctx.current_database(),
786 )?;
787 let mut active = self.mutation_active_transaction().lock().unwrap();
788 let state = active.as_mut().ok_or_else(|| {
789 Error::internal("explicit transaction disappeared during trigger compilation")
790 })?;
791 if !state.catalog.shares_working_generation(&pinned_working) {
792 return Err(Error::InvalidArgument(
793 "transaction-private catalog changed during trigger compilation; retry the statement"
794 .to_string(),
795 ));
796 }
797 state.catalog = catalog;
798 return Ok(Box::new(ExecResult::empty()));
799 }
800
801 let generation = self.mutation_engine().pin_catalog()?;
802 self.mutation_validate_trigger(stmt, generation.as_ref())?;
803 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
804 catalog.stage_statement_as(
805 Statement::CreateTrigger(Box::new(stmt.clone())),
806 ctx.effective_principal_id(),
807 ctx.current_database(),
808 )?;
809 if let Some(mutation) = catalog.pending_mutation()? {
810 let mut transaction = self.mutation_engine().begin_transaction()?;
811 transaction.stage_catalog_mutation(mutation)?;
812 transaction.commit()?;
813 }
814 Ok(Box::new(ExecResult::empty()))
815 }
816
817 fn execute_create_job(
818 &self,
819 stmt: &CreateJobStatement,
820 ctx: &ExecutionContext,
821 ) -> Result<Box<dyn QueryResult>> {
822 let active_catalog = {
823 let active = self.mutation_active_transaction().lock().unwrap();
824 active.as_ref().map(|state| {
825 (
826 state.catalog.clone(),
827 state.catalog.working_generation_shared(),
828 )
829 })
830 };
831 if let Some((mut catalog, pinned_working)) = active_catalog {
832 let definition = self.mutation_bind_job(stmt, catalog.working_generation(), ctx)?;
833 catalog.stage_bound_job_as(stmt, definition, ctx.effective_principal_id())?;
834 let mut active = self.mutation_active_transaction().lock().unwrap();
835 let state = active.as_mut().ok_or_else(|| {
836 Error::internal("explicit transaction disappeared during job binding")
837 })?;
838 if !state.catalog.shares_working_generation(&pinned_working) {
839 return Err(Error::InvalidArgument(
840 "transaction-private catalog changed during job binding; retry the statement"
841 .to_string(),
842 ));
843 }
844 state.catalog = catalog;
845 return Ok(Box::new(ExecResult::empty()));
846 }
847
848 let generation = self.mutation_engine().pin_catalog()?;
849 let definition = self.mutation_bind_job(stmt, generation.as_ref(), ctx)?;
850 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
851 catalog.stage_bound_job_as(stmt, definition, ctx.effective_principal_id())?;
852 if let Some(mutation) = catalog.pending_mutation()? {
853 let mut transaction = self.mutation_engine().begin_transaction()?;
854 transaction.stage_catalog_mutation(mutation)?;
855 transaction.commit()?;
856 }
857 Ok(Box::new(ExecResult::empty()))
858 }
859
860 fn execute_create_table(
862 &self,
863 stmt: &CreateTableStatement,
864 ctx: &ExecutionContext,
865 ) -> Result<Box<dyn QueryResult>> {
866 let table_name = &stmt.table_name.value;
867
868 let exists_in_active_transaction = {
871 let active_tx = self.mutation_active_transaction().lock().unwrap();
872 active_tx
873 .as_ref()
874 .is_some_and(|tx_state| tx_state.transaction.get_table(table_name).is_ok())
875 };
876 if exists_in_active_transaction || self.mutation_engine().table_exists(table_name)? {
877 if stmt.if_not_exists {
878 return Ok(Box::new(ExecResult::empty()));
879 }
880 return Err(Error::TableAlreadyExists(table_name.to_string()));
881 }
882
883 if self.mutation_engine().view_exists(table_name)? {
885 return Err(Error::internal(format!(
886 "cannot create table '{}': a view with the same name exists",
887 table_name
888 )));
889 }
890
891 if let Some(ref select_stmt) = stmt.as_select {
893 if self.mutation_has_active_transaction() {
894 return Err(Error::NotSupported(
895 "CREATE TABLE AS SELECT is not supported inside an explicit transaction"
896 .to_string(),
897 ));
898 }
899 return self.execute_create_table_as_select(
900 table_name,
901 select_stmt,
902 stmt.if_not_exists,
903 ctx,
904 );
905 }
906
907 let mut schema_builder = SchemaBuilder::new(table_name.as_str());
909
910 let mut table_primary_key: Option<String> = None;
911 for constraint in &stmt.table_constraints {
912 if let TableConstraint::PrimaryKey(columns) = constraint {
913 if columns.len() != 1 {
914 return Err(Error::NotSupported(
915 "composite PRIMARY KEY is not supported; use a UNIQUE multi-column constraint"
916 .to_string(),
917 ));
918 }
919 if table_primary_key
920 .replace(columns[0].value_lower.to_string())
921 .is_some()
922 {
923 return Err(Error::InvalidArgument(
924 "table declares more than one PRIMARY KEY constraint".to_string(),
925 ));
926 }
927 }
928 }
929 let column_primary_keys = stmt
930 .columns
931 .iter()
932 .filter(|column| {
933 column
934 .constraints
935 .iter()
936 .any(|constraint| matches!(constraint, ColumnConstraint::PrimaryKey))
937 })
938 .count();
939 if column_primary_keys + usize::from(table_primary_key.is_some()) > 1 {
940 return Err(Error::InvalidArgument(
941 "table declares more than one PRIMARY KEY".to_string(),
942 ));
943 }
944 let mut table_primary_key_found = table_primary_key.is_none();
945
946 let mut unique_columns: Vec<String> = Vec::new();
948
949 for col_def in &stmt.columns {
950 let col_name = &col_def.name.value;
951 let (data_type, vector_dimensions, decimal_precision, decimal_scale, external_type) =
952 super::type_binding::parse_schema_column_type(self, &col_def.data_type)?;
953 let nullable = !col_def
954 .constraints
955 .iter()
956 .any(|c| matches!(c, ColumnConstraint::NotNull));
957 let is_primary_key = col_def
958 .constraints
959 .iter()
960 .any(|c| matches!(c, ColumnConstraint::PrimaryKey))
961 || table_primary_key
962 .as_deref()
963 .is_some_and(|primary_key| primary_key == col_def.name.value_lower.as_str());
964 table_primary_key_found |= table_primary_key
965 .as_deref()
966 .is_some_and(|primary_key| primary_key == col_def.name.value_lower.as_str());
967
968 if is_primary_key && !matches!(data_type, DataType::Integer | DataType::Uuid) {
971 return Err(Error::Parse(format!(
972 "PRIMARY KEY column '{}' must be INTEGER or UUID type, got {:?}.",
973 col_name, data_type
974 )));
975 }
976
977 let is_unique = col_def
978 .constraints
979 .iter()
980 .any(|c| matches!(c, ColumnConstraint::Unique));
981
982 let is_auto_increment = col_def
983 .constraints
984 .iter()
985 .any(|c| matches!(c, ColumnConstraint::AutoIncrement));
986
987 let mut default_expr = None;
990 for constraint in &col_def.constraints {
991 if let ColumnConstraint::Default(expr) = constraint {
992 if default_expr
993 .replace(bind_persistent_expression(expr, ctx)?)
994 .is_some()
995 {
996 return Err(Error::InvalidArgument(format!(
997 "column '{}' has more than one DEFAULT constraint",
998 col_name
999 )));
1000 }
1001 }
1002 }
1003
1004 let default_value = if let Some(ref expr_str) = default_expr {
1005 if external_type.is_some() {
1006 return Err(Error::NotSupported(format!(
1007 "DEFAULT for external column '{}' requires an explicit native/text constructor",
1008 col_name
1009 )));
1010 }
1011 let val = self.evaluate_default_expression(expr_str, data_type)?;
1012 if val.is_null() {
1013 None
1014 } else {
1015 Some(val)
1016 }
1017 } else {
1018 None
1019 };
1020
1021 let mut check_expr = None;
1024 for constraint in &col_def.constraints {
1025 if let ColumnConstraint::Check(expr) = constraint {
1026 if check_expr
1027 .replace(bind_persistent_expression(expr, ctx)?)
1028 .is_some()
1029 {
1030 return Err(Error::InvalidArgument(format!(
1031 "column '{}' has more than one CHECK constraint",
1032 col_name
1033 )));
1034 }
1035 }
1036 }
1037
1038 schema_builder = schema_builder.add_with_constraints(
1040 col_name.as_str(),
1041 data_type,
1042 nullable && !is_primary_key,
1043 is_primary_key,
1044 is_auto_increment,
1045 default_expr,
1046 check_expr,
1047 );
1048 schema_builder = schema_builder.set_last_default_value(default_value);
1049
1050 if let Some((type_ref, sql_name)) = external_type {
1051 schema_builder = schema_builder.set_last_external_type(type_ref, sql_name);
1052 }
1053
1054 if vector_dimensions > 0 {
1055 schema_builder = schema_builder.set_last_vector_dimensions(vector_dimensions);
1056 }
1057 if decimal_precision > 0 {
1058 schema_builder =
1059 schema_builder.set_last_decimal_parameters(decimal_precision, decimal_scale);
1060 }
1061
1062 if is_auto_increment && !matches!(data_type, DataType::Integer | DataType::Uuid) {
1063 return Err(Error::Parse(format!(
1064 "AUTO_INCREMENT column '{}' must be INTEGER or UUID type, got {:?}.",
1065 col_name, data_type
1066 )));
1067 }
1068
1069 if is_unique && !is_primary_key {
1073 unique_columns.push(col_name.to_string());
1074 }
1075 }
1076 if !table_primary_key_found {
1077 return Err(Error::ColumnNotFound(
1078 table_primary_key.expect("missing table primary-key name"),
1079 ));
1080 }
1081
1082 for col_def in &stmt.columns {
1084 for constraint in &col_def.constraints {
1085 if let ColumnConstraint::References {
1086 table: ref ref_table,
1087 column: ref ref_col,
1088 on_delete,
1089 on_update,
1090 } = constraint
1091 {
1092 let transaction_parent = {
1093 let active_tx = self.mutation_active_transaction().lock().unwrap();
1094 active_tx.as_ref().and_then(|tx_state| {
1095 tx_state.transaction.get_table(&ref_table.value_lower).ok()
1096 })
1097 };
1098 let fk = validate_fk_reference(
1099 self.mutation_engine().as_ref(),
1100 transaction_parent.as_deref(),
1101 &schema_builder,
1102 col_def.name.value_lower.as_str(),
1103 &col_def.name.value,
1104 &ref_table.value_lower,
1105 &ref_table.value,
1106 &stmt.table_name.value_lower,
1107 ref_col.as_ref().map(|rc| rc.value_lower.as_str()),
1108 *on_delete,
1109 *on_update,
1110 )?;
1111 schema_builder = schema_builder.add_foreign_key(fk);
1112 }
1113 }
1114 }
1115
1116 let mut table_unique_constraints: Vec<Vec<String>> = Vec::new();
1120 for constraint in &stmt.table_constraints {
1121 match constraint {
1122 TableConstraint::Unique(cols) => {
1123 let col_names: Vec<String> = cols.iter().map(|c| c.value.to_string()).collect();
1124 table_unique_constraints.push(col_names);
1125 }
1126 TableConstraint::ForeignKey(fk) => {
1127 let transaction_parent = {
1128 let active_tx = self.mutation_active_transaction().lock().unwrap();
1129 active_tx.as_ref().and_then(|tx_state| {
1130 tx_state
1131 .transaction
1132 .get_table(&fk.ref_table.value_lower)
1133 .ok()
1134 })
1135 };
1136 let fk_constraint = validate_fk_reference(
1137 self.mutation_engine().as_ref(),
1138 transaction_parent.as_deref(),
1139 &schema_builder,
1140 fk.column.value_lower.as_str(),
1141 &fk.column.value,
1142 &fk.ref_table.value_lower,
1143 &fk.ref_table.value,
1144 &stmt.table_name.value_lower,
1145 fk.ref_column.as_ref().map(|rc| rc.value_lower.as_str()),
1146 fk.on_delete,
1147 fk.on_update,
1148 )?;
1149 schema_builder = schema_builder.add_foreign_key(fk_constraint);
1150 }
1151 TableConstraint::Check(expression) => {
1152 schema_builder = schema_builder
1153 .add_table_check(bind_persistent_expression(expression, ctx)?);
1154 }
1155 TableConstraint::PrimaryKey(_) => {}
1156 }
1157 }
1158
1159 let mut schema = schema_builder.build();
1160 if let Some(primary_key) = schema.primary_key_columns().first() {
1161 schema.register_primary_key_constraint(vec![primary_key.name.clone()])?;
1162 }
1163 for column in schema.columns.clone() {
1164 if let Some(expression) = column.check_expr {
1165 schema.register_check_constraint(Some(column.name), expression)?;
1166 }
1167 }
1168 let mut named_unique_constraints = Vec::new();
1169 for column in &unique_columns {
1170 let columns = vec![column.clone()];
1171 let name = schema.register_unique_constraint(columns.clone())?;
1172 named_unique_constraints.push((name, columns));
1173 }
1174 for columns in &table_unique_constraints {
1175 let name = schema.register_unique_constraint(columns.clone())?;
1176 named_unique_constraints.push((name, columns.clone()));
1177 }
1178 for foreign_key in schema.foreign_keys.clone() {
1179 schema.register_foreign_key_constraint(&foreign_key)?;
1180 }
1181 for expression in schema.table_checks.clone() {
1182 schema.register_check_constraint(None, expression)?;
1183 }
1184 schema.validate_structural_invariants()?;
1185 compile_table_check_constraints(&schema)?;
1188 schema.ensure_catalog_identity();
1193 let table_catalog_id = ObjectId::from_user_bytes(schema.catalog_id()).map_err(|error| {
1194 Error::internal(format!(
1195 "generated table catalog identity was rejected: {error}"
1196 ))
1197 })?;
1198
1199 let mut fk_index_columns: Vec<String> = Vec::new();
1201 for fk in &schema.foreign_keys {
1202 let col = &schema.columns[fk.column_index];
1203 if col.primary_key {
1205 continue;
1206 }
1207 let col_lower = col.name.to_lowercase();
1208 if unique_columns.iter().any(|u| u.to_lowercase() == col_lower) {
1209 continue;
1210 }
1211 fk_index_columns.push(col.name.clone());
1212 }
1213
1214 let catalog_statement =
1215 Statement::CreateTable(materialize_catalog_create_table(stmt, ctx)?);
1216
1217 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
1219
1220 if let Some(ref mut tx_state) = *active_tx {
1221 let mut catalog = tx_state.catalog.clone();
1224 catalog.stage_statement_with_object_ids_as(
1225 catalog_statement,
1226 [table_catalog_id],
1227 ctx.effective_principal_id(),
1228 ctx.current_database(),
1229 )?;
1230 stage_generated_foreign_key_indexes(
1231 &mut catalog,
1232 table_name,
1233 &fk_index_columns,
1234 ctx.effective_principal_id(),
1235 ctx.current_database(),
1236 )?;
1237
1238 tx_state.transaction.create_table(table_name, schema)?;
1240
1241 for (index_name, columns) in &named_unique_constraints {
1243 tx_state
1244 .transaction
1245 .create_table_index(table_name, index_name, columns, true)?;
1246 }
1247
1248 for col_name in &fk_index_columns {
1250 let index_name = format!("fk_{}_{}", table_name, col_name);
1251 tx_state.transaction.create_table_index(
1252 table_name,
1253 &index_name,
1254 std::slice::from_ref(col_name),
1255 false,
1256 )?;
1257 }
1258 tx_state.catalog = catalog;
1259 } else {
1260 let generation = self.mutation_engine().pin_catalog()?;
1264 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
1265 catalog.stage_statement_with_object_ids_as(
1266 catalog_statement,
1267 [table_catalog_id],
1268 ctx.effective_principal_id(),
1269 ctx.current_database(),
1270 )?;
1271 stage_generated_foreign_key_indexes(
1272 &mut catalog,
1273 table_name,
1274 &fk_index_columns,
1275 ctx.effective_principal_id(),
1276 ctx.current_database(),
1277 )?;
1278 let mut transaction = self.mutation_engine().begin_transaction()?;
1279 transaction.create_table(table_name, schema)?;
1280
1281 for (index_name, columns) in &named_unique_constraints {
1282 transaction.create_table_index(table_name, index_name, columns, true)?;
1283 }
1284 for col_name in &fk_index_columns {
1285 let index_name = format!("fk_{}_{}", table_name, col_name);
1286 transaction.create_table_index(
1287 table_name,
1288 &index_name,
1289 std::slice::from_ref(col_name),
1290 false,
1291 )?;
1292 }
1293 if let Some(mutation) = catalog.pending_mutation()? {
1294 transaction.stage_catalog_mutation(mutation)?;
1295 }
1296 transaction.commit()?;
1297 }
1298
1299 Ok(Box::new(ExecResult::empty()))
1300 }
1301
1302 fn execute_create_table_as_select(
1304 &self,
1305 table_name: &str,
1306 select_stmt: &SelectStatement,
1307 _if_not_exists: bool,
1308 ctx: &ExecutionContext,
1309 ) -> Result<Box<dyn QueryResult>> {
1310 use radixdb_core::Row;
1311
1312 if self.mutation_has_active_transaction() {
1313 return Err(Error::NotSupported(
1314 "CREATE TABLE AS SELECT is not supported inside an explicit transaction"
1315 .to_string(),
1316 ));
1317 }
1318
1319 let bound_columns = self.mutation_describe_select_output(select_stmt)?;
1323
1324 let mut result = self.mutation_execute_select(select_stmt, ctx)?;
1327 let columns: Vec<String> = result.columns().to_vec();
1328 if columns.len() != bound_columns.len() {
1329 return Err(Error::internal(format!(
1330 "CTAS binder produced {} columns but SELECT produced {}",
1331 bound_columns.len(),
1332 columns.len()
1333 )));
1334 }
1335
1336 let mut schema_builder = SchemaBuilder::new(table_name);
1339
1340 for (col_name, bound) in columns.iter().zip(&bound_columns) {
1341 let base_name = if let Some(pos) = col_name.rfind('.') {
1343 &col_name[pos + 1..]
1344 } else {
1345 col_name.as_str()
1346 };
1347
1348 schema_builder = schema_builder.add_nullable(base_name, bound.data_type);
1349 }
1350
1351 let mut schema = schema_builder.build();
1352 schema.ensure_catalog_identity();
1353 let table_catalog_id = ObjectId::from_user_bytes(schema.catalog_id()).map_err(|error| {
1354 Error::internal(format!(
1355 "generated CTAS catalog identity was rejected: {error}"
1356 ))
1357 })?;
1358 let catalog_statement = catalog_create_table_statement(&schema)?;
1359
1360 let generation = self.mutation_engine().pin_catalog()?;
1363 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
1364 catalog.stage_statement_with_object_ids_as(
1365 catalog_statement,
1366 [table_catalog_id],
1367 ctx.effective_principal_id(),
1368 ctx.current_database(),
1369 )?;
1370 let mut tx = self.mutation_engine().begin_transaction()?;
1371 let mut table = tx.create_table(table_name, schema)?;
1372 let mut rows_count = 0usize;
1373 while result.next() {
1374 let row: Row = result.take_row();
1375 let _ = table.insert(row)?;
1376 rows_count += 1;
1377 }
1378 if let Some(err) = result.last_error() {
1379 return Err(err);
1380 }
1381 if let Some(mutation) = catalog.pending_mutation()? {
1382 tx.stage_catalog_mutation(mutation)?;
1383 }
1384 tx.commit()?;
1385
1386 Ok(Box::new(ExecResult::with_rows_affected(rows_count as i64)))
1387 }
1388
1389 fn execute_drop_table(
1391 &self,
1392 stmt: &DropTableStatement,
1393 ctx: &ExecutionContext,
1394 ) -> Result<Box<dyn QueryResult>> {
1395 let table_name = &stmt.table_name.value;
1396
1397 let private_exists = {
1400 let active_tx = self.mutation_active_transaction().lock().unwrap();
1401 active_tx
1402 .as_ref()
1403 .is_some_and(|state| state.transaction.get_table(table_name).is_ok())
1404 };
1405 if !private_exists && !self.mutation_engine().table_exists(table_name)? {
1406 if stmt.if_exists {
1407 return Ok(Box::new(ExecResult::empty()));
1408 }
1409 return Err(Error::TableNotFound(table_name.to_string()));
1410 }
1411
1412 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
1414 let txn_id = active_tx.as_ref().map(|s| s.transaction.id());
1415
1416 crate::mutation::foreign_key::check_no_referencing_rows(
1419 self.mutation_engine(),
1420 table_name,
1421 txn_id,
1422 )?;
1423
1424 let catalog_statement = Statement::DropTable(stmt.clone());
1425
1426 if let Some(ref mut tx_state) = *active_tx {
1427 let mut catalog = tx_state.catalog.clone();
1428 catalog.stage_statement_as(
1429 catalog_statement,
1430 ctx.effective_principal_id(),
1431 ctx.current_database(),
1432 )?;
1433 tx_state.transaction.drop_table(table_name)?;
1437 tx_state.catalog = catalog;
1438 } else {
1439 let generation = self.mutation_engine().pin_catalog()?;
1440 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
1441 catalog.stage_statement_as(
1442 catalog_statement,
1443 ctx.effective_principal_id(),
1444 ctx.current_database(),
1445 )?;
1446 let mut transaction = self.mutation_engine().begin_transaction()?;
1447 transaction.drop_table(table_name)?;
1448 if let Some(mutation) = catalog.pending_mutation()? {
1449 transaction.stage_catalog_mutation(mutation)?;
1450 }
1451 transaction.commit()?;
1452 }
1453
1454 self.mutation_invalidate_query_cache(table_name);
1456 self.mutation_invalidate_semantic_cache(table_name);
1457 invalidate_semi_join_cache_for_table(table_name);
1458 invalidate_scalar_subquery_cache_for_table(table_name);
1459 invalidate_in_subquery_cache_for_table(table_name);
1460
1461 Ok(Box::new(ExecResult::empty()))
1462 }
1463
1464 fn execute_create_index(
1466 &self,
1467 stmt: &CreateIndexStatement,
1468 ctx: &ExecutionContext,
1469 ) -> Result<Box<dyn QueryResult>> {
1470 let table_name = &stmt.table_name.value;
1471 let index_name = &stmt.index_name.value;
1472
1473 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
1476 let mut validation_tx = if active_tx.is_none() {
1477 Some(self.mutation_engine().begin_transaction()?)
1478 } else {
1479 None
1480 };
1481
1482 let is_unique = stmt.is_unique;
1484
1485 let table = if let Some(tx_state) = active_tx.as_ref() {
1487 tx_state.transaction.get_table(table_name)?
1488 } else {
1489 validation_tx
1490 .as_ref()
1491 .expect("validation transaction exists")
1492 .get_table(table_name)?
1493 };
1494 let schema = table.schema();
1495
1496 for col_id in &stmt.columns {
1498 let col_name = &col_id.value;
1499 if !schema
1500 .column_index_map()
1501 .contains_key(col_id.value_lower.as_str())
1502 {
1503 return Err(Error::ColumnNotFound(col_name.to_string()));
1504 }
1505 }
1506
1507 let column_names: Vec<String> = stmt.columns.iter().map(|c| c.value.to_string()).collect();
1509
1510 let requested_index_type = stmt.index_method.map(|method| match method {
1512 radixdb_sql::ast::IndexMethod::BTree => radixdb_core::IndexType::BTree,
1513 radixdb_sql::ast::IndexMethod::Hash => radixdb_core::IndexType::Hash,
1514 radixdb_sql::ast::IndexMethod::Bitmap => radixdb_core::IndexType::Bitmap,
1515 radixdb_sql::ast::IndexMethod::Hnsw => radixdb_core::IndexType::Hnsw,
1516 });
1517 let normalized_index_type = if column_names.len() > 1 {
1518 radixdb_core::IndexType::MultiColumn
1519 } else if let Some(index_type) = requested_index_type {
1520 index_type
1521 } else {
1522 match schema
1523 .find_column(&column_names[0])
1524 .expect("CREATE INDEX column was validated")
1525 .1
1526 .data_type
1527 {
1528 DataType::Text | DataType::Json | DataType::Bytes => radixdb_core::IndexType::Hash,
1529 DataType::Boolean => radixdb_core::IndexType::Bitmap,
1530 DataType::Vector => radixdb_core::IndexType::Hnsw,
1531 _ => radixdb_core::IndexType::BTree,
1532 }
1533 };
1534
1535 if requested_index_type == Some(radixdb_core::IndexType::Hnsw) && column_names.len() > 1 {
1537 return Err(Error::invalid_argument(
1538 "HNSW index must be on a single vector column; multi-column HNSW indexes are not supported",
1539 ));
1540 }
1541
1542 let partial_predicate = if let Some(where_clause) = &stmt.where_clause {
1543 if normalized_index_type == radixdb_core::IndexType::Hnsw {
1544 return Err(Error::invalid_argument(
1545 "partial HNSW indexes are not supported",
1546 ));
1547 }
1548 let where_clause = materialize_persistent_expression(where_clause, ctx)?;
1549 Some(crate::mutation::partial_index::bind_from_ast(
1550 &where_clause,
1551 schema,
1552 )?)
1553 } else {
1554 None
1555 };
1556 let mut hnsw_m: Option<u16> = None;
1558 let mut hnsw_ef_construction: Option<u16> = None;
1559 let mut hnsw_ef_search: Option<u16> = None;
1560 let mut hnsw_distance_metric: Option<u8> = None;
1561 let is_hnsw = normalized_index_type == radixdb_core::IndexType::Hnsw;
1562 if !stmt.options.is_empty() && !is_hnsw {
1563 return Err(Error::invalid_argument(
1564 "CREATE INDEX WITH options are supported only for HNSW indexes",
1565 ));
1566 }
1567 for (key, expression) in &stmt.options {
1568 let value = bind_index_option_value(expression, ctx)?;
1569 match key.as_str() {
1570 "m" => {
1571 let v = value.parse::<u16>().map_err(|_| {
1572 Error::invalid_argument(format!(
1573 "invalid value for HNSW option 'm': '{}' (expected integer >= 2)",
1574 value
1575 ))
1576 })?;
1577 if v < 2 {
1578 return Err(Error::invalid_argument(format!(
1579 "HNSW option 'm' must be >= 2, got {}",
1580 v
1581 )));
1582 }
1583 hnsw_m = Some(v);
1584 }
1585 "ef_construction" => {
1586 let parsed = value.parse::<u16>().map_err(|_| {
1587 Error::invalid_argument(format!(
1588 "invalid value for HNSW option 'ef_construction': '{}' (expected positive integer)",
1589 value
1590 ))
1591 })?;
1592 if parsed == 0 {
1593 return Err(Error::invalid_argument(
1594 "HNSW option 'ef_construction' must be greater than zero",
1595 ));
1596 }
1597 hnsw_ef_construction = Some(parsed);
1598 }
1599 "ef_search" => {
1600 let parsed = value.parse::<u16>().map_err(|_| {
1601 Error::invalid_argument(format!(
1602 "invalid value for HNSW option 'ef_search': '{}' (expected positive integer)",
1603 value
1604 ))
1605 })?;
1606 if parsed == 0 {
1607 return Err(Error::invalid_argument(
1608 "HNSW option 'ef_search' must be greater than zero",
1609 ));
1610 }
1611 hnsw_ef_search = Some(parsed);
1612 }
1613 "metric" | "distance" => {
1614 let metric = radixdb_storage::index::HnswDistanceMetric::from_name(
1615 &value.to_lowercase(),
1616 )
1617 .ok_or_else(|| {
1618 Error::invalid_argument(format!(
1619 "unknown HNSW distance metric '{}' (expected: l2, cosine, or ip)",
1620 value
1621 ))
1622 })?;
1623 hnsw_distance_metric = Some(metric.as_u8());
1624 }
1625 other if is_hnsw => {
1626 return Err(Error::invalid_argument(format!(
1627 "unknown HNSW index option '{}' (valid options: m, ef_construction, ef_search, metric)",
1628 other
1629 )));
1630 }
1631 _ => {}
1632 }
1633 }
1634
1635 if is_hnsw {
1636 if hnsw_m.is_none() {
1637 let dims = schema
1638 .find_column(&column_names[0])
1639 .map(|(_, col)| col.vector_dimensions as usize)
1640 .unwrap_or(0);
1641 hnsw_m = Some(radixdb_storage::index::default_m_for_dims(dims) as u16);
1642 }
1643 let m = hnsw_m.unwrap() as usize;
1644 hnsw_ef_construction
1645 .get_or_insert(radixdb_storage::index::default_ef_construction(m) as u16);
1646 hnsw_ef_search.get_or_insert(radixdb_storage::index::default_ef_search(m) as u16);
1647 hnsw_distance_metric.get_or_insert(0);
1648 }
1649
1650 if let Some(existing) = table.get_index(index_name) {
1653 if stmt.if_not_exists {
1654 let requested_predicate = partial_predicate.as_ref().map(|p| p.canonical_sql());
1655 let existing_predicate = existing.partial_predicate().map(|p| p.canonical_sql());
1656 let requested_columns = column_names
1657 .iter()
1658 .map(|name| name.to_lowercase())
1659 .collect::<Vec<_>>();
1660 let existing_columns = existing
1661 .column_names()
1662 .iter()
1663 .map(|name| name.to_lowercase())
1664 .collect::<Vec<_>>();
1665 let hnsw_options_match = normalized_index_type != radixdb_core::IndexType::Hnsw
1666 || (existing.hnsw_m() == hnsw_m
1667 && existing.hnsw_ef_construction() == hnsw_ef_construction
1668 && existing
1669 .default_ef_search()
1670 .and_then(|value| u16::try_from(value).ok())
1671 == hnsw_ef_search
1672 && existing.hnsw_distance_metric() == hnsw_distance_metric);
1673 if existing_columns == requested_columns
1674 && existing.is_unique() == is_unique
1675 && existing.index_type() == normalized_index_type
1676 && existing_predicate == requested_predicate
1677 && hnsw_options_match
1678 {
1679 return Ok(Box::new(ExecResult::empty()));
1680 }
1681 return Err(Error::internal(format!(
1682 "index already exists with different definition: {}",
1683 index_name
1684 )));
1685 }
1686 return Err(Error::internal(format!(
1687 "index already exists: {}",
1688 index_name
1689 )));
1690 }
1691
1692 let definition = radixdb_storage::traits::PendingIndexDefinition {
1693 table_name: table_name.to_string(),
1694 index_name: index_name.to_string(),
1695 columns: column_names,
1696 is_unique,
1697 index_type: Some(normalized_index_type),
1698 hnsw_m,
1699 hnsw_ef_construction,
1700 hnsw_ef_search,
1701 hnsw_distance_metric,
1702 partial_predicate,
1703 key_encoder: None,
1704 };
1705 let catalog_statement =
1706 Statement::CreateIndex(materialize_catalog_create_index(stmt, ctx)?);
1707 if let Some(tx_state) = active_tx.as_mut() {
1708 let mut catalog = tx_state.catalog.clone();
1709 catalog.stage_statement_as(
1710 catalog_statement,
1711 ctx.effective_principal_id(),
1712 ctx.current_database(),
1713 )?;
1714 let mut definition = definition;
1715 let (index_type, key_encoder) = crate::catalog::bind_pending_index_semantics(
1716 catalog.working_generation(),
1717 index_name,
1718 self.mutation_plugin_registry().as_ref(),
1719 )?;
1720 definition.index_type = Some(index_type);
1721 definition.key_encoder = key_encoder;
1722 tx_state.transaction.stage_create_index(definition)?;
1723 tx_state.catalog = catalog;
1724 } else {
1725 let generation = self.mutation_engine().pin_catalog()?;
1726 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
1727 catalog.stage_statement_as(
1728 catalog_statement,
1729 ctx.effective_principal_id(),
1730 ctx.current_database(),
1731 )?;
1732 let mut definition = definition;
1733 let (index_type, key_encoder) = crate::catalog::bind_pending_index_semantics(
1734 catalog.working_generation(),
1735 index_name,
1736 self.mutation_plugin_registry().as_ref(),
1737 )?;
1738 definition.index_type = Some(index_type);
1739 definition.key_encoder = key_encoder;
1740 let mut transaction = validation_tx
1741 .take()
1742 .expect("autocommit CREATE INDEX owns a validation transaction");
1743 transaction.stage_create_index(definition)?;
1744 if let Some(mutation) = catalog.pending_mutation()? {
1745 transaction.stage_catalog_mutation(mutation)?;
1746 }
1747 transaction.commit()?;
1748 }
1749
1750 Ok(Box::new(ExecResult::empty()))
1751 }
1752
1753 fn execute_drop_index(
1755 &self,
1756 stmt: &DropIndexStatement,
1757 ctx: &ExecutionContext,
1758 ) -> Result<Box<dyn QueryResult>> {
1759 if self.mutation_active_transaction().lock().unwrap().is_some() {
1760 return Err(Error::NotSupported(
1761 "DROP INDEX is not supported inside an explicit transaction".to_string(),
1762 ));
1763 }
1764 let index_name = &stmt.index_name.value;
1765
1766 let table_name = match &stmt.table_name {
1768 Some(t) => t.value.to_string(),
1769 None => {
1770 return Err(Error::InvalidArgument(
1771 "DROP INDEX requires table name".to_string(),
1772 ))
1773 }
1774 };
1775
1776 if !self.mutation_engine().table_exists(&table_name)? {
1778 if stmt.if_exists {
1779 return Ok(Box::new(ExecResult::empty()));
1780 }
1781 return Err(Error::TableNotFound(table_name));
1782 }
1783
1784 if !self
1786 .mutation_engine()
1787 .index_exists(index_name, &table_name)?
1788 {
1789 if stmt.if_exists {
1790 return Ok(Box::new(ExecResult::empty()));
1791 }
1792 return Err(Error::IndexNotFound(index_name.to_string()));
1793 }
1794
1795 let mut tx = self.mutation_engine().begin_transaction()?;
1796 let table = tx.get_table(&table_name)?;
1797 let dropped_index = table
1798 .get_index(index_name)
1799 .ok_or_else(|| Error::IndexNotFound(index_name.to_string()))?;
1800
1801 if dropped_index.is_unique()
1806 && dropped_index.partial_predicate().is_none()
1807 && dropped_index.column_ids().len() == 1
1808 {
1809 let schema = table.schema();
1810 let dropped_column = dropped_index.column_ids()[0] as usize;
1811 if let Some(column) = schema.columns.get(dropped_column) {
1812 let referenced = self
1813 .mutation_engine()
1814 .find_referencing_fks(&table_name.to_lowercase())
1815 .iter()
1816 .any(|(_, fk)| {
1817 fk.referenced_column
1818 .eq_ignore_ascii_case(column.name.as_str())
1819 });
1820 let has_other_backing = column.primary_key
1821 || table.get_indexes().iter().any(|candidate| {
1822 candidate.name() != dropped_index.name()
1823 && candidate.is_unique()
1824 && candidate.partial_predicate().is_none()
1825 && candidate.column_ids() == dropped_index.column_ids()
1826 });
1827 if referenced && !has_other_backing {
1828 return Err(Error::InvalidArgument(format!(
1829 "cannot drop UNIQUE index '{}' because foreign keys reference '{}.{}'",
1830 index_name, table_name, column.name
1831 )));
1832 }
1833 }
1834 }
1835
1836 let generation = self.mutation_engine().pin_catalog()?;
1837 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
1838 catalog.stage_statement_as(
1839 Statement::DropIndex(stmt.clone()),
1840 ctx.effective_principal_id(),
1841 ctx.current_database(),
1842 )?;
1843 tx.stage_drop_index(PendingIndexDrop {
1844 table_name,
1845 index_name: index_name.to_string(),
1846 schema_owned: false,
1847 })?;
1848 if let Some(mutation) = catalog.pending_mutation()? {
1849 tx.stage_catalog_mutation(mutation)?;
1850 }
1851 tx.commit()?;
1852
1853 Ok(Box::new(ExecResult::empty()))
1854 }
1855
1856 fn execute_alter_index(
1865 &self,
1866 stmt: &AlterIndexStatement,
1867 ctx: &ExecutionContext,
1868 ) -> Result<Box<dyn QueryResult>> {
1869 if self.mutation_active_transaction().lock().unwrap().is_some() {
1870 return Err(Error::NotSupported(
1871 "ALTER INDEX is not supported inside an explicit transaction".to_string(),
1872 ));
1873 }
1874 let old_index_name = &stmt.index_name.value;
1875 let new_index_name = &stmt.new_index_name.value;
1876
1877 let mut tx = self.mutation_engine().begin_transaction()?;
1878 let tables = tx.list_tables()?;
1879
1880 let mut owner_table: Option<String> = None;
1881 for table_name in &tables {
1882 if self
1883 .mutation_engine()
1884 .index_exists(new_index_name, table_name)?
1885 {
1886 return Err(Error::InvalidArgument(format!(
1887 "index already exists: {}",
1888 new_index_name
1889 )));
1890 }
1891
1892 if self
1893 .mutation_engine()
1894 .index_exists(old_index_name, table_name)?
1895 {
1896 if owner_table.is_some() {
1897 return Err(Error::InvalidArgument(format!(
1898 "index name is ambiguous: {}",
1899 old_index_name
1900 )));
1901 }
1902 owner_table = Some(table_name.clone());
1903 }
1904 }
1905
1906 let table_name =
1907 owner_table.ok_or_else(|| Error::IndexNotFound(old_index_name.to_string()))?;
1908
1909 let generation = self.mutation_engine().pin_catalog()?;
1910 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
1911 catalog.stage_statement_as(
1912 Statement::AlterIndex(stmt.clone()),
1913 ctx.effective_principal_id(),
1914 ctx.current_database(),
1915 )?;
1916 tx.stage_rename_index(PendingIndexRename {
1917 table_name,
1918 old_index_name: old_index_name.to_string(),
1919 new_index_name: new_index_name.to_string(),
1920 })?;
1921 if let Some(mutation) = catalog.pending_mutation()? {
1922 tx.stage_catalog_mutation(mutation)?;
1923 }
1924 tx.commit()?;
1925
1926 Ok(Box::new(ExecResult::empty()))
1927 }
1928
1929 fn alter_index_definition(
1930 table_name: &str,
1931 name: String,
1932 columns: Vec<String>,
1933 is_unique: bool,
1934 ) -> PendingIndexDefinition {
1935 PendingIndexDefinition {
1936 table_name: table_name.to_string(),
1937 index_name: name,
1938 columns,
1939 is_unique,
1940 index_type: None,
1941 hnsw_m: None,
1942 hnsw_ef_construction: None,
1943 hnsw_ef_search: None,
1944 hnsw_distance_metric: None,
1945 partial_predicate: None,
1946 key_encoder: None,
1947 }
1948 }
1949
1950 #[allow(clippy::too_many_arguments)]
1951 fn bind_alter_foreign_key(
1952 &self,
1953 transaction: &dyn Transaction,
1954 schema: &Schema,
1955 column_name: &str,
1956 referenced_table: &Identifier,
1957 referenced_column: Option<&Identifier>,
1958 on_delete: ForeignKeyAction,
1959 on_update: ForeignKeyAction,
1960 ) -> Result<ForeignKeyConstraint> {
1961 let schema_builder = SchemaBuilder::from_schema(schema);
1962 let transaction_parent = transaction.get_table(&referenced_table.value_lower).ok();
1963 validate_fk_reference(
1964 self.mutation_engine().as_ref(),
1965 transaction_parent.as_deref(),
1966 &schema_builder,
1967 &column_name.to_lowercase(),
1968 column_name,
1969 &referenced_table.value_lower,
1970 &referenced_table.value,
1971 &schema.table_name_lower,
1972 referenced_column.map(|column| column.value_lower.as_str()),
1973 on_delete,
1974 on_update,
1975 )
1976 }
1977
1978 fn validate_alter_schema_rows(
1979 &self,
1980 transaction: &dyn Transaction,
1981 table: &dyn Table,
1982 schema: &Schema,
1983 ) -> Result<()> {
1984 let compiled_checks = compile_table_check_constraints(schema)?;
1985 let mut parent_domains = Vec::with_capacity(schema.foreign_keys.len());
1986 for fk in &schema.foreign_keys {
1987 let parent = transaction.get_table(&fk.referenced_table)?;
1988 let parent_column = parent
1989 .schema()
1990 .get_column_index(&fk.referenced_column)
1991 .ok_or_else(|| {
1992 Error::internal(format!(
1993 "foreign key references missing column '{}.{}'",
1994 fk.referenced_table, fk.referenced_column
1995 ))
1996 })?;
1997 let mut scanner = parent.scan_exact_projection(&[parent_column], None)?;
1998 let mut values = ValueSet::default();
1999 while scanner.next() {
2000 let (_, row) = scanner.take_row_with_id()?;
2001 if let Some(value) = row.get(0).filter(|value| !value.is_null()) {
2002 values.insert(value.clone());
2003 }
2004 }
2005 if let Some(error) = scanner.err() {
2006 return Err(error.clone());
2007 }
2008 parent_domains.push(values);
2009 }
2010
2011 let active_projection: Vec<usize> = (0..table.schema().columns.len()).collect();
2012 let mut scanner = table.scan_exact_projection(&active_projection, None)?;
2013 let mut vm = crate::expression::ExprVM::new();
2014 while scanner.next() {
2015 let (_, mut row) = scanner.take_row_with_id()?;
2016 while row.len() < schema.columns.len() {
2017 let column = &schema.columns[row.len()];
2018 row.push(
2019 column
2020 .default_value
2021 .clone()
2022 .unwrap_or_else(|| Value::Null(column.data_type)),
2023 );
2024 }
2025 validate_resulting_row_constraints(schema, &compiled_checks, &row, &mut vm)?;
2026 for (fk, parent_values) in schema.foreign_keys.iter().zip(&parent_domains) {
2027 let Some(value) = row.get(fk.column_index).filter(|value| !value.is_null()) else {
2028 continue;
2029 };
2030 if !parent_values.contains(value) {
2031 return Err(Error::foreign_key_violation(
2032 &schema.table_name,
2033 &fk.column_name,
2034 &fk.referenced_table,
2035 &fk.referenced_column,
2036 format!(
2037 "referenced row with {} = {} does not exist",
2038 fk.referenced_column, value
2039 ),
2040 ));
2041 }
2042 }
2043 }
2044 if let Some(error) = scanner.err() {
2045 return Err(error.clone());
2046 }
2047 Ok(())
2048 }
2049
2050 fn stage_alter_schema_operation(
2051 &self,
2052 transaction: &mut dyn Transaction,
2053 catalog: &mut crate::catalog::DdlTransaction,
2054 stmt: &AlterTableStatement,
2055 ctx: &ExecutionContext,
2056 ) -> Result<()> {
2057 let table_name = stmt.table_name.value.as_str();
2058 if stmt.operation == AlterTableOperation::RenameTable {
2059 let new_name = stmt.new_table_name.as_ref().ok_or_else(|| {
2060 Error::InvalidArgument("RENAME TABLE requires new table name".to_string())
2061 })?;
2062 catalog.stage_statement_as(
2063 Statement::AlterTable(Box::new(stmt.clone())),
2064 ctx.effective_principal_id(),
2065 ctx.current_database(),
2066 )?;
2067 transaction.rename_table(table_name, new_name.value.as_str())?;
2068 return Ok(());
2069 }
2070 let table = transaction.get_table(table_name)?;
2071 let mut schema = table.schema().clone();
2072 let table_is_empty = table
2073 .collect_rows_with_limit_unordered(None, 1, 0)?
2074 .is_empty();
2075 let mut indexes = Vec::new();
2076 let mut requires_row_normalization = false;
2077
2078 match stmt.operation {
2079 AlterTableOperation::AddColumn | AlterTableOperation::ModifyColumn => {
2080 let col_def = stmt.column_def.as_ref().ok_or_else(|| {
2081 Error::InvalidArgument(format!(
2082 "{:?} requires column definition",
2083 stmt.operation
2084 ))
2085 })?;
2086 let (data_type, vector_dimensions, decimal_precision, decimal_scale, external_type) =
2087 super::type_binding::parse_schema_column_type(self, &col_def.data_type)?;
2088 let is_add = stmt.operation == AlterTableOperation::AddColumn;
2089 let existing_index = schema.get_column_index(&col_def.name.value);
2090 if is_add && existing_index.is_some() {
2091 return Err(Error::DuplicateColumn);
2092 }
2093 let target_index = if is_add {
2094 schema.columns.len()
2095 } else {
2096 existing_index
2097 .ok_or_else(|| Error::ColumnNotFound(col_def.name.value.to_string()))?
2098 };
2099 let old_column = (!is_add).then(|| schema.columns[target_index].clone());
2100 let previous_check_expr = old_column
2101 .as_ref()
2102 .and_then(|column| column.check_expr.clone());
2103
2104 let mut not_null = false;
2105 let mut primary_key = false;
2106 let mut unique = false;
2107 let mut auto_increment = false;
2108 let mut default_expr = None;
2109 let mut check_expr = None;
2110 let mut reference = None;
2111 for constraint in &col_def.constraints {
2112 match constraint {
2113 ColumnConstraint::NotNull => not_null = true,
2114 ColumnConstraint::PrimaryKey => primary_key = true,
2115 ColumnConstraint::Unique => unique = true,
2116 ColumnConstraint::AutoIncrement => auto_increment = true,
2117 ColumnConstraint::Default(expression) => {
2118 if default_expr
2119 .replace(bind_persistent_expression(expression, ctx)?)
2120 .is_some()
2121 {
2122 return Err(Error::InvalidArgument(format!(
2123 "column '{}' has more than one DEFAULT constraint",
2124 col_def.name.value
2125 )));
2126 }
2127 }
2128 ColumnConstraint::Check(expression) => {
2129 if check_expr
2130 .replace(bind_persistent_expression(expression, ctx)?)
2131 .is_some()
2132 {
2133 return Err(Error::InvalidArgument(format!(
2134 "column '{}' has more than one CHECK constraint",
2135 col_def.name.value
2136 )));
2137 }
2138 }
2139 ColumnConstraint::References {
2140 table,
2141 column,
2142 on_delete,
2143 on_update,
2144 } => {
2145 if reference
2146 .replace((table, column.as_ref(), *on_delete, *on_update))
2147 .is_some()
2148 {
2149 return Err(Error::InvalidArgument(format!(
2150 "column '{}' has more than one REFERENCES constraint",
2151 col_def.name.value
2152 )));
2153 }
2154 }
2155 }
2156 }
2157 let adds_primary_key =
2158 primary_key && old_column.as_ref().is_none_or(|column| !column.primary_key);
2159 let adds_auto_increment = auto_increment
2160 && old_column
2161 .as_ref()
2162 .is_none_or(|column| !column.auto_increment);
2163 if (adds_primary_key || adds_auto_increment) && !table_is_empty {
2164 return Err(Error::InvalidArgument(format!(
2165 "ALTER TABLE cannot add PRIMARY KEY or AUTO_INCREMENT to populated table '{}'; rebuild the table explicitly",
2166 table_name
2167 )));
2168 }
2169 if adds_primary_key
2170 && schema
2171 .primary_key_indices()
2172 .iter()
2173 .any(|&index| index != target_index)
2174 {
2175 return Err(Error::InvalidArgument(
2176 "table already has a PRIMARY KEY".to_string(),
2177 ));
2178 }
2179 if primary_key && !matches!(data_type, DataType::Integer | DataType::Uuid) {
2180 return Err(Error::InvalidArgument(format!(
2181 "PRIMARY KEY column '{}' must be INTEGER or UUID",
2182 col_def.name.value
2183 )));
2184 }
2185 if auto_increment && !matches!(data_type, DataType::Integer | DataType::Uuid) {
2186 return Err(Error::InvalidArgument(format!(
2187 "AUTO_INCREMENT column '{}' must be INTEGER or UUID",
2188 col_def.name.value
2189 )));
2190 }
2191 let changes_type_contract = !is_add
2192 && (schema.columns[target_index].data_type != data_type
2193 || schema.columns[target_index].external_type
2194 != external_type.as_ref().map(|(type_ref, _)| *type_ref)
2195 || schema.columns[target_index].vector_dimensions != vector_dimensions
2196 || schema.columns[target_index].decimal_precision != decimal_precision
2197 || schema.columns[target_index].decimal_scale != decimal_scale);
2198 if changes_type_contract && !table_is_empty {
2199 return Err(Error::InvalidArgument(format!(
2200 "ALTER TABLE cannot change the declared type of populated column '{}'; rebuild the table explicitly",
2201 col_def.name.value
2202 )));
2203 }
2204
2205 let default_value = if let Some(expression) = &default_expr {
2206 if external_type.is_some() {
2207 return Err(Error::NotSupported(format!(
2208 "DEFAULT for external column '{}' requires an explicit native/text constructor",
2209 col_def.name.value
2210 )));
2211 }
2212 let value = self.evaluate_default_expression(expression, data_type)?;
2213 (!value.is_null()).then_some(value)
2214 } else {
2215 None
2216 };
2217 if is_add {
2218 let mut column = SchemaColumn::with_default_value(
2219 target_index,
2220 col_def.name.value.clone(),
2221 data_type,
2222 !not_null && !primary_key,
2223 primary_key,
2224 auto_increment,
2225 default_expr.clone(),
2226 default_value,
2227 check_expr.clone(),
2228 );
2229 if data_type == DataType::Vector {
2230 column.vector_dimensions = vector_dimensions;
2231 }
2232 if data_type == DataType::Decimal {
2233 column.decimal_precision = decimal_precision;
2234 column.decimal_scale = decimal_scale;
2235 }
2236 if let Some((type_ref, sql_name)) = &external_type {
2237 column.external_type = Some(*type_ref);
2238 column.external_type_name = Some(sql_name.clone());
2239 }
2240 schema.add_column(column)?;
2241 requires_row_normalization = true;
2242 } else {
2243 let old_column = old_column.expect("MODIFY target was resolved");
2244 let effective_primary_key = old_column.primary_key || primary_key;
2245 let effective_auto_increment = old_column.auto_increment || auto_increment;
2246 let column = &mut schema.columns[target_index];
2247 column.data_type = data_type;
2248 column.external_type = external_type.as_ref().map(|(type_ref, _)| *type_ref);
2249 column.external_type_name =
2250 external_type.as_ref().map(|(_, sql_name)| sql_name.clone());
2251 column.nullable = !not_null && !effective_primary_key;
2252 column.primary_key = effective_primary_key;
2253 column.auto_increment = effective_auto_increment;
2254 column.default_expr = default_expr.clone();
2258 column.default_value = default_value;
2259 column.check_expr = check_expr.clone();
2260 if data_type == DataType::Vector {
2261 column.vector_dimensions = vector_dimensions;
2262 } else {
2263 column.vector_dimensions = 0;
2264 }
2265 if data_type == DataType::Decimal {
2266 column.decimal_precision = decimal_precision;
2267 column.decimal_scale = decimal_scale;
2268 } else {
2269 column.decimal_precision = 0;
2270 column.decimal_scale = 0;
2271 }
2272 }
2273
2274 if adds_primary_key {
2275 schema.register_primary_key_constraint(vec![col_def.name.value.to_string()])?;
2276 }
2277 if previous_check_expr != check_expr {
2278 let previous_name =
2279 schema
2280 .constraints()
2281 .iter()
2282 .find_map(|constraint| match &constraint.kind {
2283 SchemaConstraintKind::Check {
2284 column_name: Some(column_name),
2285 ..
2286 } if column_name.eq_ignore_ascii_case(&col_def.name.value) => {
2287 Some(constraint.name.clone())
2288 }
2289 _ => None,
2290 });
2291 if let Some(previous_name) = previous_name {
2292 schema.take_constraint(&previous_name);
2293 }
2294 if let Some(expression) = check_expr.clone() {
2295 schema.register_check_constraint(
2296 Some(col_def.name.value.to_string()),
2297 expression,
2298 )?;
2299 }
2300 }
2301
2302 if let Some((parent, parent_column, on_delete, on_update)) = reference {
2303 if schema
2304 .foreign_keys
2305 .iter()
2306 .any(|fk| fk.column_index == target_index)
2307 {
2308 return Err(Error::InvalidArgument(format!(
2309 "column '{}' already has a FOREIGN KEY",
2310 col_def.name.value
2311 )));
2312 }
2313 let fk = self.bind_alter_foreign_key(
2314 transaction,
2315 &schema,
2316 &col_def.name.value,
2317 parent,
2318 parent_column,
2319 on_delete,
2320 on_update,
2321 )?;
2322 schema.foreign_keys.push(fk.clone());
2323 schema.register_foreign_key_constraint(&fk)?;
2324 }
2329 if unique && !primary_key {
2330 let columns = vec![col_def.name.value.to_string()];
2331 let index_name = schema.register_unique_constraint(columns.clone())?;
2332 indexes.push(Self::alter_index_definition(
2333 table_name, index_name, columns, true,
2334 ));
2335 }
2336 schema.finish_catalog_mutation()?;
2337 }
2338 AlterTableOperation::AddConstraint => {
2339 let constraint = stmt.table_constraint.as_ref().ok_or_else(|| {
2340 Error::InvalidArgument("ADD CONSTRAINT requires a constraint".to_string())
2341 })?;
2342 match constraint {
2343 TableConstraint::Check(expression) => {
2344 let expression = bind_persistent_expression(expression, ctx)?;
2345 schema.table_checks.push(expression.clone());
2346 schema.register_check_constraint(None, expression)?;
2347 }
2348 TableConstraint::Unique(columns) => {
2349 let names: Vec<String> = columns
2350 .iter()
2351 .map(|column| {
2352 if schema.find_column(&column.value).is_none() {
2353 Err(Error::ColumnNotFound(column.value.to_string()))
2354 } else {
2355 Ok(column.value.to_string())
2356 }
2357 })
2358 .collect::<Result<_>>()?;
2359 let index_name = schema.register_unique_constraint(names.clone())?;
2360 indexes.push(Self::alter_index_definition(
2361 table_name, index_name, names, true,
2362 ));
2363 }
2364 TableConstraint::PrimaryKey(columns) => {
2365 if columns.len() != 1 {
2366 return Err(Error::NotSupported(
2367 "composite PRIMARY KEY is not supported; use UNIQUE instead"
2368 .to_string(),
2369 ));
2370 }
2371 if schema.has_primary_key() {
2372 return Err(Error::InvalidArgument(
2373 "table already has a PRIMARY KEY".to_string(),
2374 ));
2375 }
2376 if !table_is_empty {
2377 return Err(Error::InvalidArgument(
2378 "ALTER TABLE cannot add PRIMARY KEY to a populated table; rebuild it explicitly"
2379 .to_string(),
2380 ));
2381 }
2382 let index = schema
2383 .get_column_index(&columns[0].value)
2384 .ok_or_else(|| Error::ColumnNotFound(columns[0].value.to_string()))?;
2385 if !matches!(
2386 schema.columns[index].data_type,
2387 DataType::Integer | DataType::Uuid
2388 ) {
2389 return Err(Error::InvalidArgument(
2390 "PRIMARY KEY must be INTEGER or UUID".to_string(),
2391 ));
2392 }
2393 schema.columns[index].primary_key = true;
2394 schema.columns[index].nullable = false;
2395 schema
2396 .register_primary_key_constraint(vec![columns[0].value.to_string()])?;
2397 }
2398 TableConstraint::ForeignKey(foreign_key) => {
2399 let local_index = schema
2400 .get_column_index(&foreign_key.column.value)
2401 .ok_or_else(|| {
2402 Error::ColumnNotFound(foreign_key.column.value.to_string())
2403 })?;
2404 if schema
2405 .foreign_keys
2406 .iter()
2407 .any(|fk| fk.column_index == local_index)
2408 {
2409 return Err(Error::InvalidArgument(format!(
2410 "column '{}' already has a FOREIGN KEY",
2411 foreign_key.column.value
2412 )));
2413 }
2414 let fk = self.bind_alter_foreign_key(
2415 transaction,
2416 &schema,
2417 &foreign_key.column.value,
2418 &foreign_key.ref_table,
2419 foreign_key.ref_column.as_ref(),
2420 foreign_key.on_delete,
2421 foreign_key.on_update,
2422 )?;
2423 schema.foreign_keys.push(fk.clone());
2424 schema.register_foreign_key_constraint(&fk)?;
2425 }
2429 }
2430 schema.finish_catalog_mutation()?;
2431 }
2432 AlterTableOperation::DropConstraint => {
2433 let constraint_name = stmt.constraint_name.as_ref().ok_or_else(|| {
2434 Error::InvalidArgument("DROP CONSTRAINT requires a constraint name".to_string())
2435 })?;
2436 let Some(constraint) = schema.find_constraint(&constraint_name.value).cloned()
2437 else {
2438 if stmt.if_exists {
2439 return Ok(());
2440 }
2441 return Err(Error::InvalidArgument(format!(
2442 "constraint '{}' does not exist on table '{}'",
2443 constraint_name.value, table_name
2444 )));
2445 };
2446
2447 let mut owned_index: Option<(String, bool)> = None;
2448 match &constraint.kind {
2449 SchemaConstraintKind::PrimaryKey { columns }
2450 | SchemaConstraintKind::Unique { columns, .. } => {
2451 let referenced_column = columns
2452 .first()
2453 .ok_or_else(|| Error::internal("key constraint has no column"))?;
2454 let catalog_alternative = schema.constraints().iter().any(|candidate| {
2455 candidate.id != constraint.id
2456 && match &candidate.kind {
2457 SchemaConstraintKind::PrimaryKey { columns } => {
2458 columns.as_slice().first().is_some_and(|column| {
2459 columns.len() == 1
2460 && column.eq_ignore_ascii_case(referenced_column)
2461 })
2462 }
2463 SchemaConstraintKind::Unique { columns, .. } => {
2464 columns.len() == 1
2465 && columns[0].eq_ignore_ascii_case(referenced_column)
2466 && schema
2467 .get_column_by_name(referenced_column)
2468 .is_some_and(|column| !column.nullable)
2469 }
2470 _ => false,
2471 }
2472 });
2473 let target_is_not_null = schema
2474 .get_column_by_name(referenced_column)
2475 .is_some_and(|column| !column.nullable);
2476 let is_current_physical_owner =
2477 |name: &str, index_type: radixdb_core::IndexType| match &constraint.kind
2478 {
2479 SchemaConstraintKind::PrimaryKey { .. } => {
2480 index_type == radixdb_core::IndexType::PrimaryKey
2481 || name.starts_with("__pk_")
2482 }
2483 SchemaConstraintKind::Unique { index_name, .. } => {
2484 name.eq_ignore_ascii_case(index_name)
2485 }
2486 _ => false,
2487 };
2488 let physical_alternative = target_is_not_null
2489 && table.get_indexes().into_iter().any(|index| {
2490 index.is_unique()
2491 && index.partial_predicate().is_none()
2492 && index.column_names().len() == 1
2493 && index.column_names()[0]
2494 .eq_ignore_ascii_case(referenced_column)
2495 && !is_current_physical_owner(index.name(), index.index_type())
2496 });
2497 let staged_alternative = target_is_not_null
2498 && transaction.staged_index_definitions(table_name).iter().any(
2499 |pending| {
2500 pending.is_unique
2501 && pending.partial_predicate.is_none()
2502 && pending.columns.len() == 1
2503 && pending.columns[0]
2504 .eq_ignore_ascii_case(referenced_column)
2505 && !match &constraint.kind {
2506 SchemaConstraintKind::Unique { index_name, .. } => {
2507 pending.index_name.eq_ignore_ascii_case(index_name)
2508 }
2509 SchemaConstraintKind::PrimaryKey { .. } => false,
2510 _ => false,
2511 }
2512 },
2513 );
2514 let has_alternative_target =
2515 catalog_alternative || physical_alternative || staged_alternative;
2516 let referencing = self
2517 .mutation_engine()
2518 .find_referencing_fks_for_txn(transaction.id(), table_name);
2519 if !has_alternative_target {
2520 if let Some((child_table, _)) = referencing.iter().find(|(_, fk)| {
2521 fk.referenced_column.eq_ignore_ascii_case(referenced_column)
2522 }) {
2523 return Err(Error::InvalidArgument(format!(
2524 "cannot drop constraint '{}' because table '{}' has a dependent foreign key",
2525 constraint.name, child_table
2526 )));
2527 }
2528 }
2529 match &constraint.kind {
2530 SchemaConstraintKind::PrimaryKey { columns } => {
2531 let column = schema
2532 .get_column_index(&columns[0])
2533 .ok_or_else(|| Error::ColumnNotFound(columns[0].clone()))?;
2534 let derived_index = table.get_indexes().into_iter().find(|index| {
2535 (index.index_type() == radixdb_core::IndexType::PrimaryKey
2536 || index.name().starts_with("__pk_"))
2537 && index.column_ids() == [column as i32]
2538 });
2539 if let Some(derived_index) = derived_index {
2540 owned_index = Some((derived_index.name().to_string(), true));
2541 } else {
2542 let existed_before_transaction = self
2543 .mutation_engine()
2544 .get_table_schema(table_name)
2545 .ok()
2546 .is_some_and(|catalog_schema| {
2547 catalog_schema
2548 .constraints()
2549 .iter()
2550 .any(|candidate| candidate.id == constraint.id)
2551 });
2552 if existed_before_transaction {
2553 return Err(Error::internal(format!(
2554 "constraint '{}' lost its derived PRIMARY KEY index",
2555 constraint.name
2556 )));
2557 }
2558 }
2563 schema.columns[column].primary_key = false;
2564 }
2565 SchemaConstraintKind::Unique { index_name, .. } => {
2566 if table.get_index(index_name).is_none()
2567 && !transaction.staged_index_definitions(table_name).iter().any(
2568 |pending| {
2569 pending.index_name.eq_ignore_ascii_case(index_name)
2570 },
2571 )
2572 {
2573 return Err(Error::IndexNotFound(index_name.clone()));
2574 }
2575 owned_index = Some((index_name.clone(), false));
2576 }
2577 _ => unreachable!(),
2578 }
2579 }
2580 SchemaConstraintKind::ForeignKey {
2581 columns,
2582 referenced_table,
2583 referenced_columns,
2584 on_delete,
2585 on_update,
2586 } => {
2587 let index = schema
2588 .foreign_keys
2589 .iter()
2590 .position(|foreign_key| {
2591 columns.len() == 1
2592 && referenced_columns.len() == 1
2593 && foreign_key.column_name.eq_ignore_ascii_case(&columns[0])
2594 && foreign_key
2595 .referenced_table
2596 .eq_ignore_ascii_case(referenced_table)
2597 && foreign_key
2598 .referenced_column
2599 .eq_ignore_ascii_case(&referenced_columns[0])
2600 && foreign_key.on_delete == *on_delete
2601 && foreign_key.on_update == *on_update
2602 })
2603 .ok_or_else(|| {
2604 Error::internal(format!(
2605 "constraint '{}' lost its FOREIGN KEY owner",
2606 constraint.name
2607 ))
2608 })?;
2609 schema.foreign_keys.remove(index);
2610 }
2611 SchemaConstraintKind::Check {
2612 column_name,
2613 expression,
2614 ..
2615 } => {
2616 if let Some(column_name) = column_name {
2617 let column = schema
2618 .get_column_index(column_name)
2619 .ok_or_else(|| Error::ColumnNotFound(column_name.clone()))?;
2620 if schema.columns[column].check_expr.as_ref() != Some(expression) {
2621 return Err(Error::internal(format!(
2622 "constraint '{}' lost its column CHECK owner",
2623 constraint.name
2624 )));
2625 }
2626 schema.columns[column].check_expr = None;
2627 } else {
2628 let occurrence = schema
2629 .constraints()
2630 .iter()
2631 .take_while(|candidate| candidate.id != constraint.id)
2632 .filter(|candidate| {
2633 matches!(
2634 &candidate.kind,
2635 SchemaConstraintKind::Check {
2636 column_name: None,
2637 expression: candidate_expression,
2638 ..
2639 } if candidate_expression == expression
2640 )
2641 })
2642 .count();
2643 let index = schema
2644 .table_checks
2645 .iter()
2646 .enumerate()
2647 .filter(|(_, candidate)| *candidate == expression)
2648 .nth(occurrence)
2649 .map(|(index, _)| index)
2650 .ok_or_else(|| {
2651 Error::internal(format!(
2652 "constraint '{}' lost its table CHECK owner",
2653 constraint.name
2654 ))
2655 })?;
2656 schema.table_checks.remove(index);
2657 }
2658 }
2659 }
2660 schema.take_constraint(&constraint.name);
2661 schema.finish_catalog_mutation()?;
2662 if let Some((index_name, schema_owned)) = owned_index {
2663 transaction.stage_drop_index(PendingIndexDrop {
2664 table_name: table_name.to_string(),
2665 index_name,
2666 schema_owned,
2667 })?;
2668 }
2669 }
2670 AlterTableOperation::DropColumn => {
2671 let column_name = stmt.column_name.as_ref().ok_or_else(|| {
2672 Error::InvalidArgument("DROP COLUMN requires column name".to_string())
2673 })?;
2674 if let Some((child_table, _)) = self
2675 .mutation_engine()
2676 .find_referencing_fks_for_txn(transaction.id(), table_name)
2677 .iter()
2678 .find(|(_, fk)| {
2679 fk.referenced_column
2680 .eq_ignore_ascii_case(column_name.value.as_str())
2681 })
2682 {
2683 return Err(Error::InvalidArgument(format!(
2684 "cannot drop referenced column '{}.{}'; foreign key exists in table '{}'",
2685 table_name, column_name.value, child_table
2686 )));
2687 }
2688 let (column_index, column) = schema
2689 .find_column(&column_name.value)
2690 .ok_or_else(|| Error::ColumnNotFound(column_name.value.to_string()))?;
2691 if column.primary_key {
2692 return Err(Error::CannotDropPrimaryKey);
2693 }
2694 schema.remove_column(&column_name.value)?;
2695 compile_table_check_constraints(&schema).map_err(|error| {
2696 Error::InvalidArgument(format!(
2697 "cannot drop column '{}' because a table CHECK would become invalid: {}",
2698 column_name.value, error
2699 ))
2700 })?;
2701 catalog.stage_table_schema_as(&schema, None, ctx.effective_principal_id())?;
2702 transaction.stage_table_schema_transition(
2703 table_name,
2704 schema,
2705 true,
2706 SchemaPhysicalTransition::DropColumn {
2707 column_name: column_name.value.to_string(),
2708 column_index,
2709 },
2710 )?;
2711 return Ok(());
2712 }
2713 AlterTableOperation::RenameColumn => {
2714 let (old_name, new_name) = stmt
2715 .column_name
2716 .as_ref()
2717 .zip(stmt.new_column_name.as_ref())
2718 .ok_or_else(|| {
2719 Error::InvalidArgument(
2720 "RENAME COLUMN requires old and new column names".to_string(),
2721 )
2722 })?;
2723 if let Some((child_table, _)) = self
2724 .mutation_engine()
2725 .find_referencing_fks_for_txn(transaction.id(), table_name)
2726 .iter()
2727 .find(|(_, fk)| {
2728 fk.referenced_column
2729 .eq_ignore_ascii_case(old_name.value.as_str())
2730 })
2731 {
2732 return Err(Error::InvalidArgument(format!(
2733 "cannot rename referenced column '{}.{}'; foreign key exists in table '{}'",
2734 table_name, old_name.value, child_table
2735 )));
2736 }
2737 schema.rename_column(&old_name.value, new_name.value.as_str())?;
2738 compile_table_check_constraints(&schema).map_err(|error| {
2739 Error::InvalidArgument(format!(
2740 "cannot rename column '{}' because a table CHECK would become invalid: {}",
2741 old_name.value, error
2742 ))
2743 })?;
2744 catalog.stage_table_schema_as(
2745 &schema,
2746 Some((old_name.value.as_str(), new_name.value.as_str())),
2747 ctx.effective_principal_id(),
2748 )?;
2749 transaction.stage_table_schema_transition(
2750 table_name,
2751 schema,
2752 false,
2753 SchemaPhysicalTransition::RenameColumn {
2754 old_name: old_name.value.to_string(),
2755 new_name: new_name.value.to_string(),
2756 },
2757 )?;
2758 return Ok(());
2759 }
2760 _ => {
2761 return Err(Error::NotSupported(
2762 "this ALTER TABLE operation cannot use the transactional schema path"
2763 .to_string(),
2764 ));
2765 }
2766 }
2767
2768 compile_table_check_constraints(&schema)?;
2769 self.validate_alter_schema_rows(transaction, table.as_ref(), &schema)?;
2770 if !catalog.stage_table_schema_as(&schema, None, ctx.effective_principal_id())? {
2776 return Ok(());
2777 }
2778 transaction.stage_table_schema_change(table_name, schema, requires_row_normalization)?;
2779 for index in indexes {
2780 transaction.stage_create_index(index)?;
2781 }
2782 Ok(())
2783 }
2784
2785 fn execute_alter_table(
2787 &self,
2788 stmt: &AlterTableStatement,
2789 ctx: &ExecutionContext,
2790 ) -> Result<Box<dyn QueryResult>> {
2791 let table_name = &stmt.table_name.value;
2792 let transactional_schema_operation = matches!(
2793 stmt.operation,
2794 AlterTableOperation::AddColumn
2795 | AlterTableOperation::ModifyColumn
2796 | AlterTableOperation::AddConstraint
2797 | AlterTableOperation::DropConstraint
2798 | AlterTableOperation::DropColumn
2799 | AlterTableOperation::RenameColumn
2800 | AlterTableOperation::RenameTable
2801 );
2802
2803 if transactional_schema_operation {
2808 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
2809 if let Some(tx_state) = active_tx.as_mut() {
2810 let mut catalog = tx_state.catalog.clone();
2811 self.stage_alter_schema_operation(
2812 tx_state.transaction.as_mut(),
2813 &mut catalog,
2814 stmt,
2815 ctx,
2816 )?;
2817 tx_state.catalog = catalog;
2818 self.mutation_invalidate_query_cache(table_name);
2819 self.mutation_invalidate_semantic_cache(table_name);
2820 invalidate_semi_join_cache_for_table(table_name);
2821 invalidate_scalar_subquery_cache_for_table(table_name);
2822 invalidate_in_subquery_cache_for_table(table_name);
2823 if let Some(new_name) = &stmt.new_table_name {
2824 self.mutation_invalidate_query_cache(&new_name.value);
2825 self.mutation_invalidate_semantic_cache(&new_name.value);
2826 invalidate_semi_join_cache_for_table(&new_name.value);
2827 invalidate_scalar_subquery_cache_for_table(&new_name.value);
2828 invalidate_in_subquery_cache_for_table(&new_name.value);
2829 }
2830 return Ok(Box::new(ExecResult::empty()));
2831 }
2832 drop(active_tx);
2833
2834 if !self.mutation_engine().table_exists(table_name)? {
2835 return Err(Error::TableNotFound(table_name.to_string()));
2836 }
2837 let mut transaction = self.mutation_engine().begin_transaction()?;
2838 let generation = self.mutation_engine().pin_catalog()?;
2839 let mut catalog = crate::catalog::DdlTransaction::begin_shared(generation);
2840 self.stage_alter_schema_operation(transaction.as_mut(), &mut catalog, stmt, ctx)?;
2841 if let Some(mutation) = catalog.pending_mutation()? {
2842 transaction.stage_catalog_mutation(mutation)?;
2843 }
2844 transaction.commit()?;
2845 self.mutation_invalidate_query_cache(table_name);
2846 self.mutation_invalidate_semantic_cache(table_name);
2847 invalidate_semi_join_cache_for_table(table_name);
2848 invalidate_scalar_subquery_cache_for_table(table_name);
2849 invalidate_in_subquery_cache_for_table(table_name);
2850 if let Some(new_name) = &stmt.new_table_name {
2851 self.mutation_invalidate_query_cache(&new_name.value);
2852 self.mutation_invalidate_semantic_cache(&new_name.value);
2853 invalidate_semi_join_cache_for_table(&new_name.value);
2854 invalidate_scalar_subquery_cache_for_table(&new_name.value);
2855 invalidate_in_subquery_cache_for_table(&new_name.value);
2856 }
2857 return Ok(Box::new(ExecResult::empty()));
2858 }
2859
2860 unreachable!("every ALTER TABLE operation uses the transactional catalog path")
2861 }
2862
2863 fn execute_create_view(
2865 &self,
2866 stmt: &CreateViewStatement,
2867 ctx: &ExecutionContext,
2868 ) -> Result<Box<dyn QueryResult>> {
2869 let view_name = &stmt.view_name.value;
2870 if let Some(mutation) = self.mutation_stage_catalog_statement_as(
2871 Statement::CreateView(stmt.clone()),
2872 ctx.effective_principal_id(),
2873 ctx.current_database(),
2874 )? {
2875 let mut transaction = self.mutation_engine().begin_transaction()?;
2876 transaction.stage_catalog_mutation(mutation)?;
2877 transaction.commit()?;
2878 }
2879
2880 self.mutation_invalidate_query_cache(view_name);
2881 self.mutation_invalidate_semantic_cache(view_name);
2882 invalidate_semi_join_cache_for_table(view_name);
2883 invalidate_scalar_subquery_cache_for_table(view_name);
2884 invalidate_in_subquery_cache_for_table(view_name);
2885
2886 Ok(Box::new(ExecResult::empty()))
2887 }
2888
2889 fn execute_drop_view(
2891 &self,
2892 stmt: &DropViewStatement,
2893 ctx: &ExecutionContext,
2894 ) -> Result<Box<dyn QueryResult>> {
2895 let view_name = &stmt.view_name.value;
2896 if let Some(mutation) = self.mutation_stage_catalog_statement_as(
2897 Statement::DropView(stmt.clone()),
2898 ctx.effective_principal_id(),
2899 ctx.current_database(),
2900 )? {
2901 let mut transaction = self.mutation_engine().begin_transaction()?;
2902 transaction.stage_catalog_mutation(mutation)?;
2903 transaction.commit()?;
2904 }
2905
2906 self.mutation_invalidate_query_cache(view_name);
2908 self.mutation_invalidate_semantic_cache(view_name);
2909 invalidate_semi_join_cache_for_table(view_name);
2910 invalidate_scalar_subquery_cache_for_table(view_name);
2911 invalidate_in_subquery_cache_for_table(view_name);
2912
2913 Ok(Box::new(ExecResult::empty()))
2914 }
2915
2916 fn evaluate_default_expression(
2918 &self,
2919 default_expr: &str,
2920 target_type: DataType,
2921 ) -> Result<Value> {
2922 use radixdb_sql::parse_sql;
2923
2924 let sql = format!("SELECT {}", default_expr);
2926 let stmts = parse_sql(&sql)
2927 .map_err(|error| Error::Parse(format!("invalid default expression: {}", error)))?;
2928 if stmts.is_empty() {
2929 return Err(Error::InvalidArgument(format!(
2930 "default expression '{}' produced no statement",
2931 default_expr
2932 )));
2933 }
2934
2935 if let Statement::Select(select) = &stmts[0] {
2937 if let Some(expr) = select.columns.first() {
2938 let mut eval = ExpressionEval::compile(expr, &[])?;
2939 let value = eval.eval_slice(&Row::new())?;
2940 return value.try_coerce_to_type(target_type);
2941 }
2942 }
2943
2944 Err(Error::InvalidArgument(format!(
2945 "default expression '{}' is not a SELECT expression",
2946 default_expr
2947 )))
2948 }
2949}
2950
2951impl<T: MutationHost + ?Sized> DdlExecutorExt for T {}