radixdb_executor/mutation/
dml_fast_path.rs

1// Copyright 2026 RadixDB Contributors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Fast-path execution for simple PK-based UPDATE and DELETE operations
16//!
17//! This module provides optimized execution paths for simple DML like:
18//! - `UPDATE table SET col = val WHERE pk = $1`
19//! - `DELETE FROM table WHERE pk = $1`
20//!
21//! By detecting these patterns early and bypassing the full executor overhead
22//! (subquery checking, memory filter setup, expression compilation), we can
23//! reduce per-operation overhead significantly.
24
25use std::sync::RwLock;
26
27use radixdb_core::{CompactArc, SmartString};
28use radixdb_core::{Error, Result, Row, Schema, Value};
29use radixdb_sql::ast::{DeleteStatement, Expression, UpdateStatement};
30use radixdb_storage::expression::{ComparisonExpr, Expression as StorageExpression};
31use radixdb_storage::traits::{Engine, QueryResult};
32
33use crate::compiled_plan::{
34    CompiledExecution, CompiledPkDelete, CompiledPkUpdate, CompiledUpdateColumn, PkValueSource,
35    UpdateValueSource,
36};
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::lookup_key::{integer_pk_admission, IntegerPkAdmission};
42use crate::mutation::host::MutationHost;
43use crate::mutation::validation::{
44    compile_table_check_constraints, validate_resulting_row_constraints,
45};
46use crate::result::ExecResult;
47
48#[doc(hidden)]
49pub trait DmlFastPathExt: MutationHost {
50    /// Try to execute an UPDATE using pre-compiled state
51    fn try_fast_pk_update_compiled(
52        &self,
53        stmt: &UpdateStatement,
54        ctx: &ExecutionContext,
55        compiled: &RwLock<CompiledExecution>,
56    ) -> Option<Result<Box<dyn QueryResult>>> {
57        // Quick reject: explicit transaction (use try_lock for fast rejection)
58        {
59            let active_tx = match self.mutation_active_transaction().try_lock() {
60                Ok(guard) => guard,
61                Err(_) => return None, // Lock contention - fall back to normal path
62            };
63            if active_tx.is_some() {
64                return None;
65            }
66        }
67
68        // Try read lock first - check if already compiled
69        {
70            let compiled_guard = match compiled.read() {
71                Ok(guard) => guard,
72                Err(_) => return None,
73            };
74            match &*compiled_guard {
75                CompiledExecution::NotOptimizable(epoch)
76                    if self.mutation_engine().schema_epoch() == *epoch =>
77                {
78                    return None
79                }
80                CompiledExecution::PkUpdate(update) => {
81                    // Fast validation using schema epoch (~1ns vs ~7ns for HashMap lookup)
82                    if self.mutation_engine().schema_epoch() == update.cached_epoch {
83                        // Fast path: extract PK value and execute
84                        let pk_value =
85                            self.extract_pk_value_from_source(&update.pk_value_source, ctx)?;
86                        return Some(self.execute_compiled_pk_update(update, pk_value, ctx));
87                    }
88                    // Epoch changed - fall through to recompile
89                }
90                CompiledExecution::NotOptimizable(_) | CompiledExecution::Unknown => {} // Epoch changed or first run - fall through to recompile
91                _ => return None, // Different type of compiled execution
92            }
93        }
94
95        // First execution or schema changed - compile and cache
96        self.compile_and_execute_pk_update(stmt, ctx, compiled)
97    }
98
99    /// Try to execute a DELETE using pre-compiled state
100    fn try_fast_pk_delete_compiled(
101        &self,
102        stmt: &DeleteStatement,
103        ctx: &ExecutionContext,
104        compiled: &RwLock<CompiledExecution>,
105    ) -> Option<Result<Box<dyn QueryResult>>> {
106        // Quick reject: explicit transaction (use try_lock for fast rejection)
107        {
108            let active_tx = match self.mutation_active_transaction().try_lock() {
109                Ok(guard) => guard,
110                Err(_) => return None, // Lock contention - fall back to normal path
111            };
112            if active_tx.is_some() {
113                return None;
114            }
115        }
116
117        // Try read lock first - check if already compiled
118        {
119            let compiled_guard = match compiled.read() {
120                Ok(guard) => guard,
121                Err(_) => return None,
122            };
123            match &*compiled_guard {
124                CompiledExecution::NotOptimizable(epoch)
125                    if self.mutation_engine().schema_epoch() == *epoch =>
126                {
127                    return None
128                }
129                CompiledExecution::PkDelete(delete) => {
130                    // Fast validation using schema epoch (~1ns vs ~7ns for HashMap lookup)
131                    if self.mutation_engine().schema_epoch() == delete.cached_epoch {
132                        // Fast path: extract PK value and execute
133                        let pk_value =
134                            self.extract_pk_value_from_source(&delete.pk_value_source, ctx)?;
135                        return Some(self.execute_compiled_pk_delete(delete, pk_value));
136                    }
137                    // Epoch changed - fall through to recompile
138                }
139                CompiledExecution::NotOptimizable(_) | CompiledExecution::Unknown => {} // Epoch changed or first run - fall through to recompile
140                _ => return None, // Different type of compiled execution
141            }
142        }
143
144        // First execution or schema changed - compile and cache
145        self.compile_and_execute_pk_delete(stmt, ctx, compiled)
146    }
147
148    // ============================================================================
149    // HELPER METHODS
150    // ============================================================================
151
152    /// Extract PK equality value from WHERE clause
153    /// Returns (pk_value, pk_source) if WHERE is `pk_col = literal` or `pk_col = $param`
154    fn extract_pk_equality_value(
155        &self,
156        expr: &Expression,
157        pk_column: &str,
158        ctx: &ExecutionContext,
159    ) -> Option<(IntegerPkAdmission, PkValueSource)> {
160        match expr {
161            Expression::Infix(infix) => {
162                if infix.operator != "=" {
163                    return None;
164                }
165
166                // Try column = value pattern
167                if let Some((col, val, source)) =
168                    self.extract_col_eq_val_dml(&infix.left, &infix.right, ctx)
169                {
170                    if col.eq_ignore_ascii_case(pk_column) {
171                        return Some((val, source));
172                    }
173                }
174
175                // Try value = column pattern
176                if let Some((col, val, source)) =
177                    self.extract_col_eq_val_dml(&infix.right, &infix.left, ctx)
178                {
179                    if col.eq_ignore_ascii_case(pk_column) {
180                        return Some((val, source));
181                    }
182                }
183
184                None
185            }
186            _ => None,
187        }
188    }
189
190    /// Extract column name, integer value, and value source from col = val pattern
191    fn extract_col_eq_val_dml(
192        &self,
193        col_expr: &Expression,
194        val_expr: &Expression,
195        ctx: &ExecutionContext,
196    ) -> Option<(String, IntegerPkAdmission, PkValueSource)> {
197        // Get column name
198        let col_name = match col_expr {
199            Expression::Identifier(id) => id.value.to_string(),
200            Expression::QualifiedIdentifier(q) => q.name.value.to_string(),
201            _ => return None,
202        };
203
204        // Get integer value and source
205        let (pk_value, pk_value_source) = match val_expr {
206            Expression::IntegerLiteral(lit) => (
207                IntegerPkAdmission::Exact(lit.value),
208                PkValueSource::Literal(lit.value),
209            ),
210            Expression::FloatLiteral(lit) => {
211                let admission = integer_pk_admission(&Value::Float(lit.value))?;
212                let cached_literal = match admission {
213                    IntegerPkAdmission::Exact(value) => value,
214                    IntegerPkAdmission::NoMatch => 0,
215                };
216                (admission, PkValueSource::Literal(cached_literal))
217            }
218            Expression::Parameter(param) => {
219                // Named parameters (e.g., :name) resolve via get_named_param() at execution time
220                // Positional parameters ($1, $2, ...) are 1-indexed, array is 0-indexed
221                if param.name.starts_with(':') {
222                    let name = &param.name[1..];
223                    let value = ctx.get_named_param(name)?;
224                    let pk_value = integer_pk_admission(value)?;
225                    (
226                        pk_value,
227                        PkValueSource::NamedParameter(SmartString::new(name)),
228                    )
229                } else {
230                    let params = ctx.params();
231                    let param_idx = if param.index > 0 {
232                        param.index - 1
233                    } else {
234                        return None;
235                    };
236                    if param_idx >= params.len() {
237                        return None;
238                    }
239                    let pk_value = integer_pk_admission(&params[param_idx])?;
240                    (pk_value, PkValueSource::Parameter(param_idx))
241                }
242            }
243            _ => return None,
244        };
245
246        Some((col_name, pk_value, pk_value_source))
247    }
248
249    /// Extract PK value from pre-compiled source
250    fn extract_pk_value_from_source(
251        &self,
252        source: &PkValueSource,
253        ctx: &ExecutionContext,
254    ) -> Option<IntegerPkAdmission> {
255        match source {
256            PkValueSource::NamedParameter(name) => integer_pk_admission(ctx.get_named_param(name)?),
257            _ => Self::extract_pk_value_from_params(source, ctx.params()),
258        }
259    }
260
261    /// Extract PK value from params slice directly (avoids ExecutionContext overhead)
262    #[inline]
263    fn extract_pk_value_from_params(
264        source: &PkValueSource,
265        params: &[Value],
266    ) -> Option<IntegerPkAdmission> {
267        match source {
268            PkValueSource::Literal(v) => Some(IntegerPkAdmission::Exact(*v)),
269            PkValueSource::Parameter(idx) => {
270                if *idx >= params.len() {
271                    return None;
272                }
273                integer_pk_admission(&params[*idx])
274            }
275            PkValueSource::NamedParameter(_) => None, // No ctx available in slice path
276        }
277    }
278
279    /// Extract update value from params slice directly
280    #[inline]
281    fn extract_update_value_from_slice(
282        source: &UpdateValueSource,
283        params: &[Value],
284    ) -> Option<Value> {
285        match source {
286            UpdateValueSource::Literal(v) => Some(v.clone()),
287            UpdateValueSource::Parameter(idx) => params.get(*idx).cloned(),
288            UpdateValueSource::NamedParameter(_) => None, // No ctx available in slice path
289        }
290    }
291
292    /// Try fast PK update with borrowed params slice (avoids Arc allocation)
293    fn try_fast_pk_update_with_params(
294        &self,
295        _stmt: &UpdateStatement,
296        params: &[Value],
297        compiled: &RwLock<CompiledExecution>,
298    ) -> Option<Result<Box<dyn QueryResult>>> {
299        // Try read lock first - check if already compiled
300        let compiled_guard = compiled.read().ok()?;
301        match &*compiled_guard {
302            CompiledExecution::NotOptimizable(_) => None,
303            CompiledExecution::PkUpdate(update) => {
304                // Fast validation using schema epoch
305                if self.mutation_engine().schema_epoch() == update.cached_epoch {
306                    let pk_value =
307                        Self::extract_pk_value_from_params(&update.pk_value_source, params)?;
308                    let IntegerPkAdmission::Exact(pk_value) = pk_value else {
309                        return Some(Ok(Box::new(ExecResult::with_rows_affected(0))));
310                    };
311                    // Extract update values
312                    let mut updates = Vec::with_capacity(update.updates.len());
313                    for u in &update.updates {
314                        let value = Self::extract_update_value_from_slice(&u.value_source, params)?;
315                        let coerced = match value.try_coerce_to_type(u.column_type) {
316                            Ok(value) => value,
317                            Err(error) => return Some(Err(error)),
318                        };
319                        updates.push((u.column_idx, coerced));
320                    }
321                    // Clone only what we need (cheap: SmartString + Arc)
322                    let table_name = update.table_name.clone();
323                    let pk_column_name = update.pk_column_name.clone();
324                    let schema = update.schema.clone();
325                    drop(compiled_guard);
326                    return Some(self.execute_pk_update_minimal(
327                        &table_name,
328                        &pk_column_name,
329                        &schema,
330                        pk_value,
331                        updates,
332                    ));
333                }
334                None // Epoch changed, use normal path
335            }
336            CompiledExecution::Unknown => None,
337            _ => None,
338        }
339    }
340
341    /// Try fast PK delete with borrowed params slice (avoids Arc allocation)
342    fn try_fast_pk_delete_with_params(
343        &self,
344        _stmt: &DeleteStatement,
345        params: &[Value],
346        compiled: &RwLock<CompiledExecution>,
347    ) -> Option<Result<Box<dyn QueryResult>>> {
348        // Try read lock first - check if already compiled
349        let compiled_guard = compiled.read().ok()?;
350        match &*compiled_guard {
351            CompiledExecution::NotOptimizable(_) => None,
352            CompiledExecution::PkDelete(delete) => {
353                // Fast validation using schema epoch
354                if self.mutation_engine().schema_epoch() == delete.cached_epoch {
355                    let pk_value =
356                        Self::extract_pk_value_from_params(&delete.pk_value_source, params)?;
357                    let IntegerPkAdmission::Exact(pk_value) = pk_value else {
358                        return Some(Ok(Box::new(ExecResult::with_rows_affected(0))));
359                    };
360                    // Clone only what we need (cheap: SmartString + Arc)
361                    let table_name = delete.table_name.clone();
362                    let pk_column_name = delete.pk_column_name.clone();
363                    let schema = delete.schema.clone();
364                    drop(compiled_guard);
365                    return Some(self.execute_pk_delete_minimal(
366                        &table_name,
367                        &pk_column_name,
368                        &schema,
369                        pk_value,
370                    ));
371                }
372                None // Epoch changed, use normal path
373            }
374            CompiledExecution::Unknown => None,
375            _ => None,
376        }
377    }
378
379    /// Execute PK update with minimal data (avoids cloning CompiledPkUpdate)
380    fn execute_pk_update_minimal(
381        &self,
382        table_name: &str,
383        pk_column_name: &str,
384        schema: &CompactArc<Schema>,
385        pk_value: i64,
386        updates: Vec<(usize, Value)>,
387    ) -> Result<Box<dyn QueryResult>> {
388        // Create auto-commit transaction
389        let tx = self.mutation_engine().begin_transaction()?;
390        let mut table = tx.get_table(table_name)?;
391
392        // Build WHERE expression for PK lookup
393        let mut pk_expr = ComparisonExpr::new(
394            pk_column_name,
395            radixdb_core::Operator::Eq,
396            Value::Integer(pk_value),
397        );
398        pk_expr.prepare_for_schema(schema);
399
400        let compiled_table_checks = compile_table_check_constraints(schema)?;
401        let mut table_check_vm = crate::expression::ExprVM::new();
402
403        // Execute update with simple setter
404        let mut setter = |mut row: Row| -> Result<(Row, bool)> {
405            for (idx, new_value) in &updates {
406                let _ = row.set(*idx, new_value.clone());
407            }
408            validate_resulting_row_constraints(
409                schema,
410                &compiled_table_checks,
411                &row,
412                &mut table_check_vm,
413            )?;
414            Ok((row, true))
415        };
416
417        let rows_affected = table.update(Some(&pk_expr), &mut setter)?;
418
419        // Invalidate caches
420        if rows_affected > 0 {
421            self.mutation_invalidate_semantic_cache(table_name);
422            invalidate_semi_join_cache_for_table(table_name);
423            invalidate_scalar_subquery_cache_for_table(table_name);
424            invalidate_in_subquery_cache_for_table(table_name);
425        }
426
427        // Commit
428        drop(table);
429        let mut tx = tx;
430        tx.commit()?;
431
432        Ok(Box::new(ExecResult::with_rows_affected(
433            rows_affected as i64,
434        )))
435    }
436
437    /// Execute PK delete with minimal data (avoids cloning CompiledPkDelete)
438    fn execute_pk_delete_minimal(
439        &self,
440        table_name: &str,
441        pk_column_name: &str,
442        schema: &CompactArc<Schema>,
443        pk_value: i64,
444    ) -> Result<Box<dyn QueryResult>> {
445        // Create auto-commit transaction
446        let tx = self.mutation_engine().begin_transaction()?;
447        let mut table = tx.get_table(table_name)?;
448
449        // Build WHERE expression for PK lookup
450        let mut pk_expr = ComparisonExpr::new(
451            pk_column_name,
452            radixdb_core::Operator::Eq,
453            Value::Integer(pk_value),
454        );
455        pk_expr.prepare_for_schema(schema);
456
457        // Execute delete
458        let rows_affected = table.delete(Some(&pk_expr))?;
459
460        // Invalidate caches
461        if rows_affected > 0 {
462            self.mutation_invalidate_semantic_cache(table_name);
463            invalidate_semi_join_cache_for_table(table_name);
464            invalidate_scalar_subquery_cache_for_table(table_name);
465            invalidate_in_subquery_cache_for_table(table_name);
466        }
467
468        // Commit
469        drop(table);
470        let mut tx = tx;
471        tx.commit()?;
472
473        Ok(Box::new(ExecResult::with_rows_affected(
474            rows_affected as i64,
475        )))
476    }
477
478    // ============================================================================
479    // EXECUTION METHODS
480    // ============================================================================
481
482    /// Execute a compiled PK update (extracts values from ctx then delegates to core impl)
483    fn execute_compiled_pk_update(
484        &self,
485        compiled: &CompiledPkUpdate,
486        pk_value: IntegerPkAdmission,
487        ctx: &ExecutionContext,
488    ) -> Result<Box<dyn QueryResult>> {
489        let IntegerPkAdmission::Exact(pk_value) = pk_value else {
490            return Ok(Box::new(ExecResult::with_rows_affected(0)));
491        };
492        // Extract values from compiled sources
493        let mut updates = Vec::with_capacity(compiled.updates.len());
494        for u in &compiled.updates {
495            let value = match &u.value_source {
496                UpdateValueSource::Literal(v) => v.clone(),
497                UpdateValueSource::Parameter(idx) => {
498                    let params = ctx.params();
499                    params.get(*idx).cloned().ok_or_else(|| {
500                        Error::InvalidArgument(format!("missing positional parameter ${}", idx + 1))
501                    })?
502                }
503                UpdateValueSource::NamedParameter(name) => {
504                    ctx.get_named_param(name).cloned().ok_or_else(|| {
505                        Error::InvalidArgument(format!("missing named parameter :{name}"))
506                    })?
507                }
508            };
509            updates.push((u.column_idx, value.try_coerce_to_type(u.column_type)?));
510        }
511
512        self.execute_pk_update_minimal(
513            &compiled.table_name,
514            &compiled.pk_column_name,
515            &compiled.schema,
516            pk_value,
517            updates,
518        )
519    }
520
521    /// Execute a compiled PK delete (delegates to core impl)
522    fn execute_compiled_pk_delete(
523        &self,
524        compiled: &CompiledPkDelete,
525        pk_value: IntegerPkAdmission,
526    ) -> Result<Box<dyn QueryResult>> {
527        let IntegerPkAdmission::Exact(pk_value) = pk_value else {
528            return Ok(Box::new(ExecResult::with_rows_affected(0)));
529        };
530        self.execute_pk_delete_minimal(
531            &compiled.table_name,
532            &compiled.pk_column_name,
533            &compiled.schema,
534            pk_value,
535        )
536    }
537
538    // ============================================================================
539    // COMPILE AND EXECUTE METHODS
540    // ============================================================================
541
542    /// Compile and execute a PK update, caching the compiled state
543    fn compile_and_execute_pk_update(
544        &self,
545        stmt: &UpdateStatement,
546        ctx: &ExecutionContext,
547        compiled: &RwLock<CompiledExecution>,
548    ) -> Option<Result<Box<dyn QueryResult>>> {
549        // Acquire write lock
550        let mut compiled_guard = match compiled.write() {
551            Ok(guard) => guard,
552            Err(_) => return None,
553        };
554
555        // Double-check after acquiring lock
556        match &*compiled_guard {
557            CompiledExecution::NotOptimizable(epoch)
558                if self.mutation_engine().schema_epoch() == *epoch =>
559            {
560                return None
561            }
562            CompiledExecution::PkUpdate(update) => {
563                let pk_value = self.extract_pk_value_from_source(&update.pk_value_source, ctx)?;
564                return Some(self.execute_compiled_pk_update(update, pk_value, ctx));
565            }
566            CompiledExecution::NotOptimizable(_) | CompiledExecution::Unknown => {} // Epoch changed or first run - recompile
567            _ => return None,
568        }
569
570        // Validate pattern
571        let where_clause = stmt.where_clause.as_ref()?;
572        if !stmt.returning.is_empty() {
573            *compiled_guard =
574                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
575            return None;
576        }
577
578        let table_name = &stmt.table_name.value_lower;
579        let schema = match self.mutation_engine().get_table_schema(table_name) {
580            Ok(s) => s,
581            Err(_) => {
582                *compiled_guard =
583                    CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
584                return None;
585            }
586        };
587
588        let pk_indices = schema.primary_key_indices();
589        if pk_indices.len() != 1 {
590            *compiled_guard =
591                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
592            return None;
593        }
594        let pk_idx = pk_indices[0];
595        if schema.columns[pk_idx].data_type != radixdb_core::DataType::Integer {
596            *compiled_guard =
597                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
598            return None;
599        }
600        let pk_column = &schema.columns[pk_idx].name;
601
602        // Reject UPDATE on primary key column (row_id == pk_value invariant)
603        {
604            let col_map = schema.column_index_map();
605            for col_name in stmt.updates.keys() {
606                let col_lower = col_name.to_lowercase();
607                if col_map.get(col_lower.as_str()).copied() == Some(pk_idx) {
608                    return Some(Err(radixdb_core::Error::invalid_argument(format!(
609                        "cannot UPDATE primary key column '{}'. Use DELETE + INSERT instead",
610                        pk_column
611                    ))));
612                }
613            }
614        }
615
616        // Bail if table has FK constraints (child table) or is referenced by other tables (parent table)
617        // FK enforcement requires cross-table lookups — fall back to normal path
618        let referencing_fks = self.mutation_active_transaction_id().map_or_else(
619            || {
620                crate::mutation::foreign_key::find_referencing_fks(
621                    self.mutation_engine(),
622                    table_name,
623                )
624            },
625            |txn_id| {
626                crate::mutation::foreign_key::find_referencing_fks_for_txn(
627                    self.mutation_engine(),
628                    txn_id,
629                    table_name,
630                )
631            },
632        );
633        if !schema.foreign_keys.is_empty() || !referencing_fks.is_empty() {
634            *compiled_guard =
635                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
636            return None;
637        }
638
639        // Bail if any column being updated has a CHECK constraint
640        // CHECK validation requires expression evaluation — fall back to normal path
641        {
642            let col_map = schema.column_index_map();
643            for col_name in stmt.updates.keys() {
644                let col_lower = col_name.to_lowercase();
645                if let Some(&idx) = col_map.get(col_lower.as_str()) {
646                    if schema.columns[idx].check_expr.is_some() {
647                        *compiled_guard = CompiledExecution::NotOptimizable(
648                            self.mutation_engine().schema_epoch(),
649                        );
650                        return None;
651                    }
652                }
653            }
654        }
655
656        // Extract PK value source
657        let (pk_value, pk_source) =
658            match self.extract_pk_equality_value(where_clause, pk_column, ctx) {
659                Some(v) => v,
660                None => {
661                    *compiled_guard =
662                        CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
663                    return None;
664                }
665            };
666
667        // Extract update value sources
668        let col_map = schema.column_index_map();
669        let mut compiled_updates = Vec::with_capacity(stmt.updates.len());
670        for (col_name, expr) in &stmt.updates {
671            let col_lower = col_name.to_lowercase();
672            let col_idx = match col_map.get(col_lower.as_str()) {
673                Some(&idx) => idx,
674                None => {
675                    *compiled_guard =
676                        CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
677                    return None;
678                }
679            };
680            let col_type = schema.columns[col_idx].data_type;
681
682            let value_source = match self.extract_value_source(expr) {
683                Some(s) => s,
684                None => {
685                    *compiled_guard =
686                        CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
687                    return None;
688                }
689            };
690
691            compiled_updates.push(CompiledUpdateColumn {
692                column_idx: col_idx,
693                column_type: col_type,
694                value_source,
695            });
696        }
697
698        // Build compiled state
699        if pk_value == IntegerPkAdmission::NoMatch
700            && matches!(&pk_source, PkValueSource::Literal(_))
701        {
702            drop(compiled_guard);
703            return Some(Ok(Box::new(ExecResult::with_rows_affected(0))));
704        }
705        let compiled_update = CompiledPkUpdate {
706            table_name: SmartString::new(table_name),
707            schema: CompactArc::new((*schema).clone()),
708            pk_column_name: SmartString::new(pk_column),
709            pk_value_source: pk_source,
710            updates: compiled_updates,
711            cached_epoch: self.mutation_engine().schema_epoch(),
712        };
713
714        *compiled_guard = CompiledExecution::PkUpdate(compiled_update.clone());
715        drop(compiled_guard);
716
717        // Execute
718        Some(self.execute_compiled_pk_update(&compiled_update, pk_value, ctx))
719    }
720
721    /// Compile and execute a PK delete, caching the compiled state
722    fn compile_and_execute_pk_delete(
723        &self,
724        stmt: &DeleteStatement,
725        ctx: &ExecutionContext,
726        compiled: &RwLock<CompiledExecution>,
727    ) -> Option<Result<Box<dyn QueryResult>>> {
728        // Acquire write lock
729        let mut compiled_guard = match compiled.write() {
730            Ok(guard) => guard,
731            Err(_) => return None,
732        };
733
734        // Double-check after acquiring lock
735        match &*compiled_guard {
736            CompiledExecution::NotOptimizable(epoch)
737                if self.mutation_engine().schema_epoch() == *epoch =>
738            {
739                return None
740            }
741            CompiledExecution::PkDelete(delete) => {
742                let pk_value = self.extract_pk_value_from_source(&delete.pk_value_source, ctx)?;
743                return Some(self.execute_compiled_pk_delete(delete, pk_value));
744            }
745            CompiledExecution::NotOptimizable(_) | CompiledExecution::Unknown => {} // Epoch changed or first run - recompile
746            _ => return None,
747        }
748
749        // Validate pattern
750        let where_clause = stmt.where_clause.as_ref()?;
751        if !stmt.returning.is_empty() {
752            *compiled_guard =
753                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
754            return None;
755        }
756
757        let table_name = &stmt.table_name.value_lower;
758        let schema = match self.mutation_engine().get_table_schema(table_name) {
759            Ok(s) => s,
760            Err(_) => {
761                *compiled_guard =
762                    CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
763                return None;
764            }
765        };
766
767        let pk_indices = schema.primary_key_indices();
768        if pk_indices.len() != 1 {
769            *compiled_guard =
770                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
771            return None;
772        }
773        let pk_idx = pk_indices[0];
774        if schema.columns[pk_idx].data_type != radixdb_core::DataType::Integer {
775            *compiled_guard =
776                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
777            return None;
778        }
779        let pk_column = &schema.columns[pk_idx].name;
780
781        // Bail if this table is referenced by child tables (FK enforcement needed)
782        let referencing_fks = self.mutation_active_transaction_id().map_or_else(
783            || {
784                crate::mutation::foreign_key::find_referencing_fks(
785                    self.mutation_engine(),
786                    table_name,
787                )
788            },
789            |txn_id| {
790                crate::mutation::foreign_key::find_referencing_fks_for_txn(
791                    self.mutation_engine(),
792                    txn_id,
793                    table_name,
794                )
795            },
796        );
797        if !referencing_fks.is_empty() {
798            *compiled_guard =
799                CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
800            return None;
801        }
802
803        // Extract PK value source
804        let (pk_value, pk_source) =
805            match self.extract_pk_equality_value(where_clause, pk_column, ctx) {
806                Some(v) => v,
807                None => {
808                    *compiled_guard =
809                        CompiledExecution::NotOptimizable(self.mutation_engine().schema_epoch());
810                    return None;
811                }
812            };
813
814        // Build compiled state
815        if pk_value == IntegerPkAdmission::NoMatch
816            && matches!(&pk_source, PkValueSource::Literal(_))
817        {
818            drop(compiled_guard);
819            return Some(Ok(Box::new(ExecResult::with_rows_affected(0))));
820        }
821        let compiled_delete = CompiledPkDelete {
822            table_name: SmartString::new(table_name),
823            schema: CompactArc::new((*schema).clone()),
824            pk_column_name: SmartString::new(pk_column),
825            pk_value_source: pk_source,
826            cached_epoch: self.mutation_engine().schema_epoch(),
827        };
828
829        *compiled_guard = CompiledExecution::PkDelete(compiled_delete.clone());
830        drop(compiled_guard);
831
832        // Execute
833        Some(self.execute_compiled_pk_delete(&compiled_delete, pk_value))
834    }
835
836    /// Extract value source (literal or parameter) from expression
837    fn extract_value_source(&self, expr: &Expression) -> Option<UpdateValueSource> {
838        match expr {
839            Expression::IntegerLiteral(lit) => {
840                Some(UpdateValueSource::Literal(Value::Integer(lit.value)))
841            }
842            Expression::FloatLiteral(lit) => {
843                Some(UpdateValueSource::Literal(Value::Float(lit.value)))
844            }
845            Expression::StringLiteral(lit) => {
846                Some(UpdateValueSource::Literal(Value::text(lit.value.as_str())))
847            }
848            Expression::BooleanLiteral(lit) => {
849                Some(UpdateValueSource::Literal(Value::Boolean(lit.value)))
850            }
851            Expression::NullLiteral(_) => Some(UpdateValueSource::Literal(Value::null_unknown())),
852            Expression::Prefix(prefix) if prefix.operator == "-" => match prefix.right.as_ref() {
853                Expression::IntegerLiteral(lit) => {
854                    Some(UpdateValueSource::Literal(Value::Integer(-lit.value)))
855                }
856                Expression::FloatLiteral(lit) => {
857                    Some(UpdateValueSource::Literal(Value::Float(-lit.value)))
858                }
859                _ => None,
860            },
861            Expression::Parameter(param) => {
862                if param.name.starts_with(':') {
863                    let name = &param.name[1..];
864                    Some(UpdateValueSource::NamedParameter(SmartString::new(name)))
865                } else {
866                    let param_idx = if param.index > 0 {
867                        param.index - 1
868                    } else {
869                        return None;
870                    };
871                    Some(UpdateValueSource::Parameter(param_idx))
872                }
873            }
874            _ => None, // Complex expression
875        }
876    }
877}
878
879impl<T: MutationHost + ?Sized> DmlFastPathExt for T {}
radixdb_executor/mutation/dml_fast_path.rs

radixdb_executor/mutation/
dml_fast_path.rs
