Repository: cberner/redb
Pull request #244 (2022-04-02 00:59:45)

Perf


Patch:
diff --git a/benches/common.rs b/benches/common.rs
index 4e44d76..85fdaa8 100644
--- a/benches/common.rs
+++ b/benches/common.rs
@@ -1,8 +1,10 @@
-use redb::{ReadableTable, TableDefinition};
+use redb::ReadableTable;
+use redb::TableDefinition;
 use std::fs;
 use std::fs::File;
 use std::path::Path;
 
+#[allow(dead_code)]
 const X: TableDefinition<[u8], [u8]> = TableDefinition::new("x");
 
 pub trait BenchDatabase {
@@ -47,6 +49,7 @@ pub struct RedbBenchDatabase<'a> {
 }
 
 impl<'a> RedbBenchDatabase<'a> {
+    #[allow(dead_code)]
     pub fn new(db: &'a redb::Database) -> Self {
         RedbBenchDatabase { db }
     }
@@ -66,9 +69,10 @@ impl<'a> BenchDatabase for RedbBenchDatabase<'a> {
     }
 
     fn read_transaction(&self) -> Self::R {
-        let txn = self.db.begin_read().unwrap();
-        let table = txn.open_table(X).unwrap();
-        RedbBenchReadTransaction { _txn: txn, table }
+        // let txn = self.db.begin_read().unwrap();
+        // let table = txn.open_table(X).unwrap();
+        // RedbBenchReadTransaction { _txn: txn, table }
+        todo!()
     }
 }
 
@@ -97,9 +101,9 @@ impl<'a, 'b> BenchWriteTransaction<'b> for RedbBenchWriteTransaction<'a> {
     type T = RedbBenchInserter<'b>;
 
     fn get_inserter(&'b self) -> Self::T {
-        RedbBenchInserter {
-            table: self.txn.open_table(X).unwrap(),
-        }
+        // let table = self.txn.open_table(X).unwrap();
+        // return RedbBenchInserter { table };
+        todo!()
     }
 
     fn commit(self) -> Result<(), ()> {
diff --git a/benches/lmdb_benchmark.rs b/benches/lmdb_benchmark.rs
index 17fb4cf..80055c5 100644
--- a/benches/lmdb_benchmark.rs
+++ b/benches/lmdb_benchmark.rs
@@ -5,6 +5,7 @@ use common::*;
 
 use rand::prelude::SliceRandom;
 use rand::Rng;
+use redb::{ReadableTable, TableDefinition};
 use std::time::{Duration, Instant};
 
 const ITERATIONS: usize = 3;
@@ -23,6 +24,172 @@ fn gen_data(count: usize, key_size: usize, value_size: usize) -> Vec<(Vec<u8>, V
     pairs
 }
 
+// TODO: merge back into benchmark()
+fn benchmark_redb(db: redb::Database) -> Vec<(&'static str, Duration)> {
+    let mut results = Vec::new();
+    let mut pairs = gen_data(1_000_000, 24, 150);
+    let mut written = 0;
+
+    let table_def: TableDefinition<[u8], [u8]> = TableDefinition::new("x");
+
+    let start = Instant::now();
+    let txn = db.begin_write().unwrap();
+    let mut inserter = txn.open_table(table_def).unwrap();
+    {
+        for _ in 0..ELEMENTS {
+            let len = pairs.len();
+            let (key, value) = &mut pairs[written % len];
+            key[16..].copy_from_slice(&(written as u64).to_le_bytes());
+            inserter.insert(key, value).unwrap();
+            written += 1;
+        }
+    }
+    drop(inserter);
+    txn.commit().unwrap();
+
+    let end = Instant::now();
+    let duration = end - start;
+    println!(
+        "{}: Bulk loaded {} items in {}ms",
+        RedbBenchDatabase::db_type_name(),
+        ELEMENTS,
+        duration.as_millis()
+    );
+    results.push(("bulk load", duration));
+
+    let start = Instant::now();
+    let writes = 100;
+    {
+        for _ in 0..writes {
+            let txn = db.begin_write().unwrap();
+            let mut inserter = txn.open_table(table_def).unwrap();
+            let len = pairs.len();
+            let (key, value) = &mut pairs[written % len];
+            key[16..].copy_from_slice(&(written as u64).to_le_bytes());
+            inserter.insert(key, value).unwrap();
+            drop(inserter);
+            txn.commit().unwrap();
+            written += 1;
+        }
+    }
+
+    let end = Instant::now();
+    let duration = end - start;
+    println!(
+        "{}: Wrote {} individual items in {}ms",
+        RedbBenchDatabase::db_type_name(),
+        writes,
+        duration.as_millis()
+    );
+    results.push(("individual writes", duration));
+
+    let start = Instant::now();
+    let batch_size = 1000;
+    {
+        for _ in 0..writes {
+            let txn = db.begin_write().unwrap();
+            let mut inserter = txn.open_table(table_def).unwrap();
+            for _ in 0..batch_size {
+                let len = pairs.len();
+                let (key, value) = &mut pairs[written % len];
+                key[16..].copy_from_slice(&(written as u64).to_le_bytes());
+                inserter.insert(key, value).unwrap();
+                written += 1;
+            }
+            drop(inserter);
+            txn.commit().unwrap();
+        }
+    }
+
+    let end = Instant::now();
+    let duration = end - start;
+    println!(
+        "{}: Wrote {} x {} items in {}ms",
+        RedbBenchDatabase::db_type_name(),
+        writes,
+        batch_size,
+        duration.as_millis()
+    );
+    results.push(("batch writes", duration));
+
+    let mut key_order: Vec<usize> = (0..ELEMENTS).collect();
+    key_order.shuffle(&mut rand::thread_rng());
+
+    let txn = db.begin_read().unwrap();
+    let table = txn.open_table(table_def).unwrap();
+    {
+        for _ in 0..ITERATIONS {
+            let start = Instant::now();
+            let mut checksum = 0u64;
+            let mut expected_checksum = 0u64;
+            for i in &key_order {
+                let len = pairs.len();
+                let (key, value) = &mut pairs[i % len];
+                key[16..].copy_from_slice(&(*i as u64).to_le_bytes());
+                let result = table.get(key).unwrap().unwrap();
+                checksum += result[0] as u64;
+                expected_checksum += value[0] as u64;
+            }
+            assert_eq!(checksum, expected_checksum);
+            let end = Instant::now();
+            let duration = end - start;
+            println!(
+                "{}: Random read {} items in {}ms",
+                RedbBenchDatabase::db_type_name(),
+                ELEMENTS,
+                duration.as_millis()
+            );
+            results.push(("random reads", duration));
+        }
+
+        for _ in 0..ITERATIONS {
+            let start = Instant::now();
+            for i in &key_order {
+                let len = pairs.len();
+                let (key, _) = &mut pairs[i % len];
+                key[16..].copy_from_slice(&(*i as u64).to_le_bytes());
+                table.range(key.as_slice()..).unwrap();
+            }
+            let end = Instant::now();
+            let duration = end - start;
+            println!(
+                "{}: Random range read {} starts in {}ms",
+                RedbBenchDatabase::db_type_name(),
+                ELEMENTS,
+                duration.as_millis()
+            );
+            results.push(("random range reads", duration));
+        }
+    }
+
+    let start = Instant::now();
+    let deletes = key_order.len() / 2;
+    {
+        let txn = db.begin_write().unwrap();
+        let mut inserter = txn.open_table(table_def).unwrap();
+        for i in 0..deletes {
+            let len = pairs.len();
+            let (key, _) = &mut pairs[i % len];
+            key[16..].copy_from_slice(&(i as u64).to_le_bytes());
+            inserter.remove(key).unwrap();
+        }
+        drop(inserter);
+        txn.commit().unwrap();
+    }
+
+    let end = Instant::now();
+    let duration = end - start;
+    println!(
+        "{}: Removed {} items in {}ms",
+        RedbBenchDatabase::db_type_name(),
+        deletes,
+        duration.as_millis()
+    );
+    results.push(("removals", duration));
+
+    results
+}
+
 fn benchmark<T: BenchDatabase>(mut db: T) -> Vec<(&'static str, Duration)> {
     let mut results = Vec::new();
     let mut pairs = gen_data(1_000_000, 24, 150);
@@ -189,8 +356,8 @@ fn main() {
     let redb_results = {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { redb::Database::create(tmpfile.path(), 4096 * 1024 * 1024).unwrap() };
-        let table = RedbBenchDatabase::new(&db);
-        benchmark(table)
+        // let table = RedbBenchDatabase::new(&db);
+        benchmark_redb(db)
     };
 
     let lmdb_results = {
diff --git a/fuzz/fuzz_targets/common.rs b/fuzz/fuzz_targets/common.rs
index 535d912..80549b4 100644
--- a/fuzz/fuzz_targets/common.rs
+++ b/fuzz/fuzz_targets/common.rs
@@ -1,3 +1,4 @@
+use std::cmp::max;
 use std::mem::size_of;
 use arbitrary::Unstructured;
 use libfuzzer_sys::arbitrary::Arbitrary;
@@ -107,6 +108,8 @@ impl FuzzConfig {
     pub(crate) fn oom_plausible(&self) -> bool {
         let mut total_entries = 0;
         let mut total_value_bytes = 0;
+        let mut consecutive_nondurable_commits = 0;
+        let mut max_consecutive_nondurable_commits = 0;
         for transaction in self.transactions.iter() {
             for write in transaction.ops.iter() {
                 total_entries += 1;
@@ -115,10 +118,16 @@ impl FuzzConfig {
                     _ => 0
                 };
             }
+            if !transaction.durable {
+                consecutive_nondurable_commits += 1;
+            } else {
+                consecutive_nondurable_commits = 0;
+            }
+            max_consecutive_nondurable_commits = max(max_consecutive_nondurable_commits, consecutive_nondurable_commits);
         }
 
         let approximate_overhead = 256_000;
-        let expected_size = total_value_bytes + (size_of::<u64>() * 4) * total_entries + approximate_overhead;
+        let expected_size = (total_value_bytes + (size_of::<u64>() * 4) * total_entries) * max_consecutive_nondurable_commits + approximate_overhead;
         // If we're within a factor of 10 of the max db size, then assume an OOM is plausible
         expected_size * 10 > self.max_db_size.value
     }
diff --git a/src/db.rs b/src/db.rs
index 219df37..1fba63a 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -1,21 +1,24 @@
-use crate::multimap_table::MultimapTable;
-use crate::table::{ReadOnlyTable, Table};
+use crate::tree_store::btree_utils::{get_mut_value, make_mut_single_leaf};
 use crate::tree_store::{
-    get_db_size, DatabaseStats, PageNumber, Storage, TableType, TransactionId, FREED_TABLE,
+    get_db_size, AllPageNumbersBtreeIter, BtreeRangeIter, FreedTableKey, InternalTableDefinition,
+    PageNumber, TableType, TransactionalMemory, FREED_TABLE,
 };
-use crate::types::{RedbKey, RedbValue};
-use crate::Result;
-use crate::{Error, ReadOnlyMultimapTable};
-use std::cell::{Cell, RefCell};
-use std::collections::HashMap;
+use crate::types::RedbValue;
+use crate::Error;
+use crate::{ReadTransaction, Result, WriteTransaction};
+use std::collections::btree_set::BTreeSet;
 use std::fs::{File, OpenOptions};
 use std::io::ErrorKind;
 use std::marker::PhantomData;
+use std::ops::RangeFull;
 use std::path::Path;
-use std::rc::Rc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Mutex;
 use std::{io, panic};
 
+pub(crate) type TransactionId = u64;
+type AtomicTransactionId = AtomicU64;
+
 // TODO: add trait bounds once const_fn_trait_bound is stable
 pub struct TableDefinition<'a, K: ?Sized, V: ?Sized> {
     name: &'a str,
@@ -25,13 +28,33 @@ pub struct TableDefinition<'a, K: ?Sized, V: ?Sized> {
 
 impl<'a, K: ?Sized, V: ?Sized> TableDefinition<'a, K, V> {
     pub const fn new(name: &'a str) -> Self {
-        // TODO: enable once const_panic is stable
+        assert!(!name.is_empty());
+        // assert_ne! isn't stable, so we have to do it manually
+        let name_bytes = name.as_bytes();
+        let freed_name_bytes = FREED_TABLE.as_bytes();
+        if name_bytes.len() == freed_name_bytes.len() {
+            let mut equal = true;
+            let mut i = 0;
+            while i < name.len() {
+                if name_bytes[i] != freed_name_bytes[i] {
+                    equal = false;
+                }
+                i += 1;
+            }
+            if equal {
+                panic!("Table name may not be equal to the internal free page table");
+            }
+        }
         Self {
             name,
             _key_type: PhantomData,
             _value_type: PhantomData,
         }
     }
+
+    pub fn name(&self) -> &str {
+        self.name
+    }
 }
 
 impl<'a, K: ?Sized, V: ?Sized> Clone for TableDefinition<'a, K, V> {
@@ -50,12 +73,33 @@ pub struct MultimapTableDefinition<'a, K: ?Sized, V: ?Sized> {
 
 impl<'a, K: ?Sized, V: ?Sized> MultimapTableDefinition<'a, K, V> {
     pub const fn new(name: &'a str) -> Self {
+        assert!(!name.is_empty());
+        // assert_ne! isn't stable, so we have to do it manually
+        let name_bytes = name.as_bytes();
+        let freed_name_bytes = FREED_TABLE.as_bytes();
+        if name_bytes.len() == freed_name_bytes.len() {
+            let mut equal = true;
+            let mut i = 0;
+            while i < name.len() {
+                if name_bytes[i] != freed_name_bytes[i] {
+                    equal = false;
+                }
+                i += 1;
+            }
+            if equal {
+                panic!("Table name may not be equal to the internal free page table");
+            }
+        }
         Self {
             name,
             _key_type: PhantomData,
             _value_type: PhantomData,
         }
     }
+
+    pub fn name(&self) -> &str {
+        self.name
+    }
 }
 
 impl<'a, K: ?Sized, V: ?Sized> Clone for MultimapTableDefinition<'a, K, V> {
@@ -67,7 +111,11 @@ impl<'a, K: ?Sized, V: ?Sized> Clone for MultimapTableDefinition<'a, K, V> {
 impl<'a, K: ?Sized, V: ?Sized> Copy for MultimapTableDefinition<'a, K, V> {}
 
 pub struct Database {
-    storage: Storage,
+    mem: TransactionalMemory,
+    next_transaction_id: AtomicTransactionId,
+    live_read_transactions: Mutex<BTreeSet<TransactionId>>,
+    live_write_transaction: Mutex<Option<TransactionId>>,
+    leaked_write_transaction: Mutex<Option<&'static panic::Location<'static>>>,
 }
 
 impl Database {
@@ -93,17 +141,14 @@ impl Database {
             }
             OpenOptions::new().read(true).write(true).open(path)?
         } else {
-            let file = OpenOptions::new()
+            OpenOptions::new()
                 .read(true)
                 .write(true)
                 .create(true)
-                .open(path)?;
-
-            file
+                .open(path)?
         };
 
-        let storage = Storage::new(file, db_size, None, true)?;
-        Ok(Database { storage })
+        Database::new(file, db_size, None, true)
     }
 
     /// Opens an existing redb database.
@@ -115,31 +160,138 @@ impl Database {
         if File::open(path.as_ref())?.metadata()?.len() > 0 {
             let existing_size = get_db_size(path.as_ref())?;
             let file = OpenOptions::new().read(true).write(true).open(path)?;
-            let storage = Storage::new(file, existing_size, None, true)?;
-            Ok(Database { storage })
+            Database::new(file, existing_size, None, true)
         } else {
             Err(Error::Io(io::Error::from(ErrorKind::InvalidData)))
         }
     }
 
-    pub fn builder() -> DatabaseBuilder {
-        DatabaseBuilder::new()
+    pub(crate) fn get_memory(&self) -> &TransactionalMemory {
+        &self.mem
+    }
+
+    fn new(
+        file: File,
+        max_capacity: usize,
+        page_size: Option<usize>,
+        dynamic_growth: bool,
+    ) -> Result<Self> {
+        let mem = TransactionalMemory::new(file, max_capacity, page_size, dynamic_growth)?;
+        if mem.needs_repair()? {
+            let root = mem
+                .get_primary_root_page()
+                .expect("Tried to repair an empty database");
+            // Clear the freed table. We're about to rebuild the allocator state by walking
+            // all the reachable pages, which will implicitly free them
+            {
+                // Safety: we own the mem object, and just created it, so there can't be other references.
+                let mut bytes = unsafe {
+                    get_mut_value::<str>(mem.get_page_mut(root), FREED_TABLE.as_bytes(), &mem)
+                        .unwrap()
+                };
+                // TODO: this needs to be done in a transaction. Otherwise a torn write could happen
+                let mut header = InternalTableDefinition::from_bytes(bytes.as_mut());
+                header.table_root = None;
+                assert_eq!(bytes.as_mut().len(), header.as_bytes().len());
+                bytes.as_mut().copy_from_slice(&header.as_bytes());
+            }
+
+            // Repair the allocator state
+            // All pages in the master table
+            let mut all_pages_iter: Box<dyn Iterator<Item = PageNumber>> =
+                Box::new(AllPageNumbersBtreeIter::new(root, &mem));
+
+            // Iterate over all other tables
+            let mut iter: BtreeRangeIter<[u8], [u8]> =
+                BtreeRangeIter::new::<RangeFull, [u8]>(.., Some(root), &mem);
+
+            // Chain all the other tables to the master table iter
+            while let Some(entry) = iter.next() {
+                let definition = InternalTableDefinition::from_bytes(entry.value());
+                if let Some(table_root) = definition.get_root() {
+                    let table_pages_iter = AllPageNumbersBtreeIter::new(table_root, &mem);
+                    all_pages_iter = Box::new(all_pages_iter.chain(table_pages_iter));
+                }
+            }
+
+            mem.repair_allocator(all_pages_iter)?;
+        }
+
+        let mut next_transaction_id = mem.get_last_committed_transaction_id()? + 1;
+        if mem.get_primary_root_page().is_none() {
+            assert_eq!(next_transaction_id, 1);
+            // Empty database, so insert the freed table.
+            let freed_table = InternalTableDefinition {
+                table_root: None,
+                table_type: TableType::Normal,
+                key_type: FreedTableKey::redb_type_name().to_string(),
+                value_type: <[u8]>::redb_type_name().to_string(),
+            };
+            let (new_root, _) =
+                make_mut_single_leaf(FREED_TABLE.as_bytes(), &freed_table.as_bytes(), &mem)?;
+            mem.set_secondary_root_page(new_root)?;
+            mem.commit(next_transaction_id, false)?;
+            next_transaction_id += 1;
+        }
+
+        Ok(Database {
+            mem,
+            next_transaction_id: AtomicTransactionId::new(next_transaction_id),
+            live_write_transaction: Mutex::new(None),
+            live_read_transactions: Mutex::new(Default::default()),
+            leaked_write_transaction: Mutex::new(Default::default()),
+        })
     }
 
-    pub fn begin_write(&self) -> Result<WriteTransaction> {
-        WriteTransaction::new(&self.storage)
+    pub(crate) fn record_leaked_write_transaction(&self, transaction_id: TransactionId) {
+        assert_eq!(
+            transaction_id,
+            self.live_write_transaction.lock().unwrap().unwrap()
+        );
+        *self.leaked_write_transaction.lock().unwrap() = Some(panic::Location::caller());
     }
 
-    pub fn begin_read(&self) -> Result<ReadTransaction> {
-        Ok(ReadTransaction::new(&self.storage))
+    pub(crate) fn deallocate_read_transaction(&self, id: TransactionId) {
+        self.live_read_transactions.lock().unwrap().remove(&id);
     }
 
-    pub fn stats(&self) -> Result<DatabaseStats> {
-        self.storage.storage_stats()
+    pub(crate) fn deallocate_write_transaction(&self, id: TransactionId) {
+        let mut live = self.live_write_transaction.lock().unwrap();
+        assert_eq!(Some(id), *live);
+        *live = None;
     }
 
-    pub fn print_debug(&self) {
-        self.storage.print_debug()
+    pub(crate) fn oldest_live_read_transaction(&self) -> Option<TransactionId> {
+        self.live_read_transactions
+            .lock()
+            .unwrap()
+            .iter()
+            .next()
+            .cloned()
+    }
+
+    pub fn builder() -> DatabaseBuilder {
+        DatabaseBuilder::new()
+    }
+
+    pub fn begin_write(&self) -> Result<WriteTransaction> {
+        let guard = self.leaked_write_transaction.lock().unwrap();
+        if let Some(leaked) = *guard {
+            return Err(Error::LeakedWriteTransaction(leaked));
+        }
+        drop(guard);
+
+        assert!(self.live_write_transaction.lock().unwrap().is_none());
+        let id = self.next_transaction_id.fetch_add(1, Ordering::AcqRel);
+        *self.live_write_transaction.lock().unwrap() = Some(id);
+        // Safety: We just asserted there was no previous write in progress
+        unsafe { WriteTransaction::new(self, id) }
+    }
+
+    pub fn begin_read(&self) -> Result<ReadTransaction> {
+        let id = self.next_transaction_id.fetch_add(1, Ordering::AcqRel);
+        self.live_read_transactions.lock().unwrap().insert(id);
+        Ok(ReadTransaction::new(self, id))
     }
 }
 
@@ -189,357 +341,6 @@ impl DatabaseBuilder {
             .create(true)
             .open(path)?;
 
-        let storage = Storage::new(file, db_size, self.page_size, self.dynamic_growth)?;
-        Ok(Database { storage })
-    }
-}
-
-#[derive(Copy, Clone)]
-pub enum Durability {
-    /// Commits with this durability level will not be persisted to disk unless followed by a
-    /// commit with a higher durability level.
-    ///
-    /// Note: Pages are only freed during commits with higher durability levels. Exclusively using
-    /// this function may result in Error::OutOfSpace.
-    None,
-    /// Commits with this durability level have been queued for persitance to disk, and should be
-    /// persistant some time after [WriteTransaction::commit] returns.
-    Eventual,
-    /// Commits with this durability level are guaranteed to be persistant as soon as
-    /// [WriteTransaction::commit] returns.
-    Immediate,
-}
-
-#[allow(clippy::type_complexity)]
-pub struct WriteTransaction<'a> {
-    storage: &'a Storage,
-    transaction_id: TransactionId,
-    root_page: Cell<Option<PageNumber>>,
-    pending_table_root_changes: RefCell<HashMap<String, Rc<Cell<Option<PageNumber>>>>>,
-    open_tables: RefCell<HashMap<String, &'static panic::Location<'static>>>,
-    completed: AtomicBool,
-    durability: Durability,
-}
-
-impl<'a> WriteTransaction<'a> {
-    fn new(storage: &'a Storage) -> Result<Self> {
-        let transaction_id = storage.allocate_write_transaction()?;
-        let root_page = storage.get_root_page_number();
-        Ok(Self {
-            storage,
-            transaction_id,
-            root_page: Cell::new(root_page),
-            pending_table_root_changes: RefCell::new(Default::default()),
-            open_tables: RefCell::new(Default::default()),
-            completed: Default::default(),
-            durability: Durability::Immediate,
-        })
-    }
-
-    pub fn set_durability(&mut self, durability: Durability) {
-        self.durability = durability;
-    }
-
-    /// Open the given table
-    ///
-    /// The table will be created if it does not exist
-    pub fn open_table<'s: 't, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &'s self,
-        definition: TableDefinition<K, V>,
-    ) -> Result<Table<'a, 't, K, V>> {
-        assert!(!definition.name.is_empty());
-        assert_ne!(definition.name, FREED_TABLE);
-        if let Some(location) = self.open_tables.borrow().get(definition.name) {
-            return Err(Error::TableAlreadyOpen(
-                definition.name.to_string(),
-                location,
-            ));
-        }
-        self.open_tables
-            .borrow_mut()
-            .insert(definition.name.to_string(), panic::Location::caller());
-
-        let (header, root) = self.storage.get_or_create_table::<K, V>(
-            definition.name,
-            TableType::Normal,
-            self.transaction_id,
-            self.root_page.get(),
-        )?;
-        self.root_page.set(Some(root));
-
-        let root = self
-            .pending_table_root_changes
-            .borrow_mut()
-            .entry(definition.name.to_string())
-            .or_insert_with(|| Rc::new(Cell::new(header.get_root())))
-            .clone();
-
-        Ok(Table::new(
-            definition.name,
-            self.transaction_id,
-            root,
-            self.storage,
-            self,
-        ))
-    }
-
-    /// Open the given table
-    ///
-    /// The table will be created if it does not exist
-    pub fn open_multimap_table<'t, K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
-        &'t self,
-        definition: MultimapTableDefinition<K, V>,
-    ) -> Result<MultimapTable<'a, 't, K, V>> {
-        assert!(!definition.name.is_empty());
-        assert_ne!(definition.name, FREED_TABLE);
-        if let Some(location) = self.open_tables.borrow().get(definition.name) {
-            return Err(Error::TableAlreadyOpen(
-                definition.name.to_string(),
-                location,
-            ));
-        }
-        self.open_tables
-            .borrow_mut()
-            .insert(definition.name.to_string(), panic::Location::caller());
-
-        let (header, root) = self.storage.get_or_create_table::<K, V>(
-            definition.name,
-            TableType::Multimap,
-            self.transaction_id,
-            self.root_page.get(),
-        )?;
-        self.root_page.set(Some(root));
-
-        let root = self
-            .pending_table_root_changes
-            .borrow_mut()
-            .entry(definition.name.to_string())
-            .or_insert_with(|| Rc::new(Cell::new(header.get_root())))
-            .clone();
-
-        Ok(MultimapTable::new(
-            definition.name,
-            self.transaction_id,
-            root,
-            self.storage,
-            self,
-        ))
-    }
-
-    pub(crate) fn close_table(&self, name: &str) {
-        self.open_tables.borrow_mut().remove(name).unwrap();
-    }
-
-    /// Delete the given table
-    ///
-    /// Returns a bool indicating whether the table existed
-    pub fn delete_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        definition: TableDefinition<K, V>,
-    ) -> Result<bool> {
-        assert!(!definition.name.is_empty());
-        assert_ne!(definition.name, FREED_TABLE);
-        let original_root = self.root_page.get();
-        let (root, found) = self.storage.delete_table::<K, V>(
-            definition.name,
-            TableType::Normal,
-            self.transaction_id,
-            original_root,
-        )?;
-
-        self.root_page.set(root);
-
-        Ok(found)
-    }
-
-    /// Delete the given table
-    ///
-    /// Returns a bool indicating whether the table existed
-    pub fn delete_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
-        &self,
-        definition: MultimapTableDefinition<K, V>,
-    ) -> Result<bool> {
-        assert!(!definition.name.is_empty());
-        let original_root = self.root_page.get();
-        let (root, found) = self.storage.delete_table::<K, V>(
-            definition.name,
-            TableType::Multimap,
-            self.transaction_id,
-            original_root,
-        )?;
-
-        self.root_page.set(root);
-
-        Ok(found)
-    }
-
-    /// List all the tables
-    // TODO: should return an iterator of &str, once GATs are available
-    pub fn list_tables(&self) -> Result<impl Iterator<Item = String> + '_> {
-        self.storage
-            .list_tables(TableType::Normal, self.root_page.get())
-    }
-
-    /// List all the multimap tables
-    // TODO: should return an iterator of &str, once GATs are available
-    pub fn list_multimap_tables(&self) -> Result<impl Iterator<Item = String> + '_> {
-        self.storage
-            .list_tables(TableType::Multimap, self.root_page.get())
-    }
-
-    pub fn commit(self) -> Result {
-        match self.commit_inner() {
-            Ok(_) => Ok(()),
-            Err(err) => match err {
-                // Rollback the transaction if we ran out of space during commit, so that user may
-                // continue with another transaction (like a delete)
-                Error::OutOfSpace => {
-                    self.abort()?;
-                    Err(err)
-                }
-                err => Err(err),
-            },
-        }
-    }
-
-    fn commit_inner(&self) -> Result {
-        // Update all the table roots in the master table, before committing
-        for (name, update) in self.pending_table_root_changes.borrow_mut().drain() {
-            let new_root = self.storage.update_table_root(
-                &name,
-                update.get(),
-                self.transaction_id,
-                self.root_page.get(),
-            )?;
-            self.root_page.set(Some(new_root));
-        }
-
-        match self.durability {
-            Durability::None => self
-                .storage
-                .non_durable_commit(self.root_page.get(), self.transaction_id)?,
-            Durability::Eventual => {
-                self.storage
-                    .durable_commit(self.root_page.get(), self.transaction_id, true)?
-            }
-            Durability::Immediate => {
-                self.storage
-                    .durable_commit(self.root_page.get(), self.transaction_id, false)?
-            }
-        }
-
-        self.completed.store(true, Ordering::Release);
-
-        Ok(())
-    }
-
-    pub fn abort(self) -> Result {
-        self.storage
-            .rollback_uncommited_writes(self.transaction_id)?;
-        self.completed.store(true, Ordering::Release);
-        Ok(())
-    }
-}
-
-impl<'a> Drop for WriteTransaction<'a> {
-    fn drop(&mut self) {
-        if !self.completed.load(Ordering::Acquire) {
-            self.storage
-                .record_leaked_write_transaction(self.transaction_id);
-        }
-    }
-}
-
-pub struct ReadTransaction<'a> {
-    storage: &'a Storage,
-    transaction_id: TransactionId,
-    root_page: Option<PageNumber>,
-}
-
-impl<'a> ReadTransaction<'a> {
-    fn new(storage: &'a Storage) -> Self {
-        let transaction_id = storage.allocate_read_transaction();
-        let root_page = storage.get_root_page_number();
-        Self {
-            storage,
-            transaction_id,
-            root_page,
-        }
-    }
-
-    /// Open the given table
-    pub fn open_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        definition: TableDefinition<K, V>,
-    ) -> Result<ReadOnlyTable<'a, K, V>> {
-        assert!(!definition.name.is_empty());
-        assert_ne!(definition.name, FREED_TABLE);
-        let header = self
-            .storage
-            .get_table::<K, V>(definition.name, TableType::Normal, self.root_page)?
-            .ok_or_else(|| Error::TableDoesNotExist(definition.name.to_string()))?;
-
-        Ok(ReadOnlyTable::new(header.get_root(), self.storage))
-    }
-
-    /// Open the given table
-    pub fn open_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
-        &self,
-        definition: MultimapTableDefinition<K, V>,
-    ) -> Result<ReadOnlyMultimapTable<'a, K, V>> {
-        assert!(!definition.name.is_empty());
-        assert_ne!(definition.name, FREED_TABLE);
-        let header = self
-            .storage
-            .get_table::<K, V>(definition.name, TableType::Multimap, self.root_page)?
-            .ok_or_else(|| Error::TableDoesNotExist(definition.name.to_string()))?;
-
-        Ok(ReadOnlyMultimapTable::new(header.get_root(), self.storage))
-    }
-
-    /// List all the tables
-    // TODO: should return an iterator of &str, once GATs are available
-    pub fn list_tables(&self) -> Result<impl Iterator<Item = String> + '_> {
-        self.storage.list_tables(TableType::Normal, self.root_page)
-    }
-
-    /// List all the multimap tables
-    // TODO: should return an iterator of &str, once GATs are available
-    pub fn list_multimap_tables(&self) -> Result<impl Iterator<Item = String> + '_> {
-        self.storage
-            .list_tables(TableType::Multimap, self.root_page)
-    }
-}
-
-impl<'a> Drop for ReadTransaction<'a> {
-    fn drop(&mut self) {
-        self.storage
-            .deallocate_read_transaction(self.transaction_id);
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use crate::{Database, TableDefinition};
-    use tempfile::NamedTempFile;
-
-    const X: TableDefinition<[u8], [u8]> = TableDefinition::new("x");
-
-    #[test]
-    fn transaction_id_persistence() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db.begin_write().unwrap();
-        {
-            let mut table = write_txn.open_table(X).unwrap();
-            table.insert(b"hello", b"world").unwrap();
-        }
-        let first_txn_id = write_txn.transaction_id;
-        write_txn.commit().unwrap();
-        drop(db);
-
-        let db2 = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
-        let write_txn = db2.begin_write().unwrap();
-        assert!(write_txn.transaction_id > first_txn_id);
+        Database::new(file, db_size, self.page_size, self.dynamic_growth)
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 3ea6a77..7c732c1 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,14 +1,12 @@
-pub use db::{
-    Database, DatabaseBuilder, Durability, MultimapTableDefinition, ReadTransaction,
-    TableDefinition, WriteTransaction,
-};
+pub use db::{Database, DatabaseBuilder, MultimapTableDefinition, TableDefinition};
 pub use error::Error;
 pub use multimap_table::{
     MultimapRangeIter, MultimapTable, MultimapValueIter, ReadOnlyMultimapTable,
     ReadableMultimapTable,
 };
 pub use table::{RangeIter, ReadOnlyTable, ReadableTable, Table};
-pub use tree_store::{AccessGuard, DatabaseStats};
+pub use transactions::{DatabaseStats, Durability, ReadTransaction, WriteTransaction};
+pub use tree_store::AccessGuard;
 
 type Result<T = (), E = Error> = std::result::Result<T, E>;
 
@@ -21,5 +19,6 @@ mod multimap_table;
 #[cfg(feature = "python")]
 mod python;
 mod table;
+mod transactions;
 mod tree_store;
 mod types;
diff --git a/src/multimap_table.rs b/src/multimap_table.rs
index f0d97a0..2539d92 100644
--- a/src/multimap_table.rs
+++ b/src/multimap_table.rs
@@ -1,15 +1,13 @@
-use crate::tree_store::{BtreeRangeIter, PageNumber, Storage, TransactionId};
+use crate::tree_store::{Btree, BtreeMut, BtreeRangeIter, PageNumber, TransactionalMemory};
 use crate::types::{
     AsBytesWithLifetime, RedbKey, RedbValue, RefAsBytesLifetime, RefLifetime, WithLifetime,
 };
 use crate::{Result, WriteTransaction};
-use std::cell::Cell;
 use std::cmp::Ordering;
 use std::collections::Bound;
 use std::convert::TryInto;
 use std::marker::PhantomData;
 use std::ops::RangeBounds;
-use std::rc::Rc;
 
 #[derive(Eq, PartialEq)]
 #[allow(clippy::enum_variant_names)]
@@ -160,13 +158,16 @@ impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> MultimapKVPairAcces
     }
 }
 
-fn make_serialized_kv<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(key: &K, value: &V) -> Vec<u8> {
+fn make_serialized_kv<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
+    key: &K,
+    value: &V,
+) -> MultimapKVPair<K, V> {
     let mut result = vec![MultimapKeyCompareOp::KeyAndValue.serialize()];
     result.extend_from_slice(&(key.as_bytes().as_ref().len() as u32).to_le_bytes());
     result.extend_from_slice(key.as_bytes().as_ref());
     result.extend_from_slice(value.as_bytes().as_ref());
 
-    result
+    MultimapKVPair::new(result)
 }
 
 fn make_serialized_key_with_op<K: RedbKey + ?Sized>(key: &K, op: MultimapKeyCompareOp) -> Vec<u8> {
@@ -289,38 +290,26 @@ impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> MultimapRangeIter<'
 pub struct MultimapTable<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
     name: String,
     transaction: &'t WriteTransaction<'s>,
-    storage: &'s Storage,
-    transaction_id: TransactionId,
-    table_root: Rc<Cell<Option<PageNumber>>>,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
+    tree: BtreeMut<'t, MultimapKVPair<K, V>, [u8]>,
 }
 
 impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'s, 't, K, V> {
     pub(crate) fn new(
         name: &str,
-        transaction_id: TransactionId,
-        table_root: Rc<Cell<Option<PageNumber>>>,
-        storage: &'s Storage,
+        table_root: Option<PageNumber>,
+        mem: &'s TransactionalMemory,
         transaction: &'t WriteTransaction<'s>,
     ) -> MultimapTable<'s, 't, K, V> {
         MultimapTable {
             name: name.to_string(),
             transaction,
-            storage,
-            transaction_id,
-            table_root,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
+            tree: BtreeMut::new(table_root, mem),
         }
     }
 
     #[allow(dead_code)]
     pub(crate) fn print_debug(&self) {
-        if let Some(page) = self.table_root.get() {
-            self.storage
-                .print_dirty_tree_debug::<MultimapKVPair<K, V>>(page);
-        }
+        self.tree.print_debug();
     }
 
     pub fn insert(&mut self, key: &K, value: &V) -> Result {
@@ -328,16 +317,7 @@ impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'s, 't, K,
         // Safety: No other references to this table can exist.
         // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
         // and we borrow &mut self.
-        let root_page = unsafe {
-            self.storage.insert::<MultimapKVPair<K, V>>(
-                &kv,
-                b"",
-                self.transaction_id,
-                self.table_root.get(),
-            )
-        }?;
-        self.table_root.set(Some(root_page));
-        Ok(())
+        unsafe { self.tree.insert(&kv, b"") }
     }
 
     pub fn remove(&mut self, key: &K, value: &V) -> Result<bool> {
@@ -345,47 +325,29 @@ impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'s, 't, K,
         // Safety: No other references to this table can exist.
         // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
         // and we borrow &mut self.
-        let (root_page, found) = unsafe {
-            self.storage.remove::<MultimapKVPair<K, V>, [u8]>(
-                &kv,
-                self.transaction_id,
-                true,
-                self.table_root.get(),
-            )?
-        };
-        self.table_root.set(root_page);
-        Ok(found.is_some())
+        unsafe { self.tree.remove(&kv).map(|x| x.is_some()) }
     }
 
     pub fn remove_all(&mut self, key: &K) -> Result<MultimapValueIter<K, V>> {
         // Match only on the key, so that we can remove all the associated values
         let key_only = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyOnly);
-        let original_root = self.table_root.get();
+        let key_only = MultimapKVPair::new(key_only);
+        // Save a snapshot of the btree. This is safe since we call remove_retain_uncommitted()
+        // instead of remove()
+        let original_tree = Btree::new(self.tree.root, self.tree.mem);
         loop {
-            // Safety: No other references to this table can exist.
-            // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
-            // and we borrow &mut self.
-            let (new_root, found) = unsafe {
-                self.storage.remove::<MultimapKVPair<K, V>, [u8]>(
-                    &key_only,
-                    self.transaction_id,
-                    // TODO: we should return an iterator that frees these pages even if they are uncommitted
-                    false,
-                    self.table_root.get(),
-                )?
-            };
+            let found = self.tree.remove_retain_uncommitted(&key_only)?;
             if found.is_none() {
                 break;
             }
-            self.table_root.set(new_root);
         }
 
         let lower_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyMinusEpsilon);
         let upper_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyPlusEpsilon);
         let lower = MultimapKVPair::<K, V>::new(lower_bytes);
         let upper = MultimapKVPair::<K, V>::new(upper_bytes);
-        self.storage
-            .get_range(lower..=upper, original_root)
+        original_tree
+            .range(lower..=upper)
             .map(MultimapValueIter::new)
     }
 }
@@ -398,9 +360,7 @@ impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K,
         let upper_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyPlusEpsilon);
         let lower = MultimapKVPair::<K, V>::new(lower_bytes);
         let upper = MultimapKVPair::<K, V>::new(upper_bytes);
-        self.storage
-            .get_range(lower..=upper, self.table_root.get())
-            .map(MultimapValueIter::new)
+        self.tree.range(lower..=upper).map(MultimapValueIter::new)
     }
 
     fn range<'a, T: RangeBounds<&'a K> + 'a>(
@@ -413,23 +373,21 @@ impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K,
         let start = make_bound(start_kv);
         let end = make_bound(end_kv);
 
-        self.storage
-            .get_range((start, end), self.table_root.get())
-            .map(MultimapRangeIter::new)
+        self.tree.range((start, end)).map(MultimapRangeIter::new)
     }
 
     fn len(&self) -> Result<usize> {
-        self.storage.len(self.table_root.get())
+        self.tree.len()
     }
 
     fn is_empty(&self) -> Result<bool> {
-        self.storage.len(self.table_root.get()).map(|x| x == 0)
+        self.len().map(|x| x == 0)
     }
 }
 
 impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> Drop for MultimapTable<'s, 't, K, V> {
     fn drop(&mut self) {
-        self.transaction.close_table(&self.name);
+        self.transaction.close_table(&self.name, &mut self.tree);
     }
 }
 
@@ -447,22 +405,16 @@ pub trait ReadableMultimapTable<K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
 }
 
 pub struct ReadOnlyMultimapTable<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
-    storage: &'s Storage,
-    table_root: Option<PageNumber>,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
+    tree: Btree<'s, MultimapKVPair<K, V>, [u8]>,
 }
 
 impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadOnlyMultimapTable<'s, K, V> {
     pub(crate) fn new(
         root_page: Option<PageNumber>,
-        storage: &'s Storage,
+        mem: &'s TransactionalMemory,
     ) -> ReadOnlyMultimapTable<'s, K, V> {
         ReadOnlyMultimapTable {
-            storage,
-            table_root: root_page,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
+            tree: Btree::new(root_page, mem),
         }
     }
 }
@@ -475,9 +427,7 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
         let upper_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyPlusEpsilon);
         let lower = MultimapKVPair::<K, V>::new(lower_bytes);
         let upper = MultimapKVPair::<K, V>::new(upper_bytes);
-        self.storage
-            .get_range(lower..=upper, self.table_root)
-            .map(MultimapValueIter::new)
+        self.tree.range(lower..=upper).map(MultimapValueIter::new)
     }
 
     fn range<'a, T: RangeBounds<&'a K> + 'a>(
@@ -490,16 +440,14 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
         let start = make_bound(start_kv);
         let end = make_bound(end_kv);
 
-        self.storage
-            .get_range((start, end), self.table_root)
-            .map(MultimapRangeIter::new)
+        self.tree.range((start, end)).map(MultimapRangeIter::new)
     }
 
     fn len(&self) -> Result<usize> {
-        self.storage.len(self.table_root)
+        self.tree.len()
     }
 
     fn is_empty(&self) -> Result<bool> {
-        self.storage.len(self.table_root).map(|x| x == 0)
+        self.len().map(|x| x == 0)
     }
 }
diff --git a/src/table.rs b/src/table.rs
index 206eacf..58bc96c 100644
--- a/src/table.rs
+++ b/src/table.rs
@@ -1,63 +1,42 @@
-use crate::tree_store::{AccessGuardMut, BtreeRangeIter, PageNumber, Storage, TransactionId};
+use crate::tree_store::{
+    AccessGuardMut, Btree, BtreeMut, BtreeRangeIter, PageNumber, TransactionalMemory,
+};
 use crate::types::{RedbKey, RedbValue, WithLifetime};
 use crate::Result;
 use crate::{AccessGuard, WriteTransaction};
 use std::borrow::Borrow;
-use std::cell::Cell;
-use std::marker::PhantomData;
 use std::ops::RangeBounds;
-use std::rc::Rc;
 
 pub struct Table<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
     name: String,
     transaction: &'t WriteTransaction<'s>,
-    storage: &'s Storage,
-    transaction_id: TransactionId,
-    table_root: Rc<Cell<Option<PageNumber>>>,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
+    tree: BtreeMut<'t, K, V>,
 }
 
 impl<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Table<'s, 't, K, V> {
     pub(crate) fn new(
         name: &str,
-        transaction_id: TransactionId,
-        table_root: Rc<Cell<Option<PageNumber>>>,
-        storage: &'s Storage,
+        table_root: Option<PageNumber>,
+        mem: &'s TransactionalMemory,
         transaction: &'t WriteTransaction<'s>,
     ) -> Table<'s, 't, K, V> {
         Table {
             name: name.to_string(),
             transaction,
-            storage,
-            transaction_id,
-            table_root,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
+            tree: BtreeMut::new(table_root, mem),
         }
     }
 
     #[allow(dead_code)]
     pub(crate) fn print_debug(&self) {
-        if let Some(page) = self.table_root.get() {
-            self.storage.print_dirty_tree_debug::<K>(page);
-        }
+        self.tree.print_debug();
     }
 
     pub fn insert(&mut self, key: &K, value: &V) -> Result {
         // Safety: No other references to this table can exist.
         // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
         // and we borrow &mut self.
-        let root_page = unsafe {
-            self.storage.insert::<K>(
-                key.as_bytes().as_ref(),
-                value.as_bytes().as_ref(),
-                self.transaction_id,
-                self.table_root.get(),
-            )
-        }?;
-        self.table_root.set(Some(root_page));
-        Ok(())
+        unsafe { self.tree.insert(key, value) }
     }
 
     /// Reserve space to insert a key-value pair
@@ -66,32 +45,14 @@ impl<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Table<'s, 't, K, V> {
         // Safety: No other references to this table can exist.
         // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
         // and we borrow &mut self.
-        let (root_page, guard) = unsafe {
-            self.storage.insert_reserve::<K>(
-                key.as_bytes().as_ref(),
-                value_length,
-                self.transaction_id,
-                self.table_root.get(),
-            )
-        }?;
-        self.table_root.set(Some(root_page));
-        Ok(guard)
+        unsafe { self.tree.insert_reserve(key, value_length) }
     }
 
     pub fn remove(&mut self, key: &K) -> Result<Option<AccessGuard<V>>> {
         // Safety: No other references to this table can exist.
         // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
         // and we borrow &mut self.
-        let (root_page, found) = unsafe {
-            self.storage.remove::<K, V>(
-                key.as_bytes().as_ref(),
-                self.transaction_id,
-                true,
-                self.table_root.get(),
-            )
-        }?;
-        self.table_root.set(root_page);
-        Ok(found)
+        unsafe { self.tree.remove(key) }
     }
 }
 
@@ -99,31 +60,28 @@ impl<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadableTable<K, V>
     for Table<'s, 't, K, V>
 {
     fn get(&self, key: &K) -> Result<Option<<<V as RedbValue>::View as WithLifetime>::Out>> {
-        self.storage
-            .get::<K, V>(key.as_bytes().as_ref(), self.table_root.get())
+        self.tree.get(key)
     }
 
     fn range<'a, T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
         &'a self,
         range: T,
     ) -> Result<RangeIter<K, V>> {
-        self.storage
-            .get_range(range, self.table_root.get())
-            .map(RangeIter::new)
+        self.tree.range(range).map(RangeIter::new)
     }
 
     fn len(&self) -> Result<usize> {
-        self.storage.len(self.table_root.get())
+        self.tree.len()
     }
 
     fn is_empty(&self) -> Result<bool> {
-        self.storage.len(self.table_root.get()).map(|x| x == 0)
+        self.len().map(|x| x == 0)
     }
 }
 
 impl<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Drop for Table<'s, 't, K, V> {
     fn drop(&mut self) {
-        self.transaction.close_table(&self.name);
+        self.transaction.close_table(&self.name, &mut self.tree);
     }
 }
 
@@ -141,22 +99,16 @@ pub trait ReadableTable<K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
 }
 
 pub struct ReadOnlyTable<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
-    storage: &'s Storage,
-    table_root: Option<PageNumber>,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
+    tree: Btree<'s, K, V>,
 }
 
 impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadOnlyTable<'s, K, V> {
     pub(crate) fn new(
         root_page: Option<PageNumber>,
-        storage: &'s Storage,
+        mem: &'s TransactionalMemory,
     ) -> ReadOnlyTable<'s, K, V> {
         ReadOnlyTable {
-            storage,
-            table_root: root_page,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
+            tree: Btree::new(root_page, mem),
         }
     }
 }
@@ -165,25 +117,22 @@ impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadableTable<K, V>
     for ReadOnlyTable<'s, K, V>
 {
     fn get(&self, key: &K) -> Result<Option<<<V as RedbValue>::View as WithLifetime>::Out>> {
-        self.storage
-            .get::<K, V>(key.as_bytes().as_ref(), self.table_root)
+        self.tree.get(key)
     }
 
     fn range<'a, T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
         &'a self,
         range: T,
     ) -> Result<RangeIter<K, V>> {
-        self.storage
-            .get_range(range, self.table_root)
-            .map(RangeIter::new)
+        self.tree.range(range).map(RangeIter::new)
     }
 
     fn len(&self) -> Result<usize> {
-        self.storage.len(self.table_root)
+        self.tree.len()
     }
 
     fn is_empty(&self) -> Result<bool> {
-        self.storage.len(self.table_root).map(|x| x == 0)
+        self.len().map(|x| x == 0)
     }
 }
 
diff --git a/src/transactions.rs b/src/transactions.rs
new file mode 100644
index 0000000..8a37f8f
--- /dev/null
+++ b/src/transactions.rs
@@ -0,0 +1,610 @@
+use crate::db::TransactionId;
+use crate::tree_store::btree_utils::{fragmented_bytes, overhead_bytes, stored_bytes, tree_height};
+use crate::tree_store::{
+    Btree, BtreeMut, FreedTableKey, InternalTableDefinition, PageNumber, TableTree, TableType,
+    TransactionalMemory, FREED_TABLE,
+};
+use crate::types::{RedbKey, RedbValue};
+use crate::{
+    Database, Error, MultimapTable, MultimapTableDefinition, ReadOnlyMultimapTable, ReadOnlyTable,
+    Result, Table, TableDefinition,
+};
+use std::cell::{Cell, RefCell};
+use std::cmp::{max, min};
+use std::collections::HashMap;
+use std::mem::size_of;
+use std::ops::RangeFull;
+use std::panic;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+#[derive(Debug)]
+pub struct DatabaseStats {
+    tree_height: usize,
+    free_pages: usize,
+    stored_leaf_bytes: usize,
+    metadata_bytes: usize,
+    fragmented_bytes: usize,
+    page_size: usize,
+}
+
+impl DatabaseStats {
+    /// Maximum traversal distance to reach the deepest (key, value) pair, across all tables
+    pub fn tree_height(&self) -> usize {
+        self.tree_height
+    }
+
+    /// Number of free pages remaining
+    pub fn free_pages(&self) -> usize {
+        self.free_pages
+    }
+
+    /// Number of bytes consumed by keys and values that have been inserted.
+    /// Does not include indexing overhead
+    pub fn stored_bytes(&self) -> usize {
+        self.stored_leaf_bytes
+    }
+
+    /// Number of bytes consumed by keys in internal index pages, plus other metadata
+    pub fn metadata_bytes(&self) -> usize {
+        self.metadata_bytes
+    }
+
+    /// Number of bytes consumed by fragmentation, both in data pages and internal metadata tables
+    pub fn fragmented_bytes(&self) -> usize {
+        self.fragmented_bytes
+    }
+
+    /// Number of bytes per page
+    pub fn page_size(&self) -> usize {
+        self.page_size
+    }
+}
+
+#[derive(Copy, Clone)]
+pub enum Durability {
+    /// Commits with this durability level will not be persisted to disk unless followed by a
+    /// commit with a higher durability level.
+    ///
+    /// Note: Pages are only freed during commits with higher durability levels. Exclusively using
+    /// this function may result in Error::OutOfSpace.
+    None,
+    /// Commits with this durability level have been queued for persitance to disk, and should be
+    /// persistant some time after [WriteTransaction::commit] returns.
+    Eventual,
+    /// Commits with this durability level are guaranteed to be persistant as soon as
+    /// [WriteTransaction::commit] returns.
+    Immediate,
+}
+
+pub struct WriteTransaction<'db> {
+    db: &'db Database,
+    mem: &'db TransactionalMemory,
+    transaction_id: TransactionId,
+    root_page: Cell<Option<PageNumber>>,
+    table_tree: RefCell<TableTree<'db>>,
+    freed_pages: RefCell<Vec<PageNumber>>,
+    open_tables: RefCell<HashMap<String, &'static panic::Location<'static>>>,
+    pending_table_updates: RefCell<HashMap<String, Option<PageNumber>>>,
+    completed: AtomicBool,
+    durability: Durability,
+}
+
+impl<'db> WriteTransaction<'db> {
+    // Safety: caller must guarantee that there is only a single WriteTransaction in existence
+    // at a time
+    pub(crate) unsafe fn new(db: &'db Database, transaction_id: TransactionId) -> Result<Self> {
+        let root_page = db.get_memory().get_primary_root_page();
+        Ok(Self {
+            db,
+            mem: db.get_memory(),
+            transaction_id,
+            root_page: Cell::new(root_page),
+            table_tree: RefCell::new(TableTree::new(root_page, db.get_memory())),
+            freed_pages: RefCell::new(vec![]),
+            open_tables: RefCell::new(Default::default()),
+            pending_table_updates: RefCell::new(Default::default()),
+            completed: Default::default(),
+            durability: Durability::Immediate,
+        })
+    }
+
+    pub fn set_durability(&mut self, durability: Durability) {
+        self.durability = durability;
+    }
+
+    /// Open the given table
+    ///
+    /// The table will be created if it does not exist
+    // TODO: simplify the number of lifetimes here. Also why is it different from multimap_table?
+    pub fn open_table<'s: 't, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &'s self,
+        definition: TableDefinition<K, V>,
+    ) -> Result<Table<'db, 't, K, V>> {
+        if let Some(location) = self.open_tables.borrow().get(definition.name()) {
+            return Err(Error::TableAlreadyOpen(
+                definition.name().to_string(),
+                location,
+            ));
+        }
+        self.open_tables
+            .borrow_mut()
+            .insert(definition.name().to_string(), panic::Location::caller());
+
+        let internal_table = self
+            .table_tree
+            .borrow_mut()
+            .get_or_create_table::<K, V>(definition.name(), TableType::Normal)?;
+
+        Ok(Table::new(
+            definition.name(),
+            internal_table.get_root(),
+            self.mem,
+            self,
+        ))
+    }
+
+    /// Open the given table
+    ///
+    /// The table will be created if it does not exist
+    pub fn open_multimap_table<'t, K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
+        &'t self,
+        definition: MultimapTableDefinition<K, V>,
+    ) -> Result<MultimapTable<'db, 't, K, V>> {
+        if let Some(location) = self.open_tables.borrow().get(definition.name()) {
+            return Err(Error::TableAlreadyOpen(
+                definition.name().to_string(),
+                location,
+            ));
+        }
+        self.open_tables
+            .borrow_mut()
+            .insert(definition.name().to_string(), panic::Location::caller());
+
+        let internal_table = self
+            .table_tree
+            .borrow_mut()
+            .get_or_create_table::<K, V>(definition.name(), TableType::Multimap)?;
+
+        Ok(MultimapTable::new(
+            definition.name(),
+            internal_table.get_root(),
+            self.mem,
+            self,
+        ))
+    }
+
+    pub(crate) fn close_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &self,
+        name: &str,
+        table: &mut BtreeMut<K, V>,
+    ) {
+        self.open_tables.borrow_mut().remove(name).unwrap();
+        self.freed_pages.borrow_mut().append(&mut table.freed_pages);
+        self.pending_table_updates
+            .borrow_mut()
+            .insert(name.to_string(), table.root);
+    }
+
+    /// Delete the given table
+    ///
+    /// Returns a bool indicating whether the table existed
+    pub fn delete_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &self,
+        definition: TableDefinition<K, V>,
+    ) -> Result<bool> {
+        self.table_tree
+            .borrow_mut()
+            .delete_table::<K, V>(definition.name(), TableType::Normal)
+    }
+
+    /// Delete the given table
+    ///
+    /// Returns a bool indicating whether the table existed
+    pub fn delete_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
+        &self,
+        definition: MultimapTableDefinition<K, V>,
+    ) -> Result<bool> {
+        self.table_tree
+            .borrow_mut()
+            .delete_table::<K, V>(definition.name(), TableType::Multimap)
+    }
+
+    /// List all the tables
+    // TODO: should return an iterator of &str, once GATs are available
+    pub fn list_tables(&self) -> Result<impl Iterator<Item = String> + '_> {
+        self.table_tree
+            .borrow()
+            .list_tables(TableType::Normal)
+            .map(|x| x.into_iter())
+    }
+
+    /// List all the multimap tables
+    // TODO: should return an iterator of &str, once GATs are available
+    pub fn list_multimap_tables(&self) -> Result<impl Iterator<Item = String> + '_> {
+        self.table_tree
+            .borrow()
+            .list_tables(TableType::Multimap)
+            .map(|x| x.into_iter())
+    }
+
+    pub fn commit(self) -> Result {
+        for (name, root) in self.pending_table_updates.borrow_mut().drain() {
+            self.table_tree
+                .borrow_mut()
+                .update_table_root(&name, root)?;
+        }
+        match self.commit_inner() {
+            Ok(_) => {
+                self.db.deallocate_write_transaction(self.transaction_id);
+                Ok(())
+            }
+            Err(err) => match err {
+                // Rollback the transaction if we ran out of space during commit, so that user may
+                // continue with another transaction (like a delete)
+                Error::OutOfSpace => {
+                    self.abort()?;
+                    Err(err)
+                }
+                err => Err(err),
+            },
+        }
+    }
+
+    fn commit_inner(&self) -> Result {
+        self.root_page.set(self.table_tree.borrow().tree.root);
+
+        match self.durability {
+            Durability::None => self.non_durable_commit()?,
+            Durability::Eventual => self.durable_commit(true)?,
+            Durability::Immediate => self.durable_commit(false)?,
+        }
+
+        self.completed.store(true, Ordering::Release);
+
+        Ok(())
+    }
+
+    pub fn abort(self) -> Result {
+        // No-op, just to avoid triggering the leak detection in BtreeMut
+        self.take_all_freed_pages();
+        self.mem.rollback_uncommited_writes()?;
+        self.db.deallocate_write_transaction(self.transaction_id);
+        self.completed.store(true, Ordering::Release);
+        Ok(())
+    }
+
+    // TODO: this method is kind of a hack
+    fn take_all_freed_pages(&self) -> Vec<PageNumber> {
+        let mut result = vec![];
+        result.append(self.freed_pages.borrow_mut().as_mut());
+        result.append(&mut self.table_tree.borrow_mut().extra_freed_pages);
+        result.append(&mut self.table_tree.borrow_mut().tree.freed_pages);
+        result
+    }
+
+    pub(crate) fn durable_commit(&self, eventual: bool) -> Result {
+        let oldest_live_read = self
+            .db
+            .oldest_live_read_transaction()
+            .unwrap_or(self.transaction_id);
+
+        self.process_freed_pages(oldest_live_read)?;
+        if oldest_live_read < self.transaction_id {
+            self.store_freed_pages()?;
+        } else {
+            for page in self.take_all_freed_pages() {
+                // Safety: The oldest live read started after this transactions, so it can't
+                // have a references to this page, since we freed it in this transaction
+                unsafe {
+                    self.mem.free(page)?;
+                }
+            }
+        }
+
+        if let Some(root) = self.table_tree.borrow().tree.root {
+            self.mem.set_secondary_root_page(root)?;
+        } else {
+            self.mem.set_secondary_root_page(PageNumber::null())?;
+        }
+
+        self.mem.commit(self.transaction_id, eventual)?;
+        Ok(())
+    }
+
+    // Commit without a durability guarantee
+    pub(crate) fn non_durable_commit(&self) -> Result {
+        // Store all freed pages for a future commit(), since we can't free pages during a
+        // non-durable commit (it's non-durable, so could be rolled back anytime in the future)
+        self.store_freed_pages()?;
+
+        if let Some(root) = self.table_tree.borrow().tree.root {
+            self.mem.set_secondary_root_page(root)?;
+        } else {
+            self.mem.set_secondary_root_page(PageNumber::null())?;
+        }
+
+        self.mem.non_durable_commit(self.transaction_id)?;
+        Ok(())
+    }
+
+    // NOTE: must be called before store_freed_pages() during commit, since this can create
+    // more pages freed by the current transaction
+    fn process_freed_pages(&self, oldest_live_read: TransactionId) -> Result {
+        // We assume below that PageNumber is length 8
+        assert_eq!(PageNumber::null().to_le_bytes().len(), 8);
+        let lookup_key = FreedTableKey {
+            transaction_id: oldest_live_read,
+            pagination_id: 0,
+        };
+
+        let mut to_remove = vec![];
+        let freed_table = self
+            .table_tree
+            .borrow()
+            .get_table::<FreedTableKey, [u8]>(FREED_TABLE, TableType::Normal)?
+            .unwrap();
+        let mut freed_tree: BtreeMut<FreedTableKey, [u8]> =
+            BtreeMut::new(freed_table.get_root(), self.mem);
+        let mut iter = freed_tree.range(..lookup_key)?;
+        while let Some(entry) = iter.next() {
+            to_remove.push(FreedTableKey::from_bytes(entry.key()));
+            let value = entry.value();
+            let length = u64::from_le_bytes(value[..size_of::<u64>()].try_into().unwrap()) as usize;
+            // 1..=length because the array is length prefixed
+            for i in 1..=length {
+                let page = PageNumber::from_le_bytes(value[i * 8..(i + 1) * 8].try_into().unwrap());
+                // Safety: we free only pages that were marked to be freed before the oldest live transaction,
+                // therefore no one can have a reference to this page still
+                unsafe {
+                    self.mem.free(page)?;
+                }
+            }
+        }
+        drop(iter);
+
+        // Remove all the old transactions. Note: this may create new pages that need to be freed
+        for key in to_remove {
+            // Safety: all references to the freed table above have already been dropped
+            unsafe {
+                freed_tree.remove(&key)?;
+            }
+        }
+        self.table_tree
+            .borrow_mut()
+            .update_table_root(FREED_TABLE, freed_tree.root)?;
+        self.freed_pages
+            .borrow_mut()
+            .append(&mut freed_tree.freed_pages);
+
+        Ok(())
+    }
+
+    fn store_freed_pages(&self) -> Result {
+        assert_eq!(PageNumber::null().to_le_bytes().len(), 8); // We assume below that PageNumber is length 8
+
+        let mut pagination_counter = 0u64;
+        let freed_table = self
+            .table_tree
+            .borrow()
+            .get_table::<FreedTableKey, [u8]>(FREED_TABLE, TableType::Normal)?
+            .unwrap();
+        let mut freed_tree: BtreeMut<FreedTableKey, [u8]> =
+            BtreeMut::new(freed_table.get_root(), self.mem);
+        // TODO: change all master_tree usages to have TableHeader as their value type
+        let mut freed_pages = self.take_all_freed_pages();
+        while !freed_pages.is_empty() {
+            let chunk_size = 100;
+            let buffer_size = size_of::<u64>() + 8 * chunk_size;
+            let key = FreedTableKey {
+                transaction_id: self.transaction_id,
+                pagination_id: pagination_counter,
+            };
+            // Safety: The freed table is only accessed from the writer, so only this function
+            // is using it. The only reference retrieved, access_guard, is dropped before the next call
+            // to this method
+            let (mut access_guard, new_root, mut freed_pages2) =
+                unsafe { freed_tree.insert_reserve_special(&key, buffer_size)? };
+            freed_pages.append(&mut freed_pages2);
+
+            if freed_pages.len() <= chunk_size {
+                // Update the master root, only on the last loop iteration (this may cause another
+                // iteration, but that's ok since it would have very few pages to process)
+                self.table_tree
+                    .borrow_mut()
+                    .update_table_root(FREED_TABLE, new_root)?;
+                freed_pages.append(&mut self.table_tree.borrow_mut().tree.freed_pages);
+                freed_pages.append(&mut self.table_tree.borrow_mut().extra_freed_pages);
+            }
+
+            let len = freed_pages.len();
+            access_guard.as_mut()[..8]
+                .copy_from_slice(&min(len as u64, chunk_size as u64).to_le_bytes());
+            for (i, page) in freed_pages.drain(len - min(len, chunk_size)..).enumerate() {
+                access_guard.as_mut()[(i + 1) * 8..(i + 2) * 8]
+                    .copy_from_slice(&page.to_le_bytes());
+            }
+            drop(access_guard);
+
+            pagination_counter += 1;
+        }
+
+        Ok(())
+    }
+
+    pub fn stats(&self) -> Result<DatabaseStats> {
+        let table_tree = self.table_tree.borrow();
+        let master_tree_height = table_tree
+            .tree
+            .root
+            .map(|p| tree_height(self.mem.get_page(p), self.mem))
+            .unwrap_or(0);
+        let mut max_subtree_height = 0;
+        let mut total_stored_bytes = 0;
+        // Include the master table in the overhead
+        let mut total_metadata_bytes = table_tree
+            .tree
+            .root
+            .map(|p| {
+                overhead_bytes(self.mem.get_page(p), self.mem)
+                    + stored_bytes(self.mem.get_page(p), self.mem)
+            })
+            .unwrap_or(0);
+        // TODO: move these stats methods into Btree?
+        let mut total_fragmented = table_tree
+            .tree
+            .root
+            .map(|p| fragmented_bytes(self.mem.get_page(p), self.mem))
+            .unwrap_or(0);
+        let mut iter = table_tree.tree.range::<RangeFull, &str>(..)?;
+        while let Some(entry) = iter.next() {
+            let definition = InternalTableDefinition::from_bytes(entry.value());
+            if let Some(table_root) = definition.get_root() {
+                if std::str::from_utf8(entry.key()).unwrap() == FREED_TABLE {
+                    // Count the stored bytes of the freed table as metadata overhead
+                    total_metadata_bytes += stored_bytes(self.mem.get_page(table_root), self.mem);
+                    total_metadata_bytes += overhead_bytes(self.mem.get_page(table_root), self.mem);
+                    total_fragmented += fragmented_bytes(self.mem.get_page(table_root), self.mem);
+                } else {
+                    let height = tree_height(self.mem.get_page(table_root), self.mem);
+                    max_subtree_height = max(max_subtree_height, height);
+                    total_stored_bytes += stored_bytes(self.mem.get_page(table_root), self.mem);
+                    total_metadata_bytes += overhead_bytes(self.mem.get_page(table_root), self.mem);
+                    total_fragmented += fragmented_bytes(self.mem.get_page(table_root), self.mem);
+                }
+            }
+        }
+        Ok(DatabaseStats {
+            tree_height: master_tree_height + max_subtree_height,
+            free_pages: self.mem.count_free_pages()?,
+            stored_leaf_bytes: total_stored_bytes,
+            metadata_bytes: total_metadata_bytes,
+            fragmented_bytes: total_fragmented,
+            page_size: self.mem.get_page_size(),
+        })
+    }
+
+    #[allow(dead_code)]
+    pub(crate) fn print_debug(&self) {
+        if let Some(page) = self.root_page.get() {
+            eprintln!("Master tree:");
+
+            let master_tree: Btree<str, [u8]> = Btree::new(Some(page), self.mem);
+            master_tree.print_debug();
+            let mut iter = master_tree.range::<RangeFull, &str>(..).unwrap();
+
+            while let Some(entry) = iter.next() {
+                eprintln!("{} tree:", String::from_utf8_lossy(entry.key()));
+                let definition = InternalTableDefinition::from_bytes(entry.value());
+                if let Some(table_root) = definition.get_root() {
+                    // Print as &[u8], since we don't know the types at compile time
+                    let tree: Btree<[u8], [u8]> = Btree::new(Some(table_root), self.mem);
+                    tree.print_debug();
+                }
+            }
+        }
+    }
+}
+
+impl<'a> Drop for WriteTransaction<'a> {
+    fn drop(&mut self) {
+        if !self.completed.load(Ordering::Acquire) {
+            self.db.record_leaked_write_transaction(self.transaction_id);
+            // No-op, just to avoid triggering the leak detection in BtreeMut
+            self.take_all_freed_pages();
+        }
+    }
+}
+
+pub struct ReadTransaction<'a> {
+    db: &'a Database,
+    tree: TableTree<'a>,
+    transaction_id: TransactionId,
+}
+
+impl<'db> ReadTransaction<'db> {
+    pub(crate) fn new(db: &'db Database, transaction_id: TransactionId) -> Self {
+        let root_page = db.get_memory().get_primary_root_page();
+        Self {
+            db,
+            tree: TableTree::new(root_page, db.get_memory()),
+            transaction_id,
+        }
+    }
+
+    /// Open the given table
+    pub fn open_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &self,
+        definition: TableDefinition<K, V>,
+    ) -> Result<ReadOnlyTable<K, V>> {
+        let header = self
+            .tree
+            .get_table::<K, V>(definition.name(), TableType::Normal)?
+            .ok_or_else(|| Error::TableDoesNotExist(definition.name().to_string()))?;
+
+        Ok(ReadOnlyTable::new(header.get_root(), self.db.get_memory()))
+    }
+
+    /// Open the given table
+    pub fn open_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
+        &self,
+        definition: MultimapTableDefinition<K, V>,
+    ) -> Result<ReadOnlyMultimapTable<K, V>> {
+        let header = self
+            .tree
+            .get_table::<K, V>(definition.name(), TableType::Multimap)?
+            .ok_or_else(|| Error::TableDoesNotExist(definition.name().to_string()))?;
+
+        Ok(ReadOnlyMultimapTable::new(
+            header.get_root(),
+            self.db.get_memory(),
+        ))
+    }
+
+    /// List all the tables
+    // TODO: should return an iterator of &str, once GATs are available
+    pub fn list_tables(&self) -> Result<impl Iterator<Item = String>> {
+        self.tree
+            .list_tables(TableType::Normal)
+            .map(|x| x.into_iter())
+    }
+
+    /// List all the multimap tables
+    // TODO: should return an iterator of &str, once GATs are available
+    pub fn list_multimap_tables(&self) -> Result<impl Iterator<Item = String>> {
+        self.tree
+            .list_tables(TableType::Multimap)
+            .map(|x| x.into_iter())
+    }
+}
+
+impl<'a> Drop for ReadTransaction<'a> {
+    fn drop(&mut self) {
+        self.db.deallocate_read_transaction(self.transaction_id);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::{Database, TableDefinition};
+    use tempfile::NamedTempFile;
+
+    const X: TableDefinition<[u8], [u8]> = TableDefinition::new("x");
+
+    #[test]
+    fn transaction_id_persistence() {
+        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+        let db = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
+        let write_txn = db.begin_write().unwrap();
+        {
+            let mut table = write_txn.open_table(X).unwrap();
+            table.insert(b"hello", b"world").unwrap();
+        }
+        let first_txn_id = write_txn.transaction_id;
+        write_txn.commit().unwrap();
+        drop(db);
+
+        let db2 = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
+        let write_txn = db2.begin_write().unwrap();
+        assert!(write_txn.transaction_id > first_txn_id);
+    }
+}
diff --git a/src/tree_store/btree.rs b/src/tree_store/btree.rs
new file mode 100644
index 0000000..5c6b18d
--- /dev/null
+++ b/src/tree_store/btree.rs
@@ -0,0 +1,225 @@
+use crate::tree_store::btree_utils::{
+    find_key, make_mut_single_leaf, print_tree, tree_delete, tree_insert,
+};
+use crate::tree_store::page_store::TransactionalMemory;
+use crate::tree_store::{AccessGuardMut, BtreeRangeIter, PageNumber};
+use crate::types::{RedbKey, RedbValue, WithLifetime};
+use crate::{AccessGuard, Result};
+use std::borrow::Borrow;
+use std::marker::PhantomData;
+use std::ops::{RangeBounds, RangeFull};
+
+pub(crate) struct BtreeMut<'a, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
+    // TODO: make these private?
+    pub(crate) mem: &'a TransactionalMemory,
+    pub(crate) root: Option<PageNumber>,
+    // TODO: storing these freed_pages is very error prone, because they need to be copied out into
+    // the Storage object before the BtreeMut is destroyed
+    pub(crate) freed_pages: Vec<PageNumber>,
+    _key_type: PhantomData<K>,
+    _value_type: PhantomData<V>,
+}
+
+impl<'a, K: RedbKey + ?Sized, V: RedbValue + ?Sized> BtreeMut<'a, K, V> {
+    pub(crate) fn new(root: Option<PageNumber>, mem: &'a TransactionalMemory) -> Self {
+        Self {
+            mem,
+            root,
+            freed_pages: vec![],
+            _key_type: Default::default(),
+            _value_type: Default::default(),
+        }
+    }
+
+    // Safety: caller must ensure that no uncommitted data is accessed within this tree, from other references
+    pub(crate) unsafe fn insert(&mut self, key: &K, value: &V) -> Result {
+        if let Some(p) = self.root {
+            let root = self.mem.get_page(p);
+            // Safety: Caller guaranteed that no uncommitted data will be used
+            let (new_root, _, freed) = tree_insert::<K>(
+                root,
+                key.as_bytes().as_ref(),
+                value.as_bytes().as_ref(),
+                self.mem,
+            )?;
+            self.freed_pages.extend_from_slice(&freed);
+            self.root = Some(new_root);
+        } else {
+            let (new_root, _) =
+                make_mut_single_leaf(key.as_bytes().as_ref(), value.as_bytes().as_ref(), self.mem)?;
+            self.root = Some(new_root);
+        }
+        Ok(())
+    }
+
+    /// Reserve space to insert a key-value pair
+    /// The returned reference will have length equal to value_length
+    // Safety: caller must ensure that no uncommitted data is accessed within this tree, from other references
+    pub(crate) unsafe fn insert_reserve(
+        &mut self,
+        key: &K,
+        value_length: usize,
+    ) -> Result<AccessGuardMut> {
+        let value = vec![0u8; value_length];
+        let (new_root, guard) = if let Some(p) = self.root {
+            let root = self.mem.get_page(p);
+            // Safety: Caller guaranteed that no uncommitted data will be used
+            let (new_root, guard, freed) =
+                tree_insert::<K>(root, key.as_bytes().as_ref(), &value, self.mem)?;
+            self.freed_pages.extend_from_slice(&freed);
+            (new_root, guard)
+        } else {
+            make_mut_single_leaf(key.as_bytes().as_ref(), &value, self.mem)?
+        };
+        self.root = Some(new_root);
+        Ok(guard)
+    }
+
+    // Special version of insert_reserve for usage in the freed table
+    // Safety: caller must ensure that no uncommitted data is accessed within this tree, from other references
+    pub(crate) unsafe fn insert_reserve_special(
+        &mut self,
+        key: &K,
+        value_length: usize,
+    ) -> Result<(AccessGuardMut, Option<PageNumber>, Vec<PageNumber>)> {
+        let value = vec![0u8; value_length];
+        let (new_root, guard) = if let Some(p) = self.root {
+            let root = self.mem.get_page(p);
+            // Safety: Caller guaranteed that no uncommitted data will be used
+            let (new_root, guard, freed) =
+                tree_insert::<K>(root, key.as_bytes().as_ref(), &value, self.mem)?;
+            self.freed_pages.extend_from_slice(&freed);
+            (new_root, guard)
+        } else {
+            make_mut_single_leaf(key.as_bytes().as_ref(), &value, self.mem)?
+        };
+        self.root = Some(new_root);
+        Ok((guard, self.root, std::mem::take(&mut self.freed_pages)))
+    }
+
+    // Safety: caller must ensure that no uncommitted data is accessed within this tree, from other references
+    pub(crate) unsafe fn remove(&mut self, key: &K) -> Result<Option<AccessGuard<V>>> {
+        if let Some(p) = self.root {
+            let root_page = self.mem.get_page(p);
+            // Safety: Caller guaranteed that no uncommitted data will be used
+            let (new_root, found, freed) =
+                tree_delete::<K, V>(root_page, key.as_bytes().as_ref(), true, self.mem)?;
+            self.freed_pages.extend_from_slice(&freed);
+            self.root = new_root;
+            Ok(found)
+        } else {
+            Ok(None)
+        }
+    }
+
+    // Like remove(), but does not free uncommitted data
+    pub(crate) fn remove_retain_uncommitted(&mut self, key: &K) -> Result<Option<AccessGuard<V>>> {
+        if let Some(p) = self.root {
+            let root_page = self.mem.get_page(p);
+            // Safety: free_uncommitted is false
+            let (new_root, found, freed) = unsafe {
+                tree_delete::<K, V>(root_page, key.as_bytes().as_ref(), false, self.mem)?
+            };
+            self.freed_pages.extend_from_slice(&freed);
+            self.root = new_root;
+            Ok(found)
+        } else {
+            Ok(None)
+        }
+    }
+
+    #[allow(dead_code)]
+    pub(crate) fn print_debug(&self) {
+        if let Some(p) = self.root {
+            print_tree::<K>(self.mem.get_page(p), self.mem);
+        }
+    }
+
+    fn read_tree(&self) -> Btree<K, V> {
+        Btree::new(self.root, self.mem)
+    }
+
+    pub(crate) fn get(
+        &self,
+        key: &K,
+    ) -> Result<Option<<<V as RedbValue>::View as WithLifetime>::Out>> {
+        self.read_tree().get(key)
+    }
+
+    pub(crate) fn range<T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
+        &self,
+        range: T,
+    ) -> Result<BtreeRangeIter<K, V>> {
+        self.read_tree().range(range)
+    }
+
+    pub(crate) fn len(&self) -> Result<usize> {
+        self.read_tree().len()
+    }
+}
+
+impl<'a, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Drop for BtreeMut<'a, K, V> {
+    fn drop(&mut self) {
+        debug_assert!(self.freed_pages.is_empty());
+        if !self.freed_pages.is_empty() {
+            eprintln!("{} pages leaked by BtreeMut", self.freed_pages.len());
+        }
+    }
+}
+
+pub(crate) struct Btree<'a, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
+    mem: &'a TransactionalMemory,
+    root: Option<PageNumber>,
+    _key_type: PhantomData<K>,
+    _value_type: PhantomData<V>,
+}
+
+impl<'a, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Btree<'a, K, V> {
+    pub(crate) fn new(root: Option<PageNumber>, mem: &'a TransactionalMemory) -> Self {
+        Self {
+            mem,
+            root,
+            _key_type: Default::default(),
+            _value_type: Default::default(),
+        }
+    }
+
+    pub(crate) fn get(
+        &self,
+        key: &K,
+    ) -> Result<Option<<<V as RedbValue>::View as WithLifetime<'a>>::Out>> {
+        if let Some(p) = self.root {
+            let root_page = self.mem.get_page(p);
+            return Ok(find_key::<K, V>(
+                root_page,
+                key.as_bytes().as_ref(),
+                self.mem,
+            ));
+        }
+        Ok(None)
+    }
+
+    pub(crate) fn range<T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
+        &self,
+        range: T,
+    ) -> Result<BtreeRangeIter<'a, K, V>> {
+        Ok(BtreeRangeIter::new(range, self.root, self.mem))
+    }
+
+    pub(crate) fn len(&self) -> Result<usize> {
+        let mut iter: BtreeRangeIter<[u8], [u8]> =
+            BtreeRangeIter::new::<RangeFull, [u8]>(.., self.root, self.mem);
+        let mut count = 0;
+        while iter.next().is_some() {
+            count += 1;
+        }
+        Ok(count)
+    }
+
+    #[allow(dead_code)]
+    pub(crate) fn print_debug(&self) {
+        if let Some(p) = self.root {
+            print_tree::<K>(self.mem.get_page(p), self.mem);
+        }
+    }
+}
diff --git a/src/tree_store/btree_base.rs b/src/tree_store/btree_base.rs
index f1a5b12..c32cd34 100644
--- a/src/tree_store/btree_base.rs
+++ b/src/tree_store/btree_base.rs
@@ -64,7 +64,7 @@ pub struct AccessGuardMut<'a> {
 }
 
 impl<'a> AccessGuardMut<'a> {
-    pub(in crate::tree_store) fn new(page: PageMut<'a>, offset: usize, len: usize) -> Self {
+    pub(crate) fn new(page: PageMut<'a>, offset: usize, len: usize) -> Self {
         AccessGuardMut { page, offset, len }
     }
 }
diff --git a/src/tree_store/btree_iters.rs b/src/tree_store/btree_iters.rs
index 7330267..c5c296d 100644
--- a/src/tree_store/btree_iters.rs
+++ b/src/tree_store/btree_iters.rs
@@ -115,7 +115,9 @@ pub(crate) struct AllPageNumbersBtreeIter<'a> {
 }
 
 impl<'a> AllPageNumbersBtreeIter<'a> {
-    pub(crate) fn new(start: RangeIterState<'a>, manager: &'a TransactionalMemory) -> Self {
+    pub(crate) fn new(root: PageNumber, manager: &'a TransactionalMemory) -> Self {
+        let root_page = manager.get_page(root);
+        let start = page_numbers_iter_start_state(root_page);
         match start {
             Leaf { entry, .. } => {
                 assert_eq!(entry, 0)
@@ -151,7 +153,7 @@ impl<'a> Iterator for AllPageNumbersBtreeIter<'a> {
     }
 }
 
-pub(crate) fn page_numbers_iter_start_state(page: PageImpl) -> RangeIterState {
+fn page_numbers_iter_start_state(page: PageImpl) -> RangeIterState {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => Leaf {
diff --git a/src/tree_store/btree_utils.rs b/src/tree_store/btree_utils.rs
index 4952bcb..0ed5056 100644
--- a/src/tree_store/btree_utils.rs
+++ b/src/tree_store/btree_utils.rs
@@ -3,7 +3,7 @@ use crate::tree_store::btree_base::{
     BTREE_ORDER, INTERNAL, LEAF,
 };
 use crate::tree_store::btree_utils::DeletionResult::{PartialInternal, PartialLeaf, Subtree};
-use crate::tree_store::page_store::{Page, PageImpl, PageNumber, TransactionalMemory};
+use crate::tree_store::page_store::{Page, PageImpl, PageMut, PageNumber, TransactionalMemory};
 use crate::tree_store::AccessGuardMut;
 use crate::types::{RedbKey, RedbValue, WithLifetime};
 use crate::Error;
@@ -1187,6 +1187,31 @@ pub(crate) fn find_key<'a, K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
     }
 }
 
+// Returns the mutable value for the queried key, if present
+// Safety: caller must ensure that not other references to the page storing the queried key exist
+pub(crate) unsafe fn get_mut_value<'a, K: RedbKey + ?Sized>(
+    page: PageMut<'a>,
+    query: &[u8],
+    manager: &'a TransactionalMemory,
+) -> Option<AccessGuardMut<'a>> {
+    let node_mem = page.memory();
+    match node_mem[0] {
+        LEAF => {
+            let accessor = LeafAccessor::new(&page);
+            let entry_index = accessor.find_key::<K>(query)?;
+            let (start, end) = accessor.value_range(entry_index).unwrap();
+            let guard = AccessGuardMut::new(page, start, end - start);
+            Some(guard)
+        }
+        INTERNAL => {
+            let accessor = InternalAccessor::new(&page);
+            let (_, child_page) = accessor.child_for_key::<K>(query);
+            return get_mut_value::<K>(manager.get_page_mut(child_page), query, manager);
+        }
+        _ => unreachable!(),
+    }
+}
+
 #[cfg(test)]
 mod test {
     use crate::tree_store::btree_base::InternalAccessor;
@@ -1268,7 +1293,8 @@ mod test {
         txn.commit().unwrap();
 
         let expected = expected_height(elements + num_internal_entries);
-        let height = db.stats().unwrap().tree_height();
+        let txn = db.begin_write().unwrap();
+        let height = txn.stats().unwrap().tree_height();
         assert!(
             height <= expected,
             "height={} expected={}",
@@ -1277,8 +1303,6 @@ mod test {
         );
 
         let reduce_to = BTREE_ORDER / 2 - num_internal_entries;
-
-        let txn = db.begin_write().unwrap();
         {
             let mut table = txn.open_table(X).unwrap();
             for i in 0..(elements - reduce_to) {
@@ -1288,7 +1312,9 @@ mod test {
         txn.commit().unwrap();
 
         let expected = expected_height(reduce_to + num_internal_entries);
-        let height = db.stats().unwrap().tree_height();
+        let txn = db.begin_write().unwrap();
+        let height = txn.stats().unwrap().tree_height();
+        txn.abort().unwrap();
         assert!(
             height <= expected,
             "height={} expected={}",
diff --git a/src/tree_store/mod.rs b/src/tree_store/mod.rs
index bf6631f..3ac2b0e 100644
--- a/src/tree_store/mod.rs
+++ b/src/tree_store/mod.rs
@@ -1,12 +1,15 @@
+mod btree;
 mod btree_base;
 mod btree_iters;
-mod btree_utils;
+pub(crate) mod btree_utils;
 mod page_store;
-mod storage;
+mod table_tree;
 
+pub(crate) use btree::{Btree, BtreeMut};
 pub use btree_base::AccessGuard;
 pub(crate) use btree_base::AccessGuardMut;
-pub(crate) use btree_iters::BtreeRangeIter;
-pub(crate) use page_store::{get_db_size, PageNumber};
-pub use storage::DatabaseStats;
-pub(crate) use storage::{Storage, TableType, TransactionId, FREED_TABLE};
+pub(crate) use btree_iters::{AllPageNumbersBtreeIter, BtreeRangeIter};
+pub(crate) use page_store::{get_db_size, PageNumber, TransactionalMemory};
+pub(crate) use table_tree::{
+    FreedTableKey, InternalTableDefinition, TableTree, TableType, FREED_TABLE,
+};
diff --git a/src/tree_store/page_store/mod.rs b/src/tree_store/page_store/mod.rs
index 8533bcc..e18b5a9 100644
--- a/src/tree_store/page_store/mod.rs
+++ b/src/tree_store/page_store/mod.rs
@@ -7,7 +7,6 @@ mod page_manager;
 mod utils;
 
 pub(crate) use base::PageNumber;
-pub(crate) use page_manager::get_db_size;
+pub(crate) use page_manager::{get_db_size, TransactionalMemory};
 
 pub(in crate::tree_store) use base::{Page, PageImpl, PageMut};
-pub(in crate::tree_store) use page_manager::TransactionalMemory;
diff --git a/src/tree_store/page_store/page_manager.rs b/src/tree_store/page_store/page_manager.rs
index afadbbd..a3afb8d 100644
--- a/src/tree_store/page_store/page_manager.rs
+++ b/src/tree_store/page_store/page_manager.rs
@@ -1395,7 +1395,9 @@ mod test {
             table.insert(b"hello", b"world").unwrap();
         }
         write_txn.commit().unwrap();
-        let free_pages = db.stats().unwrap().free_pages();
+        let write_txn = db.begin_write().unwrap();
+        let free_pages = write_txn.stats().unwrap().free_pages();
+        write_txn.abort().unwrap();
         drop(db);
 
         let file = OpenOptions::new()
@@ -1416,8 +1418,8 @@ mod test {
             .unwrap());
 
         let db2 = unsafe { Database::create(tmpfile.path(), max_size).unwrap() };
-        assert_eq!(free_pages, db2.stats().unwrap().free_pages());
         let write_txn = db2.begin_write().unwrap();
+        assert_eq!(free_pages, write_txn.stats().unwrap().free_pages());
         {
             let mut table = write_txn.open_table(X).unwrap();
             table.insert(b"hello2", b"world2").unwrap();
diff --git a/src/tree_store/storage.rs b/src/tree_store/storage.rs
deleted file mode 100644
index 3087de1..0000000
--- a/src/tree_store/storage.rs
+++ /dev/null
@@ -1,923 +0,0 @@
-use crate::tree_store::btree_base::{AccessGuard, InternalAccessor, LeafAccessor};
-use crate::tree_store::btree_iters::{page_numbers_iter_start_state, AllPageNumbersBtreeIter};
-use crate::tree_store::btree_utils::{
-    find_key, fragmented_bytes, make_mut_single_leaf, overhead_bytes, print_tree, stored_bytes,
-    tree_delete, tree_height, tree_insert,
-};
-use crate::tree_store::page_store::{Page, PageMut, PageNumber, TransactionalMemory};
-use crate::tree_store::{btree_base, AccessGuardMut, BtreeRangeIter};
-use crate::types::{
-    AsBytesWithLifetime, OwnedAsBytesLifetime, OwnedLifetime, RedbKey, RedbValue, WithLifetime,
-};
-use crate::Error;
-use crate::Result;
-use std::borrow::Borrow;
-use std::cmp::{max, min};
-use std::collections::BTreeSet;
-use std::convert::TryInto;
-use std::fs::File;
-use std::mem::size_of;
-use std::ops::{RangeBounds, RangeFull};
-use std::panic;
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::Mutex;
-
-// The table of freed pages by transaction. FreedTableKey -> binary.
-// The binary blob is a length-prefixed array of PageNumber
-pub(crate) const FREED_TABLE: &str = "$$internal$$freed";
-
-pub(crate) type TransactionId = u64;
-type AtomicTransactionId = AtomicU64;
-
-#[derive(Debug)]
-struct FreedTableKey {
-    transaction_id: u64,
-    pagination_id: u64,
-}
-
-impl RedbValue for FreedTableKey {
-    type View = OwnedLifetime<FreedTableKey>;
-    type ToBytes = OwnedAsBytesLifetime<[u8; 2 * size_of::<u64>()]>;
-
-    fn from_bytes(data: &[u8]) -> <Self::View as WithLifetime>::Out {
-        let transaction_id = u64::from_le_bytes(data[..size_of::<u64>()].try_into().unwrap());
-        let pagination_id = u64::from_le_bytes(data[size_of::<u64>()..].try_into().unwrap());
-        Self {
-            transaction_id,
-            pagination_id,
-        }
-    }
-
-    fn as_bytes(&self) -> <Self::ToBytes as AsBytesWithLifetime>::Out {
-        let mut result = [0u8; 2 * size_of::<u64>()];
-        result[..size_of::<u64>()].copy_from_slice(&self.transaction_id.to_le_bytes());
-        result[size_of::<u64>()..].copy_from_slice(&self.pagination_id.to_le_bytes());
-        result
-    }
-
-    fn redb_type_name() -> &'static str {
-        "FreedTableKey"
-    }
-}
-
-impl RedbKey for FreedTableKey {
-    fn compare(data1: &[u8], data2: &[u8]) -> std::cmp::Ordering {
-        let value1 = Self::from_bytes(data1);
-        let value2 = Self::from_bytes(data2);
-
-        match value1.transaction_id.cmp(&value2.transaction_id) {
-            std::cmp::Ordering::Greater => std::cmp::Ordering::Greater,
-            std::cmp::Ordering::Equal => value1.pagination_id.cmp(&value2.pagination_id),
-            std::cmp::Ordering::Less => std::cmp::Ordering::Less,
-        }
-    }
-}
-
-#[derive(Debug)]
-pub struct DatabaseStats {
-    pub(crate) tree_height: usize,
-    pub(crate) free_pages: usize,
-    pub(crate) stored_leaf_bytes: usize,
-    pub(crate) metadata_bytes: usize,
-    pub(crate) fragmented_bytes: usize,
-    pub(crate) page_size: usize,
-}
-
-impl DatabaseStats {
-    /// Maximum traversal distance to reach the deepest (key, value) pair, across all tables
-    pub fn tree_height(&self) -> usize {
-        self.tree_height
-    }
-
-    /// Number of free pages remaining
-    pub fn free_pages(&self) -> usize {
-        self.free_pages
-    }
-
-    /// Number of bytes consumed by keys and values that have been inserted.
-    /// Does not include indexing overhead
-    pub fn stored_bytes(&self) -> usize {
-        self.stored_leaf_bytes
-    }
-
-    /// Number of bytes consumed by keys in internal index pages, plus other metadata
-    pub fn metadata_bytes(&self) -> usize {
-        self.metadata_bytes
-    }
-
-    /// Number of bytes consumed by fragmentation, both in data pages and internal metadata tables
-    pub fn fragmented_bytes(&self) -> usize {
-        self.fragmented_bytes
-    }
-
-    /// Number of bytes per page
-    pub fn page_size(&self) -> usize {
-        self.page_size
-    }
-}
-
-#[derive(Copy, Clone, Hash, Eq, PartialEq, Debug)]
-pub(crate) enum TableType {
-    Normal,
-    Multimap,
-}
-
-#[allow(clippy::from_over_into)]
-impl Into<u8> for TableType {
-    fn into(self) -> u8 {
-        match self {
-            TableType::Normal => 1,
-            TableType::Multimap => 2,
-        }
-    }
-}
-
-impl From<u8> for TableType {
-    fn from(value: u8) -> Self {
-        match value {
-            1 => TableType::Normal,
-            2 => TableType::Multimap,
-            _ => unreachable!(),
-        }
-    }
-}
-
-#[derive(Clone)]
-pub(crate) struct TableHeader {
-    table_root: Option<PageNumber>,
-    table_type: TableType,
-    key_type: String,
-    value_type: String,
-}
-
-impl TableHeader {
-    pub(crate) fn get_root(&self) -> Option<PageNumber> {
-        self.table_root
-    }
-
-    pub(crate) fn get_type(&self) -> TableType {
-        self.table_type
-    }
-
-    fn to_bytes(&self) -> Vec<u8> {
-        let mut result = vec![self.table_type.into()];
-        result.extend_from_slice(
-            &self
-                .table_root
-                .unwrap_or_else(PageNumber::null)
-                .to_le_bytes(),
-        );
-        result.extend_from_slice(&(self.key_type.as_bytes().len() as u32).to_le_bytes());
-        result.extend_from_slice(self.key_type.as_bytes());
-        result.extend_from_slice(self.value_type.as_bytes());
-
-        result
-    }
-
-    fn from_bytes(value: &[u8]) -> Self {
-        debug_assert!(value.len() > 14);
-        let table_type = TableType::from(value[0]);
-        let table_root = PageNumber::from_le_bytes(value[1..9].try_into().unwrap());
-        let table_root = if table_root == PageNumber::null() {
-            None
-        } else {
-            Some(table_root)
-        };
-        let key_type_len = u32::from_le_bytes(value[9..13].try_into().unwrap()) as usize;
-        let key_type = std::str::from_utf8(&value[13..(13 + key_type_len)])
-            .unwrap()
-            .to_string();
-        let value_type = std::str::from_utf8(&value[(13 + key_type_len)..])
-            .unwrap()
-            .to_string();
-
-        TableHeader {
-            table_root,
-            table_type,
-            key_type,
-            value_type,
-        }
-    }
-}
-
-pub struct TableNameIter<'a> {
-    inner: BtreeRangeIter<'a, str, [u8]>,
-    table_type: TableType,
-}
-
-impl<'a> Iterator for TableNameIter<'a> {
-    type Item = String;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        while let Some(entry) = self.inner.next() {
-            if str::from_bytes(entry.key()) == FREED_TABLE {
-                continue;
-            }
-            if TableHeader::from_bytes(entry.value()).table_type == self.table_type {
-                return Some(str::from_bytes(entry.key()).to_string());
-            }
-        }
-        None
-    }
-}
-
-// Returns the mutable value for the queried key, if present
-// Safety: caller must ensure that not other references to the page storing the queried key exist
-unsafe fn get_mut_value<'a, K: RedbKey + ?Sized>(
-    page: PageMut<'a>,
-    query: &[u8],
-    manager: &'a TransactionalMemory,
-) -> Option<AccessGuardMut<'a>> {
-    let node_mem = page.memory();
-    match node_mem[0] {
-        btree_base::LEAF => {
-            let accessor = LeafAccessor::new(&page);
-            let entry_index = accessor.find_key::<K>(query)?;
-            let (start, end) = accessor.value_range(entry_index).unwrap();
-            let guard = AccessGuardMut::new(page, start, end - start);
-            Some(guard)
-        }
-        btree_base::INTERNAL => {
-            let accessor = InternalAccessor::new(&page);
-            let (_, child_page) = accessor.child_for_key::<K>(query);
-            return get_mut_value::<K>(manager.get_page_mut(child_page), query, manager);
-        }
-        _ => unreachable!(),
-    }
-}
-
-pub(crate) struct Storage {
-    mem: TransactionalMemory,
-    next_transaction_id: AtomicTransactionId,
-    live_read_transactions: Mutex<BTreeSet<TransactionId>>,
-    live_write_transaction: Mutex<Option<TransactionId>>,
-    pending_freed_pages: Mutex<Vec<PageNumber>>,
-    leaked_write_transaction: Mutex<Option<&'static panic::Location<'static>>>,
-}
-
-impl Storage {
-    pub(crate) fn new(
-        file: File,
-        max_capacity: usize,
-        page_size: Option<usize>,
-        dynamic_growth: bool,
-    ) -> Result<Storage> {
-        let mem = TransactionalMemory::new(file, max_capacity, page_size, dynamic_growth)?;
-        if mem.needs_repair()? {
-            let root = mem
-                .get_primary_root_page()
-                .expect("Tried to repair an empty database");
-            // Clear the freed table. We're about to rebuild the allocator state by walking
-            // all the reachable pages, which will implicitly free them
-            {
-                // Safety: we own the mem object, and just created it, so there can't be other references.
-                let mut bytes = unsafe {
-                    get_mut_value::<str>(mem.get_page_mut(root), FREED_TABLE.as_bytes(), &mem)
-                        .unwrap()
-                };
-                // TODO: this needs to be done in a transaction. Otherwise a torn write could happen
-                let mut header = TableHeader::from_bytes(bytes.as_mut());
-                header.table_root = None;
-                assert_eq!(bytes.as_mut().len(), header.to_bytes().len());
-                bytes.as_mut().copy_from_slice(&header.to_bytes());
-            }
-
-            // Repair the allocator state
-            let root_page = mem.get_page(root);
-            let start = page_numbers_iter_start_state(root_page);
-
-            // All pages in the master table
-            let mut all_pages_iter: Box<dyn Iterator<Item = PageNumber>> =
-                Box::new(AllPageNumbersBtreeIter::new(start, &mem));
-
-            // Iterate over all other tables
-            let mut iter: BtreeRangeIter<[u8], [u8]> =
-                BtreeRangeIter::new::<RangeFull, [u8]>(.., Some(root), &mem);
-
-            // Chain all the other tables to the master table iter
-            while let Some(entry) = iter.next() {
-                let definition = TableHeader::from_bytes(entry.value());
-                if let Some(table_root) = definition.get_root() {
-                    let page = mem.get_page(table_root);
-                    let table_start = page_numbers_iter_start_state(page);
-                    let table_pages_iter = AllPageNumbersBtreeIter::new(table_start, &mem);
-                    all_pages_iter = Box::new(all_pages_iter.chain(table_pages_iter));
-                }
-            }
-
-            mem.repair_allocator(all_pages_iter)?;
-        }
-
-        let mut next_transaction_id = mem.get_last_committed_transaction_id()? + 1;
-        if mem.get_primary_root_page().is_none() {
-            assert_eq!(next_transaction_id, 1);
-            // Empty database, so insert the freed table.
-            let freed_table = TableHeader {
-                table_root: None,
-                table_type: TableType::Normal,
-                key_type: FreedTableKey::redb_type_name().to_string(),
-                value_type: <[u8]>::redb_type_name().to_string(),
-            };
-            let (new_root, _) =
-                make_mut_single_leaf(FREED_TABLE.as_bytes(), &freed_table.to_bytes(), &mem)?;
-            mem.set_secondary_root_page(new_root)?;
-            mem.commit(next_transaction_id, false)?;
-            next_transaction_id += 1;
-        }
-
-        Ok(Storage {
-            mem,
-            next_transaction_id: AtomicTransactionId::new(next_transaction_id),
-            live_write_transaction: Mutex::new(None),
-            live_read_transactions: Mutex::new(Default::default()),
-            pending_freed_pages: Mutex::new(Default::default()),
-            leaked_write_transaction: Mutex::new(Default::default()),
-        })
-    }
-
-    pub(crate) fn record_leaked_write_transaction(&self, transaction_id: TransactionId) {
-        assert_eq!(
-            transaction_id,
-            self.live_write_transaction.lock().unwrap().unwrap()
-        );
-        *self.leaked_write_transaction.lock().unwrap() = Some(panic::Location::caller());
-    }
-
-    pub(crate) fn allocate_write_transaction(&self) -> Result<TransactionId> {
-        let guard = self.leaked_write_transaction.lock().unwrap();
-        if let Some(leaked) = *guard {
-            return Err(Error::LeakedWriteTransaction(leaked));
-        }
-        drop(guard);
-
-        assert!(self.live_write_transaction.lock().unwrap().is_none());
-        assert!(self.pending_freed_pages.lock().unwrap().is_empty());
-        let id = self.next_transaction_id.fetch_add(1, Ordering::AcqRel);
-        *self.live_write_transaction.lock().unwrap() = Some(id);
-        Ok(id)
-    }
-
-    pub(crate) fn allocate_read_transaction(&self) -> TransactionId {
-        let id = self.next_transaction_id.fetch_add(1, Ordering::AcqRel);
-        self.live_read_transactions.lock().unwrap().insert(id);
-        id
-    }
-
-    pub(crate) fn deallocate_read_transaction(&self, id: TransactionId) {
-        self.live_read_transactions.lock().unwrap().remove(&id);
-    }
-
-    pub(crate) fn update_table_root(
-        &self,
-        name: &str,
-        table_root: Option<PageNumber>,
-        transaction_id: TransactionId,
-        master_root: Option<PageNumber>,
-    ) -> Result<PageNumber> {
-        // Bypass .get_table() since the table types are dynamic
-        // TODO: optimize way this get()
-        let bytes = self
-            .get::<str, [u8]>(name.as_bytes(), master_root)
-            .unwrap()
-            .unwrap();
-        let mut definition = TableHeader::from_bytes(bytes);
-        // No-op if the root has not changed
-        if definition.table_root == table_root {
-            return Ok(master_root.unwrap());
-        }
-        definition.table_root = table_root;
-        // Safety: References into the master table are never returned to the user
-        unsafe {
-            self.insert::<str>(
-                name.as_bytes(),
-                &definition.to_bytes(),
-                transaction_id,
-                master_root,
-            )
-        }
-    }
-
-    // root_page: the root of the master table
-    pub(crate) fn list_tables(
-        &self,
-        table_type: TableType,
-        master_root_page: Option<PageNumber>,
-    ) -> Result<TableNameIter> {
-        let iter = self.get_range::<RangeFull, str, str, [u8]>(.., master_root_page)?;
-        Ok(TableNameIter {
-            inner: iter,
-            table_type,
-        })
-    }
-
-    // root_page: the root of the master table
-    pub(crate) fn get_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        name: &str,
-        table_type: TableType,
-        root_page: Option<PageNumber>,
-    ) -> Result<Option<TableHeader>> {
-        if let Some(found) = self.get::<str, [u8]>(name.as_bytes(), root_page)? {
-            let definition = TableHeader::from_bytes(found);
-            if definition.get_type() != table_type {
-                return Err(Error::TableTypeMismatch(format!(
-                    "{:?} is not of type {:?}",
-                    name, table_type
-                )));
-            }
-            if definition.key_type != K::redb_type_name()
-                || definition.value_type != V::redb_type_name()
-            {
-                return Err(Error::TableTypeMismatch(format!(
-                    "{} is of type Table<{}, {}> not Table<{}, {}>",
-                    name,
-                    &definition.key_type,
-                    &definition.value_type,
-                    K::redb_type_name(),
-                    V::redb_type_name()
-                )));
-            }
-
-            Ok(Some(definition))
-        } else {
-            Ok(None)
-        }
-    }
-
-    // root_page: the root of the master table
-    pub(crate) fn delete_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        name: &str,
-        table_type: TableType,
-        transaction_id: TransactionId,
-        root_page: Option<PageNumber>,
-    ) -> Result<(Option<PageNumber>, bool)> {
-        if let Some(definition) = self.get_table::<K, V>(name, table_type, root_page)? {
-            if let Some(table_root) = definition.get_root() {
-                let page = self.mem.get_page(table_root);
-                let start = page_numbers_iter_start_state(page);
-                let iter = AllPageNumbersBtreeIter::new(start, &self.mem);
-                let mut guard = self.pending_freed_pages.lock().unwrap();
-                for page_number in iter {
-                    guard.push(page_number);
-                }
-            }
-
-            // Safety: References into the master table are never returned to the user
-            let (new_root, found) = unsafe {
-                self.remove::<str, [u8]>(name.as_bytes(), transaction_id, true, root_page)?
-            };
-            return Ok((new_root, found.is_some()));
-        }
-
-        Ok((root_page, false))
-    }
-
-    // Returns a tuple of the table id and the new root page
-    // root_page: the root of the master table
-    pub(crate) fn get_or_create_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        name: &str,
-        table_type: TableType,
-        transaction_id: TransactionId,
-        root_page: Option<PageNumber>,
-    ) -> Result<(TableHeader, PageNumber)> {
-        if let Some(found) = self.get_table::<K, V>(name, table_type, root_page)? {
-            return Ok((found, root_page.unwrap()));
-        }
-
-        let header = TableHeader {
-            table_root: None,
-            table_type,
-            key_type: K::redb_type_name().to_string(),
-            value_type: V::redb_type_name().to_string(),
-        };
-        // Safety: References into the master table are never returned to the user
-        let new_root = unsafe {
-            self.insert::<str>(
-                name.as_bytes(),
-                &header.to_bytes(),
-                transaction_id,
-                root_page,
-            )?
-        };
-        Ok((header, new_root))
-    }
-
-    // Returns the new root page number
-    // Safety: caller must ensure that no references to uncommitted data in this transaction exist
-    // TODO: this method could be made safe, if the transaction_id was not copy and was borrowed mut
-    pub(crate) unsafe fn insert<K: RedbKey + ?Sized>(
-        &self,
-        key: &[u8],
-        value: &[u8],
-        transaction_id: TransactionId,
-        root_page: Option<PageNumber>,
-    ) -> Result<PageNumber> {
-        assert_eq!(
-            transaction_id,
-            self.live_write_transaction.lock().unwrap().unwrap()
-        );
-        if let Some(handle) = root_page {
-            let root = self.mem.get_page(handle);
-            let (new_root, _, freed) = tree_insert::<K>(root, key, value, &self.mem)?;
-            self.pending_freed_pages
-                .lock()
-                .unwrap()
-                .extend_from_slice(&freed);
-            Ok(new_root)
-        } else {
-            let (new_root, _) = make_mut_single_leaf(key, value, &self.mem)?;
-            Ok(new_root)
-        }
-    }
-
-    // Returns the new root page number, and accessor for writing the value
-    // Safety: caller must ensure that no references to uncommitted data in this transaction exist
-    // TODO: this method could be made safe, if the transaction_id was not copy and was borrowed mut
-    pub(crate) unsafe fn insert_reserve<K: RedbKey + ?Sized>(
-        &self,
-        key: &[u8],
-        value_len: usize,
-        transaction_id: TransactionId,
-        root_page: Option<PageNumber>,
-    ) -> Result<(PageNumber, AccessGuardMut)> {
-        assert_eq!(
-            transaction_id,
-            self.live_write_transaction.lock().unwrap().unwrap()
-        );
-        let value = vec![0u8; value_len];
-        let (new_root, guard) = if let Some(handle) = root_page {
-            let root = self.mem.get_page(handle);
-            let (new_root, guard, freed) = tree_insert::<K>(root, key, &value, &self.mem)?;
-            self.pending_freed_pages
-                .lock()
-                .unwrap()
-                .extend_from_slice(&freed);
-            (new_root, guard)
-        } else {
-            make_mut_single_leaf(key, &value, &self.mem)?
-        };
-        Ok((new_root, guard))
-    }
-
-    pub(crate) fn len(&self, root_page: Option<PageNumber>) -> Result<usize> {
-        let mut iter: BtreeRangeIter<[u8], [u8]> =
-            self.get_range::<RangeFull, [u8], [u8], [u8]>(.., root_page)?;
-        let mut count = 0;
-        while iter.next().is_some() {
-            count += 1;
-        }
-        Ok(count)
-    }
-
-    pub(crate) fn get_root_page_number(&self) -> Option<PageNumber> {
-        self.mem.get_primary_root_page()
-    }
-
-    pub(crate) fn durable_commit(
-        &self,
-        mut new_master_root: Option<PageNumber>,
-        transaction_id: TransactionId,
-        eventual: bool,
-    ) -> Result {
-        let oldest_live_read = self
-            .live_read_transactions
-            .lock()
-            .unwrap()
-            .iter()
-            .next()
-            .cloned()
-            .unwrap_or_else(|| self.next_transaction_id.load(Ordering::Acquire));
-
-        new_master_root =
-            self.process_freed_pages(oldest_live_read, transaction_id, new_master_root)?;
-        if oldest_live_read < transaction_id {
-            new_master_root = self.store_freed_pages(transaction_id, new_master_root)?;
-        } else {
-            for page in self.pending_freed_pages.lock().unwrap().drain(..) {
-                // Safety: The oldest live read started after this transactions, so it can't
-                // have a references to this page, since we freed it in this transaction
-                unsafe {
-                    self.mem.free(page)?;
-                }
-            }
-        }
-
-        if let Some(root) = new_master_root {
-            self.mem.set_secondary_root_page(root)?;
-        } else {
-            self.mem.set_secondary_root_page(PageNumber::null())?;
-        }
-
-        self.mem.commit(transaction_id, eventual)?;
-        assert_eq!(
-            Some(transaction_id),
-            self.live_write_transaction.lock().unwrap().take()
-        );
-        Ok(())
-    }
-
-    // Commit without a durability guarantee
-    pub(crate) fn non_durable_commit(
-        &self,
-        mut new_master_root: Option<PageNumber>,
-        transaction_id: TransactionId,
-    ) -> Result {
-        // Store all freed pages for a future commit(), since we can't free pages during a
-        // non-durable commit (it's non-durable, so could be rolled back anytime in the future)
-        new_master_root = self.store_freed_pages(transaction_id, new_master_root)?;
-
-        if let Some(root) = new_master_root {
-            self.mem.set_secondary_root_page(root)?;
-        } else {
-            self.mem.set_secondary_root_page(PageNumber::null())?;
-        }
-
-        self.mem.non_durable_commit(transaction_id)?;
-        assert_eq!(
-            Some(transaction_id),
-            self.live_write_transaction.lock().unwrap().take()
-        );
-        Ok(())
-    }
-
-    // NOTE: must be called before store_freed_pages() during commit, since this can create
-    // more pages freed by the current transaction
-    fn process_freed_pages(
-        &self,
-        oldest_live_read: TransactionId,
-        transaction_id: TransactionId,
-        mut master_root: Option<PageNumber>,
-    ) -> Result<Option<PageNumber>> {
-        assert_eq!(
-            transaction_id,
-            self.live_write_transaction.lock().unwrap().unwrap()
-        );
-        // We assume below that PageNumber is length 8
-        assert_eq!(PageNumber::null().to_le_bytes().len(), 8);
-        let lookup_key = FreedTableKey {
-            transaction_id: oldest_live_read,
-            pagination_id: 0,
-        };
-
-        let mut to_remove = vec![];
-        let mut freed_table = self
-            .get_table::<FreedTableKey, [u8]>(FREED_TABLE, TableType::Normal, master_root)?
-            .unwrap();
-        #[allow(clippy::type_complexity)]
-        let mut iter: BtreeRangeIter<'_, FreedTableKey, [u8]> =
-            self.get_range(..lookup_key, freed_table.get_root())?;
-        while let Some(entry) = iter.next() {
-            to_remove.push(FreedTableKey::from_bytes(entry.key()));
-            let value = entry.value();
-            let length = u64::from_le_bytes(value[..size_of::<u64>()].try_into().unwrap()) as usize;
-            // 1..=length because the array is length prefixed
-            for i in 1..=length {
-                let page = PageNumber::from_le_bytes(value[i * 8..(i + 1) * 8].try_into().unwrap());
-                // Safety: we free only pages that were marked to be freed before the oldest live transaction,
-                // therefore no one can have a reference to this page still
-                unsafe {
-                    self.mem.free(page)?;
-                }
-            }
-        }
-        drop(iter);
-
-        // Remove all the old transactions. Note: this may create new pages that need to be freed
-        for key in to_remove {
-            // Safety: all references to the freed table above have already been dropped
-            let (new_root, _) = unsafe {
-                self.remove::<FreedTableKey, [u8]>(
-                    key.as_bytes().as_slice(),
-                    transaction_id,
-                    true,
-                    freed_table.table_root,
-                )?
-            };
-            freed_table.table_root = new_root;
-        }
-        // Safety: References into the master table are never returned to the user
-        unsafe {
-            master_root = Some(self.insert::<str>(
-                FREED_TABLE.as_bytes(),
-                &freed_table.to_bytes(),
-                transaction_id,
-                master_root,
-            )?);
-        }
-
-        Ok(master_root)
-    }
-
-    fn store_freed_pages(
-        &self,
-        transaction_id: TransactionId,
-        mut master_root: Option<PageNumber>,
-    ) -> Result<Option<PageNumber>> {
-        assert_eq!(
-            transaction_id,
-            self.live_write_transaction.lock().unwrap().unwrap()
-        );
-        assert_eq!(PageNumber::null().to_le_bytes().len(), 8); // We assume below that PageNumber is length 8
-
-        let mut pagination_counter = 0u64;
-        let mut freed_table = self
-            .get_table::<FreedTableKey, [u8]>(FREED_TABLE, TableType::Normal, master_root)?
-            .unwrap();
-        while !self.pending_freed_pages.lock().unwrap().is_empty() {
-            let chunk_size = 100;
-            let buffer_size = size_of::<u64>() + 8 * chunk_size;
-            let key = FreedTableKey {
-                transaction_id,
-                pagination_id: pagination_counter,
-            };
-            // Safety: The freed table is only accessed from the writer, so only this function
-            // is using it. The only reference retrieved, access_guard, is dropped before the next call
-            // to this method
-            let (r, mut access_guard) = unsafe {
-                self.insert_reserve::<FreedTableKey>(
-                    key.as_bytes().as_ref(),
-                    buffer_size,
-                    transaction_id,
-                    freed_table.table_root,
-                )?
-            };
-            freed_table.table_root = Some(r);
-
-            if self.pending_freed_pages.lock().unwrap().len() <= chunk_size {
-                // Update the master root, only on the last loop iteration (this may cause another
-                // iteration, but that's ok since it would have very few pages to process)
-                // Safety: References into the master table are never returned to the user
-                unsafe {
-                    master_root = Some(self.insert::<str>(
-                        FREED_TABLE.as_bytes(),
-                        &freed_table.to_bytes(),
-                        transaction_id,
-                        master_root,
-                    )?);
-                }
-            }
-
-            let len = self.pending_freed_pages.lock().unwrap().len();
-            access_guard.as_mut()[..8]
-                .copy_from_slice(&min(len as u64, chunk_size as u64).to_le_bytes());
-            for (i, page) in self
-                .pending_freed_pages
-                .lock()
-                .unwrap()
-                .drain(len - min(len, chunk_size)..)
-                .enumerate()
-            {
-                access_guard.as_mut()[(i + 1) * 8..(i + 2) * 8]
-                    .copy_from_slice(&page.to_le_bytes());
-            }
-            drop(access_guard);
-
-            pagination_counter += 1;
-        }
-
-        Ok(master_root)
-    }
-
-    pub(crate) fn rollback_uncommited_writes(&self, transaction_id: TransactionId) -> Result {
-        self.pending_freed_pages.lock().unwrap().clear();
-        let result = self.mem.rollback_uncommited_writes();
-        assert_eq!(
-            Some(transaction_id),
-            self.live_write_transaction.lock().unwrap().take()
-        );
-
-        result
-    }
-
-    pub(crate) fn storage_stats(&self) -> Result<DatabaseStats> {
-        let master_tree_height = self
-            .get_root_page_number()
-            .map(|p| tree_height(self.mem.get_page(p), &self.mem))
-            .unwrap_or(0);
-        let mut max_subtree_height = 0;
-        let mut total_stored_bytes = 0;
-        // Include the master table in the overhead
-        let mut total_metadata_bytes = self
-            .get_root_page_number()
-            .map(|p| {
-                overhead_bytes(self.mem.get_page(p), &self.mem)
-                    + stored_bytes(self.mem.get_page(p), &self.mem)
-            })
-            .unwrap_or(0);
-        let mut total_fragmented = self
-            .get_root_page_number()
-            .map(|p| fragmented_bytes(self.mem.get_page(p), &self.mem))
-            .unwrap_or(0);
-        let mut iter: BtreeRangeIter<[u8], [u8]> =
-            self.get_range::<RangeFull, [u8], [u8], [u8]>(.., self.get_root_page_number())?;
-        while let Some(entry) = iter.next() {
-            let definition = TableHeader::from_bytes(entry.value());
-            if let Some(table_root) = definition.get_root() {
-                if std::str::from_utf8(entry.key()).unwrap() == FREED_TABLE {
-                    // Count the stored bytes of the freed table as metadata overhead
-                    total_metadata_bytes += stored_bytes(self.mem.get_page(table_root), &self.mem);
-                    total_metadata_bytes +=
-                        overhead_bytes(self.mem.get_page(table_root), &self.mem);
-                    total_fragmented += fragmented_bytes(self.mem.get_page(table_root), &self.mem);
-                } else {
-                    let height = tree_height(self.mem.get_page(table_root), &self.mem);
-                    max_subtree_height = max(max_subtree_height, height);
-                    total_stored_bytes += stored_bytes(self.mem.get_page(table_root), &self.mem);
-                    total_metadata_bytes +=
-                        overhead_bytes(self.mem.get_page(table_root), &self.mem);
-                    total_fragmented += fragmented_bytes(self.mem.get_page(table_root), &self.mem);
-                }
-            }
-        }
-        Ok(DatabaseStats {
-            tree_height: master_tree_height + max_subtree_height,
-            free_pages: self.mem.count_free_pages()?,
-            stored_leaf_bytes: total_stored_bytes,
-            metadata_bytes: total_metadata_bytes,
-            fragmented_bytes: total_fragmented,
-            page_size: self.mem.get_page_size(),
-        })
-    }
-
-    #[allow(dead_code)]
-    pub(crate) fn print_debug(&self) {
-        if let Some(page) = self.get_root_page_number() {
-            eprintln!("Master tree:");
-            print_tree::<str>(self.mem.get_page(page), &self.mem);
-
-            let mut iter: BtreeRangeIter<[u8], [u8]> = self
-                .get_range::<RangeFull, [u8], [u8], [u8]>(.., Some(page))
-                .unwrap();
-
-            while let Some(entry) = iter.next() {
-                eprintln!("{} tree:", String::from_utf8_lossy(entry.key()));
-                let definition = TableHeader::from_bytes(entry.value());
-                if let Some(table_root) = definition.get_root() {
-                    // Print as &[u8], since we don't know the types at compile time
-                    self.print_dirty_tree_debug::<[u8]>(table_root);
-                }
-            }
-        }
-    }
-
-    #[allow(dead_code)]
-    pub(crate) fn print_dirty_tree_debug<K: RedbKey + ?Sized>(&self, root_page: PageNumber) {
-        print_tree::<K>(self.mem.get_page(root_page), &self.mem);
-    }
-
-    pub(crate) fn get<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        key: &[u8],
-        root_page_handle: Option<PageNumber>,
-    ) -> Result<Option<<<V as RedbValue>::View as WithLifetime>::Out>> {
-        if let Some(handle) = root_page_handle {
-            let root_page = self.mem.get_page(handle);
-            return Ok(find_key::<K, V>(root_page, key, &self.mem));
-        }
-        Ok(None)
-    }
-
-    pub(crate) fn get_range<
-        'a,
-        T: RangeBounds<KR>,
-        KR: Borrow<K> + ?Sized + 'a,
-        K: RedbKey + ?Sized + 'a,
-        V: RedbValue + ?Sized + 'a,
-    >(
-        &'a self,
-        range: T,
-        root_page: Option<PageNumber>,
-    ) -> Result<BtreeRangeIter<K, V>> {
-        Ok(BtreeRangeIter::new(range, root_page, &self.mem))
-    }
-
-    // Returns the new root page, and a bool indicating whether the entry existed
-    // Safety: caller must ensure that no references to uncommitted data in this transaction exist,
-    // if free_uncommitted = true
-    // TODO: this method could be made safe, if the transaction_id was not copy and was borrowed mut
-    pub(crate) unsafe fn remove<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        key: &[u8],
-        transaction_id: TransactionId,
-        free_uncommitted: bool,
-        root_handle: Option<PageNumber>,
-    ) -> Result<(Option<PageNumber>, Option<AccessGuard<V>>)> {
-        assert_eq!(
-            transaction_id,
-            self.live_write_transaction.lock().unwrap().unwrap()
-        );
-        if let Some(handle) = root_handle {
-            let root_page = self.mem.get_page(handle);
-            let (new_root, found, freed) =
-                tree_delete::<K, V>(root_page, key, free_uncommitted, &self.mem)?;
-            self.pending_freed_pages
-                .lock()
-                .unwrap()
-                .extend_from_slice(&freed);
-            return Ok((new_root, found));
-        }
-        Ok((root_handle, None))
-    }
-}
diff --git a/src/tree_store/table_tree.rs b/src/tree_store/table_tree.rs
new file mode 100644
index 0000000..7bc1417
--- /dev/null
+++ b/src/tree_store/table_tree.rs
@@ -0,0 +1,296 @@
+use crate::tree_store::btree_iters::AllPageNumbersBtreeIter;
+use crate::tree_store::{BtreeMut, BtreeRangeIter, PageNumber, TransactionalMemory};
+use crate::types::{
+    AsBytesWithLifetime, OwnedAsBytesLifetime, OwnedLifetime, RedbKey, RedbValue, WithLifetime,
+};
+use crate::{Error, Result};
+use std::mem::size_of;
+use std::ops::RangeFull;
+
+// The table of freed pages by transaction. FreedTableKey -> binary.
+// The binary blob is a length-prefixed array of PageNumber
+// TODO: create a separate root in the metapage for this
+pub(crate) const FREED_TABLE: &str = "$$internal$$freed";
+
+#[derive(Debug)]
+pub(crate) struct FreedTableKey {
+    pub(crate) transaction_id: u64,
+    pub(crate) pagination_id: u64,
+}
+
+impl RedbValue for FreedTableKey {
+    type View = OwnedLifetime<FreedTableKey>;
+    type ToBytes = OwnedAsBytesLifetime<[u8; 2 * size_of::<u64>()]>;
+
+    fn from_bytes(data: &[u8]) -> <Self::View as WithLifetime>::Out {
+        let transaction_id = u64::from_le_bytes(data[..size_of::<u64>()].try_into().unwrap());
+        let pagination_id = u64::from_le_bytes(data[size_of::<u64>()..].try_into().unwrap());
+        Self {
+            transaction_id,
+            pagination_id,
+        }
+    }
+
+    fn as_bytes(&self) -> <Self::ToBytes as AsBytesWithLifetime>::Out {
+        let mut result = [0u8; 2 * size_of::<u64>()];
+        result[..size_of::<u64>()].copy_from_slice(&self.transaction_id.to_le_bytes());
+        result[size_of::<u64>()..].copy_from_slice(&self.pagination_id.to_le_bytes());
+        result
+    }
+
+    fn redb_type_name() -> &'static str {
+        "FreedTableKey"
+    }
+}
+
+impl RedbKey for FreedTableKey {
+    fn compare(data1: &[u8], data2: &[u8]) -> std::cmp::Ordering {
+        let value1 = Self::from_bytes(data1);
+        let value2 = Self::from_bytes(data2);
+
+        match value1.transaction_id.cmp(&value2.transaction_id) {
+            std::cmp::Ordering::Greater => std::cmp::Ordering::Greater,
+            std::cmp::Ordering::Equal => value1.pagination_id.cmp(&value2.pagination_id),
+            std::cmp::Ordering::Less => std::cmp::Ordering::Less,
+        }
+    }
+}
+
+#[derive(Copy, Clone, Hash, Eq, PartialEq, Debug)]
+pub(crate) enum TableType {
+    Normal,
+    Multimap,
+}
+
+#[allow(clippy::from_over_into)]
+impl Into<u8> for TableType {
+    fn into(self) -> u8 {
+        match self {
+            TableType::Normal => 1,
+            TableType::Multimap => 2,
+        }
+    }
+}
+
+impl From<u8> for TableType {
+    fn from(value: u8) -> Self {
+        match value {
+            1 => TableType::Normal,
+            2 => TableType::Multimap,
+            _ => unreachable!(),
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct InternalTableDefinition {
+    // TODO: make private
+    pub(crate) table_root: Option<PageNumber>,
+    pub(crate) table_type: TableType,
+    pub(crate) key_type: String,
+    pub(crate) value_type: String,
+}
+
+impl InternalTableDefinition {
+    pub(crate) fn get_root(&self) -> Option<PageNumber> {
+        self.table_root
+    }
+
+    pub(crate) fn get_type(&self) -> TableType {
+        self.table_type
+    }
+}
+
+impl RedbValue for InternalTableDefinition {
+    type View = OwnedLifetime<InternalTableDefinition>;
+    type ToBytes = OwnedAsBytesLifetime<Vec<u8>>;
+
+    fn from_bytes(data: &[u8]) -> <Self::View as WithLifetime>::Out {
+        debug_assert!(data.len() > 14);
+        let table_type = TableType::from(data[0]);
+        let table_root = PageNumber::from_le_bytes(data[1..9].try_into().unwrap());
+        let table_root = if table_root == PageNumber::null() {
+            None
+        } else {
+            Some(table_root)
+        };
+        let key_type_len = u32::from_le_bytes(data[9..13].try_into().unwrap()) as usize;
+        let key_type = std::str::from_utf8(&data[13..(13 + key_type_len)])
+            .unwrap()
+            .to_string();
+        let value_type = std::str::from_utf8(&data[(13 + key_type_len)..])
+            .unwrap()
+            .to_string();
+
+        InternalTableDefinition {
+            table_root,
+            table_type,
+            key_type,
+            value_type,
+        }
+    }
+
+    fn as_bytes(&self) -> <Self::ToBytes as AsBytesWithLifetime>::Out {
+        let mut result = vec![self.table_type.into()];
+        result.extend_from_slice(
+            &self
+                .table_root
+                .unwrap_or_else(PageNumber::null)
+                .to_le_bytes(),
+        );
+        result.extend_from_slice(&(self.key_type.as_bytes().len() as u32).to_le_bytes());
+        result.extend_from_slice(self.key_type.as_bytes());
+        result.extend_from_slice(self.value_type.as_bytes());
+
+        result
+    }
+
+    fn redb_type_name() -> &'static str {
+        "InternalTableDefinition"
+    }
+}
+
+pub struct TableNameIter<'a> {
+    inner: BtreeRangeIter<'a, str, InternalTableDefinition>,
+    table_type: TableType,
+}
+
+impl<'a> Iterator for TableNameIter<'a> {
+    type Item = String;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        while let Some(entry) = self.inner.next() {
+            if str::from_bytes(entry.key()) == FREED_TABLE {
+                continue;
+            }
+            if InternalTableDefinition::from_bytes(entry.value()).table_type == self.table_type {
+                return Some(str::from_bytes(entry.key()).to_string());
+            }
+        }
+        None
+    }
+}
+
+pub(crate) struct TableTree<'txn> {
+    // TODO: make private?
+    pub(crate) tree: BtreeMut<'txn, str, InternalTableDefinition>,
+    mem: &'txn TransactionalMemory,
+    // TODO: unify with the freed pages in self.tree
+    pub(crate) extra_freed_pages: Vec<PageNumber>,
+}
+
+impl<'txn> TableTree<'txn> {
+    pub(crate) fn new(master_root: Option<PageNumber>, mem: &'txn TransactionalMemory) -> Self {
+        Self {
+            tree: BtreeMut::new(master_root, mem),
+            mem,
+            extra_freed_pages: vec![],
+        }
+    }
+
+    pub(crate) fn update_table_root(
+        &mut self,
+        name: &str,
+        table_root: Option<PageNumber>,
+    ) -> Result {
+        // Bypass .get_table() since the table types are dynamic
+        // TODO: optimize way this get()
+        let mut definition = self.tree.get(name).unwrap().unwrap();
+        // No-op if the root has not changed
+        if definition.table_root == table_root {
+            return Ok(());
+        }
+        definition.table_root = table_root;
+        // Safety: References into the master table are never returned to the user
+        unsafe {
+            self.tree.insert(name, &definition)?;
+        }
+        Ok(())
+    }
+
+    // root_page: the root of the master table
+    pub(crate) fn list_tables(&self, table_type: TableType) -> Result<Vec<String>> {
+        let iter = self.tree.range::<RangeFull, &str>(..)?;
+        let iter = TableNameIter {
+            inner: iter,
+            table_type,
+        };
+        Ok(iter.collect())
+    }
+
+    // root_page: the root of the master table
+    pub(crate) fn get_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &self,
+        name: &str,
+        table_type: TableType,
+    ) -> Result<Option<InternalTableDefinition>> {
+        if let Some(definition) = self.tree.get(name)? {
+            if definition.get_type() != table_type {
+                return Err(Error::TableTypeMismatch(format!(
+                    "{:?} is not of type {:?}",
+                    name, table_type
+                )));
+            }
+            if definition.key_type != K::redb_type_name()
+                || definition.value_type != V::redb_type_name()
+            {
+                return Err(Error::TableTypeMismatch(format!(
+                    "{} is of type Table<{}, {}> not Table<{}, {}>",
+                    name,
+                    &definition.key_type,
+                    &definition.value_type,
+                    K::redb_type_name(),
+                    V::redb_type_name()
+                )));
+            }
+
+            Ok(Some(definition))
+        } else {
+            Ok(None)
+        }
+    }
+
+    // root_page: the root of the master table
+    pub(crate) fn delete_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &mut self,
+        name: &str,
+        table_type: TableType,
+    ) -> Result<bool> {
+        if let Some(definition) = self.get_table::<K, V>(name, table_type)? {
+            if let Some(table_root) = definition.get_root() {
+                let iter = AllPageNumbersBtreeIter::new(table_root, self.mem);
+                for page_number in iter {
+                    self.extra_freed_pages.push(page_number);
+                }
+            }
+
+            // Safety: References into the master table are never returned to the user
+            let found = unsafe { self.tree.remove(name)?.is_some() };
+            return Ok(found);
+        }
+
+        Ok(false)
+    }
+
+    // Returns a tuple of the table id and the new root page
+    // root_page: the root of the master table
+    pub(crate) fn get_or_create_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &mut self,
+        name: &str,
+        table_type: TableType,
+    ) -> Result<InternalTableDefinition> {
+        if let Some(found) = self.get_table::<K, V>(name, table_type)? {
+            return Ok(found);
+        }
+
+        let table = InternalTableDefinition {
+            table_root: None,
+            table_type,
+            key_type: K::redb_type_name().to_string(),
+            value_type: V::redb_type_name().to_string(),
+        };
+        // Safety: References into the master table are never returned to the user
+        unsafe { self.tree.insert(name, &table)? };
+        Ok(table)
+    }
+}


Test patch:
diff --git a/tests/basic_tests.rs b/tests/basic_tests.rs
index c9adc04..4535f96 100644
--- a/tests/basic_tests.rs
+++ b/tests/basic_tests.rs
@@ -34,9 +34,11 @@ fn stored_size() {
     }
     write_txn.commit().unwrap();
 
-    assert_eq!(db.stats().unwrap().stored_bytes(), 10);
-    assert!(db.stats().unwrap().fragmented_bytes() > 0);
-    assert!(db.stats().unwrap().metadata_bytes() > 0);
+    let write_txn = db.begin_write().unwrap();
+    assert_eq!(write_txn.stats().unwrap().stored_bytes(), 10);
+    assert!(write_txn.stats().unwrap().fragmented_bytes() > 0);
+    assert!(write_txn.stats().unwrap().metadata_bytes() > 0);
+    write_txn.abort().unwrap();
 }
 
 #[test]
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index 7e45452..49ec5e3 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -156,9 +156,9 @@ fn free() {
     }
 
     txn.commit().unwrap();
-    let free_pages = db.stats().unwrap().free_pages();
 
     let txn = db.begin_write().unwrap();
+    let free_pages = txn.stats().unwrap().free_pages();
 
     let key = vec![0; 100];
     let value = vec![0; 1024];
@@ -194,7 +194,9 @@ fn free() {
         }
     }
 
-    assert_eq!(free_pages, db.stats().unwrap().free_pages());
+    let txn = db.begin_write().unwrap();
+    assert_eq!(free_pages, txn.stats().unwrap().free_pages());
+    txn.abort().unwrap();
 }
 
 #[test]