BTree delete implementation

COMPAT.md

@@ -114,7 +114,7 @@ The current status of Limbo is:
 | PRAGMA foreign_key_check         | No         |                                                 |
 | PRAGMA foreign_key_list          | No         |                                                 |
 | PRAGMA foreign_keys              | No         |                                                 |
-| PRAGMA freelist_count            | No         |                                                 |
+| PRAGMA freelist_count            | Yes        |                                                 |
 | PRAGMA full_column_names         | Not Needed | deprecated in SQLite                            |
 | PRAGMA fullsync                  | No         |                                                 |
 | PRAGMA function_list             | No         |                                                 |

core/storage/pager.rs

@@ -3,7 +3,7 @@ use crate::storage::buffer_pool::BufferPool;
 use crate::storage::database::DatabaseStorage;
 use crate::storage::sqlite3_ondisk::{self, DatabaseHeader, PageContent};
 use crate::storage::wal::Wal;
-use crate::{Buffer, Result};
+use crate::{Buffer, LimboError, Result};
 use log::trace;
 use std::cell::{RefCell, UnsafeCell};
 use std::collections::HashSet;
@@ -436,6 +436,71 @@ impl Pager {
         self.page_cache.write().unwrap().clear();
     }
 
+    // Providing a page is optional, if provided it will be used to avoid reading the page from disk.
+    // This is implemented in accordance with sqlite freepage2() function.
+    pub fn free_page(&self, page: Option<PageRef>, page_id: usize) -> Result<()> {
+        if page_id < 2 || page_id > self.db_header.borrow().database_size as usize {
+            return Err(LimboError::Corrupt(format!(
+                "Invalid page number {} for free operation",
+                page_id
+            )));
+        }
+
+        let page = match page {
+            Some(page) => {
+                assert_eq!(page.get().id, page_id, "Page id mismatch");
+                page
+            }
+            None => self.read_page(page_id)?,
+        };
+
+        let page_1 = self.read_page(1)?;
+        page_1.set_dirty();
+        self.add_dirty(1);
+
+        self.db_header.borrow_mut().freelist_pages += 1;
+
+        let trunk_page_id = self.db_header.borrow().freelist_trunk_page;
+
+        if trunk_page_id != 0 {
+            // Add as leaf to current trunk
+            let trunk_page = self.read_page(trunk_page_id as usize)?;
+            let trunk_page_contents = trunk_page.get().contents.as_ref().unwrap();
+            let number_of_leaf_pages = trunk_page_contents.read_u32(4);
+
+            let max_free_list_entries = (self.usable_size() / 4) - 8;
+
+            if number_of_leaf_pages < max_free_list_entries as u32 {
+                trunk_page.set_dirty();
+                self.add_dirty(trunk_page_id as usize);
+
+                trunk_page_contents.write_u32(4, number_of_leaf_pages + 1);
+                trunk_page_contents
+                    .write_u32((8 + (number_of_leaf_pages * 4)) as usize, page_id as u32);
+                page.clear_uptodate();
+                page.clear_loaded();
+
+                return Ok(());
+            }
+        }
+
+        // If we get here, need to make this page a new trunk
+        page.set_dirty();
+        self.add_dirty(page_id);
+
+        let contents = page.get().contents.as_mut().unwrap();
+        // Point to previous trunk
+        contents.write_u32(0, trunk_page_id);
+        // Zero leaf count
+        contents.write_u32(4, 0);
+        // Update page 1 to point to new trunk
+        self.db_header.borrow_mut().freelist_trunk_page = page_id as u32;
+        // Clear flags
+        page.clear_uptodate();
+        page.clear_loaded();
+        Ok(())
+    }
+
     /*
         Gets a new page that increasing the size of the page or uses a free page.
         Currently free list pages are not yet supported.

core/storage/sqlite3_ondisk.rs

@@ -105,10 +105,10 @@ pub struct DatabaseHeader {
     pub database_size: u32,
 
     /// Page number of the first freelist trunk page.
-    freelist_trunk_page: u32,
+    pub freelist_trunk_page: u32,
 
     /// Total number of freelist pages.
-    freelist_pages: u32,
+    pub freelist_pages: u32,
 
     /// The schema cookie. Incremented when the database schema changes.
     schema_cookie: u32,
@@ -332,7 +332,7 @@ pub fn begin_write_database_header(header: &DatabaseHeader, pager: &Pager) -> Re
     Ok(())
 }
 
-fn write_header_to_buf(buf: &mut [u8], header: &DatabaseHeader) {
+pub(crate) fn write_header_to_buf(buf: &mut [u8], header: &DatabaseHeader) {
     buf[0..16].copy_from_slice(&header.magic);
     buf[16..18].copy_from_slice(&header.page_size.to_be_bytes());
     buf[18] = header.write_version;
@@ -428,7 +428,7 @@ impl PageContent {
         }
     }
 
-    fn read_u8(&self, pos: usize) -> u8 {
+    pub fn read_u8(&self, pos: usize) -> u8 {
         let buf = self.as_ptr();
         buf[self.offset + pos]
     }
@@ -438,7 +438,7 @@ impl PageContent {
         u16::from_be_bytes([buf[self.offset + pos], buf[self.offset + pos + 1]])
     }
 
-    fn read_u32(&self, pos: usize) -> u32 {
+    pub fn read_u32(&self, pos: usize) -> u32 {
         let buf = self.as_ptr();
         u32::from_be_bytes([
             buf[self.offset + pos],

core/translate/mod.rs

@@ -600,6 +600,10 @@ fn update_pragma(
             query_pragma("wal_checkpoint", header, program)?;
             Ok(())
         }
+        PragmaName::FreelistCount => {
+            query_pragma("freelist_count", header, program)?;
+            Ok(())
+        }
     }
 }
 
@@ -629,28 +633,20 @@ fn query_pragma(
                 value: "wal".into(),
                 dest: register,
             });
-            program.emit_insn(Insn::ResultRow {
-                start_reg: register,
-                count: 1,
-            });
         }
-        PragmaName::WalCheckpoint => {
-            // Checkpoint uses 3 registers: P1, P2, P3. Ref Insn::Checkpoint for more info.
-            // Allocate two more here as one was allocated at the top.
-            program.alloc_register();
-            program.alloc_register();
-            program.emit_insn(Insn::Checkpoint {
-                database: 0,
-                checkpoint_mode: CheckpointMode::Passive,
+        PragmaName::FreelistCount => {
+            let free_page_count = database_header.borrow().freelist_pages;
+            program.emit_insn(Insn::Integer {
+                value: free_page_count as i64,
                 dest: register,
             });
-            program.emit_insn(Insn::ResultRow {
-                start_reg: register,
-                count: 3,
-            });
         }
     }
 
+    program.emit_insn(Insn::ResultRow {
+        start_reg: register,
+        count: 1,
+    });
     Ok(())
 }