1use radixdb_catalog::TriggerTiming;
23use radixdb_core::CompactArc;
24use radixdb_core::I64Set;
25use radixdb_core::SmartString;
26use radixdb_core::{DataType, Error, Result, Row, Schema, Value};
27use radixdb_sql::ast::*;
28use radixdb_storage::expression::Expression as StorageExpr;
29use radixdb_storage::traits::{Engine, QueryResult, Table};
30use rustc_hash::FxHashMap;
31use std::sync::Arc;
32
33use super::dml_support::*;
34use super::returning::build_returning_result;
35use super::upsert::{apply_on_duplicate_update, compile_upsert};
36use crate::compiled_plan::{CompiledExecution, CompiledInsert};
37use crate::context::{
38 invalidate_in_subquery_cache_for_table, invalidate_scalar_subquery_cache_for_table,
39 invalidate_semi_join_cache_for_table, ExecutionContext,
40};
41use crate::expression::CompiledEvaluator;
42use crate::mutation::host::MutationHost;
43use crate::mutation::validation::{
44 compile_table_check_constraints, prepare_insert_row_constraints,
45 validate_resulting_row_constraints,
46};
47use crate::procedural::{DmlTriggerEvent, DmlTriggerPlan};
48use crate::pushdown;
49use crate::result::ExecResult;
50use crate::utils::dummy_token_clone;
51use std::sync::RwLock;
52
53#[doc(hidden)]
54pub trait DmlExecutorExt: MutationHost {
55 fn execute_with_trigger_statement<T>(
56 &self,
57 plan: &DmlTriggerPlan,
58 ctx: &ExecutionContext,
59 execute: impl FnOnce(&DmlTriggerPlan) -> Result<T>,
60 ) -> Result<T> {
61 let boundary = self.mutation_begin_trigger_boundary()?;
62 let outcome = (|| {
63 self.mutation_fire_statement_triggers(plan, TriggerTiming::Before, ctx)?;
64 let value = execute(plan)?;
65 self.mutation_fire_statement_triggers(plan, TriggerTiming::After, ctx)?;
66 Ok(value)
67 })();
68 match outcome {
69 Ok(value) => match self.mutation_complete_trigger_boundary(&boundary) {
70 Ok(()) => Ok(value),
71 Err(primary) => {
72 match self.mutation_abort_trigger_boundary(&boundary) {
73 Ok(()) => Err(primary),
74 Err(cleanup) => Err(Error::internal(format!(
75 "trigger statement commit failed: {primary}; rollback also failed: {cleanup}"
76 ))),
77 }
78 }
79 },
80 Err(primary) => match self.mutation_abort_trigger_boundary(&boundary) {
81 Ok(()) => Err(primary),
82 Err(cleanup) => Err(Error::internal(format!(
83 "trigger statement failed: {primary}; rollback also failed: {cleanup}"
84 ))),
85 },
86 }
87 }
88
89 fn accept_inserted_row(
90 &self,
91 triggers: Option<&DmlTriggerPlan>,
92 inserted: Option<Row>,
93 has_returning: bool,
94 returning_rows: &mut Vec<Row>,
95 ctx: &ExecutionContext,
96 ) -> Result<()> {
97 if let Some(inserted) = inserted {
98 if let Some(plan) = triggers {
99 self.mutation_fire_after_row_triggers(plan, None, Some(&inserted), None, ctx)?;
100 }
101 if has_returning {
102 returning_rows.push(inserted);
103 }
104 }
105 Ok(())
106 }
107
108 fn select_row_ids_for_dml(
118 &self,
119 table_name: &str,
120 where_clause: &Expression,
121 schema: &Schema,
122 table: &dyn Table,
123 ctx: &ExecutionContext,
124 ) -> Result<Option<Vec<i64>>> {
125 let pk_indices = schema.primary_key_indices();
127 if pk_indices.len() != 1 {
128 return Ok(None);
129 }
130
131 let pk_idx = pk_indices[0];
132 let pk_col = &schema.columns[pk_idx];
133
134 if pk_col.data_type != DataType::Integer {
136 return Ok(None);
137 }
138
139 let pk_column_name = &pk_col.name;
140 let pk_column_lower = pk_column_name.to_lowercase();
141
142 if let Expression::InHashSet(in_expr) = where_clause {
145 if let Expression::Identifier(id) = in_expr.column.as_ref() {
146 if id.value_lower == pk_column_lower {
147 if in_expr.not {
148 let excluded: I64Set = in_expr
150 .values
151 .iter()
152 .filter_map(|v| match v {
153 Value::Integer(i) => Some(*i),
154 _ => None,
155 })
156 .collect();
157
158 let mut row_ids: Vec<i64> = table
159 .get_active_row_ids()
160 .into_iter()
161 .filter(|id| !excluded.contains(*id))
162 .collect();
163 row_ids.sort_unstable();
164 return Ok(Some(row_ids));
165 } else {
166 let mut row_ids: Vec<i64> = in_expr
168 .values
169 .iter()
170 .filter_map(|v| match v {
171 Value::Integer(i) => Some(*i),
172 _ => None,
173 })
174 .collect();
175 row_ids.sort_unstable();
176 return Ok(Some(row_ids));
177 }
178 }
179 }
180 }
181
182 let select_stmt = SelectStatement {
184 token: dummy_token_clone(),
185 distinct: false,
186 distinct_on: vec![],
187 columns: vec![Expression::Identifier(Identifier::new(
188 dummy_token_clone(),
189 pk_column_name.clone(),
190 ))],
191 with: None,
192 table_expr: Some(Box::new(Expression::TableSource(Box::new(
193 SimpleTableSource {
194 token: dummy_token_clone(),
195 name: Identifier::new(dummy_token_clone(), table_name.to_string()),
196 alias: None,
197 as_of: None,
198 },
199 )))),
200 where_clause: Some(Box::new(where_clause.clone())),
201 group_by: GroupByClause::default(),
202 having: None,
203 window_defs: Vec::new(),
204 order_by: Vec::new(),
205 limit: None,
206 offset: None,
207 set_operations: Vec::new(),
208 };
209
210 let mut result = self.mutation_execute_select(&select_stmt, ctx)?;
212
213 let mut row_ids = Vec::new();
215 while result.next() {
216 let row = result.row();
217 if let Some(Value::Integer(id)) = row.get(0) {
218 row_ids.push(*id);
219 }
220 }
221 if let Some(err) = result.last_error() {
222 return Err(err);
223 }
224
225 row_ids.sort_unstable();
227
228 Ok(Some(row_ids))
229 }
230
231 fn execute_insert(
233 &self,
234 stmt: &InsertStatement,
235 ctx: &ExecutionContext,
236 ) -> Result<Box<dyn QueryResult>> {
237 let plan = self.mutation_prepare_dml_triggers(
238 stmt.table_name.value_lower.as_str(),
239 DmlTriggerEvent::Insert,
240 &[],
241 ctx,
242 )?;
243 if plan.is_empty() {
244 return self.execute_insert_body(stmt, ctx, None);
245 }
246 self.execute_with_trigger_statement(&plan, ctx, |plan| {
247 self.execute_insert_body(stmt, ctx, Some(plan))
248 })
249 }
250
251 fn execute_insert_body(
252 &self,
253 stmt: &InsertStatement,
254 ctx: &ExecutionContext,
255 triggers: Option<&DmlTriggerPlan>,
256 ) -> Result<Box<dyn QueryResult>> {
257 let table_name = &stmt.table_name.value_lower;
259
260 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
262
263 let (mut table, should_auto_commit, standalone_tx) =
264 if let Some(ref mut tx_state) = *active_tx {
265 let table = tx_state.transaction.get_table(table_name)?;
268
269 if !tx_state.tables.contains_key(table_name.as_str()) {
271 tx_state.tables.insert(
272 table_name.to_string(),
273 tx_state.transaction.get_table(table_name)?,
274 );
275 }
276
277 (table, false, None)
278 } else {
279 let tx = self.mutation_engine().begin_transaction()?;
281 let table = tx.get_table(table_name)?;
282 (table, true, Some(tx))
283 };
284
285 drop(active_tx);
287
288 let resulting_schema = table.schema().clone();
294
295 let schema_column_count: usize;
297 let column_indices: Vec<usize>;
298 let column_types: Vec<radixdb_core::DataType>;
300 let column_vector_dims: Vec<u16>;
302 let column_names: Vec<String>;
304 let all_column_types: Vec<radixdb_core::DataType>;
306 let default_exprs: Vec<Option<String>>;
308 let auto_increment_pk_idx: Option<usize>;
309 {
310 let schema = table.schema();
311 schema_column_count = schema.columns.len();
312 auto_increment_pk_idx = auto_increment_pk_index(schema);
313
314 default_exprs = schema
316 .columns
317 .iter()
318 .map(|c| c.default_expr.clone())
319 .collect();
320 all_column_types = schema.columns.iter().map(|c| c.data_type).collect();
321
322 if stmt.columns.is_empty() {
325 column_indices = (0..schema_column_count).collect();
326 column_types = all_column_types.clone();
327 column_vector_dims = schema.columns.iter().map(|c| c.vector_dimensions).collect();
328 column_names = schema.column_names_owned().to_vec();
330 } else {
331 let col_map = schema.column_index_map();
334 column_indices = stmt
335 .columns
336 .iter()
337 .map(|id| {
338 col_map
340 .get(id.value_lower.as_str())
341 .copied()
342 .ok_or_else(|| Error::ColumnNotFound(id.value.to_string()))
343 })
344 .collect::<Result<Vec<_>>>()?;
345 column_types = column_indices
347 .iter()
348 .map(|&idx| schema.columns[idx].data_type)
349 .collect();
350 column_vector_dims = column_indices
351 .iter()
352 .map(|&idx| schema.columns[idx].vector_dimensions)
353 .collect();
354 column_names = column_indices
356 .iter()
357 .map(|&idx| schema.columns[idx].name.clone())
358 .collect();
359 }
360 }
361
362 let mut seen_insert_targets = rustc_hash::FxHashSet::default();
363 for &column_index in &column_indices {
364 if !seen_insert_targets.insert(column_index) {
365 return Err(Error::InvalidArgument(
366 "INSERT target column is specified more than once".to_string(),
367 ));
368 }
369 }
370
371 let fk_schema = if !table.schema().foreign_keys.is_empty() {
373 Some(table.schema().clone())
374 } else {
375 None
376 };
377 let compiled_table_checks = compile_table_check_constraints(table.schema())?;
378 validate_conflict_target(&stmt.conflict_target, table.schema(), &*table)?;
379 use crate::expression::{compile_expression, ExecuteContext, ExprVM};
381 let mut vm = ExprVM::new();
382 let params = ctx.params();
383 let named_params = ctx.named_params();
384 let empty_row = Row::new();
385
386 let mut base_exec_ctx = ExecuteContext::new(&empty_row);
388 if !params.is_empty() {
389 base_exec_ctx = base_exec_ctx.with_params(params);
390 }
391 if !named_params.is_empty() {
392 base_exec_ctx = base_exec_ctx.with_named_params(named_params);
393 }
394 base_exec_ctx = base_exec_ctx
395 .with_transaction_id(ctx.transaction_id())
396 .with_stored_function_invoker(ctx.stored_function_invoker());
397
398 let mut rows_affected = 0i64;
399 let mut last_insert_id = 0i64;
400
401 let has_returning = !stmt.returning.is_empty();
403 let needs_inserted_row =
404 has_returning || triggers.is_some_and(DmlTriggerPlan::has_after_row_triggers);
405 let mut returning_rows: Vec<Row> = Vec::new();
406 let schema_column_names_arc = if has_returning {
408 Some(table.schema().column_names_arc())
409 } else {
410 None
411 };
412
413 if let Some(ref select_stmt) = stmt.select {
415 let select_schema = if stmt.on_duplicate || stmt.do_nothing {
417 Some(self.mutation_engine().get_table_schema(table_name)?)
418 } else {
419 None
420 };
421
422 let compiled_upsert = if stmt.on_duplicate {
425 if let Some(ref s) = select_schema {
426 Some(compile_upsert(self, s, stmt)?)
427 } else {
428 None
429 }
430 } else {
431 None
432 };
433
434 let mut select_result = self.mutation_execute_select(select_stmt, ctx)?;
439 if !should_auto_commit {
440 let columns = select_result.columns().to_vec();
444 let rows = <Self as MutationHost>::mutation_materialize_result(select_result)?;
445 select_result = Box::new(crate::result::ExecutorResult::new(columns, rows));
446 }
447
448 while select_result.next() {
450 let select_row = select_result.row();
451 if select_row.len() != column_indices.len() {
452 return Err(Error::InvalidArgument(format!(
453 "INSERT has {} columns but SELECT returns {} columns",
454 column_indices.len(),
455 select_row.len()
456 )));
457 }
458
459 let mut row_values = Vec::with_capacity(schema_column_count);
462 for i in 0..schema_column_count {
463 if let Some(ref default_expr) = default_exprs[i] {
464 let default_type = all_column_types[i];
465 row_values.push(evaluate_default_expr(default_expr, default_type)?);
466 } else {
467 row_values.push(Value::null_unknown());
468 }
469 }
470
471 for (i, value) in select_row.iter().enumerate() {
473 let coerced = value.coerce_to_type(column_types[i]);
475 validate_coercion(
477 value,
478 &coerced,
479 &column_names[i],
480 column_types[i],
481 column_vector_dims[i],
482 )?;
483 row_values[column_indices[i]] = coerced;
484 }
485
486 let mut row = Row::from_values(row_values);
487 prepare_insert_row_constraints(
488 &mut *table,
489 &resulting_schema,
490 &compiled_table_checks,
491 &mut row,
492 &mut vm,
493 )?;
494 let row = if let Some(plan) = triggers {
495 let Some(row) =
496 self.mutation_fire_before_row_triggers(plan, None, Some(row), None, ctx)?
497 else {
498 continue;
499 };
500 validate_resulting_row_constraints(
501 &resulting_schema,
502 &compiled_table_checks,
503 &row,
504 &mut vm,
505 )?;
506 row
507 } else {
508 row
509 };
510 let saved_row_values = stmt.on_duplicate.then(|| row.as_slice().to_vec());
513
514 if let Some(ref fks) = fk_schema {
516 crate::mutation::foreign_key::check_parent_exists(
517 self.mutation_engine(),
518 table.txn_id(),
519 fks,
520 &row,
521 )?;
522 }
523
524 if stmt.do_nothing {
525 let schema_ref = select_schema.as_ref().unwrap();
526 let insert_result = insert_row_for_command_result(
528 &mut *table,
529 row,
530 needs_inserted_row,
531 auto_increment_pk_idx,
532 &mut last_insert_id,
533 );
534 match insert_result {
535 Ok(opt_row) => {
536 self.accept_inserted_row(
537 triggers,
538 opt_row,
539 has_returning,
540 &mut returning_rows,
541 ctx,
542 )?;
543 rows_affected += 1;
544 }
545 Err(ref e @ Error::PrimaryKeyConstraint { .. })
546 | Err(ref e @ Error::UniqueConstraint { .. }) => {
547 if !conflict_matches_target(&stmt.conflict_target, schema_ref, e) {
548 return Err(e.clone());
549 }
550 }
552 Err(e) => return Err(e),
553 }
554 } else if stmt.on_duplicate {
555 let row_values = saved_row_values.as_ref().unwrap();
556 let schema_ref = select_schema.as_ref().unwrap();
557 let compiled_upsert = compiled_upsert.as_ref().ok_or_else(|| {
558 Error::internal("missing precompiled ON CONFLICT update plan")
559 })?;
560 let insert_result = insert_row_for_command_result(
562 &mut *table,
563 row,
564 needs_inserted_row,
565 auto_increment_pk_idx,
566 &mut last_insert_id,
567 );
568 match insert_result {
569 Ok(opt_row) => {
570 self.accept_inserted_row(
571 triggers,
572 opt_row,
573 has_returning,
574 &mut returning_rows,
575 ctx,
576 )?;
577 rows_affected += 1;
578 }
579 Err(ref e @ Error::PrimaryKeyConstraint { row_id }) => {
580 if !conflict_matches_target(&stmt.conflict_target, schema_ref, e) {
581 return Err(Error::PrimaryKeyConstraint { row_id });
582 }
583 match apply_on_duplicate_update(
584 self,
585 &mut table,
586 schema_ref,
587 row_id,
588 None,
589 row_values,
590 compiled_upsert,
591 ctx,
592 has_returning,
593 ) {
594 Ok(Some(updated_row)) => {
595 returning_rows.push(updated_row);
596 rows_affected += 1;
597 }
598 Ok(None) => {
599 rows_affected += 1;
600 }
601 Err(e) => return Err(e),
602 }
603 }
604 Err(
605 ref e @ Error::UniqueConstraint {
606 ref index,
607 ref column,
608 ref value,
609 row_id: conflict_rid,
610 },
611 ) => {
612 if !conflict_matches_target(&stmt.conflict_target, schema_ref, e) {
613 return Err(Error::UniqueConstraint {
614 index: index.clone(),
615 column: column.clone(),
616 value: value.clone(),
617 row_id: conflict_rid,
618 });
619 }
620 let found_row_id = if conflict_rid >= 0 {
624 Ok(Some(conflict_rid))
625 } else {
626 find_row_by_unique_index(
627 &*table, schema_ref, index, column, row_values,
628 )
629 };
630 match found_row_id {
631 Ok(Some(row_id)) => {
632 match apply_on_duplicate_update(
633 self,
634 &mut table,
635 schema_ref,
636 row_id,
637 Some(column),
638 row_values,
639 compiled_upsert,
640 ctx,
641 has_returning,
642 ) {
643 Ok(Some(updated_row)) => {
644 returning_rows.push(updated_row);
645 rows_affected += 1;
646 }
647 Ok(None) => {
648 rows_affected += 1;
649 }
650 Err(e) => return Err(e),
651 }
652 }
653 Ok(None) => {
654 return Err(Error::UniqueConstraint {
655 index: index.clone(),
656 column: column.clone(),
657 value: value.clone(),
658 row_id: -1,
659 });
660 }
661 Err(e) => return Err(e),
662 }
663 }
664 Err(e) => return Err(e),
665 }
666 } else {
667 let inserted = insert_row_for_command_result(
668 &mut *table,
669 row,
670 needs_inserted_row,
671 auto_increment_pk_idx,
672 &mut last_insert_id,
673 )?;
674 self.accept_inserted_row(
675 triggers,
676 inserted,
677 has_returning,
678 &mut returning_rows,
679 ctx,
680 )?;
681 rows_affected += 1;
682 }
683 }
684 if let Some(err) = select_result.last_error() {
686 return Err(err);
687 }
688
689 if rows_affected > 0 {
693 self.mutation_invalidate_semantic_cache(table_name);
694 invalidate_semi_join_cache_for_table(table_name);
695 invalidate_scalar_subquery_cache_for_table(table_name);
696 invalidate_in_subquery_cache_for_table(table_name);
697 }
698
699 let mut returning_result = if has_returning {
700 Some(build_returning_result(
701 &stmt.returning,
702 std::mem::take(&mut returning_rows),
703 schema_column_names_arc.as_ref().unwrap(),
704 ctx,
705 )?)
706 } else {
707 None
708 };
709
710 if should_auto_commit {
712 if let Some(mut tx) = standalone_tx {
713 match tx.commit() {
714 Ok(()) => {}
715 Err(e)
716 if (stmt.on_duplicate || stmt.do_nothing)
717 && e.is_pk_or_unique_violation() =>
718 {
719 if stmt.on_duplicate && ctx.query_depth() == 0 {
720 let retry_ctx = ctx.with_incremented_query_depth();
723 return self.execute_insert(stmt, &retry_ctx);
724 }
725 if stmt.do_nothing {
726 rows_affected = 0;
728 last_insert_id = 0;
729 returning_result = Some(build_returning_result(
730 &stmt.returning,
731 Vec::new(),
732 schema_column_names_arc.as_ref().unwrap(),
733 ctx,
734 )?);
735 } else {
736 return Err(e);
737 }
738 }
739 Err(e) => return Err(e),
740 }
741 }
742 }
743
744 if let Some(result) = returning_result {
746 return Ok(result);
747 }
748
749 return Ok(Box::new(ExecResult::with_last_insert_id(
750 rows_affected,
751 last_insert_id,
752 )));
753 }
754
755 if stmt.do_nothing || stmt.on_duplicate {
757 let schema = self.mutation_engine().get_table_schema(table_name)?;
759
760 let compiled_upsert = if stmt.on_duplicate {
763 Some(compile_upsert(self, &schema, stmt)?)
764 } else {
765 None
766 };
767
768 for value_row in &stmt.values {
769 if value_row.len() != column_indices.len() {
770 return Err(Error::InvalidArgument(format!(
771 "INSERT has {} columns but {} values",
772 column_indices.len(),
773 value_row.len()
774 )));
775 }
776
777 let mut row_values = Vec::with_capacity(schema_column_count);
779 for i in 0..schema_column_count {
780 if let Some(ref default_expr) = default_exprs[i] {
781 let default_type = all_column_types[i];
782 row_values.push(evaluate_default_expr(default_expr, default_type)?);
783 } else {
784 row_values.push(Value::null_unknown());
785 }
786 }
787 for (i, expr) in value_row.iter().enumerate() {
789 if matches!(expr, Expression::Default(_)) {
791 continue;
792 }
793 let value = if let Some(lit_value) = try_extract_literal(expr) {
796 lit_value
797 } else {
798 let program = compile_expression(expr, &[])?;
800 vm.execute_cow(&program, &base_exec_ctx)?
801 };
802 let coerced = value.coerce_to_type(column_types[i]);
804 validate_coercion(
806 &value,
807 &coerced,
808 &column_names[i],
809 column_types[i],
810 column_vector_dims[i],
811 )?;
812 row_values[column_indices[i]] = coerced;
813 }
814
815 let mut row = Row::from_values(row_values.clone());
817 prepare_insert_row_constraints(
818 &mut *table,
819 &resulting_schema,
820 &compiled_table_checks,
821 &mut row,
822 &mut vm,
823 )?;
824 let row = if let Some(plan) = triggers {
825 let Some(row) =
826 self.mutation_fire_before_row_triggers(plan, None, Some(row), None, ctx)?
827 else {
828 continue;
829 };
830 validate_resulting_row_constraints(
831 &resulting_schema,
832 &compiled_table_checks,
833 &row,
834 &mut vm,
835 )?;
836 row
837 } else {
838 row
839 };
840 row_values = row.as_slice().to_vec();
841
842 if let Some(ref fks) = fk_schema {
844 crate::mutation::foreign_key::check_parent_exists(
845 self.mutation_engine(),
846 table.txn_id(),
847 fks,
848 &row,
849 )?;
850 }
851
852 if stmt.do_nothing {
853 let insert_result = insert_row_for_command_result(
855 &mut *table,
856 row,
857 needs_inserted_row,
858 auto_increment_pk_idx,
859 &mut last_insert_id,
860 );
861 match insert_result {
862 Ok(opt_row) => {
863 self.accept_inserted_row(
864 triggers,
865 opt_row,
866 has_returning,
867 &mut returning_rows,
868 ctx,
869 )?;
870 rows_affected += 1;
871 }
872 Err(ref e @ Error::PrimaryKeyConstraint { .. })
873 | Err(ref e @ Error::UniqueConstraint { .. }) => {
874 if !conflict_matches_target(&stmt.conflict_target, &schema, e) {
875 return Err(e.clone());
876 }
877 }
879 Err(e) => return Err(e),
880 }
881 } else {
882 let compiled_upsert = compiled_upsert.as_ref().ok_or_else(|| {
883 Error::internal("missing precompiled ON CONFLICT update plan")
884 })?;
885 let insert_result = insert_row_for_command_result(
887 &mut *table,
888 row,
889 needs_inserted_row,
890 auto_increment_pk_idx,
891 &mut last_insert_id,
892 );
893 match insert_result {
894 Ok(opt_row) => {
895 self.accept_inserted_row(
896 triggers,
897 opt_row,
898 has_returning,
899 &mut returning_rows,
900 ctx,
901 )?;
902 rows_affected += 1;
903 }
904 Err(ref e @ Error::PrimaryKeyConstraint { row_id }) => {
905 if !conflict_matches_target(&stmt.conflict_target, &schema, e) {
906 return Err(Error::PrimaryKeyConstraint { row_id });
907 }
908 match apply_on_duplicate_update(
909 self,
910 &mut table,
911 &schema,
912 row_id,
913 None,
914 &row_values,
915 compiled_upsert,
916 ctx,
917 has_returning,
918 ) {
919 Ok(Some(updated_row)) => {
920 returning_rows.push(updated_row);
921 rows_affected += 1;
922 }
923 Ok(None) => {
924 rows_affected += 1;
925 }
926 Err(e) => return Err(e),
927 }
928 }
929 Err(
930 ref e @ Error::UniqueConstraint {
931 ref index,
932 ref column,
933 ref value,
934 row_id: conflict_rid,
935 },
936 ) => {
937 if !conflict_matches_target(&stmt.conflict_target, &schema, e) {
938 return Err(Error::UniqueConstraint {
939 index: index.clone(),
940 column: column.clone(),
941 value: value.clone(),
942 row_id: conflict_rid,
943 });
944 }
945 let found_row_id = if conflict_rid >= 0 {
946 Ok(Some(conflict_rid))
947 } else {
948 find_row_by_unique_index(
949 &*table,
950 &schema,
951 index,
952 column,
953 &row_values,
954 )
955 };
956 match found_row_id {
957 Ok(Some(row_id)) => {
958 match apply_on_duplicate_update(
959 self,
960 &mut table,
961 &schema,
962 row_id,
963 Some(column),
964 &row_values,
965 compiled_upsert,
966 ctx,
967 has_returning,
968 ) {
969 Ok(Some(updated_row)) => {
970 returning_rows.push(updated_row);
971 rows_affected += 1;
972 }
973 Ok(None) => {
974 rows_affected += 1;
975 }
976 Err(e) => return Err(e),
977 }
978 }
979 Ok(None) => {
980 return Err(Error::UniqueConstraint {
981 index: index.clone(),
982 column: column.clone(),
983 value: value.clone(),
984 row_id: -1,
985 });
986 }
987 Err(e) => return Err(e),
988 }
989 }
990 Err(e) => return Err(e),
991 }
992 }
993 }
994 } else {
995 for value_row in &stmt.values {
997 if value_row.len() != column_indices.len() {
998 return Err(Error::InvalidArgument(format!(
999 "INSERT has {} columns but {} values",
1000 column_indices.len(),
1001 value_row.len()
1002 )));
1003 }
1004
1005 let mut row_values = Vec::with_capacity(schema_column_count);
1007 for i in 0..schema_column_count {
1008 if let Some(ref default_expr) = default_exprs[i] {
1009 let default_type = all_column_types[i];
1011 row_values.push(evaluate_default_expr(default_expr, default_type)?);
1012 } else {
1013 row_values.push(Value::null_unknown());
1014 }
1015 }
1016
1017 for (i, expr) in value_row.iter().enumerate() {
1019 if matches!(expr, Expression::Default(_)) {
1021 continue;
1022 }
1023 let value = if let Some(lit_value) = try_extract_literal(expr) {
1026 lit_value
1027 } else {
1028 let program = compile_expression(expr, &[])?;
1030 vm.execute_cow(&program, &base_exec_ctx)?
1031 };
1032 let coerced = value.coerce_to_type(column_types[i]);
1034 validate_coercion(
1036 &value,
1037 &coerced,
1038 &column_names[i],
1039 column_types[i],
1040 column_vector_dims[i],
1041 )?;
1042 row_values[column_indices[i]] = coerced;
1043 }
1044
1045 let mut row = Row::from_values(row_values);
1047 prepare_insert_row_constraints(
1048 &mut *table,
1049 &resulting_schema,
1050 &compiled_table_checks,
1051 &mut row,
1052 &mut vm,
1053 )?;
1054 let row = if let Some(plan) = triggers {
1055 let Some(row) =
1056 self.mutation_fire_before_row_triggers(plan, None, Some(row), None, ctx)?
1057 else {
1058 continue;
1059 };
1060 validate_resulting_row_constraints(
1061 &resulting_schema,
1062 &compiled_table_checks,
1063 &row,
1064 &mut vm,
1065 )?;
1066 row
1067 } else {
1068 row
1069 };
1070
1071 if let Some(ref fks) = fk_schema {
1073 crate::mutation::foreign_key::check_parent_exists(
1074 self.mutation_engine(),
1075 table.txn_id(),
1076 fks,
1077 &row,
1078 )?;
1079 }
1080
1081 let inserted_row = insert_row_for_command_result(
1082 &mut *table,
1083 row,
1084 needs_inserted_row,
1085 auto_increment_pk_idx,
1086 &mut last_insert_id,
1087 )?;
1088 self.accept_inserted_row(
1089 triggers,
1090 inserted_row,
1091 has_returning,
1092 &mut returning_rows,
1093 ctx,
1094 )?;
1095 rows_affected += 1;
1096 }
1097 }
1098
1099 if rows_affected > 0 {
1102 self.mutation_invalidate_semantic_cache(table_name);
1103 invalidate_semi_join_cache_for_table(table_name);
1104 invalidate_scalar_subquery_cache_for_table(table_name);
1105 invalidate_in_subquery_cache_for_table(table_name);
1106 }
1107
1108 let mut returning_result = if has_returning {
1109 Some(build_returning_result(
1110 &stmt.returning,
1111 std::mem::take(&mut returning_rows),
1112 schema_column_names_arc.as_ref().unwrap(),
1113 ctx,
1114 )?)
1115 } else {
1116 None
1117 };
1118
1119 if should_auto_commit {
1121 if let Some(mut tx) = standalone_tx {
1122 match tx.commit() {
1123 Ok(()) => {}
1124 Err(e)
1125 if (stmt.on_duplicate || stmt.do_nothing)
1126 && e.is_pk_or_unique_violation() =>
1127 {
1128 if stmt.on_duplicate && ctx.query_depth() == 0 {
1129 let retry_ctx = ctx.with_incremented_query_depth();
1130 return self.execute_insert(stmt, &retry_ctx);
1131 }
1132 if stmt.do_nothing {
1133 rows_affected = 0;
1135 returning_result = Some(build_returning_result(
1136 &stmt.returning,
1137 Vec::new(),
1138 schema_column_names_arc.as_ref().unwrap(),
1139 ctx,
1140 )?);
1141 } else {
1142 return Err(e);
1143 }
1144 }
1145 Err(e) => return Err(e),
1146 }
1147 }
1148 }
1149
1150 if let Some(result) = returning_result {
1152 return Ok(result);
1153 }
1154
1155 Ok(Box::new(ExecResult::with_last_insert_id(
1156 rows_affected,
1157 last_insert_id,
1158 )))
1159 }
1160
1161 fn execute_insert_with_compiled_cache(
1165 &self,
1166 stmt: &InsertStatement,
1167 ctx: &ExecutionContext,
1168 compiled_cache: &Arc<RwLock<CompiledExecution>>,
1169 ) -> Result<Box<dyn QueryResult>> {
1170 let trigger_plan = self.mutation_prepare_dml_triggers(
1171 stmt.table_name.value_lower.as_str(),
1172 DmlTriggerEvent::Insert,
1173 &[],
1174 ctx,
1175 )?;
1176 if !trigger_plan.is_empty() {
1177 return self.execute_insert(stmt, ctx);
1178 }
1179 if stmt.on_duplicate || stmt.do_nothing {
1181 return self.execute_insert(stmt, ctx);
1182 }
1183
1184 let table_name = &stmt.table_name.value_lower;
1186
1187 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
1189
1190 let (mut table, should_auto_commit, standalone_tx) =
1191 if let Some(ref mut tx_state) = *active_tx {
1192 let table = tx_state.transaction.get_table(table_name)?;
1194
1195 if !tx_state.tables.contains_key(table_name.as_str()) {
1197 tx_state.tables.insert(
1198 table_name.to_string(),
1199 tx_state.transaction.get_table(table_name)?,
1200 );
1201 }
1202
1203 (table, false, None)
1204 } else {
1205 let tx = self.mutation_engine().begin_transaction()?;
1207 let table = tx.get_table(table_name)?;
1208 (table, true, Some(tx))
1209 };
1210
1211 drop(active_tx);
1213 let resulting_schema = table.schema().clone();
1214
1215 let current_epoch = self.mutation_engine().schema_epoch();
1217 let cached_insert = {
1218 let cache_read = compiled_cache.read().unwrap();
1219 if let CompiledExecution::Insert(ref cached) = *cache_read {
1220 if cached.cached_epoch == current_epoch && *cached.table_name == *table_name {
1221 Some(cached.clone())
1222 } else {
1223 None }
1225 } else {
1226 None
1227 }
1228 };
1229
1230 let (
1232 column_indices,
1233 column_types,
1234 column_vector_dims,
1235 column_names,
1236 all_column_types,
1237 default_row_template,
1238 ) = if let Some(cached) = cached_insert {
1239 (
1241 cached.column_indices,
1242 cached.column_types,
1243 cached.column_vector_dims,
1244 cached.column_names,
1245 cached.all_column_types,
1246 cached.default_row_template,
1247 )
1248 } else {
1249 let schema = table.schema();
1251 let schema_column_count = schema.columns.len();
1252
1253 let all_column_types: Vec<DataType> =
1254 schema.columns.iter().map(|c| c.data_type).collect();
1255
1256 let default_row_template = vec![Value::null_unknown(); schema.columns.len()];
1260
1261 let (column_indices, column_types, column_vector_dims, column_names) =
1262 if stmt.columns.is_empty() {
1263 let indices: Vec<usize> = (0..schema_column_count).collect();
1265 let types = all_column_types.clone();
1266 let dims: Vec<u16> =
1267 schema.columns.iter().map(|c| c.vector_dimensions).collect();
1268 let names: Vec<SmartString> = schema
1269 .columns
1270 .iter()
1271 .map(|c| SmartString::new(&c.name))
1272 .collect();
1273 (indices, types, dims, names)
1274 } else {
1275 let col_map = schema.column_index_map();
1277 let indices: Vec<usize> = stmt
1278 .columns
1279 .iter()
1280 .map(|id| {
1281 col_map
1282 .get(id.value_lower.as_str())
1283 .copied()
1284 .ok_or_else(|| Error::ColumnNotFound(id.value.to_string()))
1285 })
1286 .collect::<Result<Vec<_>>>()?;
1287 let types: Vec<DataType> = indices
1288 .iter()
1289 .map(|&idx| schema.columns[idx].data_type)
1290 .collect();
1291 let dims: Vec<u16> = indices
1292 .iter()
1293 .map(|&idx| schema.columns[idx].vector_dimensions)
1294 .collect();
1295 let names: Vec<SmartString> = indices
1296 .iter()
1297 .map(|&idx| SmartString::new(&schema.columns[idx].name))
1298 .collect();
1299 (indices, types, dims, names)
1300 };
1301
1302 let compiled = CompiledInsert {
1304 table_name: SmartString::new(table_name),
1305 column_indices: Arc::new(column_indices.clone()),
1306 column_types: Arc::new(column_types.clone()),
1307 column_vector_dims: Arc::new(column_vector_dims.clone()),
1308 column_names: Arc::new(column_names.clone()),
1309 all_column_types: Arc::new(all_column_types.clone()),
1310 default_row_template: Arc::new(default_row_template.clone()),
1311 cached_epoch: current_epoch,
1312 };
1313
1314 if let Ok(mut cache_write) = compiled_cache.write() {
1316 *cache_write = CompiledExecution::Insert(compiled);
1317 }
1318
1319 (
1320 Arc::new(column_indices),
1321 Arc::new(column_types),
1322 Arc::new(column_vector_dims),
1323 Arc::new(column_names),
1324 Arc::new(all_column_types),
1325 Arc::new(default_row_template),
1326 )
1327 };
1328
1329 let mut seen_insert_targets = rustc_hash::FxHashSet::default();
1330 for &column_index in column_indices.iter() {
1331 if !seen_insert_targets.insert(column_index) {
1332 return Err(Error::InvalidArgument(
1333 "INSERT target column is specified more than once".to_string(),
1334 ));
1335 }
1336 }
1337
1338 let fk_schema = if !table.schema().foreign_keys.is_empty() {
1340 Some(table.schema().clone())
1341 } else {
1342 None
1343 };
1344 let compiled_table_checks = compile_table_check_constraints(table.schema())?;
1345 let auto_increment_pk_idx = auto_increment_pk_index(table.schema());
1346 let default_exprs: Vec<Option<String>> = table
1347 .schema()
1348 .columns
1349 .iter()
1350 .map(|column| column.default_expr.clone())
1351 .collect();
1352
1353 use crate::expression::{compile_expression, ExecuteContext, ExprVM};
1355 let mut vm = ExprVM::new();
1356 let params = ctx.params();
1357 let named_params = ctx.named_params();
1358 let empty_row = Row::new();
1359
1360 let mut base_exec_ctx = ExecuteContext::new(&empty_row);
1362 if !params.is_empty() {
1363 base_exec_ctx = base_exec_ctx.with_params(params);
1364 }
1365 if !named_params.is_empty() {
1366 base_exec_ctx = base_exec_ctx.with_named_params(named_params);
1367 }
1368 base_exec_ctx = base_exec_ctx
1369 .with_transaction_id(ctx.transaction_id())
1370 .with_stored_function_invoker(ctx.stored_function_invoker());
1371
1372 let mut rows_affected = 0i64;
1373 let mut last_insert_id = 0i64;
1374
1375 let has_returning = !stmt.returning.is_empty();
1377 let mut returning_rows: Vec<Row> = Vec::new();
1378 let schema_column_names_arc = if has_returning {
1379 Some(table.schema().column_names_arc())
1380 } else {
1381 None
1382 };
1383
1384 let col_to_value_pos: Vec<Option<usize>> = {
1387 let mut lookup = vec![None; default_row_template.len()];
1388 for (value_pos, &col_idx) in column_indices.iter().enumerate() {
1389 lookup[col_idx] = Some(value_pos);
1390 }
1391 lookup
1392 };
1393 let num_columns = default_row_template.len();
1394
1395 if let Some(ref select_stmt) = stmt.select {
1397 let mut select_result = self.mutation_execute_select(select_stmt, ctx)?;
1398 if !should_auto_commit {
1399 let columns = select_result.columns().to_vec();
1401 let rows = <Self as MutationHost>::mutation_materialize_result(select_result)?;
1402 select_result = Box::new(crate::result::ExecutorResult::new(columns, rows));
1403 }
1404
1405 while select_result.next() {
1407 let select_row = select_result.row();
1408 if select_row.len() != column_indices.len() {
1409 return Err(Error::InvalidArgument(format!(
1410 "INSERT has {} columns but SELECT returns {} columns",
1411 column_indices.len(),
1412 select_row.len()
1413 )));
1414 }
1415
1416 let mut row_values = Vec::with_capacity(num_columns);
1419 for (col_idx, default_val) in default_row_template.iter().enumerate() {
1420 if let Some(value_pos) = col_to_value_pos[col_idx] {
1421 let value = &select_row[value_pos];
1423 let coerced = value.coerce_to_type(column_types[value_pos]);
1424 validate_coercion(
1425 value,
1426 &coerced,
1427 &column_names[value_pos],
1428 column_types[value_pos],
1429 column_vector_dims[value_pos],
1430 )?;
1431 row_values.push(coerced);
1432 } else if let Some(expression) = &default_exprs[col_idx] {
1433 row_values.push(evaluate_default_expr(
1434 expression,
1435 all_column_types[col_idx],
1436 )?);
1437 } else {
1438 row_values.push(default_val.clone());
1439 }
1440 }
1441
1442 let mut row = Row::from_values(row_values);
1444 prepare_insert_row_constraints(
1445 &mut *table,
1446 &resulting_schema,
1447 &compiled_table_checks,
1448 &mut row,
1449 &mut vm,
1450 )?;
1451
1452 if let Some(ref fks) = fk_schema {
1454 crate::mutation::foreign_key::check_parent_exists(
1455 self.mutation_engine(),
1456 table.txn_id(),
1457 fks,
1458 &row,
1459 )?;
1460 }
1461
1462 if let Some(inserted_row) = insert_row_for_command_result(
1463 &mut *table,
1464 row,
1465 has_returning,
1466 auto_increment_pk_idx,
1467 &mut last_insert_id,
1468 )? {
1469 returning_rows.push(inserted_row);
1470 }
1471 rows_affected += 1;
1472 }
1473 if let Some(err) = select_result.last_error() {
1474 return Err(err);
1475 }
1476 } else {
1477 for value_list in &stmt.values {
1479 if value_list.len() != column_indices.len() {
1480 return Err(Error::InvalidArgument(format!(
1481 "INSERT has {} columns but {} values provided",
1482 column_indices.len(),
1483 value_list.len()
1484 )));
1485 }
1486
1487 let mut row_values = Vec::with_capacity(num_columns);
1490 for (col_idx, default_val) in default_row_template.iter().enumerate() {
1491 if let Some(value_pos) = col_to_value_pos[col_idx] {
1492 let expr = &value_list[value_pos];
1494 if matches!(expr, Expression::Default(_)) {
1495 if let Some(expression) = &default_exprs[col_idx] {
1496 row_values.push(evaluate_default_expr(
1497 expression,
1498 all_column_types[col_idx],
1499 )?);
1500 } else {
1501 row_values.push(default_val.clone());
1502 }
1503 } else {
1504 let value = if let Some(lit_val) = try_extract_literal(expr) {
1506 lit_val
1507 } else {
1508 let program = compile_expression(expr, &[])?;
1510 vm.execute_cow(&program, &base_exec_ctx)?
1511 };
1512
1513 let target_type = column_types[value_pos];
1514 let coerced = value.coerce_to_type(target_type);
1515 validate_coercion(
1516 &value,
1517 &coerced,
1518 &column_names[value_pos],
1519 target_type,
1520 column_vector_dims[value_pos],
1521 )?;
1522 row_values.push(coerced);
1523 }
1524 } else if let Some(expression) = &default_exprs[col_idx] {
1525 row_values.push(evaluate_default_expr(
1526 expression,
1527 all_column_types[col_idx],
1528 )?);
1529 } else {
1530 row_values.push(default_val.clone());
1531 }
1532 }
1533
1534 let mut row = Row::from_values(row_values);
1536 prepare_insert_row_constraints(
1537 &mut *table,
1538 &resulting_schema,
1539 &compiled_table_checks,
1540 &mut row,
1541 &mut vm,
1542 )?;
1543
1544 if let Some(ref fks) = fk_schema {
1546 crate::mutation::foreign_key::check_parent_exists(
1547 self.mutation_engine(),
1548 table.txn_id(),
1549 fks,
1550 &row,
1551 )?;
1552 }
1553
1554 if let Some(inserted_row) = insert_row_for_command_result(
1555 &mut *table,
1556 row,
1557 has_returning,
1558 auto_increment_pk_idx,
1559 &mut last_insert_id,
1560 )? {
1561 returning_rows.push(inserted_row);
1562 }
1563 rows_affected += 1;
1564 }
1565 }
1566
1567 if rows_affected > 0 {
1569 self.mutation_invalidate_semantic_cache(table_name);
1570 invalidate_semi_join_cache_for_table(table_name);
1571 invalidate_scalar_subquery_cache_for_table(table_name);
1572 invalidate_in_subquery_cache_for_table(table_name);
1573 }
1574
1575 let returning_result = if has_returning {
1576 Some(build_returning_result(
1577 &stmt.returning,
1578 returning_rows,
1579 schema_column_names_arc.as_ref().unwrap(),
1580 ctx,
1581 )?)
1582 } else {
1583 None
1584 };
1585
1586 if should_auto_commit {
1588 if let Some(mut tx) = standalone_tx {
1589 tx.commit()?;
1590 }
1591 }
1592
1593 if let Some(result) = returning_result {
1595 return Ok(result);
1596 }
1597
1598 Ok(Box::new(ExecResult::with_last_insert_id(
1599 rows_affected,
1600 last_insert_id,
1601 )))
1602 }
1603
1604 fn execute_update(
1606 &self,
1607 stmt: &UpdateStatement,
1608 ctx: &ExecutionContext,
1609 ) -> Result<Box<dyn QueryResult>> {
1610 let updated_columns = stmt
1611 .updates
1612 .keys()
1613 .map(ToString::to_string)
1614 .collect::<Vec<_>>();
1615 let plan = self.mutation_prepare_dml_triggers(
1616 stmt.table_name.value_lower.as_str(),
1617 DmlTriggerEvent::Update,
1618 &updated_columns,
1619 ctx,
1620 )?;
1621 if plan.is_empty() {
1622 return self.execute_update_body(stmt, ctx, None);
1623 }
1624 self.execute_with_trigger_statement(&plan, ctx, |plan| {
1625 self.execute_update_body(stmt, ctx, Some(plan))
1626 })
1627 }
1628
1629 fn execute_update_body(
1630 &self,
1631 stmt: &UpdateStatement,
1632 ctx: &ExecutionContext,
1633 triggers: Option<&DmlTriggerPlan>,
1634 ) -> Result<Box<dyn QueryResult>> {
1635 let table_name = &stmt.table_name.value_lower;
1637
1638 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
1640
1641 let (mut table, should_auto_commit, standalone_tx) =
1642 if let Some(ref mut tx_state) = *active_tx {
1643 let table = tx_state.transaction.get_table(table_name)?;
1646
1647 if !tx_state.tables.contains_key(table_name.as_str()) {
1649 tx_state.tables.insert(
1650 table_name.to_string(),
1651 tx_state.transaction.get_table(table_name)?,
1652 );
1653 }
1654
1655 (table, false, None)
1656 } else {
1657 let tx = self.mutation_engine().begin_transaction()?;
1659 let table = tx.get_table(table_name)?;
1660 (table, true, Some(tx))
1661 };
1662
1663 drop(active_tx);
1665
1666 let has_returning = !stmt.returning.is_empty();
1668
1669 let schema = table.schema();
1671 let constraint_schema = schema.clone();
1672 let compiled_table_checks = compile_table_check_constraints(schema)?;
1673 let column_names = schema.column_names_arc();
1675 let col_map = schema.column_index_map();
1676 for (column, expression) in &stmt.updates {
1677 if !col_map.contains_key(column.to_lowercase().as_str()) {
1678 return Err(Error::ColumnNotFound(column.to_string()));
1679 }
1680 if !<Self as MutationHost>::mutation_has_subqueries(expression) {
1681 crate::expression::compile_expression(expression, &column_names)?;
1682 }
1683 }
1684
1685 let fk_cols_in_update: Vec<(usize, radixdb_core::ForeignKeyConstraint)> = {
1688 let col_map = schema.column_index_map();
1689 let updated_col_indices: Vec<usize> = stmt
1690 .updates
1691 .keys()
1692 .filter_map(|col_name| {
1693 let col_lower = col_name.to_lowercase();
1694 col_map.get(col_lower.as_str()).copied()
1695 })
1696 .collect();
1697 schema
1698 .foreign_keys
1699 .iter()
1700 .filter(|fk| updated_col_indices.contains(&fk.column_index))
1701 .map(|fk| (fk.column_index, fk.clone()))
1702 .collect()
1703 };
1704 let has_fk_updates = !fk_cols_in_update.is_empty();
1705
1706 if let Some(pk_idx) = schema.pk_column_index() {
1711 let col_map = schema.column_index_map();
1712 for col_name in stmt.updates.keys() {
1713 let col_lower = col_name.to_lowercase();
1714 if col_map.get(col_lower.as_str()).copied() == Some(pk_idx) {
1715 let pk_col_name = &schema.columns[pk_idx].name;
1716 return Err(radixdb_core::Error::invalid_argument(format!(
1717 "cannot UPDATE primary key column '{}'. Use DELETE + INSERT instead",
1718 pk_col_name
1719 )));
1720 }
1721 }
1722 }
1723
1724 let all_referencing_fks = crate::mutation::foreign_key::find_referencing_fks_for_txn(
1727 self.mutation_engine(),
1728 table.txn_id(),
1729 table_name,
1730 );
1731 let referencing_fks_for_update: Arc<Vec<(String, radixdb_core::ForeignKeyConstraint)>> =
1732 if all_referencing_fks.is_empty() {
1733 Arc::new(Vec::new())
1734 } else {
1735 let col_map = schema.column_index_map();
1736 let updated_cols: Vec<usize> = stmt
1737 .updates
1738 .iter()
1739 .filter_map(|(column, expression)| {
1740 if Self::assignment_preserves_column(expression, column, table_name) {
1741 None
1742 } else {
1743 col_map.get(column.to_lowercase().as_str()).copied()
1744 }
1745 })
1746 .collect();
1747 let relevant: Vec<_> = all_referencing_fks
1748 .iter()
1749 .filter(|(_, fk)| {
1750 col_map
1751 .get(fk.referenced_column.to_lowercase().as_str())
1752 .is_some_and(|&idx| updated_cols.contains(&idx))
1753 })
1754 .cloned()
1755 .collect();
1756 Arc::new(relevant)
1757 };
1758
1759 let fk_update_schema = if has_fk_updates {
1761 Some(self.mutation_engine().get_table_schema(table_name)?)
1762 } else {
1763 None
1764 };
1765
1766 if has_fk_updates && !should_auto_commit {
1771 let col_map = schema.column_index_map();
1772 for (col_name, expr) in &stmt.updates {
1773 let col_lower = col_name.to_lowercase();
1774 if let Some(&col_idx) = col_map.get(col_lower.as_str()) {
1775 if let Some(fk) = schema
1776 .foreign_keys
1777 .iter()
1778 .find(|f| f.column_index == col_idx)
1779 {
1780 if let Some(value) = Self::try_extract_constant_fk_value(expr, ctx) {
1781 if !value.is_null() {
1782 crate::mutation::foreign_key::validate_fk_value(
1783 self.mutation_engine(),
1784 table.txn_id(),
1785 fk,
1786 &value,
1787 table_name,
1788 )?;
1789 }
1790 }
1791 }
1792 }
1793 }
1794 }
1795
1796 let has_update_subqueries = stmt
1798 .updates
1799 .iter()
1800 .any(|(_, expr)| <Self as MutationHost>::mutation_has_subqueries(expr));
1801
1802 let has_correlated_updates = stmt.updates.iter().any(|(_, expr)| {
1804 <Self as MutationHost>::mutation_has_subqueries(expr)
1805 && <Self as MutationHost>::mutation_has_correlated_subqueries(expr)
1806 });
1807
1808 let processed_updates: Option<Vec<(String, Expression)>> =
1811 if has_update_subqueries && !has_correlated_updates {
1812 let processed: Result<Vec<_>> = stmt
1813 .updates
1814 .iter()
1815 .map(|(col_name, expr)| {
1816 let processed_expr = self.mutation_process_where_subqueries(expr, ctx)?;
1817 Ok((col_name.to_string(), processed_expr))
1818 })
1819 .collect();
1820 Some(processed?)
1821 } else {
1822 None
1823 };
1824
1825 let requires_row_staging = has_correlated_updates || triggers.is_some();
1828
1829 let update_indices: Vec<(usize, radixdb_core::DataType, u16, Expression, bool)> =
1833 if requires_row_staging {
1834 {
1837 let col_map = schema.column_index_map();
1838 let staged_updates = processed_updates.clone().unwrap_or_else(|| {
1839 stmt.updates
1840 .iter()
1841 .map(|(column, expression)| (column.to_string(), expression.clone()))
1842 .collect()
1843 });
1844 staged_updates
1845 .iter()
1846 .filter_map(|(col_name, expr)| {
1847 let is_correlated =
1848 <Self as MutationHost>::mutation_has_subqueries(expr)
1849 && <Self as MutationHost>::mutation_has_correlated_subqueries(
1850 expr,
1851 );
1852 let col_lower = col_name.to_lowercase();
1853 col_map.get(&col_lower).map(|&idx| {
1854 (
1855 idx,
1856 schema.columns[idx].data_type,
1857 schema.columns[idx].vector_dimensions,
1858 expr.clone(),
1859 is_correlated,
1860 )
1861 })
1862 })
1863 .collect()
1864 }
1865 } else {
1866 Vec::new()
1868 };
1869
1870 let (where_expr, needs_memory_filter, memory_where_clause): (
1876 Option<Box<dyn StorageExpr>>,
1877 bool,
1878 Option<Expression>,
1879 ) = if let Some(ref where_clause) = stmt.where_clause {
1880 let has_correlated_where =
1881 <Self as MutationHost>::mutation_has_subqueries(where_clause)
1882 && <Self as MutationHost>::mutation_has_correlated_subqueries(where_clause);
1883
1884 let processed_where = if has_correlated_where {
1885 let outer_tables = vec![table_name.to_string()];
1888
1889 let exists_optimized = self
1891 .mutation_optimize_exists_to_semi_join(where_clause, ctx, &outer_tables, None)
1892 .ok()
1893 .flatten();
1894
1895 let expr_for_in = exists_optimized.as_ref().unwrap_or(where_clause.as_ref());
1897 let in_optimized = self
1898 .mutation_optimize_in_to_semi_join(expr_for_in, ctx, &outer_tables)
1899 .ok()
1900 .flatten();
1901
1902 let current_expr = in_optimized
1904 .or(exists_optimized)
1905 .unwrap_or_else(|| (**where_clause).clone());
1906
1907 if <Self as MutationHost>::mutation_has_subqueries(¤t_expr) {
1909 self.mutation_process_where_subqueries(¤t_expr, ctx)?
1910 } else {
1911 current_expr
1912 }
1913 } else if <Self as MutationHost>::mutation_has_subqueries(where_clause) {
1914 self.mutation_process_where_subqueries(where_clause, ctx)?
1915 } else {
1916 (**where_clause).clone()
1917 };
1918
1919 let plan = pushdown::try_pushdown_plan(&processed_where, schema, Some(ctx));
1921 let needs_mem = plan.needs_memory_filter();
1922 (plan.storage_expr, needs_mem, plan.residual)
1923 } else {
1924 (None, false, None)
1925 };
1926
1927 let function_registry = self.mutation_function_registry();
1928
1929 let mut evaluator = CompiledEvaluator::new(function_registry).with_context(ctx);
1931 evaluator.init_columns_arc(CompactArc::clone(&column_names));
1932
1933 use std::cell::RefCell;
1935 let returning_rows: RefCell<Vec<Row>> = RefCell::new(Vec::new());
1936 let fk_new_values: RefCell<Vec<Row>> = RefCell::new(Vec::new());
1938 let ref_col_changes: RefCell<Vec<(usize, Value, Value)>> = RefCell::new(Vec::new());
1940
1941 let ref_col_indices_for_fk: Vec<usize> = if !referencing_fks_for_update.is_empty() {
1944 let col_map = schema.column_index_map();
1945 referencing_fks_for_update
1946 .iter()
1947 .filter_map(|(_, fk)| {
1948 col_map
1949 .get(fk.referenced_column.to_lowercase().as_str())
1950 .copied()
1951 })
1952 .collect::<rustc_hash::FxHashSet<usize>>()
1953 .into_iter()
1954 .collect()
1955 } else {
1956 Vec::new()
1957 };
1958
1959 let fks_by_ref_col: FxHashMap<usize, Vec<(String, radixdb_core::ForeignKeyConstraint)>> =
1962 if !referencing_fks_for_update.is_empty() {
1963 let col_map = schema.column_index_map();
1964 let mut map: FxHashMap<usize, Vec<(String, radixdb_core::ForeignKeyConstraint)>> =
1965 FxHashMap::default();
1966 for (tbl, fk) in referencing_fks_for_update.iter() {
1967 if let Some(&idx) = col_map.get(fk.referenced_column.to_lowercase().as_str()) {
1968 map.entry(idx).or_default().push((tbl.clone(), fk.clone()));
1969 }
1970 }
1971 map
1972 } else {
1973 FxHashMap::default()
1974 };
1975
1976 if !fks_by_ref_col.is_empty() {
1980 let any_needs_precheck = fks_by_ref_col.values().any(|fks| {
1981 crate::mutation::foreign_key::fk_tree_needs_precheck(
1982 self.mutation_engine(),
1983 table.txn_id(),
1984 fks,
1985 )
1986 });
1987 if any_needs_precheck {
1988 let parent_rows = table.collect_all_rows(where_expr.as_deref())?;
1989 for (_rid, row) in parent_rows.iter() {
1990 if needs_memory_filter {
1991 if let Some(ref mem_where) = memory_where_clause {
1992 evaluator.set_row_array(row);
1993 match evaluator.evaluate_bool(mem_where) {
1994 Ok(true) => {}
1995 _ => continue,
1996 }
1997 }
1998 }
1999 for (&col_idx, fks_for_col) in &fks_by_ref_col {
2000 if let Some(old_val) = row.get(col_idx) {
2001 if !old_val.is_null() {
2002 crate::mutation::foreign_key::pre_check_restrict_for_update(
2003 self.mutation_engine(),
2004 table.txn_id(),
2005 table_name,
2006 old_val,
2007 fks_for_col,
2008 )?;
2009 }
2010 }
2011 }
2012 }
2013 }
2014 }
2015
2016 let rows_affected = if requires_row_staging {
2020 type StagedUpdate = (i64, Row, Vec<(usize, Value)>);
2030 let mut precomputed: Vec<StagedUpdate> = Vec::new();
2031
2032 let all_col_indices: Vec<usize> = (0..column_names.len()).collect();
2034
2035 let column_names_lower = schema.column_names_lower_arc();
2038 let col_name_pairs: Vec<(CompactArc<str>, CompactArc<str>)> = column_names_lower
2039 .iter()
2040 .map(|col_lower| {
2041 let qualified =
2042 CompactArc::from(format!("{}.{}", table_name, col_lower).as_str());
2043 (CompactArc::from(col_lower.as_str()), qualified)
2044 })
2045 .collect();
2046
2047 let mut outer_row_map: FxHashMap<CompactArc<str>, Value> =
2050 FxHashMap::with_capacity_and_hasher(col_name_pairs.len() * 2, Default::default());
2051
2052 let mut scanner = table.scan(&all_col_indices, where_expr.as_deref())?;
2055 while scanner.next() {
2056 let row = scanner.row();
2057 let row_id = scanner.current_row_id()?;
2058
2059 evaluator.set_row_array(row);
2061 if needs_memory_filter {
2062 if let Some(ref where_clause) = memory_where_clause {
2063 match evaluator.evaluate_bool(where_clause) {
2064 Ok(true) => {}
2065 Ok(false) => continue,
2066 Err(error) => return Err(error),
2067 }
2068 }
2069 }
2070
2071 outer_row_map.clear();
2073 for (i, (col_lower, qualified)) in col_name_pairs.iter().enumerate() {
2074 if let Some(value) = row.get(i) {
2075 outer_row_map.insert(col_lower.clone(), value.clone());
2076 outer_row_map.insert(qualified.clone(), value.clone());
2077 }
2078 }
2079
2080 let mut correlated_ctx = ctx.with_outer_row(
2083 std::mem::take(&mut outer_row_map),
2084 CompactArc::clone(&column_names),
2085 );
2086
2087 let mut new_values: Vec<(usize, Value)> = Vec::with_capacity(update_indices.len());
2089 for (idx, col_type, vec_dims, expr, is_correlated) in update_indices.iter() {
2090 let evaluated = if *is_correlated {
2091 let processed_expr =
2093 self.mutation_process_correlated_expression(expr, &correlated_ctx)?;
2094 let mut eval =
2096 CompiledEvaluator::new(function_registry).with_context(&correlated_ctx);
2097 eval.init_columns_arc(CompactArc::clone(&column_names));
2098 eval.set_row_array(row);
2099 Some(eval.evaluate(&processed_expr)?)
2100 } else {
2101 Some(evaluator.evaluate(expr)?)
2102 };
2103
2104 if let Some(new_value) = evaluated {
2105 let coerced = new_value.coerce_to_type(*col_type);
2106 validate_coercion(
2107 &new_value,
2108 &coerced,
2109 &schema.columns[*idx].name,
2110 *col_type,
2111 *vec_dims,
2112 )?;
2113 new_values.push((*idx, coerced));
2114 }
2115 }
2116
2117 outer_row_map = correlated_ctx.take_outer_row().unwrap_or_default();
2119
2120 if !new_values.is_empty() {
2121 precomputed.push((row_id, row.clone(), new_values));
2122 }
2123 }
2124 drop(scanner);
2125
2126 let mut table_check_vm = crate::expression::ExprVM::new();
2128 let mut updated = 0i32;
2129 for (row_id, old_row, updates) in precomputed {
2130 let mut new_row = old_row.clone();
2131 for (idx, new_value) in &updates {
2132 let _ = new_row.set(*idx, new_value.clone());
2133 }
2134 if let Some(plan) = triggers {
2135 let Some(trigger_row) = self.mutation_fire_before_row_triggers(
2136 plan,
2137 Some(&old_row),
2138 Some(new_row),
2139 Some(row_id),
2140 ctx,
2141 )?
2142 else {
2143 continue;
2144 };
2145 new_row = trigger_row;
2146 }
2147
2148 validate_resulting_row_constraints(
2149 &constraint_schema,
2150 &compiled_table_checks,
2151 &new_row,
2152 &mut table_check_vm,
2153 )?;
2154
2155 for &column_index in &ref_col_indices_for_fk {
2156 let old_value = old_row.get(column_index);
2157 let new_value = new_row.get(column_index);
2158 if let (Some(old_value), Some(new_value)) = (old_value, new_value) {
2159 if old_value != new_value {
2160 ref_col_changes.borrow_mut().push((
2161 column_index,
2162 old_value.clone(),
2163 new_value.clone(),
2164 ));
2165 }
2166 }
2167 }
2168 if has_fk_updates {
2169 fk_new_values.borrow_mut().push(new_row.clone());
2170 }
2171
2172 let final_row = new_row.clone();
2173 let mut setter =
2174 |_row: Row| -> Result<(Row, bool)> { Ok((final_row.clone(), true)) };
2175 let row_count = table.update_by_row_ids(&[row_id], &mut setter)?;
2176 if row_count > 0 {
2177 if has_returning {
2178 returning_rows.borrow_mut().push(new_row.clone());
2179 }
2180 if let Some(plan) = triggers {
2181 self.mutation_fire_after_row_triggers(
2182 plan,
2183 Some(&old_row),
2184 Some(&new_row),
2185 Some(row_id),
2186 ctx,
2187 )?;
2188 }
2189 }
2190 updated += row_count;
2191 }
2192 updated
2193 } else {
2194 use crate::expression::{compile_expression, ExecuteContext, ExprVM, SharedProgram};
2198
2199 let col_map = schema.column_index_map();
2200 let compiled_updates: Vec<(usize, String, radixdb_core::DataType, u16, SharedProgram)> =
2201 if let Some(ref processed) = processed_updates {
2202 processed
2204 .iter()
2205 .map(|(col_name, expr)| {
2206 let col_lower = col_name.to_lowercase();
2207 let &idx = col_map
2208 .get(&col_lower)
2209 .ok_or_else(|| Error::ColumnNotFound(col_name.clone()))?;
2210 let program = compile_expression(expr, &column_names)?;
2211 Ok((
2212 idx,
2213 schema.columns[idx].name.clone(),
2214 schema.columns[idx].data_type,
2215 schema.columns[idx].vector_dimensions,
2216 program,
2217 ))
2218 })
2219 .collect::<Result<_>>()?
2220 } else {
2221 stmt.updates
2223 .iter()
2224 .map(|(col_name, expr)| {
2225 let col_lower: String = col_name.to_lowercase().into();
2226 let &idx = col_map
2227 .get(&col_lower)
2228 .ok_or_else(|| Error::ColumnNotFound(col_name.to_string()))?;
2229 let program = compile_expression(expr, &column_names)?;
2230 Ok((
2231 idx,
2232 schema.columns[idx].name.clone(),
2233 schema.columns[idx].data_type,
2234 schema.columns[idx].vector_dimensions,
2235 program,
2236 ))
2237 })
2238 .collect::<Result<_>>()?
2239 };
2240
2241 if compiled_updates.len() != stmt.updates.len() {
2242 return Err(Error::internal(
2243 "UPDATE compiler did not produce one program per assignment",
2244 ));
2245 }
2246
2247 let mut vm = ExprVM::new();
2249 let params = ctx.params();
2251 let named_params = ctx.named_params();
2252 let mut setter = |mut row: Row| -> Result<(Row, bool)> {
2253 if needs_memory_filter {
2255 evaluator.set_row_array(&row);
2256 if let Some(ref where_expr) = memory_where_clause {
2257 match evaluator.evaluate_bool(where_expr) {
2258 Ok(true) => {}
2259 Ok(false) => return Ok((row, false)),
2260 Err(error) => return Err(error),
2261 }
2262 }
2263 }
2264
2265 let updates_to_apply: Vec<(usize, Value)> = {
2267 let exec_ctx = ExecuteContext::new(&row)
2268 .with_params(params)
2269 .with_named_params(named_params)
2270 .with_transaction_id(ctx.transaction_id())
2271 .with_stored_function_invoker(ctx.stored_function_invoker());
2272
2273 let mut updates = Vec::with_capacity(compiled_updates.len());
2274 for (idx, col_name, col_type, vec_dims, program) in &compiled_updates {
2275 let v = vm.execute_cow(program, &exec_ctx)?;
2276 let coerced = v.try_coerce_to_type(*col_type)?;
2277 validate_coercion(&v, &coerced, col_name, *col_type, *vec_dims)?;
2278 updates.push((*idx, coerced));
2279 }
2280 updates
2281 };
2282
2283 let changed = !updates_to_apply.is_empty();
2285
2286 let ref_old_values: Vec<(usize, Value)> =
2288 if changed && !ref_col_indices_for_fk.is_empty() {
2289 ref_col_indices_for_fk
2290 .iter()
2291 .filter_map(|&ci| row.get(ci).map(|v| (ci, v.clone())))
2292 .collect()
2293 } else {
2294 Vec::new()
2295 };
2296
2297 for (idx, new_value) in updates_to_apply {
2298 let _ = row.set(idx, new_value);
2299 }
2300
2301 if changed {
2302 validate_resulting_row_constraints(
2303 &constraint_schema,
2304 &compiled_table_checks,
2305 &row,
2306 &mut vm,
2307 )?;
2308 }
2309
2310 if changed && has_fk_updates {
2312 fk_new_values.borrow_mut().push(row.clone());
2313 }
2314
2315 if !ref_old_values.is_empty() {
2317 for (ci, old_val) in &ref_old_values {
2318 if let Some(new_val) = row.get(*ci) {
2319 if old_val != new_val {
2320 ref_col_changes.borrow_mut().push((
2321 *ci,
2322 old_val.clone(),
2323 new_val.clone(),
2324 ));
2325 }
2326 }
2327 }
2328 }
2329
2330 if changed && has_returning {
2332 returning_rows.borrow_mut().push(row.clone());
2333 }
2334
2335 Ok((row, changed))
2336 };
2337
2338 let rows = if where_expr.is_none() {
2341 if let Some(ref where_clause) = memory_where_clause {
2342 if let Some(row_ids) = self.select_row_ids_for_dml(
2343 table_name,
2344 where_clause,
2345 schema,
2346 table.as_ref(),
2347 ctx,
2348 )? {
2349 table.update_by_row_ids(&row_ids, &mut setter)?
2350 } else {
2351 table.update(where_expr.as_deref(), &mut setter)?
2353 }
2354 } else {
2355 table.update(None, &mut setter)?
2356 }
2357 } else {
2358 table.update(where_expr.as_deref(), &mut setter)?
2363 };
2364 rows
2365 };
2366
2367 if has_fk_updates {
2369 let fk_rows = fk_new_values.into_inner();
2370 if let Some(ref fk_schema) = fk_update_schema {
2371 for row in &fk_rows {
2372 crate::mutation::foreign_key::check_parent_exists(
2373 self.mutation_engine(),
2374 table.txn_id(),
2375 fk_schema,
2376 row,
2377 )?;
2378 }
2379 }
2380 }
2381
2382 if !fks_by_ref_col.is_empty() {
2386 let changes = ref_col_changes.into_inner();
2387 for (col_idx, old_val, new_val) in &changes {
2388 if let Some(fks_for_col) = fks_by_ref_col.get(col_idx) {
2389 crate::mutation::foreign_key::enforce_update_actions(
2390 self.mutation_engine(),
2391 table.txn_id(),
2392 old_val,
2393 new_val,
2394 fks_for_col,
2395 )?;
2396 }
2397 }
2398 }
2399
2400 if rows_affected > 0 {
2403 self.mutation_invalidate_semantic_cache(table_name);
2404 invalidate_semi_join_cache_for_table(table_name);
2405 invalidate_scalar_subquery_cache_for_table(table_name);
2406 invalidate_in_subquery_cache_for_table(table_name);
2407 }
2408
2409 let returning_result = if has_returning {
2410 let rows = returning_rows.into_inner();
2411 Some(build_returning_result(
2412 &stmt.returning,
2413 rows,
2414 &column_names,
2415 ctx,
2416 )?)
2417 } else {
2418 None
2419 };
2420
2421 if should_auto_commit {
2423 if let Some(mut tx) = standalone_tx {
2425 tx.commit()?;
2426 }
2427 }
2428
2429 if let Some(result) = returning_result {
2431 return Ok(result);
2432 }
2433
2434 Ok(Box::new(ExecResult::with_rows_affected(
2435 rows_affected as i64,
2436 )))
2437 }
2438
2439 fn try_extract_constant_fk_value(expr: &Expression, ctx: &ExecutionContext) -> Option<Value> {
2443 match expr {
2444 Expression::IntegerLiteral(lit) => Some(Value::Integer(lit.value)),
2445 Expression::FloatLiteral(lit) => Some(Value::Float(lit.value)),
2446 Expression::StringLiteral(lit) => Some(Value::text(lit.value.as_str())),
2447 Expression::BooleanLiteral(lit) => Some(Value::Boolean(lit.value)),
2448 Expression::NullLiteral(_) => Some(Value::null_unknown()),
2449 Expression::Prefix(prefix) if prefix.operator == "-" => match prefix.right.as_ref() {
2450 Expression::IntegerLiteral(lit) => Some(Value::Integer(-lit.value)),
2451 Expression::FloatLiteral(lit) => Some(Value::Float(-lit.value)),
2452 _ => None,
2453 },
2454 Expression::Parameter(param) => {
2455 if param.name.starts_with(':') {
2456 ctx.get_named_param(¶m.name[1..]).cloned()
2457 } else if param.index > 0 {
2458 ctx.params().get(param.index - 1).cloned()
2459 } else {
2460 None
2461 }
2462 }
2463 _ => None, }
2465 }
2466
2467 fn assignment_preserves_column(
2470 expression: &Expression,
2471 column_name: &str,
2472 table_name: &str,
2473 ) -> bool {
2474 match expression {
2475 Expression::Identifier(identifier) => {
2476 identifier.value.eq_ignore_ascii_case(column_name)
2477 }
2478 Expression::QualifiedIdentifier(identifier) => {
2479 identifier.name.value.eq_ignore_ascii_case(column_name)
2480 && identifier.qualifier.value.eq_ignore_ascii_case(table_name)
2481 }
2482 _ => false,
2483 }
2484 }
2485
2486 fn execute_delete(
2488 &self,
2489 stmt: &DeleteStatement,
2490 ctx: &ExecutionContext,
2491 ) -> Result<Box<dyn QueryResult>> {
2492 let plan = self.mutation_prepare_dml_triggers(
2493 stmt.table_name.value_lower.as_str(),
2494 DmlTriggerEvent::Delete,
2495 &[],
2496 ctx,
2497 )?;
2498 if plan.is_empty() {
2499 return self.execute_delete_body(stmt, ctx, None);
2500 }
2501 self.execute_with_trigger_statement(&plan, ctx, |plan| {
2502 self.execute_delete_body(stmt, ctx, Some(plan))
2503 })
2504 }
2505
2506 fn execute_delete_body(
2507 &self,
2508 stmt: &DeleteStatement,
2509 ctx: &ExecutionContext,
2510 triggers: Option<&DmlTriggerPlan>,
2511 ) -> Result<Box<dyn QueryResult>> {
2512 let table_name = &stmt.table_name.value_lower;
2514 let effective_name = stmt
2516 .alias
2517 .as_ref()
2518 .map(|a| a.value_lower.as_str())
2519 .unwrap_or(table_name.as_str());
2520
2521 let mut active_tx = self.mutation_active_transaction().lock().unwrap();
2523
2524 let (mut table, should_auto_commit, standalone_tx) =
2525 if let Some(ref mut tx_state) = *active_tx {
2526 let table = tx_state.transaction.get_table(table_name)?;
2529
2530 if !tx_state.tables.contains_key(table_name.as_str()) {
2532 tx_state.tables.insert(
2533 table_name.to_string(),
2534 tx_state.transaction.get_table(table_name)?,
2535 );
2536 }
2537
2538 (table, false, None)
2539 } else {
2540 let tx = self.mutation_engine().begin_transaction()?;
2542 let table = tx.get_table(table_name)?;
2543 (table, true, Some(tx))
2544 };
2545
2546 drop(active_tx);
2548
2549 let has_returning = !stmt.returning.is_empty();
2551 let mut returning_rows: Vec<Row> = Vec::new();
2552
2553 let schema = table.schema();
2556
2557 let mut has_correlated = if let Some(ref where_clause) = stmt.where_clause {
2560 <Self as MutationHost>::mutation_has_subqueries(where_clause)
2561 && <Self as MutationHost>::mutation_has_correlated_subqueries(where_clause)
2562 } else {
2563 false
2564 };
2565
2566 let (where_expr, needs_memory_filter, memory_where_clause): (
2569 Option<Box<dyn StorageExpr>>,
2570 bool,
2571 Option<Expression>,
2572 ) = if let Some(ref where_clause) = stmt.where_clause {
2573 if has_correlated {
2574 let outer_tables = vec![table_name.to_string()];
2577
2578 let exists_optimized = self
2580 .mutation_optimize_exists_to_semi_join(where_clause, ctx, &outer_tables, None)
2581 .ok()
2582 .flatten();
2583
2584 let expr_for_in = exists_optimized.as_ref().unwrap_or(where_clause.as_ref());
2586 let in_optimized = self
2587 .mutation_optimize_in_to_semi_join(expr_for_in, ctx, &outer_tables)
2588 .ok()
2589 .flatten();
2590
2591 let (current_expr, any_optimized) = match (&exists_optimized, &in_optimized) {
2593 (_, Some(_)) => (in_optimized.unwrap(), true),
2594 (Some(_), None) => (exists_optimized.unwrap(), true),
2595 (None, None) => ((**where_clause).clone(), false),
2596 };
2597
2598 let still_correlated =
2600 <Self as MutationHost>::mutation_has_correlated_subqueries(¤t_expr);
2601
2602 if any_optimized && !still_correlated {
2603 has_correlated = false;
2605 let processed =
2606 if <Self as MutationHost>::mutation_has_subqueries(¤t_expr) {
2607 self.mutation_process_where_subqueries(¤t_expr, ctx)?
2608 } else {
2609 current_expr
2610 };
2611 let plan = pushdown::try_pushdown_plan(&processed, schema, Some(ctx));
2612 let needs_mem = plan.needs_memory_filter();
2613 (plan.storage_expr, needs_mem, plan.residual)
2614 } else {
2615 (None, true, Some(current_expr))
2617 }
2618 } else {
2619 let processed_where =
2620 if <Self as MutationHost>::mutation_has_subqueries(where_clause) {
2621 self.mutation_process_where_subqueries(where_clause, ctx)?
2622 } else {
2623 (**where_clause).clone()
2624 };
2625
2626 let plan = pushdown::try_pushdown_plan(&processed_where, schema, Some(ctx));
2628 let needs_mem = plan.needs_memory_filter();
2629 (plan.storage_expr, needs_mem, plan.residual)
2630 }
2631 } else {
2632 (None, false, None)
2633 };
2634
2635 let referencing_fks = crate::mutation::foreign_key::find_referencing_fks_for_txn(
2637 self.mutation_engine(),
2638 table.txn_id(),
2639 table_name,
2640 );
2641
2642 let column_names_owned = schema.column_names_owned().to_vec();
2644 let column_count = schema.columns.len();
2645 let has_referencing_fks = !referencing_fks.is_empty();
2646
2647 let needs_trigger_rows = triggers.is_some_and(DmlTriggerPlan::has_row_triggers);
2649 let rows_affected =
2650 if needs_memory_filter || has_returning || has_referencing_fks || needs_trigger_rows {
2651 let schema_arc = self.mutation_engine().get_table_schema(table_name)?;
2655
2656 let column_names_with_prefix: Vec<String> = column_names_owned
2660 .iter()
2661 .map(|c| format!("{}.{}", effective_name, c))
2662 .collect();
2663
2664 let mut evaluator =
2666 CompiledEvaluator::new(self.mutation_function_registry()).with_context(ctx);
2667 evaluator.init_columns(&column_names_with_prefix);
2669
2670 let column_indices: Vec<usize> = (0..column_count).collect();
2675 let mut scanner = table.scan(&column_indices, where_expr.as_deref())?;
2676 let mut rows_to_delete: Vec<(i64, Row)> = Vec::new();
2677
2678 let column_names_arc = if has_correlated {
2680 Some(CompactArc::new(column_names_owned.clone()))
2681 } else {
2682 None
2683 };
2684
2685 let column_names_lower = schema.column_names_lower_arc();
2689 #[allow(clippy::type_complexity)]
2690 let col_name_triples: Vec<(
2691 CompactArc<str>,
2692 CompactArc<str>,
2693 Option<CompactArc<str>>,
2694 )> = column_names_lower
2695 .iter()
2696 .map(|col_lower| {
2697 let effective_qualified =
2698 CompactArc::from(format!("{}.{}", effective_name, col_lower).as_str());
2699 let table_qualified = if effective_name != table_name {
2700 Some(CompactArc::from(
2701 format!("{}.{}", table_name, col_lower).as_str(),
2702 ))
2703 } else {
2704 None
2705 };
2706 (
2707 CompactArc::from(col_lower.as_str()),
2708 effective_qualified,
2709 table_qualified,
2710 )
2711 })
2712 .collect();
2713
2714 let estimated_entries = col_name_triples.len() * 3; let mut outer_row_map: FxHashMap<CompactArc<str>, Value> =
2718 FxHashMap::with_capacity_and_hasher(estimated_entries, Default::default());
2719
2720 while scanner.next() {
2721 let row = scanner.row();
2722
2723 let matches = if needs_memory_filter {
2725 evaluator.set_row_array(row);
2726 if let Some(ref where_expr) = memory_where_clause {
2727 if has_correlated {
2728 outer_row_map.clear();
2730 for (i, (col_lower, effective_qualified, table_qualified)) in
2731 col_name_triples.iter().enumerate()
2732 {
2733 if let Some(value) = row.get(i) {
2734 outer_row_map.insert(col_lower.clone(), value.clone());
2735 outer_row_map
2736 .insert(effective_qualified.clone(), value.clone());
2737 if let Some(tq) = table_qualified {
2738 outer_row_map.insert(tq.clone(), value.clone());
2739 }
2740 }
2741 }
2742
2743 let mut correlated_ctx = ctx.with_outer_row(
2745 std::mem::take(&mut outer_row_map),
2746 column_names_arc.clone().unwrap(),
2747 );
2748
2749 let processed = self.mutation_process_correlated_where(
2751 where_expr,
2752 &correlated_ctx,
2753 )?;
2754 evaluator.set_outer_row_owned(
2756 correlated_ctx.take_outer_row().unwrap_or_default(),
2757 );
2758 let result = evaluator.evaluate_bool(&processed)?;
2759 outer_row_map = evaluator.take_outer_row();
2761 result
2762 } else {
2763 evaluator.evaluate_bool(where_expr)?
2764 }
2765 } else {
2766 true
2767 }
2768 } else {
2769 true };
2771
2772 if matches {
2773 rows_to_delete.push((scanner.current_row_id()?, row.clone()));
2774 }
2775 }
2776 drop(scanner);
2778
2779 if let Some(plan) = triggers {
2780 let mut admitted = Vec::with_capacity(rows_to_delete.len());
2781 for (row_id, old_row) in rows_to_delete {
2782 if self
2783 .mutation_fire_before_row_triggers(
2784 plan,
2785 Some(&old_row),
2786 None,
2787 Some(row_id),
2788 ctx,
2789 )?
2790 .is_some()
2791 {
2792 admitted.push((row_id, old_row));
2793 }
2794 }
2795 rows_to_delete = admitted;
2796 }
2797
2798 if has_referencing_fks && !rows_to_delete.is_empty() {
2800 crate::mutation::foreign_key::enforce_delete_actions_iter(
2801 self.mutation_engine(),
2802 table.txn_id(),
2803 table_name,
2804 &schema_arc,
2805 rows_to_delete.iter().map(|(_, row)| row),
2806 &referencing_fks,
2807 )?;
2808 }
2809
2810 let row_ids: Vec<i64> = rows_to_delete.iter().map(|(row_id, _)| *row_id).collect();
2814 let mut deleted_row_ids = Vec::with_capacity(row_ids.len());
2815 let delete_count =
2816 table.delete_candidate_row_ids_collect(&row_ids, None, &mut deleted_row_ids)?;
2817 let needs_deleted_rows =
2818 has_returning || triggers.is_some_and(DmlTriggerPlan::has_after_row_triggers);
2819 if needs_deleted_rows {
2820 let mut rows_by_id: FxHashMap<i64, Row> = rows_to_delete.into_iter().collect();
2821 if has_returning {
2822 returning_rows.reserve(deleted_row_ids.len());
2823 }
2824 for row_id in deleted_row_ids {
2825 if let Some(row) = rows_by_id.remove(&row_id) {
2826 if let Some(plan) = triggers {
2827 self.mutation_fire_after_row_triggers(
2828 plan,
2829 Some(&row),
2830 None,
2831 Some(row_id),
2832 ctx,
2833 )?;
2834 }
2835 if has_returning {
2836 returning_rows.push(row);
2837 }
2838 }
2839 }
2840 }
2841 delete_count
2842 } else {
2843 let row_ids = table.collect_delete_candidate_row_ids(where_expr.as_deref())?;
2849 table.delete_candidate_row_ids(&row_ids, where_expr.as_deref())?
2850 };
2851
2852 if rows_affected > 0 {
2855 self.mutation_invalidate_semantic_cache(table_name);
2856 invalidate_semi_join_cache_for_table(table_name);
2857 invalidate_scalar_subquery_cache_for_table(table_name);
2858 invalidate_in_subquery_cache_for_table(table_name);
2859 }
2860
2861 let returning_result = if has_returning {
2862 Some(build_returning_result(
2863 &stmt.returning,
2864 returning_rows,
2865 &column_names_owned,
2866 ctx,
2867 )?)
2868 } else {
2869 None
2870 };
2871
2872 if should_auto_commit {
2874 if let Some(mut tx) = standalone_tx {
2876 tx.commit()?;
2877 }
2878 }
2879
2880 if let Some(result) = returning_result {
2882 return Ok(result);
2883 }
2884
2885 Ok(Box::new(ExecResult::with_rows_affected(
2886 rows_affected as i64,
2887 )))
2888 }
2889
2890 fn execute_truncate(
2902 &self,
2903 stmt: &TruncateStatement,
2904 _ctx: &ExecutionContext,
2905 ) -> Result<Box<dyn QueryResult>> {
2906 let table_name = &stmt.table_name.value_lower;
2908
2909 let active_tx = self.mutation_active_transaction().lock().unwrap();
2911
2912 let (txn_id, standalone_tx) = if let Some(tx_state) = active_tx.as_ref() {
2913 if tx_state.tables.contains_key(table_name.as_str()) {
2914 return Err(Error::TableHasActiveTransactions);
2915 }
2916 let _ = tx_state.transaction.get_table(table_name)?;
2920 (tx_state.transaction.id(), None)
2921 } else {
2922 let tx = self.mutation_engine().begin_transaction()?;
2923 let txn_id = tx.id();
2924 let _ = tx.get_table(table_name)?;
2925 (txn_id, Some(tx))
2926 };
2927
2928 drop(active_tx);
2930
2931 crate::mutation::foreign_key::check_no_referencing_rows(
2934 self.mutation_engine(),
2935 table_name,
2936 Some(txn_id),
2937 )?;
2938
2939 let rows_affected = self
2940 .mutation_engine()
2941 .truncate_table_under_ddl_fence(table_name, txn_id)?;
2942
2943 self.mutation_invalidate_semantic_cache(table_name);
2947 invalidate_semi_join_cache_for_table(table_name);
2948 invalidate_scalar_subquery_cache_for_table(table_name);
2949 invalidate_in_subquery_cache_for_table(table_name);
2950
2951 if let Some(mut tx) = standalone_tx {
2953 tx.commit()?;
2954 }
2955
2956 Ok(Box::new(ExecResult::with_rows_affected(
2957 rows_affected as i64,
2958 )))
2959 }
2960}
2961
2962impl<T: MutationHost + ?Sized> DmlExecutorExt for T {}