test/fuzz: add UPDATE/DELETE fuzz test

Our current UPDATE/DELETE fuzz tests are coupled to the btree and do
not exercise the VDBE code paths at all, so a separate one makes sense.

This test repeats the following:

- Creates one table with n columns
- Creates (0..=n) indexes
- Executes UPDATE/DELETE statements
- Asserts that both sqlite and tursodb have the same DB state after each stmt

tests/integration/common.rs

@@ -208,6 +208,45 @@ pub(crate) fn limbo_exec_rows(
     rows
 }
 
+pub(crate) fn limbo_exec_rows_fallible(
+    _db: &TempDatabase,
+    conn: &Arc<turso_core::Connection>,
+    query: &str,
+) -> Result<Vec<Vec<rusqlite::types::Value>>, turso_core::LimboError> {
+    let mut stmt = conn.prepare(query)?;
+    let mut rows = Vec::new();
+    'outer: loop {
+        let row = loop {
+            let result = stmt.step()?;
+            match result {
+                turso_core::StepResult::Row => {
+                    let row = stmt.row().unwrap();
+                    break row;
+                }
+                turso_core::StepResult::IO => {
+                    stmt.run_once()?;
+                    continue;
+                }
+
+                turso_core::StepResult::Done => break 'outer,
+                r => panic!("unexpected result {r:?}: expecting single row"),
+            }
+        };
+        let row = row
+            .get_values()
+            .map(|x| match x {
+                turso_core::Value::Null => rusqlite::types::Value::Null,
+                turso_core::Value::Integer(x) => rusqlite::types::Value::Integer(*x),
+                turso_core::Value::Float(x) => rusqlite::types::Value::Real(*x),
+                turso_core::Value::Text(x) => rusqlite::types::Value::Text(x.as_str().to_string()),
+                turso_core::Value::Blob(x) => rusqlite::types::Value::Blob(x.to_vec()),
+            })
+            .collect();
+        rows.push(row);
+    }
+    Ok(rows)
+}
+
 pub(crate) fn limbo_exec_rows_error(
     _db: &TempDatabase,
     conn: &Arc<turso_core::Connection>,

tests/integration/fuzz/mod.rs

@@ -10,7 +10,10 @@ mod tests {
     use rusqlite::{params, types::Value};
 
     use crate::{
-        common::{limbo_exec_rows, rng_from_time, sqlite_exec_rows, TempDatabase},
+        common::{
+            limbo_exec_rows, limbo_exec_rows_fallible, rng_from_time, sqlite_exec_rows,
+            TempDatabase,
+        },
         fuzz::grammar_generator::{const_str, rand_int, rand_str, GrammarGenerator},
     };
 
@@ -504,6 +507,136 @@ mod tests {
         }
     }
 
+    #[test]
+    /// Create a table with a random number of columns and indexes, and then randomly update or delete rows from the table.
+    /// Verify that the results are the same for SQLite and Turso.
+    pub fn table_index_mutation_fuzz() {
+        let _ = env_logger::try_init();
+        let (mut rng, seed) = rng_from_time();
+        println!("index_scan_single_key_mutation_fuzz seed: {seed}");
+
+        const OUTER_ITERATIONS: usize = 30;
+        for i in 0..OUTER_ITERATIONS {
+            println!(
+                "table_index_mutation_fuzz iteration {}/{}",
+                i + 1,
+                OUTER_ITERATIONS
+            );
+            let limbo_db = TempDatabase::new_empty(true);
+            let sqlite_db = TempDatabase::new_empty(true);
+            let num_cols = rng.random_range(1..=10);
+            let table_def = (0..num_cols)
+                .map(|i| format!("c{i} INTEGER"))
+                .collect::<Vec<_>>();
+            let table_def = table_def.join(", ");
+            let table_def = format!("CREATE TABLE t ({table_def})");
+
+            let num_indexes = rng.random_range(0..=num_cols);
+            let indexes = (0..num_indexes)
+                .map(|i| format!("CREATE INDEX idx_{i} ON t(c{i})"))
+                .collect::<Vec<_>>();
+
+            // Create tables and indexes in both databases
+            let limbo_conn = limbo_db.connect_limbo();
+            limbo_exec_rows(&limbo_db, &limbo_conn, &table_def);
+            for t in indexes.iter() {
+                limbo_exec_rows(&limbo_db, &limbo_conn, t);
+            }
+
+            let sqlite_conn = rusqlite::Connection::open(sqlite_db.path.clone()).unwrap();
+            sqlite_conn.execute(&table_def, params![]).unwrap();
+            for t in indexes.iter() {
+                sqlite_conn.execute(t, params![]).unwrap();
+            }
+
+            // Generate initial data
+            let num_inserts = rng.random_range(10..=1000);
+            let mut tuples = HashSet::new();
+            while tuples.len() < num_inserts {
+                tuples.insert(
+                    (0..num_cols)
+                        .map(|_| rng.random_range(0..1000))
+                        .collect::<Vec<_>>(),
+                );
+            }
+            let mut insert_values = Vec::new();
+            for tuple in tuples {
+                insert_values.push(format!(
+                    "({})",
+                    tuple
+                        .iter()
+                        .map(|x| x.to_string())
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                ));
+            }
+            // Track executed statements in case we fail
+            let mut dml_statements = Vec::new();
+            let insert = format!("INSERT INTO t VALUES {}", insert_values.join(", "));
+            dml_statements.push(insert.clone());
+
+            // Insert initial data into both databases
+            sqlite_conn.execute(&insert, params![]).unwrap();
+            limbo_exec_rows(&limbo_db, &limbo_conn, &insert);
+
+            const COMPARISONS: [&str; 3] = ["=", "<", ">"];
+            const INNER_ITERATIONS: usize = 100;
+
+            for _ in 0..INNER_ITERATIONS {
+                let do_update = rng.random_range(0..2) == 0;
+
+                let comparison = COMPARISONS[rng.random_range(0..COMPARISONS.len())];
+                let affected_col = rng.random_range(0..num_cols);
+                let predicate_col = rng.random_range(0..num_cols);
+                let predicate_value = rng.random_range(0..1000);
+
+                let query = if do_update {
+                    let new_y = rng.random_range(0..1000);
+                    format!("UPDATE t SET c{affected_col} = {new_y} WHERE c{predicate_col} {comparison} {predicate_value}")
+                } else {
+                    format!("DELETE FROM t WHERE c{predicate_col} {comparison} {predicate_value}")
+                };
+
+                dml_statements.push(query.clone());
+
+                // Execute on both databases
+                sqlite_conn.execute(&query, params![]).unwrap();
+                let limbo_res = limbo_exec_rows_fallible(&limbo_db, &limbo_conn, &query);
+                if let Err(e) = &limbo_res {
+                    // print all the DDL and DML statements
+                    println!("{table_def};");
+                    for t in indexes.iter() {
+                        println!("{t};");
+                    }
+                    for t in dml_statements.iter() {
+                        println!("{t};");
+                    }
+                    panic!("Error executing query: {e}");
+                }
+
+                // Verify results match exactly
+                let verify_query = format!(
+                    "SELECT * FROM t ORDER BY {}",
+                    (0..num_cols)
+                        .map(|i| format!("c{i}"))
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                );
+                let sqlite_rows = sqlite_exec_rows(&sqlite_conn, &verify_query);
+                let limbo_rows = limbo_exec_rows(&limbo_db, &limbo_conn, &verify_query);
+
+                assert_eq!(
+                    sqlite_rows, limbo_rows,
+                    "Different results after mutation! limbo: {limbo_rows:?}, sqlite: {sqlite_rows:?}, seed: {seed}, query: {query}",
+                );
+
+                if sqlite_rows.is_empty() {
+                    break;
+                }
+            }
+        }
+    }
+
     #[test]
     pub fn compound_select_fuzz() {
         let _ = env_logger::try_init();