Merge 'Stmt reset cursors' from Nikita Sivukhin

This PR reset cursor state in the `stmt.reset()` method because under
the hood statement caches some BTree state which can be no longer valid
at the moment of next statement run.

Reviewed-by: Preston Thorpe <preston@turso.tech>

Closes #3859

core/vdbe/mod.rs

@@ -270,7 +270,7 @@ pub enum TxnCleanup {
 pub struct ProgramState {
     pub io_completions: Option<IOCompletions>,
     pub pc: InsnReference,
-    cursors: Vec<Option<Cursor>>,
+    pub(crate) cursors: Vec<Option<Cursor>>,
     cursor_seqs: Vec<i64>,
     registers: Vec<Register>,
     pub(crate) result_row: Option<Row>,
@@ -417,10 +417,14 @@ impl ProgramState {
             self.cursors.resize_with(max_cursors, || None);
             self.cursor_seqs.resize(max_cursors, 0);
         }
-        if let Some(max_resgisters) = max_registers {
+        if let Some(max_registers) = max_registers {
             self.registers
-                .resize_with(max_resgisters, || Register::Value(Value::Null));
+                .resize_with(max_registers, || Register::Value(Value::Null));
         }
+        // reset cursors as they can have cached information which will be no longer relevant on next program execution
+        self.cursors.iter_mut().for_each(|c| {
+            let _ = c.take();
+        });
         self.registers
             .iter_mut()
             .for_each(|r| *r = Register::Value(Value::Null));

tests/integration/query_processing/test_read_path.rs

@@ -882,6 +882,49 @@ fn test_multiple_connections_visibility() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[test]
+fn test_stmt_reset() -> anyhow::Result<()> {
+    let tmp_db = TempDatabase::new_with_rusqlite("CREATE TABLE test (x);");
+    let conn1 = tmp_db.connect_limbo();
+    let mut stmt1 = conn1.prepare("INSERT INTO test VALUES (?)").unwrap();
+    for _ in 0..3 {
+        stmt1.reset();
+        stmt1.bind_at(1.try_into().unwrap(), Value::Blob(vec![0u8; 1024]));
+        loop {
+            match stmt1.step().unwrap() {
+                StepResult::Done => break,
+                _ => tmp_db.io.step().unwrap(),
+            }
+        }
+    }
+
+    // force btree-page split which will be "unnoticed" by stmt1 if it will cache something in between of calls
+    conn1
+        .execute("INSERT INTO test VALUES (randomblob(1024))")
+        .unwrap();
+
+    stmt1.reset();
+    stmt1.bind_at(1.try_into().unwrap(), Value::Blob(vec![0u8; 1024]));
+    loop {
+        match stmt1.step().unwrap() {
+            StepResult::Done => break,
+            _ => tmp_db.io.step().unwrap(),
+        }
+    }
+    let rows = limbo_exec_rows(&tmp_db, &conn1, "SELECT rowid FROM test");
+    assert_eq!(
+        rows,
+        vec![
+            vec![rusqlite::types::Value::Integer(1)],
+            vec![rusqlite::types::Value::Integer(2)],
+            vec![rusqlite::types::Value::Integer(3)],
+            vec![rusqlite::types::Value::Integer(4)],
+            vec![rusqlite::types::Value::Integer(5)],
+        ]
+    );
+    Ok(())
+}
+
 #[test]
 /// Test that we can only join up to 63 tables, and trying to join more should fail with an error instead of panicing.
 fn test_max_joined_tables_limit() {