Add ResetSorter instruction

This instruction isn't used yet, but it will be needed for window
functions, since they heavily rely on ephemeral tables.

COMPAT.md

@@ -519,6 +519,7 @@ Modifiers:
 | RealAffinity   | Yes    |         |
 | Remainder      | Yes    |         |
 | ResetCount     | No     |         |
+| ResetSorter    | Partial| sorter cursors are not supported yet; only ephemeral tables are |
 | ResultRow      | Yes    |         |
 | Return         | Yes    |         |
 | Rewind         | Yes    |         |

core/storage/btree.rs

@@ -5124,10 +5124,32 @@ impl BTreeCursor {
         }
     }
 
-    /// Destroys a B-tree by freeing all its pages in an iterative depth-first order.
+    /// Deletes all content from the B-Tree but preserves the root page.
+    ///
+    /// Unlike [`btree_destroy`], which frees all pages including the root,
+    /// this method only clears the tree’s contents. The root page remains
+    /// allocated and is reset to an empty leaf page.
+    pub fn clear_btree(&mut self) -> Result<IOResult<Option<usize>>> {
+        self.destroy_btree_contents(true)
+    }
+
+    /// Destroys the entire B-Tree, including the root page.
+    ///
+    /// All pages belonging to the tree are freed, leaving no trace of the B-Tree.
+    /// Use this when the structure itself is no longer needed.
+    ///
+    /// For cases where the B-Tree should remain allocated but emptied, see [`btree_clear`].
+    #[instrument(skip(self), level = Level::DEBUG)]
+    pub fn btree_destroy(&mut self) -> Result<IOResult<Option<usize>>> {
+        self.destroy_btree_contents(false)
+    }
+
+    /// Deletes all contents of the B-tree by freeing all its pages in an iterative depth-first order.
     /// This ensures child pages are freed before their parents
     /// Uses a state machine to keep track of the operation to ensure IO doesn't cause repeated traversals
     ///
+    /// Depending on the caller, the root page may either be freed as well or left allocated but emptied.
+    ///
     /// # Example
     /// For a B-tree with this structure (where 4' is an overflow page):
     /// ```text
@@ -5139,8 +5161,7 @@ impl BTreeCursor {
     /// ```
     ///
     /// The destruction order would be: [4',4,5,2,6,7,3,1]
-    #[instrument(skip(self), level = Level::DEBUG)]
-    pub fn btree_destroy(&mut self) -> Result<IOResult<Option<usize>>> {
+    fn destroy_btree_contents(&mut self, keep_root: bool) -> Result<IOResult<Option<usize>>> {
         if let CursorState::None = &self.state {
             let c = self.move_to_root()?;
             self.state = CursorState::Destroy(DestroyInfo {
@@ -5302,9 +5323,9 @@ impl BTreeCursor {
                     let page = self.stack.top();
                     let page_id = page.get().id;
 
-                    return_if_io!(self.pager.free_page(Some(page), page_id));
-
                     if self.stack.has_parent() {
+                        return_if_io!(self.pager.free_page(Some(page), page_id));
+
                         self.stack.pop();
                         let destroy_info = self
                             .state
@@ -5312,6 +5333,12 @@ impl BTreeCursor {
                             .expect("unable to get a mut reference to destroy state in cursor");
                         destroy_info.state = DestroyState::ProcessPage;
                     } else {
+                        if keep_root {
+                            self.clear_root(&page);
+                        } else {
+                            return_if_io!(self.pager.free_page(Some(page), page_id));
+                        }
+
                         self.state = CursorState::None;
                         //  TODO: For now, no-op the result return None always. This will change once [AUTO_VACUUM](https://www.sqlite.org/lang_vacuum.html) is introduced
                         //  At that point, the last root page(call this x) will be moved into the position of the root page of this table and the value returned will be x
@@ -5322,6 +5349,19 @@ impl BTreeCursor {
         }
     }
 
+    fn clear_root(&mut self, root_page: &PageRef) {
+        let page_ref = root_page.get();
+        let contents = page_ref.contents.as_ref().unwrap();
+
+        let page_type = match contents.page_type() {
+            PageType::TableLeaf | PageType::TableInterior => PageType::TableLeaf,
+            PageType::IndexLeaf | PageType::IndexInterior => PageType::IndexLeaf,
+        };
+
+        self.pager.add_dirty(root_page);
+        btree_init_page(root_page, page_type, 0, self.pager.usable_space());
+    }
+
     pub fn table_id(&self) -> usize {
         self.root_page
     }
@@ -7751,6 +7791,36 @@ mod tests {
         payload
     }
 
+    fn insert_record(
+        cursor: &mut BTreeCursor,
+        pager: &Rc<Pager>,
+        rowid: i64,
+        val: Value,
+    ) -> Result<(), LimboError> {
+        let regs = &[Register::Value(val)];
+        let record = ImmutableRecord::from_registers(regs, regs.len());
+
+        run_until_done(
+            || {
+                let key = SeekKey::TableRowId(rowid);
+                cursor.seek(key, SeekOp::GE { eq_only: true })
+            },
+            pager.deref(),
+        )?;
+        run_until_done(
+            || cursor.insert(&BTreeKey::new_table_rowid(rowid, Some(&record))),
+            pager.deref(),
+        )?;
+        Ok(())
+    }
+
+    fn assert_btree_empty(cursor: &mut BTreeCursor, pager: &Pager) -> Result<()> {
+        let _c = cursor.move_to_root()?;
+        let empty = !run_until_done(|| cursor.next(), pager)?;
+        assert!(empty, "expected B-tree to be empty");
+        Ok(())
+    }
+
     #[test]
     fn test_insert_cell() {
         let db = get_database();
@@ -9193,6 +9263,162 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    pub fn test_clear_btree_with_single_page() -> Result<()> {
+        let (pager, root_page, _, _) = empty_btree();
+        let num_columns = 5;
+        let record_count = 10;
+
+        let mut cursor = BTreeCursor::new_table(None, pager.clone(), root_page, num_columns);
+
+        for rowid in 1..=record_count {
+            insert_record(&mut cursor, &pager, rowid, Value::Integer(rowid))?;
+        }
+
+        let page_count = pager
+            .io
+            .block(|| pager.with_header(|header| header.database_size.get()))?;
+        assert_eq!(
+            page_count, 2,
+            "expected two pages (header + root), got {page_count}"
+        );
+
+        run_until_done(|| cursor.clear_btree(), &pager)?;
+
+        assert_btree_empty(&mut cursor, pager.deref())
+    }
+
+    #[test]
+    pub fn test_clear_btree_with_multiple_pages() -> Result<()> {
+        let (pager, root_page, _, _) = empty_btree();
+        let num_columns = 5;
+        let record_count = 1000;
+
+        let mut cursor = BTreeCursor::new_table(None, pager.clone(), root_page, num_columns);
+
+        for rowid in 1..=record_count {
+            insert_record(&mut cursor, &pager, rowid, Value::Integer(rowid))?;
+        }
+
+        // Ensure enough records were created so the tree spans multiple pages.
+        let page_count = pager
+            .io
+            .block(|| pager.with_header(|header| header.database_size.get()))?;
+        assert!(
+            page_count > 2,
+            "expected more pages than just header + root, got {page_count}"
+        );
+
+        run_until_done(|| cursor.clear_btree(), &pager)?;
+
+        assert_btree_empty(&mut cursor, pager.deref())
+    }
+
+    #[test]
+    pub fn test_clear_btree_reinsertion() -> Result<()> {
+        let (pager, root_page, _, _) = empty_btree();
+        let num_columns = 5;
+        let record_count = 1000;
+
+        let mut cursor = BTreeCursor::new_table(None, pager.clone(), root_page, num_columns);
+
+        for rowid in 1..=record_count {
+            insert_record(&mut cursor, &pager, rowid, Value::Integer(rowid))?;
+        }
+
+        run_until_done(|| cursor.clear_btree(), &pager)?;
+
+        // Reinsert into cleared B-tree to ensure it’s still functional
+        for rowid in 1..=record_count {
+            insert_record(&mut cursor, &pager, rowid, Value::Integer(rowid))?;
+        }
+
+        if let (_, false) = validate_btree(pager.clone(), root_page) {
+            panic!("Invalid B-tree after reinsertion");
+        }
+
+        let _c = cursor.move_to_root()?;
+        for i in 1..=record_count {
+            let exists = run_until_done(|| cursor.next(), &pager)?;
+            assert!(exists, "Record {i} not found");
+
+            let record = run_until_done(|| cursor.record(), &pager)?;
+            let value = record.unwrap().get_value(0)?;
+            assert_eq!(
+                value,
+                RefValue::Integer(i),
+                "Unexpected value for record {i}",
+            );
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    pub fn test_clear_btree_multiple_cursors() -> Result<()> {
+        let (pager, root_page, _, _) = empty_btree();
+        let num_columns = 5;
+        let record_count = 1000;
+
+        let mut cursor1 = BTreeCursor::new_table(None, pager.clone(), root_page, num_columns);
+        let mut cursor2 = BTreeCursor::new_table(None, pager.clone(), root_page, num_columns);
+
+        // Use cursor1 to insert records
+        for rowid in 1..=record_count {
+            insert_record(&mut cursor1, &pager, rowid, Value::Integer(rowid))?;
+        }
+
+        // Use cursor1 to clear the btree
+        run_until_done(|| cursor1.clear_btree(), &pager)?;
+
+        // Verify that cursor2 works correctly
+        assert_btree_empty(&mut cursor2, pager.deref())?;
+
+        // Insert using cursor2
+        insert_record(&mut cursor1, &pager, 1, Value::Integer(123))?;
+
+        if let (_, false) = validate_btree(pager.clone(), root_page) {
+            panic!("Invalid B-tree after insertion");
+        }
+
+        let key = Value::Integer(1);
+        let exists = run_until_done(|| cursor2.exists(&key), pager.deref())?;
+        assert!(exists, "key not found {key}");
+
+        Ok(())
+    }
+
+    #[test]
+    pub fn test_clear_btree_with_overflow_pages() -> Result<()> {
+        let (pager, root_page, _, _) = empty_btree();
+        let num_columns = 5;
+        let record_count = 100;
+
+        let mut cursor = BTreeCursor::new_table(None, pager.clone(), root_page, num_columns);
+
+        let initial_page_count = pager
+            .io
+            .block(|| pager.with_header(|header| header.database_size.get()))?;
+
+        for rowid in 1..=record_count {
+            let large_blob = vec![b'A'; 8192];
+            insert_record(&mut cursor, &pager, rowid, Value::Blob(large_blob))?;
+        }
+
+        let page_count_after_inserts = pager
+            .io
+            .block(|| pager.with_header(|header| header.database_size.get()))?;
+        let created_pages = page_count_after_inserts - initial_page_count;
+        assert!(
+            created_pages > record_count as u32,
+            "expected more pages to be created than records, got {created_pages}"
+        );
+
+        run_until_done(|| cursor.clear_btree(), &pager)?;
+
+        assert_btree_empty(&mut cursor, pager.deref())
+    }
+
     #[test]
     pub fn test_defragment() {
         let db = get_database();

core/vdbe/execute.rs

@@ -6500,6 +6500,33 @@ pub fn op_destroy(
     Ok(InsnFunctionStepResult::Step)
 }
 
+pub fn op_reset_sorter(
+    program: &Program,
+    state: &mut ProgramState,
+    insn: &Insn,
+    pager: &Rc<Pager>,
+    mv_store: Option<&Arc<MvStore>>,
+) -> Result<InsnFunctionStepResult> {
+    load_insn!(ResetSorter { cursor_id }, insn);
+
+    let (_, cursor_type) = program.cursor_ref.get(*cursor_id).unwrap();
+    let cursor = state.get_cursor(*cursor_id);
+
+    match cursor_type {
+        CursorType::BTreeTable(table) => {
+            let cursor = cursor.as_btree_mut();
+            return_if_io!(cursor.clear_btree());
+        }
+        CursorType::Sorter => {
+            unimplemented!("ResetSorter is not supported for sorter cursors yet")
+        }
+        _ => panic!("ResetSorter is not supported for {cursor_type:?}"),
+    }
+
+    state.pc += 1;
+    Ok(InsnFunctionStepResult::Step)
+}
+
 pub fn op_drop_table(
     program: &Program,
     state: &mut ProgramState,

core/vdbe/explain.rs

@@ -1271,6 +1271,15 @@ pub fn insn_to_row(
                     "root iDb={root} former_root={former_root_reg} is_temp={is_temp}"
                 ),
             ),
+            Insn::ResetSorter { cursor_id } => (
+                "ResetSorter",
+                *cursor_id as i32,
+                0,
+                0,
+                Value::build_text(""),
+                0,
+                format!("cursor={cursor_id}"),
+            ),
             Insn::DropTable {
                 db,
                 _p2,

core/vdbe/insn.rs

@@ -840,6 +840,18 @@ pub enum Insn {
         is_temp: usize,
     },
 
+    /// Deletes all contents from the ephemeral table that the cursor points to.
+    ///
+    /// In Turso, we do not currently distinguish strictly between ephemeral
+    /// and standard tables at the type level. Therefore, it is the caller’s
+    /// responsibility to ensure that `ResetSorter` is applied only to ephemeral
+    /// tables.
+    ///
+    /// SQLite also supports sorter cursors, but this is not yet implemented in Turso.
+    ResetSorter {
+        cursor_id: CursorID,
+    },
+
     ///  Drop a table
     DropTable {
         ///  The database within which this b-tree needs to be dropped (P1).
@@ -1207,6 +1219,7 @@ impl Insn {
             Insn::Copy { .. } => execute::op_copy,
             Insn::CreateBtree { .. } => execute::op_create_btree,
             Insn::Destroy { .. } => execute::op_destroy,
+            Insn::ResetSorter { .. } => execute::op_reset_sorter,
 
             Insn::DropTable { .. } => execute::op_drop_table,
             Insn::DropView { .. } => execute::op_drop_view,