Repository: cberner/redb
Pull request #101 (2021-12-28 02:29:52)

Refactor transaction API


Patch:
diff --git a/benches/common.rs b/benches/common.rs
index 39a84a2..ccbd51e 100644
--- a/benches/common.rs
+++ b/benches/common.rs
@@ -1,5 +1,5 @@
-pub trait BenchTable {
-    type W: BenchWriteTransaction;
+pub trait BenchDatabase {
+    type W: for<'a> BenchWriteTransaction<'a>;
     type R: for<'a> BenchReadTransaction<'a>;
 
     fn db_type_name() -> &'static str;
@@ -9,31 +9,35 @@ pub trait BenchTable {
     fn read_transaction(&self) -> Self::R;
 }
 
-pub trait BenchWriteTransaction {
-    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()>;
+pub trait BenchWriteTransaction<'a> {
+    type T: BenchInserter + 'a;
+
+    fn get_inserter(&'a self) -> Self::T;
 
     fn commit(self) -> Result<(), ()>;
 }
 
+pub trait BenchInserter {
+    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()>;
+}
+
 pub trait BenchReadTransaction<'a> {
     type Output: AsRef<[u8]> + 'a;
 
     fn get(&'a self, key: &[u8]) -> Option<Self::Output>;
 }
 
-pub struct RedbBenchTable<'a> {
-    table: redb::Table<'a, [u8], [u8]>,
+pub struct RedbBenchDatabase<'a> {
+    db: &'a redb::Database,
 }
 
-impl<'a> RedbBenchTable<'a> {
+impl<'a> RedbBenchDatabase<'a> {
     pub fn new(db: &'a redb::Database) -> Self {
-        RedbBenchTable {
-            table: db.open_table(b"x").unwrap(),
-        }
+        RedbBenchDatabase { db }
     }
 }
 
-impl<'a> BenchTable for RedbBenchTable<'a> {
+impl<'a> BenchDatabase for RedbBenchDatabase<'a> {
     type W = RedbBenchWriteTransaction<'a>;
     type R = RedbBenchReadTransaction<'a>;
 
@@ -42,37 +46,41 @@ impl<'a> BenchTable for RedbBenchTable<'a> {
     }
 
     fn write_transaction(&mut self) -> Self::W {
-        RedbBenchWriteTransaction {
-            txn: self.table.begin_write().unwrap(),
-        }
+        let txn = self.db.begin_write().unwrap();
+        RedbBenchWriteTransaction { txn }
     }
 
     fn read_transaction(&self) -> Self::R {
-        RedbBenchReadTransaction {
-            txn: self.table.read_transaction().unwrap(),
-        }
+        let txn = self.db.begin_read().unwrap();
+        let table = txn.open_table("x").unwrap();
+        RedbBenchReadTransaction { _txn: txn, table }
     }
 }
 
 pub struct RedbBenchReadTransaction<'a> {
-    txn: redb::ReadOnlyTransaction<'a, [u8], [u8]>,
+    _txn: redb::ReadOnlyDatabaseTransaction<'a>,
+    table: redb::ReadOnlyTable<'a, [u8], [u8]>,
 }
 
 impl<'a, 'b> BenchReadTransaction<'b> for RedbBenchReadTransaction<'a> {
     type Output = redb::AccessGuard<'b, [u8]>;
 
     fn get(&'b self, key: &[u8]) -> Option<redb::AccessGuard<'b, [u8]>> {
-        self.txn.get(key).unwrap()
+        self.table.get(key).unwrap()
     }
 }
 
 pub struct RedbBenchWriteTransaction<'a> {
-    txn: redb::WriteTransaction<'a, [u8], [u8]>,
+    txn: redb::DatabaseTransaction<'a>,
 }
 
-impl BenchWriteTransaction for RedbBenchWriteTransaction<'_> {
-    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()> {
-        self.txn.insert(key, value).map_err(|_| ())
+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(),
+        }
     }
 
     fn commit(self) -> Result<(), ()> {
@@ -80,17 +88,27 @@ impl BenchWriteTransaction for RedbBenchWriteTransaction<'_> {
     }
 }
 
-pub struct SledBenchTable<'a> {
+pub struct RedbBenchInserter<'a> {
+    table: redb::Table<'a, 'a, [u8], [u8]>,
+}
+
+impl BenchInserter for RedbBenchInserter<'_> {
+    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()> {
+        self.table.insert(key, value).map_err(|_| ())
+    }
+}
+
+pub struct SledBenchDatabase<'a> {
     db: &'a sled::Db,
 }
 
-impl<'a> SledBenchTable<'a> {
+impl<'a> SledBenchDatabase<'a> {
     pub fn new(db: &'a sled::Db) -> Self {
-        SledBenchTable { db }
+        SledBenchDatabase { db }
     }
 }
 
-impl<'a> BenchTable for SledBenchTable<'a> {
+impl<'a> BenchDatabase for SledBenchDatabase<'a> {
     type W = SledBenchWriteTransaction<'a>;
     type R = SledBenchReadTransaction<'a>;
 
@@ -123,9 +141,11 @@ pub struct SledBenchWriteTransaction<'a> {
     db: &'a sled::Db,
 }
 
-impl BenchWriteTransaction for SledBenchWriteTransaction<'_> {
-    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()> {
-        self.db.insert(key, value).map(|_| ()).map_err(|_| ())
+impl<'a, 'b> BenchWriteTransaction<'b> for SledBenchWriteTransaction<'a> {
+    type T = SledBenchInserter<'b>;
+
+    fn get_inserter(&'b self) -> Self::T {
+        SledBenchInserter { db: self.db }
     }
 
     fn commit(self) -> Result<(), ()> {
@@ -133,19 +153,29 @@ impl BenchWriteTransaction for SledBenchWriteTransaction<'_> {
     }
 }
 
-pub struct LmdbRkvBenchTable<'a> {
+pub struct SledBenchInserter<'a> {
+    db: &'a sled::Db,
+}
+
+impl<'a> BenchInserter for SledBenchInserter<'a> {
+    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()> {
+        self.db.insert(key, value).map(|_| ()).map_err(|_| ())
+    }
+}
+
+pub struct LmdbRkvBenchDatabase<'a> {
     env: &'a lmdb::Environment,
     db: lmdb::Database,
 }
 
-impl<'a> LmdbRkvBenchTable<'a> {
-    pub fn new(env: &'a lmdb::Environment) -> Self {
+impl<'a> LmdbRkvBenchDatabase<'a> {
+    pub fn _new(env: &'a lmdb::Environment) -> Self {
         let db = env.open_db(None).unwrap();
-        LmdbRkvBenchTable { env, db }
+        LmdbRkvBenchDatabase { env, db }
     }
 }
 
-impl<'a> BenchTable for LmdbRkvBenchTable<'a> {
+impl<'a> BenchDatabase for LmdbRkvBenchDatabase<'a> {
     type W = LmdbRkvBenchWriteTransaction<'a>;
     type R = LmdbRkvBenchReadTransaction<'a>;
 
@@ -169,11 +199,14 @@ pub struct LmdbRkvBenchWriteTransaction<'a> {
     txn: lmdb::RwTransaction<'a>,
 }
 
-impl BenchWriteTransaction for LmdbRkvBenchWriteTransaction<'_> {
-    fn insert(&mut self, key: &[u8], value: &[u8]) -> Result<(), ()> {
-        self.txn
-            .put(self.db, &key, &value, lmdb::WriteFlags::empty())
-            .map_err(|_| ())
+impl<'a, 'b> BenchWriteTransaction<'b> for LmdbRkvBenchWriteTransaction<'a> {
+    type T = LmdbRkvBenchInserter<'b>;
+
+    fn get_inserter(&'b self) -> Self::T {
+        LmdbRkvBenchInserter {
+            _db: self.db,
+            _txn: &self.txn,
+        }
     }
 
     fn commit(self) -> Result<(), ()> {
@@ -182,6 +215,20 @@ impl BenchWriteTransaction for LmdbRkvBenchWriteTransaction<'_> {
     }
 }
 
+pub struct LmdbRkvBenchInserter<'a> {
+    _db: lmdb::Database,
+    _txn: &'a lmdb::RwTransaction<'a>,
+}
+
+impl BenchInserter for LmdbRkvBenchInserter<'_> {
+    fn insert(&mut self, _key: &[u8], _value: &[u8]) -> Result<(), ()> {
+        todo!("Fix this")
+        // self.txn
+        //     .put(self.db, &key, &value, lmdb::WriteFlags::empty())
+        //     .map_err(|_| ())
+    }
+}
+
 pub struct LmdbRkvBenchReadTransaction<'a> {
     db: lmdb::Database,
     txn: lmdb::RoTransaction<'a>,
diff --git a/benches/lmdb_benchmark.rs b/benches/lmdb_benchmark.rs
index cfbc04f..ce3af0a 100644
--- a/benches/lmdb_benchmark.rs
+++ b/benches/lmdb_benchmark.rs
@@ -23,21 +23,23 @@ fn gen_data(count: usize, key_size: usize, value_size: usize) -> Vec<(Vec<u8>, V
     pairs
 }
 
-fn benchmark<T: BenchTable>(mut db: T) {
+fn benchmark<T: BenchDatabase>(mut db: T) {
     let pairs = gen_data(1000, 16, 1500);
     let mut written = 0;
 
     let start = SystemTime::now();
-    let mut txn = db.write_transaction();
+    let txn = db.write_transaction();
+    let mut inserter = txn.get_inserter();
     {
         for _ in 0..ELEMENTS {
             let (key, value) = &pairs[written % pairs.len()];
             let mut mut_key = key.clone();
             mut_key.extend_from_slice(&written.to_be_bytes());
-            txn.insert(&mut_key, value).unwrap();
+            inserter.insert(&mut_key, value).unwrap();
             written += 1;
         }
     }
+    drop(inserter);
     txn.commit().unwrap();
 
     let end = SystemTime::now();
@@ -53,11 +55,13 @@ fn benchmark<T: BenchTable>(mut db: T) {
     let writes = 100;
     {
         for _ in 0..writes {
-            let mut txn = db.write_transaction();
+            let txn = db.write_transaction();
+            let mut inserter = txn.get_inserter();
             let (key, value) = &pairs[written % pairs.len()];
             let mut mut_key = key.clone();
             mut_key.extend_from_slice(&written.to_be_bytes());
-            txn.insert(&mut_key, value).unwrap();
+            inserter.insert(&mut_key, value).unwrap();
+            drop(inserter);
             txn.commit().unwrap();
             written += 1;
         }
@@ -76,14 +80,16 @@ fn benchmark<T: BenchTable>(mut db: T) {
     let batch_size = 1000;
     {
         for _ in 0..writes {
-            let mut txn = db.write_transaction();
+            let txn = db.write_transaction();
+            let mut inserter = txn.get_inserter();
             for _ in 0..batch_size {
                 let (key, value) = &pairs[written % pairs.len()];
                 let mut mut_key = key.clone();
                 mut_key.extend_from_slice(&written.to_be_bytes());
-                txn.insert(&mut_key, value).unwrap();
+                inserter.insert(&mut_key, value).unwrap();
                 written += 1;
             }
+            drop(inserter);
             txn.commit().unwrap();
         }
     }
@@ -129,23 +135,23 @@ fn benchmark<T: BenchTable>(mut db: T) {
 }
 
 fn main() {
-    {
-        let tmpfile: TempDir = tempfile::tempdir().unwrap();
-        let env = lmdb::Environment::new().open(tmpfile.path()).unwrap();
-        env.set_map_size(4096 * 1024 * 1024).unwrap();
-        let table = LmdbRkvBenchTable::new(&env);
-        benchmark(table);
-    }
+    // {
+    //     let tmpfile: TempDir = tempfile::tempdir().unwrap();
+    //     let env = lmdb::Environment::new().open(tmpfile.path()).unwrap();
+    //     env.set_map_size(4096 * 1024 * 1024).unwrap();
+    //     let table = LmdbRkvBenchDatabase::new(&env);
+    //     benchmark(table);
+    // }
     {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { redb::Database::open(tmpfile.path(), 4096 * 1024 * 1024).unwrap() };
-        let table = RedbBenchTable::new(&db);
+        let table = RedbBenchDatabase::new(&db);
         benchmark(table);
     }
     {
         let tmpfile: TempDir = tempfile::tempdir().unwrap();
         let db = sled::Config::new().path(tmpfile.path()).open().unwrap();
-        let table = SledBenchTable::new(&db);
+        let table = SledBenchDatabase::new(&db);
         benchmark(table);
     }
 }
diff --git a/src/db.rs b/src/db.rs
index 36d1543..93257df 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -1,9 +1,10 @@
 use crate::multimap_table::MultiMapTable;
-use crate::table::Table;
-use crate::tree_store::{DbStats, Storage, TableType};
+use crate::table::{ReadOnlyTable, Table};
+use crate::tree_store::{DbStats, NodeHandle, Storage, TableType};
 use crate::types::{RedbKey, RedbValue};
-use crate::Error;
+use crate::{Error, ReadOnlyMultiMapTable};
 use memmap2::MmapMut;
+use std::cell::Cell;
 use std::fs::OpenOptions;
 use std::path::Path;
 
@@ -39,38 +40,12 @@ impl Database {
         Ok(Database { storage })
     }
 
-    pub fn open_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
-        &self,
-        name: impl AsRef<[u8]>,
-    ) -> Result<Table<K, V>, Error> {
-        assert!(!name.as_ref().is_empty());
-        // TODO: this could conflict with an on-going write
-        let id = self.storage.allocate_write_transaction();
-        let (definition, root) = self.storage.get_or_create_table(
-            name.as_ref(),
-            TableType::Normal,
-            id,
-            self.storage.get_root_page_number(),
-        )?;
-        self.storage.commit(Some(root), id)?;
-        Table::new(definition.get_id(), &self.storage)
+    pub fn begin_write(&self) -> Result<DatabaseTransaction, Error> {
+        Ok(DatabaseTransaction::new(&self.storage))
     }
 
-    pub fn open_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
-        &self,
-        name: impl AsRef<[u8]>,
-    ) -> Result<MultiMapTable<K, V>, Error> {
-        assert!(!name.as_ref().is_empty());
-        // TODO: this could conflict with an on-going write
-        let id = self.storage.allocate_write_transaction();
-        let (definition, root) = self.storage.get_or_create_table(
-            name.as_ref(),
-            TableType::MultiMap,
-            id,
-            self.storage.get_root_page_number(),
-        )?;
-        self.storage.commit(Some(root), id)?;
-        MultiMapTable::new(definition.get_id(), &self.storage)
+    pub fn begin_read(&self) -> Result<ReadOnlyDatabaseTransaction, Error> {
+        Ok(ReadOnlyDatabaseTransaction::new(&self.storage))
     }
 
     pub fn stats(&self) -> Result<DbStats, Error> {
@@ -126,9 +101,164 @@ impl DatabaseBuilder {
     }
 }
 
+pub struct DatabaseTransaction<'a> {
+    storage: &'a Storage,
+    transaction_id: u128,
+    root_page: Cell<Option<NodeHandle>>,
+}
+
+impl<'a> DatabaseTransaction<'a> {
+    fn new(storage: &'a Storage) -> Self {
+        let transaction_id = storage.allocate_write_transaction();
+        let root_page = storage.get_root_page_number();
+        Self {
+            storage,
+            transaction_id,
+            root_page: Cell::new(root_page),
+        }
+    }
+
+    /// Open the given table
+    ///
+    /// The table will be created if it does not exist
+    pub fn open_table<K: RedbKey + ?Sized, V: RedbValue + ?Sized>(
+        &self,
+        name: impl AsRef<[u8]>,
+    ) -> Result<Table<'a, '_, K, V>, Error> {
+        assert!(!name.as_ref().is_empty());
+        let (definition, root) = self.storage.get_or_create_table(
+            name.as_ref(),
+            TableType::Normal,
+            self.transaction_id,
+            self.root_page.get(),
+        )?;
+        self.root_page.set(Some(root));
+
+        Ok(Table::new(
+            definition.get_id(),
+            self.transaction_id,
+            &self.root_page,
+            self.storage,
+        ))
+    }
+
+    /// Open the given table
+    ///
+    /// The table will be created if it does not exist
+    pub fn open_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
+        &self,
+        name: impl AsRef<[u8]>,
+    ) -> Result<MultiMapTable<'a, '_, K, V>, Error> {
+        assert!(!name.as_ref().is_empty());
+        let (definition, root) = self.storage.get_or_create_table(
+            name.as_ref(),
+            TableType::MultiMap,
+            self.transaction_id,
+            self.root_page.get(),
+        )?;
+        self.root_page.set(Some(root));
+
+        Ok(MultiMapTable::new(
+            definition.get_id(),
+            self.transaction_id,
+            &self.root_page,
+            self.storage,
+        ))
+    }
+
+    pub fn commit(self) -> Result<(), Error> {
+        self.storage
+            .commit(self.root_page.get(), self.transaction_id)?;
+        Ok(())
+    }
+
+    /// Note: pages are only freed during commit(). So exclusively using this function may result
+    /// in an out-of-memory error
+    pub fn non_durable_commit(self) -> Result<(), Error> {
+        self.storage
+            .non_durable_commit(self.root_page.get(), self.transaction_id)?;
+        Ok(())
+    }
+
+    pub fn abort(self) -> Result<(), Error> {
+        self.storage.rollback_uncommited_writes(self.transaction_id)
+    }
+}
+
+pub struct ReadOnlyDatabaseTransaction<'a> {
+    storage: &'a Storage,
+    transaction_id: u128,
+    root_page: Option<NodeHandle>,
+}
+
+impl<'a> ReadOnlyDatabaseTransaction<'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,
+        name: impl AsRef<[u8]>,
+    ) -> Result<ReadOnlyTable<'a, K, V>, Error> {
+        assert!(!name.as_ref().is_empty());
+        let definition = self
+            .storage
+            .get_table(name.as_ref(), TableType::Normal, self.root_page)?
+            .ok_or_else(|| {
+                Error::DoesNotExist(format!(
+                    "Table '{}' does not exist",
+                    String::from_utf8_lossy(name.as_ref())
+                ))
+            })?;
+
+        Ok(ReadOnlyTable::new(
+            definition.get_id(),
+            self.root_page,
+            self.storage,
+        ))
+    }
+
+    /// Open the given table
+    pub fn open_multimap_table<K: RedbKey + ?Sized, V: RedbKey + ?Sized>(
+        &self,
+        name: impl AsRef<[u8]>,
+    ) -> Result<ReadOnlyMultiMapTable<'a, K, V>, Error> {
+        assert!(!name.as_ref().is_empty());
+        let definition = self
+            .storage
+            .get_table(name.as_ref(), TableType::MultiMap, self.root_page)?
+            .ok_or_else(|| {
+                Error::DoesNotExist(format!(
+                    "Table '{}' does not exist",
+                    String::from_utf8_lossy(name.as_ref())
+                ))
+            })?;
+
+        Ok(ReadOnlyMultiMapTable::new(
+            definition.get_id(),
+            self.root_page,
+            self.storage,
+        ))
+    }
+}
+
+impl<'a> Drop for ReadOnlyDatabaseTransaction<'a> {
+    fn drop(&mut self) {
+        self.storage
+            .deallocate_read_transaction(self.transaction_id);
+    }
+}
+
 #[cfg(test)]
 mod test {
-    use crate::{Database, Table};
+    use crate::{Database, ReadOnlyTable, Table};
     use tempfile::NamedTempFile;
 
     #[test]
@@ -137,15 +267,32 @@ mod test {
 
         let db_size = 1024 * 1024 + 1;
         let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table("x").unwrap();
+        let txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
         let key = vec![0; 1024];
         let value = vec![0; 1];
-        let mut txn = table.begin_write().unwrap();
-        txn.insert(&key, &value).unwrap();
+        table.insert(&key, &value).unwrap();
         txn.commit().unwrap();
 
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(read_txn.len().unwrap(), 1);
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(table.len().unwrap(), 1);
+    }
+
+    #[test]
+    fn transaction_id_persistence() {
+        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"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::open(tmpfile.path(), 1024 * 1024).unwrap() };
+        let write_txn = db2.begin_write().unwrap();
+        assert!(write_txn.transaction_id > first_txn_id);
     }
 }
diff --git a/src/error.rs b/src/error.rs
index aafa064..1b97197 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -5,6 +5,7 @@ use std::io;
 pub enum Error {
     Corrupted(String),
     TableTypeMismatch(String),
+    DoesNotExist(String),
     Io(io::Error),
 }
 
diff --git a/src/lib.rs b/src/lib.rs
index 43bf018..48502f7 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,10 +1,7 @@
-pub use db::{Database, DatabaseBuilder};
+pub use db::{Database, DatabaseBuilder, DatabaseTransaction, ReadOnlyDatabaseTransaction};
 pub use error::Error;
-pub use multimap_table::{
-    MultiMapRangeIter, MultiMapReadOnlyTransaction, MultiMapTable, MultiMapWriteTransaction,
-};
-pub use table::Table;
-pub use transactions::{ReadOnlyTransaction, WriteTransaction};
+pub use multimap_table::{MultiMapRangeIter, MultiMapTable, ReadOnlyMultiMapTable};
+pub use table::{ReadOnlyTable, Table};
 pub use tree_store::AccessGuard;
 
 #[cfg(feature = "python")]
@@ -16,6 +13,5 @@ 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 32125ad..e5345c8 100644
--- a/src/multimap_table.rs
+++ b/src/multimap_table.rs
@@ -3,6 +3,7 @@ use crate::tree_store::{BtreeEntry, BtreeRangeIter, NodeHandle, Storage};
 use crate::types::{
     AsBytesWithLifetime, RedbKey, RedbValue, RefAsBytesLifetime, RefLifetime, WithLifetime,
 };
+use std::cell::Cell;
 use std::cmp::Ordering;
 use std::collections::Bound;
 use std::convert::TryInto;
@@ -258,58 +259,27 @@ impl<
     }
 }
 
-pub struct MultiMapTable<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
-    storage: &'mmap Storage,
-    table_id: u64,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
-}
-
-impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapTable<'mmap, K, V> {
-    pub(in crate) fn new(
-        table_id: u64,
-        storage: &'mmap Storage,
-    ) -> Result<MultiMapTable<'mmap, K, V>, Error> {
-        Ok(MultiMapTable {
-            storage,
-            table_id,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
-        })
-    }
-
-    pub fn begin_write(&'_ mut self) -> Result<MultiMapWriteTransaction<'mmap, K, V>, Error> {
-        Ok(MultiMapWriteTransaction::new(self.table_id, self.storage))
-    }
-
-    pub fn read_transaction(&'_ self) -> Result<MultiMapReadOnlyTransaction<'mmap, K, V>, Error> {
-        Ok(MultiMapReadOnlyTransaction::new(
-            self.table_id,
-            self.storage,
-        ))
-    }
-}
-
-pub struct MultiMapWriteTransaction<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
-    storage: &'mmap Storage,
+pub struct MultiMapTable<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
+    storage: &'s Storage,
     table_id: u64,
     transaction_id: u128,
-    root_page: Option<NodeHandle>,
+    root_page: &'t Cell<Option<NodeHandle>>,
     _key_type: PhantomData<K>,
     _value_type: PhantomData<V>,
 }
 
-impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapWriteTransaction<'mmap, K, V> {
+impl<'s, 't, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapTable<'s, 't, K, V> {
     pub(in crate) fn new(
         table_id: u64,
-        storage: &'mmap Storage,
-    ) -> MultiMapWriteTransaction<'mmap, K, V> {
-        let transaction_id = storage.allocate_write_transaction();
-        MultiMapWriteTransaction {
+        transaction_id: u128,
+        root_page: &'t Cell<Option<NodeHandle>>,
+        storage: &'s Storage,
+    ) -> MultiMapTable<'s, 't, K, V> {
+        MultiMapTable {
             storage,
             table_id,
             transaction_id,
-            root_page: storage.get_root_page_number(),
+            root_page,
             _key_type: Default::default(),
             _value_type: Default::default(),
         }
@@ -317,20 +287,21 @@ impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapWriteTransaction<'
 
     #[allow(dead_code)]
     pub(in crate) fn print_debug(&self) {
-        if let Some(page) = self.root_page {
+        if let Some(page) = self.root_page.get() {
             self.storage.print_dirty_debug(page);
         }
     }
 
     pub fn insert(&mut self, key: &K, value: &V) -> Result<(), Error> {
         let kv = make_serialized_kv(key, value);
-        self.root_page = Some(self.storage.insert::<MultiMapKVPair<K, V>>(
+        let page = Some(self.storage.insert::<MultiMapKVPair<K, V>>(
             self.table_id,
             &kv,
             b"",
             self.transaction_id,
-            self.root_page,
+            self.root_page.get(),
         )?);
+        self.root_page.set(page);
         Ok(())
     }
 
@@ -343,18 +314,19 @@ impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapWriteTransaction<'
         let lower = MultiMapKVPair::<K, V>::new(lower_bytes);
         let upper = MultiMapKVPair::<K, V>::new(upper_bytes);
         self.storage
-            .get_range(self.table_id, lower..=upper, self.root_page)
+            .get_range(self.table_id, lower..=upper, self.root_page.get())
             .map(MultiMapRangeIter::new)
     }
 
     pub fn remove(&mut self, key: &K, value: &V) -> Result<(), Error> {
         let kv = make_serialized_kv(key, value);
-        self.root_page = self.storage.remove::<MultiMapKVPair<K, V>>(
+        let page = self.storage.remove::<MultiMapKVPair<K, V>>(
             self.table_id,
             &kv,
             self.transaction_id,
-            self.root_page,
+            self.root_page.get(),
         )?;
+        self.root_page.set(page);
         Ok(())
     }
 
@@ -362,59 +334,40 @@ impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapWriteTransaction<'
         // 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);
         loop {
-            let old_root = self.root_page;
-            self.root_page = self.storage.remove::<MultiMapKVPair<K, V>>(
+            let old_root = self.root_page.get();
+            let new_root = self.storage.remove::<MultiMapKVPair<K, V>>(
                 self.table_id,
                 &key_only,
                 self.transaction_id,
-                self.root_page,
+                self.root_page.get(),
             )?;
-            if old_root == self.root_page {
+            if old_root == new_root {
                 break;
             }
+            self.root_page.set(new_root);
         }
         Ok(())
     }
-
-    pub fn commit(self) -> Result<(), Error> {
-        self.storage.commit(self.root_page, self.transaction_id)?;
-        Ok(())
-    }
-
-    /// Note: pages are only freed during commit(). So exclusively using this function may result
-    /// in an out-of-memory error
-    pub fn non_durable_commit(self) -> Result<(), Error> {
-        self.storage
-            .non_durable_commit(self.root_page, self.transaction_id)?;
-        Ok(())
-    }
-
-    pub fn abort(self) -> Result<(), Error> {
-        self.storage.rollback_uncommited_writes(self.transaction_id)
-    }
 }
 
-pub struct MultiMapReadOnlyTransaction<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
-    storage: &'mmap Storage,
+pub struct ReadOnlyMultiMapTable<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
+    storage: &'s Storage,
     root_page: Option<NodeHandle>,
     table_id: u64,
-    transaction_id: u128,
     _key_type: PhantomData<K>,
     _value_type: PhantomData<V>,
 }
 
-impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapReadOnlyTransaction<'mmap, K, V> {
+impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadOnlyMultiMapTable<'s, K, V> {
     pub(in crate) fn new(
         table_id: u64,
-        storage: &'mmap Storage,
-    ) -> MultiMapReadOnlyTransaction<'mmap, K, V> {
-        let root_page = storage.get_root_page_number();
-        let transaction_id = storage.allocate_read_transaction();
-        MultiMapReadOnlyTransaction {
+        root_page: Option<NodeHandle>,
+        storage: &'s Storage,
+    ) -> ReadOnlyMultiMapTable<'s, K, V> {
+        ReadOnlyMultiMapTable {
             storage,
             root_page,
             table_id,
-            transaction_id,
             _key_type: Default::default(),
             _value_type: Default::default(),
         }
@@ -474,23 +427,14 @@ impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultiMapReadOnlyTransactio
     }
 }
 
-impl<'mmap, K: RedbKey + ?Sized, V: RedbKey + ?Sized> Drop
-    for MultiMapReadOnlyTransaction<'mmap, K, V>
-{
-    fn drop(&mut self) {
-        self.storage
-            .deallocate_read_transaction(self.transaction_id);
-    }
-}
-
 #[cfg(test)]
 mod test {
-    use crate::{Database, MultiMapReadOnlyTransaction, MultiMapTable};
+    use crate::{Database, MultiMapTable, ReadOnlyMultiMapTable};
     use tempfile::NamedTempFile;
 
-    fn get_vec(txn: &MultiMapReadOnlyTransaction<[u8], [u8]>, key: &[u8]) -> Vec<Vec<u8>> {
+    fn get_vec(table: &ReadOnlyMultiMapTable<[u8], [u8]>, key: &[u8]) -> Vec<Vec<u8>> {
         let mut result = vec![];
-        let mut iter = txn.get(key).unwrap();
+        let mut iter = table.get(key).unwrap();
         loop {
             let item = iter.next();
             if let Some((item_key, item_value)) = item {
@@ -506,71 +450,78 @@ mod test {
     fn len() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: MultiMapTable<[u8], [u8]> = db.open_multimap_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        write_txn.insert(b"hello", b"world2").unwrap();
-        write_txn.insert(b"hi", b"world").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+
+        table.insert(b"hello", b"world").unwrap();
+        table.insert(b"hello", b"world2").unwrap();
+        table.insert(b"hi", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(read_txn.len().unwrap(), 3);
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+        assert_eq!(table.len().unwrap(), 3);
     }
 
     #[test]
     fn is_empty() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: MultiMapTable<[u8], [u8]> = db.open_multimap_table(b"x").unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.is_empty().unwrap());
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
+
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(!read_txn.is_empty().unwrap());
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+        assert!(!table.is_empty().unwrap());
     }
 
     #[test]
     fn insert() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: MultiMapTable<[u8], [u8]> = db.open_multimap_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        write_txn.insert(b"hello", b"world2").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
+        table.insert(b"hello", b"world2").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
         assert_eq!(
             vec![b"world".to_vec(), b"world2".to_vec()],
-            get_vec(&read_txn, b"hello")
+            get_vec(&table, b"hello")
         );
-        assert_eq!(read_txn.len().unwrap(), 2);
+        assert_eq!(table.len().unwrap(), 2);
     }
 
     #[test]
     fn range_query() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: MultiMapTable<[u8], [u8]> = db.open_multimap_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
         for i in 0..5u8 {
             let value = vec![i];
-            write_txn.insert(b"0", &value).unwrap();
+            table.insert(b"0", &value).unwrap();
         }
         for i in 5..10u8 {
             let value = vec![i];
-            write_txn.insert(b"1", &value).unwrap();
+            table.insert(b"1", &value).unwrap();
         }
         for i in 10..15u8 {
             let value = vec![i];
-            write_txn.insert(b"2", &value).unwrap();
+            table.insert(b"2", &value).unwrap();
         }
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
         let start = b"0".as_ref();
         let end = b"1".as_ref();
-        let mut iter = read_txn.get_range(start..=end).unwrap();
+        let mut iter = table.get_range(start..=end).unwrap();
         for i in 0..10u8 {
             let (key, value) = iter.next().unwrap();
             if i < 5 {
@@ -587,34 +538,39 @@ mod test {
     fn delete() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: MultiMapTable<[u8], [u8]> = db.open_multimap_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        write_txn.insert(b"hello", b"world2").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
+        table.insert(b"hello", b"world2").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
         assert_eq!(
             vec![b"world".to_vec(), b"world2".to_vec()],
-            get_vec(&read_txn, b"hello")
+            get_vec(&table, b"hello")
         );
-        assert_eq!(read_txn.len().unwrap(), 2);
+        assert_eq!(table.len().unwrap(), 2);
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.remove(b"hello", b"world2").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+        table.remove(b"hello", b"world2").unwrap();
         write_txn.commit().unwrap();
 
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(vec![b"world".to_vec()], get_vec(&read_txn, b"hello"));
-        assert_eq!(read_txn.len().unwrap(), 1);
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+        assert_eq!(vec![b"world".to_vec()], get_vec(&table, b"hello"));
+        assert_eq!(table.len().unwrap(), 1);
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.remove_all(b"hello").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: MultiMapTable<[u8], [u8]> = write_txn.open_multimap_table(b"x").unwrap();
+        table.remove_all(b"hello").unwrap();
         write_txn.commit().unwrap();
 
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.is_empty().unwrap());
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyMultiMapTable<[u8], [u8]> = read_txn.open_multimap_table(b"x").unwrap();
+        assert!(table.is_empty().unwrap());
         let empty: Vec<Vec<u8>> = vec![];
-        assert_eq!(empty, get_vec(&read_txn, b"hello"));
+        assert_eq!(empty, get_vec(&table, b"hello"));
     }
 }
diff --git a/src/table.rs b/src/table.rs
index c65ab47..48463c1 100644
--- a/src/table.rs
+++ b/src/table.rs
@@ -1,36 +1,142 @@
 use crate::error::Error;
-use crate::transactions::WriteTransaction;
-use crate::tree_store::Storage;
+use crate::tree_store::{AccessGuardMut, BtreeRangeIter, NodeHandle, Storage};
 use crate::types::{RedbKey, RedbValue};
-use crate::ReadOnlyTransaction;
+use crate::AccessGuard;
+use std::cell::Cell;
 use std::marker::PhantomData;
+use std::ops::RangeBounds;
 
-pub struct Table<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
-    storage: &'mmap Storage,
+pub struct Table<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
+    storage: &'s Storage,
     table_id: u64,
+    transaction_id: u128,
+    root_page: &'t Cell<Option<NodeHandle>>,
     _key_type: PhantomData<K>,
     _value_type: PhantomData<V>,
 }
 
-impl<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Table<'mmap, K, V> {
+impl<'s, 't, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Table<'s, 't, K, V> {
     pub(in crate) fn new(
         table_id: u64,
-        storage: &'mmap Storage,
-    ) -> Result<Table<'mmap, K, V>, Error> {
-        Ok(Table {
+        transaction_id: u128,
+        root_page: &'t Cell<Option<NodeHandle>>,
+        storage: &'s Storage,
+    ) -> Table<'s, 't, K, V> {
+        Table {
             storage,
             table_id,
+            transaction_id,
+            root_page,
             _key_type: Default::default(),
             _value_type: Default::default(),
-        })
+        }
+    }
+
+    #[allow(dead_code)]
+    pub(in crate) fn print_debug(&self) {
+        if let Some(page) = self.root_page.get() {
+            self.storage.print_dirty_debug(page);
+        }
+    }
+
+    pub fn insert(&mut self, key: &K, value: &V) -> Result<(), Error> {
+        let page = Some(self.storage.insert::<K>(
+            self.table_id,
+            key.as_bytes().as_ref(),
+            value.as_bytes().as_ref(),
+            self.transaction_id,
+            self.root_page.get(),
+        )?);
+        self.root_page.set(page);
+        Ok(())
+    }
+
+    /// Reserve space to insert a key-value pair
+    /// The returned reference will have length equal to value_length
+    pub fn insert_reserve(
+        &mut self,
+        key: &K,
+        value_length: usize,
+    ) -> Result<AccessGuardMut, Error> {
+        let (root_page, guard) = self.storage.insert_reserve::<K>(
+            self.table_id,
+            key.as_bytes().as_ref(),
+            value_length,
+            self.transaction_id,
+            self.root_page.get(),
+        )?;
+        self.root_page.set(Some(root_page));
+        Ok(guard)
     }
 
-    pub fn begin_write(&'_ mut self) -> Result<WriteTransaction<'mmap, K, V>, Error> {
-        Ok(WriteTransaction::new(self.table_id, self.storage))
+    pub fn get(&self, key: &K) -> Result<Option<AccessGuard<V>>, Error> {
+        self.storage
+            .get::<K, V>(self.table_id, key.as_bytes().as_ref(), self.root_page.get())
     }
 
-    pub fn read_transaction(&'_ self) -> Result<ReadOnlyTransaction<'mmap, K, V>, Error> {
-        Ok(ReadOnlyTransaction::new(self.table_id, self.storage))
+    pub fn remove(&mut self, key: &K) -> Result<(), Error> {
+        let page = self.storage.remove::<K>(
+            self.table_id,
+            key.as_bytes().as_ref(),
+            self.transaction_id,
+            self.root_page.get(),
+        )?;
+        self.root_page.set(page);
+        Ok(())
+    }
+}
+
+pub struct ReadOnlyTable<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
+    storage: &'s Storage,
+    root_page: Option<NodeHandle>,
+    table_id: u64,
+    _key_type: PhantomData<K>,
+    _value_type: PhantomData<V>,
+}
+
+impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadOnlyTable<'s, K, V> {
+    pub(in crate) fn new(
+        table_id: u64,
+        root_page: Option<NodeHandle>,
+        storage: &'s Storage,
+    ) -> ReadOnlyTable<'s, K, V> {
+        ReadOnlyTable {
+            storage,
+            root_page,
+            table_id,
+            _key_type: Default::default(),
+            _value_type: Default::default(),
+        }
+    }
+
+    pub fn get(&self, key: &K) -> Result<Option<AccessGuard<'s, V>>, Error> {
+        self.storage
+            .get::<K, V>(self.table_id, key.as_bytes().as_ref(), self.root_page)
+    }
+
+    pub fn get_range<'a, T: RangeBounds<KR> + 'a, KR: AsRef<K>>(
+        &'a self,
+        range: T,
+    ) -> Result<BtreeRangeIter<T, KR, K, V>, Error> {
+        self.storage.get_range(self.table_id, range, self.root_page)
+    }
+
+    pub fn get_range_reversed<'a, T: RangeBounds<KR> + 'a, KR: AsRef<K>>(
+        &'a self,
+        range: T,
+    ) -> Result<BtreeRangeIter<T, KR, K, V>, Error> {
+        self.storage
+            .get_range_reversed(self.table_id, range, self.root_page)
+    }
+
+    pub fn len(&self) -> Result<usize, Error> {
+        self.storage.len(self.table_id, self.root_page)
+    }
+
+    pub fn is_empty(&self) -> Result<bool, Error> {
+        self.storage
+            .len(self.table_id, self.root_page)
+            .map(|x| x == 0)
     }
 }
 
@@ -40,7 +146,7 @@ mod test {
     use crate::types::{
         AsBytesWithLifetime, RedbKey, RedbValue, RefAsBytesLifetime, RefLifetime, WithLifetime,
     };
-    use crate::{Database, Table};
+    use crate::{Database, Error, ReadOnlyTable, Table};
     use std::cmp::Ordering;
     use tempfile::NamedTempFile;
 
@@ -48,145 +154,151 @@ mod test {
     fn len() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        write_txn.insert(b"hello2", b"world2").unwrap();
-        write_txn.insert(b"hi", b"world").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
+        table.insert(b"hello2", b"world2").unwrap();
+        table.insert(b"hi", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(read_txn.len().unwrap(), 3);
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(table.len().unwrap(), 3);
     }
 
     #[test]
     fn multiple_tables() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"1").unwrap();
-        let mut table2: Table<[u8], [u8]> = db.open_table(b"2").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"1").unwrap();
+        let mut table2: Table<[u8], [u8]> = write_txn.open_table(b"2").unwrap();
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
+        table.insert(b"hello", b"world").unwrap();
+        table2.insert(b"hello", b"world2").unwrap();
         write_txn.commit().unwrap();
-        let mut write_txn2 = table2.begin_write().unwrap();
-        write_txn2.insert(b"hello", b"world2").unwrap();
-        write_txn2.commit().unwrap();
-
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(read_txn.len().unwrap(), 1);
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
-        let read_txn2 = table2.read_transaction().unwrap();
-        assert_eq!(read_txn2.len().unwrap(), 1);
-        assert_eq!(
-            b"world2",
-            read_txn2.get(b"hello").unwrap().unwrap().as_ref()
-        );
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"1").unwrap();
+        let table2: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"2").unwrap();
+        assert_eq!(table.len().unwrap(), 1);
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+        assert_eq!(table2.len().unwrap(), 1);
+        assert_eq!(b"world2", table2.get(b"hello").unwrap().unwrap().as_ref());
     }
 
     #[test]
     fn is_empty() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.is_empty().unwrap());
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
+
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(!read_txn.is_empty().unwrap());
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert!(!table.is_empty().unwrap());
     }
 
     #[test]
     fn abort() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.is_empty().unwrap());
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"aborted").unwrap();
-        assert_eq!(
-            b"aborted",
-            write_txn.get(b"hello").unwrap().unwrap().as_ref()
-        );
+
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"aborted").unwrap();
+        assert_eq!(b"aborted", table.get(b"hello").unwrap().unwrap().as_ref());
         write_txn.abort().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.is_empty().unwrap());
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: Result<ReadOnlyTable<[u8], [u8]>, Error> = read_txn.open_table(b"x");
+        assert!(table.is_err());
+
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
-        assert_eq!(read_txn.len().unwrap(), 1);
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+        assert_eq!(table.len().unwrap(), 1);
     }
 
     #[test]
     fn insert_overwrite() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"replaced").unwrap();
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"replaced").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(
-            b"replaced",
-            read_txn.get(b"hello").unwrap().unwrap().as_ref()
-        );
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(b"replaced", table.get(b"hello").unwrap().unwrap().as_ref());
     }
 
     #[test]
     fn u64_type() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<u64, u64> = db.open_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(&0, &1).unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<u64, u64> = write_txn.open_table(b"x").unwrap();
+        table.insert(&0, &1).unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(1, read_txn.get(&0).unwrap().unwrap().to_value());
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<u64, u64> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(1, table.get(&0).unwrap().unwrap().to_value());
     }
 
     #[test]
     fn insert_reserve() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
         let value = b"world";
-        let mut reserved = write_txn.insert_reserve(b"hello", value.len()).unwrap();
+        let mut reserved = table.insert_reserve(b"hello", value.len()).unwrap();
         reserved.as_mut().copy_from_slice(value);
         drop(reserved);
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(value, read_txn.get(b"hello").unwrap().unwrap().as_ref());
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(value, table.get(b"hello").unwrap().unwrap().as_ref());
     }
 
     #[test]
     fn range_query() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
         for i in 0..10u8 {
             let key = vec![i];
-            write_txn.insert(&key, b"value").unwrap();
+            table.insert(&key, b"value").unwrap();
         }
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
         let start = vec![3u8];
         let end = vec![7u8];
-        let mut iter = read_txn
-            .get_range(start.as_slice()..end.as_slice())
-            .unwrap();
+        let mut iter = table.get_range(start.as_slice()..end.as_slice()).unwrap();
         for i in 3..7u8 {
             let entry = iter.next().unwrap();
             assert_eq!(&[i], entry.key());
@@ -199,18 +311,19 @@ mod test {
     fn range_query_reversed() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
         for i in 0..10u8 {
             let key = vec![i];
-            write_txn.insert(&key, b"value").unwrap();
+            table.insert(&key, b"value").unwrap();
         }
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
         let start = vec![3u8];
         let end = vec![7u8];
-        let mut iter = read_txn
+        let mut iter = table
             .get_range_reversed(start.as_slice()..end.as_slice())
             .unwrap();
         for i in (3..7u8).rev() {
@@ -252,18 +365,19 @@ mod test {
 
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<ReverseKey, [u8]> = db.open_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<ReverseKey, [u8]> = write_txn.open_table(b"x").unwrap();
         for i in 0..10u8 {
             let key = vec![i];
-            write_txn.insert(&ReverseKey(key), b"value").unwrap();
+            table.insert(&ReverseKey(key), b"value").unwrap();
         }
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
+
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<ReverseKey, [u8]> = read_txn.open_table(b"x").unwrap();
         let start = ReverseKey(vec![7u8]); // ReverseKey is used, so 7 < 3
         let end = ReverseKey(vec![3u8]);
-        let mut iter = read_txn.get_range(start..=end).unwrap();
+        let mut iter = table.get_range(start..=end).unwrap();
         for i in (3..=7u8).rev() {
             let entry = iter.next().unwrap();
             assert_eq!(&[i], entry.key());
@@ -276,76 +390,76 @@ mod test {
     fn delete() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        write_txn.insert(b"hello2", b"world").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
+        table.insert(b"hello2", b"world").unwrap();
         write_txn.commit().unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
-        assert_eq!(read_txn.len().unwrap(), 2);
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.remove(b"hello").unwrap();
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+        assert_eq!(table.len().unwrap(), 2);
+
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.remove(b"hello").unwrap();
         write_txn.commit().unwrap();
 
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.get(b"hello").unwrap().is_none());
-        assert_eq!(read_txn.len().unwrap(), 1);
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert!(table.get(b"hello").unwrap().is_none());
+        assert_eq!(table.len().unwrap(), 1);
     }
 
     #[test]
     fn no_dirty_reads() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        let read_txn = table.read_transaction().unwrap();
-        assert!(read_txn.get(b"hello").unwrap().is_none());
-        assert!(read_txn.is_empty().unwrap());
+        let read_txn = db.begin_read().unwrap();
+        let table: Result<ReadOnlyTable<[u8], [u8]>, Error> = read_txn.open_table(b"x");
+        assert!(table.is_err());
         write_txn.commit().unwrap();
 
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
     }
 
     #[test]
     fn read_isolation() {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
         let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        table.insert(b"hello", b"world").unwrap();
         write_txn.commit().unwrap();
 
-        let read_txn = table.read_transaction().unwrap();
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
+        let read_txn = db.begin_read().unwrap();
+        let table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
 
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.remove(b"hello").unwrap();
-        write_txn.insert(b"hello2", b"world2").unwrap();
-        write_txn.insert(b"hello3", b"world3").unwrap();
+        let write_txn = db.begin_write().unwrap();
+        let mut write_table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+        write_table.remove(b"hello").unwrap();
+        write_table.insert(b"hello2", b"world2").unwrap();
+        write_table.insert(b"hello3", b"world3").unwrap();
         write_txn.commit().unwrap();
 
-        let read_txn2 = table.read_transaction().unwrap();
-        assert!(read_txn2.get(b"hello").unwrap().is_none());
-        assert_eq!(
-            b"world2",
-            read_txn2.get(b"hello2").unwrap().unwrap().as_ref()
-        );
-        assert_eq!(
-            b"world3",
-            read_txn2.get(b"hello3").unwrap().unwrap().as_ref()
-        );
-        assert_eq!(read_txn2.len().unwrap(), 2);
-
-        assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
-        assert!(read_txn.get(b"hello2").unwrap().is_none());
-        assert!(read_txn.get(b"hello3").unwrap().is_none());
-        assert_eq!(read_txn.len().unwrap(), 1);
+        let read_txn2 = db.begin_read().unwrap();
+        let table2: ReadOnlyTable<[u8], [u8]> = read_txn2.open_table(b"x").unwrap();
+        assert!(table2.get(b"hello").unwrap().is_none());
+        assert_eq!(b"world2", table2.get(b"hello2").unwrap().unwrap().as_ref());
+        assert_eq!(b"world3", table2.get(b"hello3").unwrap().unwrap().as_ref());
+        assert_eq!(table2.len().unwrap(), 2);
+
+        assert_eq!(b"world", table.get(b"hello").unwrap().unwrap().as_ref());
+        assert!(table.get(b"hello2").unwrap().is_none());
+        assert!(table.get(b"hello3").unwrap().is_none());
+        assert_eq!(table.len().unwrap(), 1);
     }
 }
diff --git a/src/transactions.rs b/src/transactions.rs
deleted file mode 100644
index c64e21f..0000000
--- a/src/transactions.rs
+++ /dev/null
@@ -1,183 +0,0 @@
-use crate::error::Error;
-use crate::tree_store::{AccessGuard, AccessGuardMut, BtreeRangeIter, NodeHandle, Storage};
-use crate::types::{RedbKey, RedbValue};
-use std::marker::PhantomData;
-use std::ops::RangeBounds;
-
-pub struct WriteTransaction<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
-    storage: &'mmap Storage,
-    table_id: u64,
-    transaction_id: u128,
-    root_page: Option<NodeHandle>,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
-}
-
-impl<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> WriteTransaction<'mmap, K, V> {
-    pub(in crate) fn new(table_id: u64, storage: &'mmap Storage) -> WriteTransaction<'mmap, K, V> {
-        let transaction_id = storage.allocate_write_transaction();
-        WriteTransaction {
-            storage,
-            table_id,
-            transaction_id,
-            root_page: storage.get_root_page_number(),
-            _key_type: Default::default(),
-            _value_type: Default::default(),
-        }
-    }
-
-    #[allow(dead_code)]
-    pub(in crate) fn print_debug(&self) {
-        if let Some(page) = self.root_page {
-            self.storage.print_dirty_debug(page);
-        }
-    }
-
-    pub fn insert(&mut self, key: &K, value: &V) -> Result<(), Error> {
-        self.root_page = Some(self.storage.insert::<K>(
-            self.table_id,
-            key.as_bytes().as_ref(),
-            value.as_bytes().as_ref(),
-            self.transaction_id,
-            self.root_page,
-        )?);
-        Ok(())
-    }
-
-    /// Reserve space to insert a key-value pair
-    /// The returned reference will have length equal to value_length
-    pub fn insert_reserve(
-        &mut self,
-        key: &K,
-        value_length: usize,
-    ) -> Result<AccessGuardMut, Error> {
-        let (root_page, guard) = self.storage.insert_reserve::<K>(
-            self.table_id,
-            key.as_bytes().as_ref(),
-            value_length,
-            self.transaction_id,
-            self.root_page,
-        )?;
-        self.root_page = Some(root_page);
-        Ok(guard)
-    }
-
-    pub fn get(&self, key: &K) -> Result<Option<AccessGuard<V>>, Error> {
-        self.storage
-            .get::<K, V>(self.table_id, key.as_bytes().as_ref(), self.root_page)
-    }
-
-    pub fn remove(&mut self, key: &K) -> Result<(), Error> {
-        self.root_page = self.storage.remove::<K>(
-            self.table_id,
-            key.as_bytes().as_ref(),
-            self.transaction_id,
-            self.root_page,
-        )?;
-        Ok(())
-    }
-
-    pub fn commit(self) -> Result<(), Error> {
-        self.storage.commit(self.root_page, self.transaction_id)?;
-        Ok(())
-    }
-
-    /// Note: pages are only freed during commit(). So exclusively using this function may result
-    /// in an out-of-memory error
-    pub fn non_durable_commit(self) -> Result<(), Error> {
-        self.storage
-            .non_durable_commit(self.root_page, self.transaction_id)?;
-        Ok(())
-    }
-
-    pub fn abort(self) -> Result<(), Error> {
-        self.storage.rollback_uncommited_writes(self.transaction_id)
-    }
-}
-
-pub struct ReadOnlyTransaction<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
-    storage: &'mmap Storage,
-    root_page: Option<NodeHandle>,
-    table_id: u64,
-    transaction_id: u128,
-    _key_type: PhantomData<K>,
-    _value_type: PhantomData<V>,
-}
-
-impl<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadOnlyTransaction<'mmap, K, V> {
-    pub(in crate) fn new(
-        table_id: u64,
-        storage: &'mmap Storage,
-    ) -> ReadOnlyTransaction<'mmap, K, V> {
-        let root_page = storage.get_root_page_number();
-        let transaction_id = storage.allocate_read_transaction();
-        ReadOnlyTransaction {
-            storage,
-            root_page,
-            table_id,
-            transaction_id,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
-        }
-    }
-
-    pub fn get(&self, key: &K) -> Result<Option<AccessGuard<'mmap, V>>, Error> {
-        self.storage
-            .get::<K, V>(self.table_id, key.as_bytes().as_ref(), self.root_page)
-    }
-
-    pub fn get_range<'a, T: RangeBounds<KR> + 'a, KR: AsRef<K>>(
-        &'a self,
-        range: T,
-    ) -> Result<BtreeRangeIter<T, KR, K, V>, Error> {
-        self.storage.get_range(self.table_id, range, self.root_page)
-    }
-
-    pub fn get_range_reversed<'a, T: RangeBounds<KR> + 'a, KR: AsRef<K>>(
-        &'a self,
-        range: T,
-    ) -> Result<BtreeRangeIter<T, KR, K, V>, Error> {
-        self.storage
-            .get_range_reversed(self.table_id, range, self.root_page)
-    }
-
-    pub fn len(&self) -> Result<usize, Error> {
-        self.storage.len(self.table_id, self.root_page)
-    }
-
-    pub fn is_empty(&self) -> Result<bool, Error> {
-        self.storage
-            .len(self.table_id, self.root_page)
-            .map(|x| x == 0)
-    }
-}
-
-impl<'mmap, K: RedbKey + ?Sized, V: RedbValue + ?Sized> Drop for ReadOnlyTransaction<'mmap, K, V> {
-    fn drop(&mut self) {
-        self.storage
-            .deallocate_read_transaction(self.transaction_id);
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use crate::{Database, Table};
-    use tempfile::NamedTempFile;
-
-    #[test]
-    fn transaction_id_persistence() {
-        let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
-        let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
-        let mut write_txn = table.begin_write().unwrap();
-        write_txn.insert(b"hello", b"world").unwrap();
-        let first_txn_id = write_txn.transaction_id;
-        write_txn.commit().unwrap();
-        drop(db);
-
-        let db2 = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db2.open_table(b"x").unwrap();
-        let write_txn = table.begin_write().unwrap();
-        assert!(write_txn.transaction_id > first_txn_id);
-    }
-}
diff --git a/src/tree_store/btree_utils.rs b/src/tree_store/btree_utils.rs
index 52250b9..993c678 100644
--- a/src/tree_store/btree_utils.rs
+++ b/src/tree_store/btree_utils.rs
@@ -1998,13 +1998,13 @@ mod test {
         let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
 
         let db = unsafe { Database::open(tmpfile.path(), 16 * 1024 * 1024).unwrap() };
-        let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+        let txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
         let elements = (BTREE_ORDER / 2).pow(2) as usize;
 
-        let mut txn = table.begin_write().unwrap();
         for i in (0..elements).rev() {
-            txn.insert(&i.to_be_bytes(), b"").unwrap();
+            table.insert(&i.to_be_bytes(), b"").unwrap();
         }
         txn.commit().unwrap();
 
@@ -2019,9 +2019,10 @@ mod test {
 
         let reduce_to = BTREE_ORDER / 2;
 
-        let mut txn = table.begin_write().unwrap();
+        let txn = db.begin_write().unwrap();
+        let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
         for i in 0..(elements - reduce_to) {
-            txn.remove(&i.to_be_bytes()).unwrap();
+            table.remove(&i.to_be_bytes()).unwrap();
         }
         txn.commit().unwrap();
 
diff --git a/src/tree_store/page_store/page_manager.rs b/src/tree_store/page_store/page_manager.rs
index 94fdf44..c00d05f 100644
--- a/src/tree_store/page_store/page_manager.rs
+++ b/src/tree_store/page_store/page_manager.rs
@@ -17,7 +17,7 @@ const MAGICNUMBER: [u8; 4] = [b'r', b'e', b'd', b'b'];
 const VERSION_OFFSET: usize = MAGICNUMBER.len();
 const PAGE_SIZE_OFFSET: usize = VERSION_OFFSET + 1;
 const DB_SIZE_OFFSET: usize = PAGE_SIZE_OFFSET + size_of::<u64>();
-const PRIMARY_BIT_OFFSET: usize = DB_SIZE_OFFSET + 1;
+const PRIMARY_BIT_OFFSET: usize = DB_SIZE_OFFSET + size_of::<u64>();
 const TRANSACTION_SIZE: usize = 128;
 const TRANSACTION_0_OFFSET: usize = 128;
 const TRANSACTION_1_OFFSET: usize = TRANSACTION_0_OFFSET + TRANSACTION_SIZE;
@@ -299,7 +299,7 @@ pub(crate) struct PageMut<'a> {
 
 impl<'a> PageMut<'a> {
     pub(crate) fn memory_mut(&mut self) -> &mut [u8] {
-        &mut self.mem
+        self.mem
     }
 }
 
@@ -474,7 +474,7 @@ impl TransactionalMemory {
     ) -> Result<(&mut [u8], MutexGuard<'a, MetapageGuard>), Error> {
         // The allocator is a cache, and therefore can only be modified when it's marked dirty
         if !mutator.get_allocator_dirty() {
-            mutator.set_allocator_dirty(false);
+            mutator.set_allocator_dirty(true);
             self.mmap.flush()?
         }
 
diff --git a/src/tree_store/storage.rs b/src/tree_store/storage.rs
index 82a3538..8dbb1f1 100644
--- a/src/tree_store/storage.rs
+++ b/src/tree_store/storage.rs
@@ -146,6 +146,27 @@ impl Storage {
         self.live_read_transactions.borrow_mut().remove(&id);
     }
 
+    pub(in crate) fn get_table(
+        &self,
+        name: &[u8],
+        table_type: TableType,
+        root_page: Option<NodeHandle>,
+    ) -> Result<Option<TableDefinition>, Error> {
+        if let Some(found) = self.get::<[u8], [u8]>(TABLE_TABLE_ID, name, root_page)? {
+            let definition = TableDefinition::from_bytes(found.as_ref());
+            if definition.get_type() != table_type {
+                return Err(Error::TableTypeMismatch(format!(
+                    "{:?} is not of type {:?}",
+                    name, table_type
+                )));
+            }
+
+            Ok(Some(definition))
+        } else {
+            Ok(None)
+        }
+    }
+
     // Returns a tuple of the table id and the new root page
     pub(in crate) fn get_or_create_table(
         &self,


Test patch:
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index 8f9cde6..faf1335 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -2,7 +2,7 @@ use tempfile::NamedTempFile;
 
 use rand::prelude::SliceRandom;
 use rand::Rng;
-use redb::{Database, Table};
+use redb::{Database, ReadOnlyTable, Table};
 
 const ELEMENTS: usize = 100;
 
@@ -25,33 +25,32 @@ fn non_durable_commit_persistence() {
 
     let db_size = 16 * 1024 * 1024;
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
     let pairs = gen_data(100, 16, 20);
 
-    let mut txn = table.begin_write().unwrap();
     {
         for i in 0..ELEMENTS {
             let (key, value) = &pairs[i % pairs.len()];
-            txn.insert(key, value).unwrap();
+            table.insert(key, value).unwrap();
         }
     }
     txn.non_durable_commit().unwrap();
 
-    drop(table);
     // Check that cleanly closing the database persists the non-durable commit
     drop(db);
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
     let mut key_order: Vec<usize> = (0..ELEMENTS).collect();
     key_order.shuffle(&mut rand::thread_rng());
 
-    let txn = table.read_transaction().unwrap();
     {
         for i in &key_order {
             let (key, value) = &pairs[*i % pairs.len()];
-            let result = &txn.get(key).unwrap().unwrap();
+            let result = &table.get(key).unwrap().unwrap();
             assert_eq!(result.as_ref(), value);
         }
     }
@@ -63,32 +62,31 @@ fn persistence() {
 
     let db_size = 16 * 1024 * 1024;
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
     let pairs = gen_data(100, 16, 20);
 
-    let mut txn = table.begin_write().unwrap();
     {
         for i in 0..ELEMENTS {
             let (key, value) = &pairs[i % pairs.len()];
-            txn.insert(key, value).unwrap();
+            table.insert(key, value).unwrap();
         }
     }
     txn.commit().unwrap();
 
-    drop(table);
     drop(db);
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
     let mut key_order: Vec<usize> = (0..ELEMENTS).collect();
     key_order.shuffle(&mut rand::thread_rng());
 
-    let txn = table.read_transaction().unwrap();
     {
         for i in &key_order {
             let (key, value) = &pairs[*i % pairs.len()];
-            let result = &txn.get(key).unwrap().unwrap();
+            let result = &table.get(key).unwrap().unwrap();
             assert_eq!(result.as_ref(), value);
         }
     }
@@ -100,7 +98,8 @@ fn free() {
 
     let db_size = 512 * 1024;
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
     let free_pages = db.stats().unwrap().free_pages();
 
@@ -111,18 +110,18 @@ fn free() {
     // Make sure an internal index page is required
     assert!(num_writes > 64);
 
-    let mut txn = table.begin_write().unwrap();
     {
         for _ in 0..num_writes {
-            txn.insert(&key, &value).unwrap();
+            table.insert(&key, &value).unwrap();
         }
     }
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
     {
         for _ in 0..num_writes {
-            txn.remove(&key).unwrap();
+            table.remove(&key).unwrap();
         }
     }
     txn.commit().unwrap();
@@ -136,24 +135,25 @@ fn large_keys() {
 
     let db_size = 1024_1024;
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
 
     let mut key = vec![0; 1024];
     let value = vec![0; 1];
-    let mut txn = table.begin_write().unwrap();
     {
         for i in 0..100 {
             key[0] = i;
-            txn.insert(&key, &value).unwrap();
+            table.insert(&key, &value).unwrap();
         }
     }
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = txn.open_table(b"x").unwrap();
     {
         for i in 0..100 {
             key[0] = i;
-            txn.remove(&key).unwrap();
+            table.remove(&key).unwrap();
         }
     }
     txn.commit().unwrap();
@@ -166,38 +166,45 @@ fn regression() {
 
     let db_size = 1024 * 1024;
     let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
-    let mut table: Table<u64, u64> = db.open_table(b"x").unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.insert(&1, &1).unwrap();
+    table.insert(&1, &1).unwrap();
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.insert(&6, &9).unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
+    table.insert(&6, &9).unwrap();
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.insert(&12, &10).unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
+    table.insert(&12, &10).unwrap();
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.insert(&18, &27).unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
+    table.insert(&18, &27).unwrap();
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.insert(&24, &33).unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
+    table.insert(&24, &33).unwrap();
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.insert(&30, &14).unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
+    table.insert(&30, &14).unwrap();
     txn.commit().unwrap();
 
-    let mut txn = table.begin_write().unwrap();
-    txn.remove(&30).unwrap();
+    let txn = db.begin_write().unwrap();
+    let mut table: Table<u64, u64> = txn.open_table(b"x").unwrap();
+    table.remove(&30).unwrap();
     txn.commit().unwrap();
 
-    let txn = table.read_transaction().unwrap();
-    let v = txn.get(&6).unwrap().unwrap().to_value();
+    let txn = db.begin_read().unwrap();
+    let table: ReadOnlyTable<u64, u64> = txn.open_table(b"x").unwrap();
+    let v = table.get(&6).unwrap().unwrap().to_value();
     assert_eq!(v, 9);
 }
 
@@ -205,35 +212,44 @@ fn regression() {
 fn non_durable_read_isolation() {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
     let db = unsafe { Database::open(tmpfile.path(), 1024 * 1024).unwrap() };
-    let mut table: Table<[u8], [u8]> = db.open_table(b"x").unwrap();
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
 
-    let mut write_txn = table.begin_write().unwrap();
-    write_txn.insert(b"hello", b"world").unwrap();
+    table.insert(b"hello", b"world").unwrap();
     write_txn.non_durable_commit().unwrap();
 
-    let read_txn = table.read_transaction().unwrap();
-    assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
+    let read_txn = db.begin_read().unwrap();
+    let read_table: ReadOnlyTable<[u8], [u8]> = read_txn.open_table(b"x").unwrap();
+    assert_eq!(
+        b"world",
+        read_table.get(b"hello").unwrap().unwrap().as_ref()
+    );
 
-    let mut write_txn = table.begin_write().unwrap();
-    write_txn.remove(b"hello").unwrap();
-    write_txn.insert(b"hello2", b"world2").unwrap();
-    write_txn.insert(b"hello3", b"world3").unwrap();
+    let write_txn = db.begin_write().unwrap();
+    let mut table: Table<[u8], [u8]> = write_txn.open_table(b"x").unwrap();
+    table.remove(b"hello").unwrap();
+    table.insert(b"hello2", b"world2").unwrap();
+    table.insert(b"hello3", b"world3").unwrap();
     write_txn.non_durable_commit().unwrap();
 
-    let read_txn2 = table.read_transaction().unwrap();
-    assert!(read_txn2.get(b"hello").unwrap().is_none());
+    let read_txn2 = db.begin_read().unwrap();
+    let read_table2: ReadOnlyTable<[u8], [u8]> = read_txn2.open_table(b"x").unwrap();
+    assert!(read_table2.get(b"hello").unwrap().is_none());
     assert_eq!(
         b"world2",
-        read_txn2.get(b"hello2").unwrap().unwrap().as_ref()
+        read_table2.get(b"hello2").unwrap().unwrap().as_ref()
     );
     assert_eq!(
         b"world3",
-        read_txn2.get(b"hello3").unwrap().unwrap().as_ref()
+        read_table2.get(b"hello3").unwrap().unwrap().as_ref()
     );
-    assert_eq!(read_txn2.len().unwrap(), 2);
+    assert_eq!(read_table2.len().unwrap(), 2);
 
-    assert_eq!(b"world", read_txn.get(b"hello").unwrap().unwrap().as_ref());
-    assert!(read_txn.get(b"hello2").unwrap().is_none());
-    assert!(read_txn.get(b"hello3").unwrap().is_none());
-    assert_eq!(read_txn.len().unwrap(), 1);
+    assert_eq!(
+        b"world",
+        read_table.get(b"hello").unwrap().unwrap().as_ref()
+    );
+    assert!(read_table.get(b"hello2").unwrap().is_none());
+    assert!(read_table.get(b"hello3").unwrap().is_none());
+    assert_eq!(read_table.len().unwrap(), 1);
 }