triggers: add capability for DeletePlan to write the write set into a RowSet first

This is needed for safe DELETE when there are DELETE triggers on the affected
table.

core/translate/delete.rs

@@ -2,13 +2,16 @@ use crate::schema::{Schema, Table};
 use crate::translate::emitter::{emit_program, Resolver};
 use crate::translate::expr::process_returning_clause;
 use crate::translate::optimizer::optimize_plan;
-use crate::translate::plan::{DeletePlan, Operation, Plan};
+use crate::translate::plan::{
+    DeletePlan, JoinOrderMember, Operation, Plan, QueryDestination, ResultSetColumn, SelectPlan,
+};
 use crate::translate::planner::{parse_limit, parse_where};
+use crate::translate::trigger_exec::has_relevant_triggers_type_only;
 use crate::util::normalize_ident;
 use crate::vdbe::builder::{ProgramBuilder, ProgramBuilderOpts};
 use crate::Result;
 use std::sync::Arc;
-use turso_parser::ast::{Expr, Limit, QualifiedName, ResultColumn};
+use turso_parser::ast::{Expr, Limit, QualifiedName, ResultColumn, TriggerEvent};
 
 use super::plan::{ColumnUsedMask, JoinedTable, TableReferences};
 
@@ -93,6 +96,8 @@ pub fn prepare_delete_plan(
         );
     }
 
+    let btree_table_for_triggers = table.btree();
+
     let table = if let Some(table) = table.virtual_table() {
         Table::Virtual(table.clone())
     } else if let Some(table) = table.btree() {
@@ -130,18 +135,84 @@ pub fn prepare_delete_plan(
     let (resolved_limit, resolved_offset) =
         limit.map_or(Ok((None, None)), |l| parse_limit(l, connection))?;
 
-    let plan = DeletePlan {
-        table_references,
-        result_columns,
-        where_clause: where_predicates,
-        order_by: vec![],
-        limit: resolved_limit,
-        offset: resolved_offset,
-        contains_constant_false_condition: false,
-        indexes,
-    };
+    // Check if there are DELETE triggers. If so, we need to materialize the write set into a RowSet first.
+    // This is done in SQLite for all DELETE triggers on the affected table even if the trigger would not have an impact
+    // on the target table -- presumably due to lack of static analysis capabilities to determine whether it's safe
+    // to skip the rowset materialization.
+    let has_delete_triggers = btree_table_for_triggers
+        .as_ref()
+        .map(|bt| has_relevant_triggers_type_only(schema, TriggerEvent::Delete, None, bt))
+        .unwrap_or(false);
 
-    Ok(Plan::Delete(plan))
+    if has_delete_triggers {
+        // Create a SelectPlan that materializes rowids into a RowSet
+        let rowid_internal_id = table_references
+            .joined_tables()
+            .first()
+            .unwrap()
+            .internal_id;
+        let rowset_reg = program.alloc_register();
+
+        let rowset_plan = SelectPlan {
+            table_references: table_references.clone(),
+            result_columns: vec![ResultSetColumn {
+                expr: Expr::RowId {
+                    database: None,
+                    table: rowid_internal_id,
+                },
+                alias: None,
+                contains_aggregates: false,
+            }],
+            where_clause: where_predicates,
+            group_by: None,
+            order_by: vec![],
+            aggregates: vec![],
+            limit: resolved_limit,
+            query_destination: QueryDestination::RowSet { rowset_reg },
+            join_order: table_references
+                .joined_tables()
+                .iter()
+                .enumerate()
+                .map(|(i, t)| JoinOrderMember {
+                    table_id: t.internal_id,
+                    original_idx: i,
+                    is_outer: false,
+                })
+                .collect(),
+            offset: resolved_offset,
+            contains_constant_false_condition: false,
+            distinctness: super::plan::Distinctness::NonDistinct,
+            values: vec![],
+            window: None,
+            non_from_clause_subqueries: vec![],
+        };
+
+        Ok(Plan::Delete(DeletePlan {
+            table_references,
+            result_columns,
+            where_clause: vec![],
+            order_by: vec![],
+            limit: None,
+            offset: None,
+            contains_constant_false_condition: false,
+            indexes,
+            rowset_plan: Some(rowset_plan),
+            rowset_reg: Some(rowset_reg),
+        }))
+    } else {
+        Ok(Plan::Delete(DeletePlan {
+            table_references,
+            result_columns,
+            where_clause: where_predicates,
+            order_by: vec![],
+            limit: resolved_limit,
+            offset: resolved_offset,
+            contains_constant_false_condition: false,
+            indexes,
+            rowset_plan: None,
+            rowset_reg: None,
+        }))
+    }
 }
 
 fn estimate_num_instructions(plan: &DeletePlan) -> usize {

core/translate/plan.rs

@@ -259,6 +259,12 @@ pub enum QueryDestination {
         /// The number of registers that hold the result of the subquery.
         num_regs: usize,
     },
+    /// The results of the query are stored in a RowSet (for DELETE operations with triggers).
+    /// Rowids are added to the RowSet using RowSetAdd, then read back using RowSetRead.
+    RowSet {
+        /// The register that holds the RowSet object.
+        rowset_reg: usize,
+    },
     /// Decision made at some point after query plan construction.
     Unset,
 }
@@ -474,6 +480,11 @@ pub struct DeletePlan {
     pub contains_constant_false_condition: bool,
     /// Indexes that must be updated by the delete operation.
     pub indexes: Vec<Arc<Index>>,
+    /// If there are DELETE triggers, materialize rowids into a RowSet first.
+    /// This ensures triggers see a stable set of rows to delete.
+    pub rowset_plan: Option<SelectPlan>,
+    /// Register ID for the RowSet (if rowset_plan is Some)
+    pub rowset_reg: Option<usize>,
 }
 
 #[derive(Debug, Clone)]

core/translate/result_row.rs

@@ -160,6 +160,18 @@ pub fn emit_result_row_and_limit(
                 extra_amount: num_regs - 1,
             });
         }
+        QueryDestination::RowSet { rowset_reg } => {
+            // For RowSet, we add the rowid (which should be the only result column) to the RowSet
+            assert_eq!(
+                plan.result_columns.len(),
+                1,
+                "RowSet should only have one result column (rowid)"
+            );
+            program.emit_insn(Insn::RowSetAdd {
+                rowset_reg: *rowset_reg,
+                value_reg: result_columns_start_reg,
+            });
+        }
         QueryDestination::Unset => unreachable!("Unset query destination should not be reached"),
     }
 

core/translate/values.rs

@@ -34,6 +34,9 @@ pub fn emit_values(
         QueryDestination::RowValueSubqueryResult { .. } => {
             emit_toplevel_values(program, plan, t_ctx)?
         }
+        QueryDestination::RowSet { .. } => {
+            unreachable!("RowSet query destination should not be used in values emission")
+        }
         QueryDestination::Unset => unreachable!("Unset query destination should not be reached"),
     };
     Ok(reg_result_cols_start)
@@ -212,6 +215,9 @@ fn emit_values_to_destination(
                 extra_amount: num_regs - 1,
             });
         }
+        QueryDestination::RowSet { .. } => {
+            unreachable!("RowSet query destination should not be used in values emission")
+        }
         QueryDestination::Unset => unreachable!("Unset query destination should not be reached"),
     }
 }

core/vdbe/builder.rs

@@ -7,11 +7,16 @@ use tracing::{instrument, Level};
 use turso_parser::ast::{self, ResolveType, TableInternalId};
 
 use crate::{
-    CaptureDataChangesMode, Connection, Value, VirtualTable, index_method::IndexMethodAttachment, numeric::Numeric, parameters::Parameters, schema::{BTreeTable, Index, PseudoCursorType, Schema, Table, Trigger}, translate::{
+    index_method::IndexMethodAttachment,
+    numeric::Numeric,
+    parameters::Parameters,
+    schema::{BTreeTable, Index, PseudoCursorType, Schema, Table, Trigger},
+    translate::{
         collate::CollationSeq,
         emitter::TransactionMode,
         plan::{ResultSetColumn, TableReferences},
-    }
+    },
+    CaptureDataChangesMode, Connection, Value, VirtualTable,
 };
 
 #[derive(Default)]

core/vdbe/execute.rs

@@ -39,7 +39,9 @@ use crate::{
     },
     translate::emitter::TransactionMode,
 };
-use crate::{CheckpointMode, Completion, Connection, DatabaseStorage, MvCursor, StepResult, get_cursor};
+use crate::{
+    get_cursor, CheckpointMode, Completion, Connection, DatabaseStorage, MvCursor, StepResult,
+};
 use either::Either;
 use std::any::Any;
 use std::env::temp_dir;

core/vdbe/insn.rs

@@ -5,7 +5,12 @@ use std::{
 
 use super::{execute, AggFunc, BranchOffset, CursorID, FuncCtx, InsnFunction, PageIdx};
 use crate::{
-    Statement, Value, schema::{BTreeTable, Column, Index}, storage::{pager::CreateBTreeFlags, wal::CheckpointMode}, translate::{collate::CollationSeq, emitter::TransactionMode}, types::KeyInfo, vdbe::affinity::Affinity
+    schema::{BTreeTable, Column, Index},
+    storage::{pager::CreateBTreeFlags, wal::CheckpointMode},
+    translate::{collate::CollationSeq, emitter::TransactionMode},
+    types::KeyInfo,
+    vdbe::affinity::Affinity,
+    Statement, Value,
 };
 use parking_lot::RwLock;
 use strum::EnumCount;

core/vdbe/mod.rs

@@ -29,12 +29,24 @@ pub mod sorter;
 pub mod value;
 
 use crate::{
-    ValueRef, error::LimboError, function::{AggFunc, FuncCtx}, mvcc::{LocalClock, database::CommitStateMachine}, return_if_io, schema::Trigger, state_machine::StateMachine, storage::{pager::PagerCommitResult, sqlite3_ondisk::SmallVec}, translate::{collate::CollationSeq, plan::TableReferences}, types::{IOCompletions, IOResult}, vdbe::{
+    error::LimboError,
+    function::{AggFunc, FuncCtx},
+    mvcc::{database::CommitStateMachine, LocalClock},
+    return_if_io,
+    schema::Trigger,
+    state_machine::StateMachine,
+    storage::{pager::PagerCommitResult, sqlite3_ondisk::SmallVec},
+    translate::{collate::CollationSeq, plan::TableReferences},
+    types::{IOCompletions, IOResult},
+    vdbe::{
         execute::{
-            OpCheckpointState, OpColumnState, OpDeleteState, OpDeleteSubState, OpDestroyState, OpIdxInsertState, OpInsertState, OpInsertSubState, OpNewRowidState, OpNoConflictState, OpProgramState, OpRowIdState, OpSeekState, OpTransactionState
+            OpCheckpointState, OpColumnState, OpDeleteState, OpDeleteSubState, OpDestroyState,
+            OpIdxInsertState, OpInsertState, OpInsertSubState, OpNewRowidState, OpNoConflictState,
+            OpProgramState, OpRowIdState, OpSeekState, OpTransactionState,
         },
         metrics::StatementMetrics,
-    }
+    },
+    ValueRef,
 };
 
 use crate::{