Repository: cberner/redb
Pull request #343 (2022-09-01 04:49:26)

File format improvements


Patch:
diff --git a/src/tree_store/page_store/base.rs b/src/tree_store/page_store/base.rs
index fed989d..58d82d1 100644
--- a/src/tree_store/page_store/base.rs
+++ b/src/tree_store/page_store/base.rs
@@ -1,4 +1,5 @@
 use crate::tree_store::page_store::page_manager::MAX_MAX_PAGE_ORDER;
+#[cfg(debug_assertions)]
 use std::collections::HashMap;
 #[cfg(debug_assertions)]
 use std::collections::HashSet;
diff --git a/src/tree_store/page_store/buddy_allocator.rs b/src/tree_store/page_store/buddy_allocator.rs
index 4916338..cf6c257 100644
--- a/src/tree_store/page_store/buddy_allocator.rs
+++ b/src/tree_store/page_store/buddy_allocator.rs
@@ -15,8 +15,9 @@ pub(crate) struct BuddyAllocator {
 // Pages are marked free at only a single order, and it must always be the largest order
 //
 // Data structure format:
-// max_order: u64
-// order_offsets: array of (u64, u64), with offset & length for PageAllocator structure for the given order
+// max_order: u8
+// padding: 3 bytes
+// order_ends: array of u32, with ending offset for PageAllocator structure for the given order
 // ... PageAllocator structures
 impl BuddyAllocator {
     pub(crate) fn new(num_pages: usize, max_capacity: usize, max_order: usize) -> Self {
@@ -41,26 +42,24 @@ impl BuddyAllocator {
         max_order: usize,
     ) -> Self {
         assert!(data.len() >= Self::required_space(max_page_capacity, max_order));
-        data[..size_of::<u64>()].copy_from_slice(&(max_order as u64).to_le_bytes());
+        data[0] = max_order.try_into().unwrap();
 
-        let mut metadata_offset = size_of::<u64>();
-        let mut data_offset = size_of::<u64>() + 2 * (max_order + 1) * size_of::<u64>();
+        let mut metadata_offset = 4;
+        let mut data_offset = 4 + (max_order + 1) * size_of::<u32>();
 
         let mut orders = vec![];
         let mut pages_for_order = max_page_capacity;
         for order in 0..=max_order {
             let required = PageAllocator::required_space(pages_for_order);
-            data[metadata_offset..metadata_offset + size_of::<u64>()]
-                .copy_from_slice(&(data_offset as u64).to_le_bytes());
-            data[metadata_offset + size_of::<u64>()..metadata_offset + 2 * size_of::<u64>()]
-                .copy_from_slice(&(required as u64).to_le_bytes());
+            data_offset += required;
+            data[metadata_offset..metadata_offset + size_of::<u32>()]
+                .copy_from_slice(&(data_offset as u32).to_le_bytes());
             orders.push(PageAllocator::init_new(
                 Self::get_order_bytes_mut(data, order),
                 pages_for_order,
             ));
             pages_for_order = Self::next_higher_order(pages_for_order as u64) as usize;
-            metadata_offset += 2 * size_of::<u64>();
-            data_offset += required;
+            metadata_offset += size_of::<u32>();
         }
 
         // Mark the available pages, starting with the highest order
@@ -179,7 +178,7 @@ impl BuddyAllocator {
 
     /// Returns the number of bytes required for the data argument of new()
     pub(crate) fn required_space(mut capacity: usize, max_order: usize) -> usize {
-        let mut required = size_of::<u64>() + 2 * (max_order + 1) * size_of::<u64>();
+        let mut required = 4 + (max_order + 1) * size_of::<u32>();
         for _ in 0..=max_order {
             required += PageAllocator::required_space(capacity);
             capacity = Self::next_higher_order(capacity as u64) as usize;
@@ -188,6 +187,17 @@ impl BuddyAllocator {
         required
     }
 
+    pub(crate) fn highest_free_order(&self, data: &[u8]) -> Option<usize> {
+        for (order, allocator) in self.orders.iter().enumerate().rev() {
+            // TODO: optimize this. We don't need to count all the free pages, just determine whether there is at least one
+            if allocator.count_free_pages(Self::get_order_bytes(data, order)) > 0 {
+                return Some(order);
+            }
+        }
+
+        None
+    }
+
     pub(crate) fn count_free_pages(&self, data: &[u8]) -> usize {
         let mut pages = 0;
         for (order, allocator) in self.orders.iter().enumerate() {
@@ -228,19 +238,28 @@ impl BuddyAllocator {
         free_pages
     }
 
+    fn get_order_start_offset(data: &[u8], order: usize) -> usize {
+        if order == 0 {
+            let max_order = data[0] as usize;
+            4 + (max_order + 1) * size_of::<u32>()
+        } else {
+            Self::get_order_end_offset(data, order - 1)
+        }
+    }
+
+    fn get_order_end_offset(data: &[u8], order: usize) -> usize {
+        let base = 4 + order * size_of::<u32>();
+        u32::from_le_bytes(data[base..base + size_of::<u32>()].try_into().unwrap()) as usize
+    }
+
     fn get_order_offset_and_length(data: &[u8], order: usize) -> (usize, usize) {
-        let max_order = u64::from_le_bytes(data[..size_of::<u64>()].try_into().unwrap()) as usize;
+        let max_order = data[0] as usize;
         assert!(order <= max_order);
-        let base = size_of::<u64>() + order * 2 * size_of::<u64>();
-        let offset =
-            u64::from_le_bytes(data[base..base + size_of::<u64>()].try_into().unwrap()) as usize;
-        let length = u64::from_le_bytes(
-            data[base + size_of::<u64>()..base + 2 * size_of::<u64>()]
-                .try_into()
-                .unwrap(),
-        ) as usize;
+        let offset = Self::get_order_start_offset(data, order);
+        let end = Self::get_order_end_offset(data, order);
+        let length = end - offset;
         // Data starts with max_order
-        assert!(offset > size_of::<u64>());
+        assert!(offset > size_of::<u32>());
 
         (offset, length)
     }
diff --git a/src/tree_store/page_store/grouped_bitmap.rs b/src/tree_store/page_store/grouped_bitmap.rs
index c89590e..0245fca 100644
--- a/src/tree_store/page_store/grouped_bitmap.rs
+++ b/src/tree_store/page_store/grouped_bitmap.rs
@@ -20,6 +20,19 @@ impl<'a> U64GroupedBitMap<'a> {
         self.data.iter().map(|x| x.count_ones() as usize).sum()
     }
 
+    pub(crate) fn first_unset(&self, start_bit: usize, end_bit: usize) -> Option<usize> {
+        assert_eq!(end_bit, (start_bit - start_bit % 64) + 64);
+
+        let (index, bit) = self.data_index_of(start_bit);
+        let mask = !((1 << bit) - 1);
+        let group = u64::from_le_bytes(self.data[index..(index + 8)].try_into().unwrap());
+        let group = group & mask;
+        match group.trailing_zeros() {
+            64 => None,
+            x => Some(start_bit + x as usize - bit),
+        }
+    }
+
     pub(crate) fn get(&self, bit: usize) -> bool {
         let (index, bit_index) = self.data_index_of(bit);
         let group = u64::from_le_bytes(self.data[index..(index + 8)].try_into().unwrap());
@@ -66,12 +79,6 @@ impl<'a> U64GroupedBitMapMut<'a> {
         group == 0
     }
 
-    pub(crate) fn get(&self, bit: usize) -> bool {
-        let (index, bit_index) = self.data_index_of(bit);
-        let group = u64::from_le_bytes(self.data[index..(index + 8)].try_into().unwrap());
-        group & Self::select_mask(bit_index) == 0
-    }
-
     pub(crate) fn clear(&mut self, bit: usize) {
         let (index, bit_index) = self.data_index_of(bit);
         let mut group = u64::from_le_bytes(self.data[index..(index + 8)].try_into().unwrap());
@@ -79,18 +86,6 @@ impl<'a> U64GroupedBitMapMut<'a> {
         self.data[index..(index + 8)].copy_from_slice(&group.to_le_bytes());
     }
 
-    pub(crate) fn first_unset(&self, start_bit: usize, end_bit: usize) -> Option<usize> {
-        assert_eq!(start_bit % 64, 0);
-        assert_eq!(end_bit, start_bit + 64);
-
-        let (index, _) = self.data_index_of(start_bit);
-        let group = u64::from_le_bytes(self.data[index..(index + 8)].try_into().unwrap());
-        match group.trailing_zeros() {
-            64 => None,
-            x => Some(start_bit + x as usize),
-        }
-    }
-
     fn data_index_of(&self, bit: usize) -> (usize, usize) {
         ((bit / 64) as usize * size_of::<u64>(), (bit % 64) as usize)
     }
diff --git a/src/tree_store/page_store/layout.rs b/src/tree_store/page_store/layout.rs
index 2a87d64..0fee180 100644
--- a/src/tree_store/page_store/layout.rs
+++ b/src/tree_store/page_store/layout.rs
@@ -1,7 +1,6 @@
 use crate::tree_store::page_store::buddy_allocator::BuddyAllocator;
-use crate::tree_store::page_store::grouped_bitmap::U64GroupedBitMapMut;
 use crate::tree_store::page_store::page_manager::{
-    DB_HEADER_SIZE, MAX_MAX_PAGE_ORDER, MIN_USABLE_PAGES,
+    RegionAllocator, DB_HEADER_SIZE, MAX_MAX_PAGE_ORDER, MIN_USABLE_PAGES,
 };
 use crate::{Error, Result};
 use std::cmp::min;
@@ -22,7 +21,6 @@ pub(super) struct RegionLayout {
     num_pages: usize,
     // Offset where pages start
     pages_start: usize,
-    allocator_state_len: usize,
     max_order: usize,
     page_size: usize,
 }
@@ -31,14 +29,12 @@ impl RegionLayout {
     pub(super) fn new(
         num_pages: usize,
         header_size: usize,
-        allocator_state_len: usize,
         max_order: usize,
         page_size: usize,
     ) -> Self {
         Self {
             num_pages,
             pages_start: header_size,
-            allocator_state_len,
             max_order,
             page_size,
         }
@@ -104,7 +100,6 @@ impl RegionLayout {
         Some(RegionLayout {
             num_pages,
             pages_start: required_header_size,
-            allocator_state_len: Self::header_size(max_usable_region_bytes, page_size)?,
             max_order,
             page_size,
         })
@@ -123,10 +118,6 @@ impl RegionLayout {
         .unwrap()
     }
 
-    pub(super) fn header_len(&self) -> usize {
-        self.allocator_state_len
-    }
-
     pub(super) fn data_section(&self) -> Range<usize> {
         self.pages_start..(self.pages_start + self.usable_bytes())
     }
@@ -183,24 +174,31 @@ impl DatabaseLayout {
         desired_usable_bytes = min(desired_usable_bytes, db_capacity);
         let full_region_layout =
             RegionLayout::full_region_layout(max_usable_region_bytes, page_size);
-        let min_header_size = DB_HEADER_SIZE + U64GroupedBitMapMut::required_bytes(1);
-        let region_allocator_range = DB_HEADER_SIZE..min_header_size;
+        let min_header_size =
+            DB_HEADER_SIZE + RegionAllocator::required_bytes(1, MAX_MAX_PAGE_ORDER + 1);
+        let max_regions = (db_capacity - min_header_size + full_region_layout.len() - 1)
+            / full_region_layout.len();
+        let db_header_bytes =
+            DB_HEADER_SIZE + RegionAllocator::required_bytes(max_regions, MAX_MAX_PAGE_ORDER + 1);
+        let region_allocator_range = DB_HEADER_SIZE..db_header_bytes;
         // Pad to be page aligned
-        let min_header_size = round_up_to_multiple_of(min_header_size, page_size);
-        if db_capacity < min_header_size + MIN_USABLE_PAGES * page_size {
+        let super_header_size = round_up_to_multiple_of(db_header_bytes, page_size);
+        if db_capacity < super_header_size + MIN_USABLE_PAGES * page_size {
             return Err(Error::OutOfSpace);
         }
-        let result = if desired_usable_bytes <= full_region_layout.usable_bytes() {
+        let result = if desired_usable_bytes <= full_region_layout.usable_bytes()
+            || db_capacity - super_header_size <= full_region_layout.len()
+        {
             // Single region layout
             let region_layout = RegionLayout::calculate(
-                db_capacity - min_header_size,
+                db_capacity - super_header_size,
                 desired_usable_bytes,
                 max_usable_region_bytes,
                 page_size,
             )
             .ok_or(Error::OutOfSpace)?;
             DatabaseLayout {
-                db_header_bytes: min_header_size,
+                db_header_bytes: super_header_size,
                 region_allocator_range,
                 full_region_layout,
                 num_full_regions: 0,
@@ -208,17 +206,11 @@ impl DatabaseLayout {
             }
         } else {
             // Multi region layout
-            let max_regions = (db_capacity - min_header_size + full_region_layout.len() - 1)
-                / full_region_layout.len();
-            let db_header_bytes = DB_HEADER_SIZE + U64GroupedBitMapMut::required_bytes(max_regions);
-            let region_allocator_range = DB_HEADER_SIZE..db_header_bytes;
-            // Pad to be page aligned
-            let db_header_bytes = round_up_to_multiple_of(db_header_bytes, page_size);
-            let max_full_regions = (db_capacity - db_header_bytes) / full_region_layout.len();
+            let max_full_regions = (db_capacity - super_header_size) / full_region_layout.len();
             let desired_full_regions = desired_usable_bytes / max_usable_region_bytes;
             let num_full_regions = min(max_full_regions, desired_full_regions);
             let remaining_space =
-                db_capacity - db_header_bytes - num_full_regions * full_region_layout.len();
+                db_capacity - super_header_size - num_full_regions * full_region_layout.len();
             let remaining_desired =
                 desired_usable_bytes - num_full_regions * max_usable_region_bytes;
             assert!(num_full_regions > 0);
@@ -230,9 +222,7 @@ impl DatabaseLayout {
             );
             // TODO: change the calculation to use a fixed header size for all regions, including the trailing one
             let trailing_region = if let Some(region) = trailing_region {
-                if region.pages_start != full_region_layout.pages_start
-                    || region.allocator_state_len != full_region_layout.allocator_state_len
-                {
+                if region.pages_start != full_region_layout.pages_start {
                     None
                 } else {
                     Some(region)
@@ -241,7 +231,7 @@ impl DatabaseLayout {
                 None
             };
             DatabaseLayout {
-                db_header_bytes,
+                db_header_bytes: super_header_size,
                 region_allocator_range,
                 full_region_layout,
                 num_full_regions,
diff --git a/src/tree_store/page_store/mod.rs b/src/tree_store/page_store/mod.rs
index e8a6869..353ac46 100644
--- a/src/tree_store/page_store/mod.rs
+++ b/src/tree_store/page_store/mod.rs
@@ -5,6 +5,7 @@ mod layout;
 mod mmap;
 mod page_allocator;
 mod page_manager;
+mod region;
 mod utils;
 #[allow(dead_code)]
 mod xxh3;
diff --git a/src/tree_store/page_store/page_allocator.rs b/src/tree_store/page_store/page_allocator.rs
index ade90a6..20bf11a 100644
--- a/src/tree_store/page_store/page_allocator.rs
+++ b/src/tree_store/page_store/page_allocator.rs
@@ -3,8 +3,14 @@ use crate::Error;
 use crate::Result;
 use std::mem::size_of;
 
+const ELEMENTS_OFFSET: usize = 0;
+const CAPACITY_OFFSET: usize = ELEMENTS_OFFSET + size_of::<u32>();
+const HEIGHT_OFFSET: usize = CAPACITY_OFFSET + size_of::<u32>();
+const END_OFFSETS: usize = HEIGHT_OFFSET + size_of::<u32>();
+
 #[derive(Clone)]
 pub(crate) struct PageAllocator {
+    // TODO: we can read these fields from the on-disk format, so remove them
     num_pages: usize,
     tree_level_offsets: Vec<(usize, usize)>,
 }
@@ -14,14 +20,20 @@ pub(crate) struct PageAllocator {
 // borrowing the data array
 //
 // Data structure format:
+// height: u32
+// TODO: maybe remove the elements & capacity fields?
+// elements: u32
+// capacity: u32
+// layer_ends: array of u32, ending offset in bytes of layers. Does not include the root layer
 // root: u64
 // subtree layer: 2-64 u64s
 // ...consecutive layers. Except for the last level, all sub-trees of the root must be complete
 impl PageAllocator {
+    // TODO: this should take capacity as well as num pages
     pub(crate) fn new(num_pages: usize) -> Self {
         let mut tree_level_offsets = vec![];
 
-        let mut offset = 0;
+        let mut offset = Self::tree_data_start(num_pages);
         // root level
         tree_level_offsets.push((offset, offset + size_of::<u64>()));
         offset += size_of::<u64>();
@@ -56,11 +68,25 @@ impl PageAllocator {
 
     pub(crate) fn init_new(data: &mut [u8], num_pages: usize) -> Self {
         assert!(data.len() >= Self::required_space(num_pages));
+        // TODO: elements and capacity should be separate, and there should be a resize() method to change elements
+        data[ELEMENTS_OFFSET..(ELEMENTS_OFFSET + size_of::<u32>())]
+            .copy_from_slice(&(num_pages as u32).to_le_bytes());
+        data[CAPACITY_OFFSET..(CAPACITY_OFFSET + size_of::<u32>())]
+            .copy_from_slice(&(num_pages as u32).to_le_bytes());
+        let height = Self::required_tree_height(num_pages);
+        data[HEIGHT_OFFSET..(HEIGHT_OFFSET + size_of::<u32>())]
+            .copy_from_slice(&(height as u32).to_le_bytes());
         // Initialize the memory, so that all pages are allocated
-        U64GroupedBitMapMut::init_full(data);
+        U64GroupedBitMapMut::init_full(&mut data[Self::tree_data_start(num_pages)..]);
 
         let result = Self::new(num_pages);
 
+        let mut index = END_OFFSETS;
+        for (_, end) in result.tree_level_offsets.iter().skip(1) {
+            data[index..(index + size_of::<u32>())].copy_from_slice(&(*end as u32).to_le_bytes());
+            index += size_of::<u32>();
+        }
+
         // Mark all the subtrees that don't exist
         for i in Self::required_subtrees(num_pages)..64 {
             result.get_level_mut(data, 0).set(i);
@@ -88,7 +114,7 @@ impl PageAllocator {
 
     /// Returns the number of bytes required for the data argument of new()
     pub(crate) fn required_space(num_pages: usize) -> usize {
-        if Self::required_tree_height(num_pages) == 1 {
+        let tree_space = if Self::required_tree_height(num_pages) == 1 {
             assert!(num_pages <= 64);
             // Space for root
             size_of::<u64>()
@@ -96,15 +122,21 @@ impl PageAllocator {
             // Space for root
             size_of::<u64>() +
                 // Space for the leaves
-                (num_pages + 63) / 64 * size_of::<u64>()
+                U64GroupedBitMapMut::required_bytes(num_pages)
         } else {
             // Space for root
             size_of::<u64>() +
                 // Space for the subtrees
                 Self::required_subtrees(num_pages) * Self::required_interior_bytes_per_subtree(num_pages) +
                 // Space for the leaves
-                (num_pages + 63) / 64 * size_of::<u64>()
-        }
+                U64GroupedBitMapMut::required_bytes(num_pages)
+        };
+
+        Self::tree_data_start(num_pages) + tree_space
+    }
+
+    fn tree_data_start(capacity: usize) -> usize {
+        END_OFFSETS + (Self::required_tree_height(capacity) - 1) * size_of::<u32>()
     }
 
     fn required_interior_bytes_per_subtree(num_pages: usize) -> usize {
@@ -187,20 +219,26 @@ impl PageAllocator {
 
     /// data must have been initialized by Self::init_new()
     pub(crate) fn alloc(&self, data: &mut [u8]) -> Result<u64> {
-        if let Some(mut entry) = self.get_level_mut(data, 0).first_unset(0, 64) {
+        let entry = self.find_free(data)?;
+        self.record_alloc(data, entry as u64);
+        Ok(entry)
+    }
+
+    /// data must have been initialized by Self::init_new(). Returns the first free id, after (inclusive) of start
+    pub(crate) fn find_free(&self, data: &[u8]) -> Result<u64> {
+        if let Some(mut entry) = self.get_level(data, 0).first_unset(0, 64) {
             let mut height = 0;
 
             while height < self.get_height() - 1 {
                 height += 1;
                 entry *= 64;
                 entry = self
-                    .get_level_mut(data, height)
+                    .get_level(data, height)
                     .first_unset(entry, entry + 64)
                     .unwrap();
             }
 
             assert!(entry < self.get_num_pages() as usize);
-            self.record_alloc(data, entry as u64);
             Ok(entry as u64)
         } else {
             Err(Error::OutOfSpace)
@@ -319,6 +357,25 @@ mod test {
         );
     }
 
+    #[test]
+    fn find_free() {
+        let num_pages = 129;
+        let mut data = vec![0; PageAllocator::required_space(num_pages)];
+        let allocator = PageAllocator::init_new(&mut data, num_pages);
+        assert!(matches!(
+            allocator.find_free(&data).unwrap_err(),
+            Error::OutOfSpace
+        ));
+        allocator.free(&mut data, 128);
+        assert_eq!(allocator.find_free(&data).unwrap(), 128);
+        allocator.free(&mut data, 65);
+        assert_eq!(allocator.find_free(&data).unwrap(), 65);
+        allocator.free(&mut data, 8);
+        assert_eq!(allocator.find_free(&data).unwrap(), 8);
+        allocator.free(&mut data, 0);
+        assert_eq!(allocator.find_free(&data).unwrap(), 0);
+    }
+
     #[test]
     fn random_pattern() {
         let seed = rand::thread_rng().gen();
diff --git a/src/tree_store/page_store/page_manager.rs b/src/tree_store/page_store/page_manager.rs
index a772395..7497dbb 100644
--- a/src/tree_store/page_store/page_manager.rs
+++ b/src/tree_store/page_store/page_manager.rs
@@ -1,15 +1,18 @@
 use crate::tree_store::btree_base::Checksum;
 use crate::tree_store::page_store::buddy_allocator::BuddyAllocator;
-use crate::tree_store::page_store::grouped_bitmap::U64GroupedBitMapMut;
 use crate::tree_store::page_store::layout::{DatabaseLayout, RegionLayout};
 use crate::tree_store::page_store::mmap::Mmap;
+use crate::tree_store::page_store::page_allocator::PageAllocator;
+use crate::tree_store::page_store::region::{RegionHeaderAccessor, RegionHeaderMutator};
 use crate::tree_store::page_store::utils::get_page_size;
 use crate::tree_store::page_store::{hash128_with_seed, PageImpl, PageMut};
 use crate::tree_store::PageNumber;
 use crate::Error;
 use crate::Result;
 use std::cmp::{max, min};
-use std::collections::{HashMap, HashSet};
+#[cfg(debug_assertions)]
+use std::collections::HashMap;
+use std::collections::HashSet;
 use std::convert::TryInto;
 use std::fs::File;
 use std::io;
@@ -29,13 +32,12 @@ use std::sync::{Mutex, MutexGuard};
 // 1 byte: max page order
 // 1 byte: god byte
 // 1 byte: checksum type
-// 4 bytes: super-header length
+// 4 bytes: super-header pages
 // 4 bytes: page size
 // 4 bytes: region allocator state length
 // 8 bytes: database max size
 // Definition of region
-// 4 bytes: region header length
-// 4 bytes: regional allocator state length
+// 4 bytes: region header pages
 // 4 bytes: region max data pages
 //
 // Commit slot 0 (next 192 bytes):
@@ -70,20 +72,19 @@ pub(super) const MIN_USABLE_PAGES: usize = 10;
 const MIN_DESIRED_USABLE_BYTES: usize = 1024 * 1024;
 
 // TODO: set to 1, when version 1.0 is released
-const FILE_FORMAT_VERSION: u8 = 104;
+const FILE_FORMAT_VERSION: u8 = 105;
 
 // Inspired by PNG's magic number
 const MAGICNUMBER: [u8; 9] = [b'r', b'e', b'd', b'b', 0x1A, 0x0A, 0xA9, 0x0D, 0x0A];
 const MAX_PAGE_ORDER_OFFSET: usize = MAGICNUMBER.len();
 const GOD_BYTE_OFFSET: usize = MAX_PAGE_ORDER_OFFSET + size_of::<u8>();
 const CHECKSUM_TYPE_OFFSET: usize = GOD_BYTE_OFFSET + size_of::<u8>();
-const SUPERHEADER_LENGTH_OFFSET: usize = CHECKSUM_TYPE_OFFSET + size_of::<u8>();
-const PAGE_SIZE_OFFSET: usize = SUPERHEADER_LENGTH_OFFSET + size_of::<u32>();
+const SUPERHEADER_PAGES_OFFSET: usize = CHECKSUM_TYPE_OFFSET + size_of::<u8>();
+const PAGE_SIZE_OFFSET: usize = SUPERHEADER_PAGES_OFFSET + size_of::<u32>();
 const REGION_ALLOCATOR_LENGTH_OFFSET: usize = PAGE_SIZE_OFFSET + size_of::<u32>();
 const DB_SIZE_OFFSET: usize = REGION_ALLOCATOR_LENGTH_OFFSET + size_of::<u32>();
-const REGION_HEADER_LENGTH_OFFSET: usize = DB_SIZE_OFFSET + size_of::<u64>();
-const REGIONAL_ALLOCATOR_LENGTH_OFFSET: usize = REGION_HEADER_LENGTH_OFFSET + size_of::<u32>();
-const REGION_MAX_DATA_PAGES_OFFSET: usize = REGIONAL_ALLOCATOR_LENGTH_OFFSET + size_of::<u32>();
+const REGION_HEADER_PAGES_OFFSET: usize = DB_SIZE_OFFSET + size_of::<u64>();
+const REGION_MAX_DATA_PAGES_OFFSET: usize = REGION_HEADER_PAGES_OFFSET + size_of::<u32>();
 const TRANSACTION_SIZE: usize = 192;
 const TRANSACTION_0_OFFSET: usize = 128;
 const TRANSACTION_1_OFFSET: usize = TRANSACTION_0_OFFSET + TRANSACTION_SIZE;
@@ -185,7 +186,6 @@ impl<'a> MetadataAccessor<'a> {
         RegionLayout::new(
             num_pages,
             self.get_region_header_length(),
-            self.get_regional_allocator_state_length(),
             self.get_max_page_order() as usize,
             self.get_page_size(),
         )
@@ -298,16 +298,19 @@ impl<'a> MetadataAccessor<'a> {
     }
 
     fn get_superheader_length(&self) -> usize {
-        u32::from_le_bytes(
-            self.header[SUPERHEADER_LENGTH_OFFSET..(SUPERHEADER_LENGTH_OFFSET + size_of::<u32>())]
+        let pages = u32::from_le_bytes(
+            self.header[SUPERHEADER_PAGES_OFFSET..(SUPERHEADER_PAGES_OFFSET + size_of::<u32>())]
                 .try_into()
                 .unwrap(),
-        ) as usize
+        ) as usize;
+        pages * self.get_page_size()
     }
 
     fn set_superheader_length(&mut self, length: usize) {
-        self.header[SUPERHEADER_LENGTH_OFFSET..(SUPERHEADER_LENGTH_OFFSET + size_of::<u32>())]
-            .copy_from_slice(&(length as u32).to_le_bytes());
+        assert_eq!(length % self.get_page_size(), 0);
+        let pages = length / self.get_page_size();
+        self.header[SUPERHEADER_PAGES_OFFSET..(SUPERHEADER_PAGES_OFFSET + size_of::<u32>())]
+            .copy_from_slice(&(pages as u32).to_le_bytes());
     }
 
     fn get_region_allocator_state_length(&self) -> usize {
@@ -326,32 +329,20 @@ impl<'a> MetadataAccessor<'a> {
     }
 
     fn get_region_header_length(&self) -> usize {
-        u32::from_le_bytes(
+        let pages = u32::from_le_bytes(
             self.header
-                [REGION_HEADER_LENGTH_OFFSET..(REGION_HEADER_LENGTH_OFFSET + size_of::<u32>())]
+                [REGION_HEADER_PAGES_OFFSET..(REGION_HEADER_PAGES_OFFSET + size_of::<u32>())]
                 .try_into()
                 .unwrap(),
-        ) as usize
+        ) as usize;
+        pages * self.get_page_size()
     }
 
     fn set_region_header_length(&mut self, length: usize) {
-        self.header[REGION_HEADER_LENGTH_OFFSET..(REGION_HEADER_LENGTH_OFFSET + size_of::<u32>())]
-            .copy_from_slice(&(length as u32).to_le_bytes());
-    }
-
-    fn get_regional_allocator_state_length(&self) -> usize {
-        u32::from_le_bytes(
-            self.header[REGIONAL_ALLOCATOR_LENGTH_OFFSET
-                ..(REGIONAL_ALLOCATOR_LENGTH_OFFSET + size_of::<u32>())]
-                .try_into()
-                .unwrap(),
-        ) as usize
-    }
-
-    fn set_regional_allocator_state_length(&mut self, length: usize) {
-        self.header[REGIONAL_ALLOCATOR_LENGTH_OFFSET
-            ..(REGIONAL_ALLOCATOR_LENGTH_OFFSET + size_of::<u32>())]
-            .copy_from_slice(&(length as u32).to_le_bytes());
+        assert_eq!(length % self.get_page_size(), 0);
+        let pages = length / self.get_page_size();
+        self.header[REGION_HEADER_PAGES_OFFSET..(REGION_HEADER_PAGES_OFFSET + size_of::<u32>())]
+            .copy_from_slice(&(pages as u32).to_le_bytes());
     }
 
     // TODO: remove this method
@@ -407,15 +398,37 @@ impl<'a> MetadataAccessor<'a> {
         }
     }
 
-    fn get_regional_allocator(&mut self, region: usize, layout: &DatabaseLayout) -> &[u8] {
+    fn get_region(&mut self, region: usize, layout: &DatabaseLayout) -> RegionHeaderAccessor {
         let base = layout.region_base_address(region);
-        let len = layout.region_layout(region).header_len();
+        let len = layout.region_layout(region).data_section().start;
         let absolute = base..(base + len);
 
         // Safety: We own the metadata lock, so there can't be any other references
         // and this function takes &mut self, so the returned lifetime can't overlap with any other
         // calls into MetadataAccessor
-        unsafe { self.mmap.get_memory(absolute) }
+        let mem = unsafe { self.mmap.get_memory(absolute) };
+
+        RegionHeaderAccessor::new(mem)
+    }
+
+    fn initialize_region_allocator(&mut self, layout: &DatabaseLayout) {
+        let max_regions = DatabaseLayout::calculate(
+            self.get_max_capacity(),
+            self.get_max_capacity(),
+            self.get_region_max_usable_bytes(),
+            self.get_page_size(),
+        )
+        .unwrap()
+        .num_regions();
+
+        let range = layout.region_allocator_address_range();
+
+        // Safety: We own the metadata lock, so there can't be any other references
+        // and this function takes &mut self, so the returned lifetime can't overlap with any other
+        // calls into MetadataAccessor
+        assert!(range.start >= DB_HEADER_SIZE);
+        let mem = unsafe { self.mmap.get_memory_mut(range) };
+        RegionAllocator::init_new(max_regions, self.get_max_page_order() as usize + 1, mem);
     }
 
     // Note: It's very important that the lifetime of the returned allocator accessors is the same
@@ -423,7 +436,7 @@ impl<'a> MetadataAccessor<'a> {
     fn allocators_mut(
         &mut self,
         layout: &DatabaseLayout,
-    ) -> Result<(U64GroupedBitMapMut, RegionsAccessor)> {
+    ) -> Result<(RegionAllocator, RegionsAccessor)> {
         if !self.get_recovery_required() {
             self.set_recovery(true);
             self.mmap.flush()?
@@ -443,7 +456,96 @@ impl<'a> MetadataAccessor<'a> {
             mmap: self.mmap,
             layout: layout.clone(),
         };
-        Ok((U64GroupedBitMapMut::new(mem), region_accessor))
+        Ok((RegionAllocator::new(mem), region_accessor))
+    }
+}
+
+// Tracks the page orders that MAY BE free in each region. This data structure is optimistic, so
+// a region may not actually have a page free for a given order
+//
+// Format:
+// num_allocators: u32 number of allocators
+// allocator_len: u32 length of each allocator
+// data: PageAllocator data for each order
+pub(crate) struct RegionAllocator<'a> {
+    // TODO: remove this field
+    accessor: PageAllocator,
+    data: &'a mut [u8],
+}
+
+impl<'a> RegionAllocator<'a> {
+    pub(crate) fn new(data: &'a mut [u8]) -> Self {
+        // TODO: this is hacky since we rely on private details of the PageAllocator implementation
+        let regions = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
+        Self {
+            accessor: PageAllocator::new(regions),
+            data,
+        }
+    }
+
+    pub(crate) fn required_bytes(regions: usize, orders: usize) -> usize {
+        2 * size_of::<u32>() + orders * PageAllocator::required_space(regions)
+    }
+
+    pub(crate) fn init_new(regions: usize, orders: usize, data: &'a mut [u8]) -> Self {
+        assert!(data.len() >= Self::required_bytes(regions, orders));
+        data[..4].copy_from_slice(&(orders as u32).to_le_bytes());
+        data[4..8].copy_from_slice(&(PageAllocator::required_space(regions) as u32).to_le_bytes());
+        let mut result = Self {
+            accessor: PageAllocator::new(regions),
+            data,
+        };
+        for i in 0..orders {
+            PageAllocator::init_new(result.get_order_mut(i), regions);
+        }
+
+        result
+    }
+
+    pub(crate) fn find_free(&self, order: usize) -> Result<u64> {
+        self.accessor.find_free(self.get_order(order))
+    }
+
+    pub(crate) fn mark_free(&mut self, order: usize, region: u64) {
+        assert!(order < self.suballocators());
+        for i in 0..=order {
+            let start = 8 + i * self.suballocator_len();
+            let end = start + self.suballocator_len();
+            let mem = &mut self.data[start..end];
+            self.accessor.free(mem, region);
+        }
+    }
+
+    pub(crate) fn mark_full(&mut self, order: usize, region: u64) {
+        assert!(order < self.suballocators());
+        for i in order..self.suballocators() {
+            let start = 8 + i * self.suballocator_len();
+            let end = start + self.suballocator_len();
+            let mem = &mut self.data[start..end];
+            self.accessor.record_alloc(mem, region);
+        }
+    }
+
+    fn suballocator_len(&self) -> usize {
+        u32::from_le_bytes(self.data[4..8].try_into().unwrap()) as usize
+    }
+
+    fn suballocators(&self) -> usize {
+        u32::from_le_bytes(self.data[..4].try_into().unwrap()) as usize
+    }
+
+    fn get_order_mut(&mut self, order: usize) -> &mut [u8] {
+        assert!(order < self.suballocators());
+        let start = 8 + order * self.suballocator_len();
+        let end = start + self.suballocator_len();
+        &mut self.data[start..end]
+    }
+
+    fn get_order(&self, order: usize) -> &[u8] {
+        assert!(order < self.suballocators());
+        let start = 8 + order * self.suballocator_len();
+        let end = start + self.suballocator_len();
+        &self.data[start..end]
     }
 }
 
@@ -454,14 +556,16 @@ struct RegionsAccessor<'a> {
 }
 
 impl<'a> RegionsAccessor<'a> {
-    fn get_regional_allocator_mut(&mut self, region: usize) -> &mut [u8] {
+    fn get_region_mut(&mut self, region: usize) -> RegionHeaderMutator {
         // Safety: We have exclusive access to regional metadata
         let base = self.layout.region_base_address(region);
-        let len = &self.layout.region_layout(region).header_len();
-        let absolute = base..(base + len);
+        let region_header_len = &self.layout.region_layout(region).data_section().start;
+        let absolute = base..(base + region_header_len);
 
         assert!(absolute.start >= self.layout.header_bytes());
-        unsafe { self.mmap.get_memory_mut(absolute) }
+        let mem = unsafe { self.mmap.get_memory_mut(absolute) };
+
+        RegionHeaderMutator::new(mem)
     }
 }
 
@@ -686,11 +790,11 @@ impl TransactionalMemory {
         let mutex = Mutex::new(MetadataGuard {});
         let mut metadata = unsafe { MetadataAccessor::new(&mmap, mutex.lock().unwrap()) };
 
-        if metadata.get_magic_number() != MAGICNUMBER {
-            let region_size = requested_region_size.unwrap_or(MAX_USABLE_REGION_SPACE);
-            assert!(region_size.is_power_of_two());
-            let max_usable_region_bytes = min(region_size, max_capacity.next_power_of_two());
+        let region_size = requested_region_size.unwrap_or(MAX_USABLE_REGION_SPACE);
+        assert!(region_size.is_power_of_two());
+        let max_usable_region_bytes = min(region_size, max_capacity.next_power_of_two());
 
+        if metadata.get_magic_number() != MAGICNUMBER {
             let starting_size = if dynamic_growth {
                 MIN_DESIRED_USABLE_BYTES
             } else {
@@ -713,23 +817,37 @@ impl TransactionalMemory {
             // Explicitly zero the header
             metadata.header.fill(0);
 
+            // Store the page & db size. These are immutable
+            metadata.set_page_size(page_size);
+            metadata.set_max_page_order(layout.full_region_layout().max_order() as u8);
+            metadata.set_superheader_length(layout.header_bytes());
+            metadata
+                .set_region_allocator_state_length(layout.region_allocator_address_range().len());
+            metadata.set_region_header_length(layout.full_region_layout().data_section().start);
+            metadata.set_region_max_data_pages(layout.full_region_layout().num_pages());
+            metadata.set_max_capacity(max_capacity);
+            let checksum_type = if use_checksums.unwrap_or(true) {
+                ChecksumType::XXH3_128
+            } else {
+                ChecksumType::Zero
+            };
+            metadata.set_checksum_type(checksum_type);
+
+            // Initialize the region allocator. Must be done after writing the page_size and other
+            // immutable fields, since it relies on those fields.
+            metadata.initialize_region_allocator(&layout);
             let (mut region_allocator, mut regions) = metadata.allocators_mut(&layout)?;
 
-            // Initialize the region allocator
             let num_regions = layout.num_regions();
             for i in 0..num_regions {
-                region_allocator.clear(i);
-            }
-            for i in num_regions..region_allocator.len() {
-                region_allocator.set(i);
+                region_allocator.mark_free(layout.full_region_layout().max_order(), i as u64);
             }
 
             // Initialize all the regional allocators
             for i in 0..num_regions {
-                let mem = regions.get_regional_allocator_mut(i);
+                let mut region = regions.get_region_mut(i);
                 let region_layout = layout.region_layout(i);
-                BuddyAllocator::init_new(
-                    mem,
+                region.initialize(
                     region_layout.num_pages(),
                     layout.full_region_layout().num_pages(),
                     region_layout.max_order(),
@@ -739,23 +857,6 @@ impl TransactionalMemory {
             // dirtied it
             metadata.set_recovery(false);
 
-            // Store the page & db size. These are immutable
-            metadata.set_page_size(page_size);
-            metadata.set_max_page_order(layout.full_region_layout().max_order() as u8);
-            metadata.set_superheader_length(layout.header_bytes());
-            metadata
-                .set_region_allocator_state_length(layout.region_allocator_address_range().len());
-            metadata.set_region_header_length(layout.full_region_layout().data_section().start);
-            metadata.set_regional_allocator_state_length(layout.full_region_layout().header_len());
-            metadata.set_region_max_data_pages(layout.full_region_layout().num_pages());
-            metadata.set_max_capacity(max_capacity);
-            let checksum_type = if use_checksums.unwrap_or(true) {
-                ChecksumType::XXH3_128
-            } else {
-                ChecksumType::Zero
-            };
-            metadata.set_checksum_type(checksum_type);
-
             let mut mutator = metadata.secondary_slot_mut();
             mutator.set_root_page(None);
             mutator.set_freed_root(None);
@@ -885,27 +986,31 @@ impl TransactionalMemory {
         }
 
         let layout = self.layout.lock().unwrap();
+        metadata.initialize_region_allocator(&layout);
         let (mut region_allocator, mut regions) = metadata.allocators_mut(&layout)?;
 
-        // Initialize the region allocator
         let num_regions = layout.num_regions();
-        for i in num_regions..region_allocator.len() {
-            region_allocator.set(i);
-        }
-        // Since the region allocator is lazily set, we can leave it all free, and it will be
-        // populated lazily when a region is discovered to be full
-
         // Initialize all the regional allocators
         let mut regional_allocators = vec![];
         for i in 0..num_regions {
-            let mem = regions.get_regional_allocator_mut(i);
+            let mut region = regions.get_region_mut(i);
             let region_layout = layout.region_layout(i);
-            regional_allocators.push(BuddyAllocator::init_new(
-                mem,
+            region.initialize(
+                region_layout.num_pages(),
+                layout.full_region_layout().num_pages(),
+                region_layout.max_order(),
+            );
+            let allocator = BuddyAllocator::new(
                 region_layout.num_pages(),
                 layout.full_region_layout().num_pages(),
                 region_layout.max_order(),
-            ));
+            );
+            let highest_free = allocator
+                .highest_free_order(region.allocator_state_mut())
+                .unwrap();
+            // Initialize the region allocator
+            region_allocator.mark_free(highest_free, i as u64);
+            regional_allocators.push(allocator);
         }
 
         let mut guard = self.regional_allocators.lock().unwrap();
@@ -925,9 +1030,10 @@ impl TransactionalMemory {
         let regional_allocators = self.regional_allocators.lock().unwrap();
 
         for page_number in allocated_pages {
-            let region = page_number.region as usize;
-            let mem = regions.get_regional_allocator_mut(region);
-            regional_allocators.as_ref().unwrap()[region].record_alloc(
+            let region_index = page_number.region as usize;
+            let mut region = regions.get_region_mut(region_index);
+            let mem = region.allocator_state_mut();
+            regional_allocators.as_ref().unwrap()[region_index].record_alloc(
                 mem,
                 page_number.page_index as u64,
                 page_number.page_order as usize,
@@ -1069,19 +1175,22 @@ impl TransactionalMemory {
         for op in self.log_since_commit.lock().unwrap().drain(..).rev() {
             match op {
                 AllocationOp::Allocate(page_number) => {
-                    let region = page_number.region as usize;
-                    region_allocator.clear(region);
-                    let mem = regions.get_regional_allocator_mut(region);
-                    regional_guard.as_ref().unwrap()[region].free(
+                    let region_index = page_number.region as usize;
+                    region_allocator
+                        .mark_free(page_number.page_order as usize, region_index as u64);
+                    let mut region = regions.get_region_mut(region_index);
+                    let mem = region.allocator_state_mut();
+                    regional_guard.as_ref().unwrap()[region_index].free(
                         mem,
                         page_number.page_index as u64,
                         page_number.page_order as usize,
                     );
                 }
                 AllocationOp::Free(page_number) | AllocationOp::FreeUncommitted(page_number) => {
-                    let region = page_number.region as usize;
-                    let mem = regions.get_regional_allocator_mut(region);
-                    regional_guard.as_ref().unwrap()[region].record_alloc(
+                    let region_index = page_number.region as usize;
+                    let mut region = regions.get_region_mut(region_index);
+                    let mem = region.allocator_state_mut();
+                    regional_guard.as_ref().unwrap()[region_index].record_alloc(
                         mem,
                         page_number.page_index as u64,
                         page_number.page_order as usize,
@@ -1103,7 +1212,8 @@ impl TransactionalMemory {
             // Restore the size of the last region's allocator
             let last_region_index = restore.num_regions() - 1;
             let last_region = restore.region_layout(last_region_index);
-            let allocator_data = regions.get_regional_allocator_mut(last_region_index);
+            let mut region = regions.get_region_mut(last_region_index);
+            let allocator_data = region.allocator_state_mut();
             let last_allocator = &mut regional_guard.as_mut().unwrap()[last_region_index];
             last_allocator.resize(allocator_data, last_region.num_pages());
 
@@ -1220,16 +1330,17 @@ impl TransactionalMemory {
         let mut metadata = self.lock_metadata();
         let layout = self.layout.lock().unwrap();
         let (mut region_allocator, mut regions) = metadata.allocators_mut(&layout)?;
-        let region = page.region as usize;
+        let region_index = page.region as usize;
         // Free in the regional allocator
-        let mem = regions.get_regional_allocator_mut(region);
-        self.regional_allocators.lock().unwrap().as_ref().unwrap()[region].free(
+        let mut region = regions.get_region_mut(region_index);
+        let mem = region.allocator_state_mut();
+        self.regional_allocators.lock().unwrap().as_ref().unwrap()[region_index].free(
             mem,
             page.page_index as u64,
             page.page_order as usize,
         );
         // Ensure that the region is marked as having free space
-        region_allocator.clear(region);
+        region_allocator.mark_free(page.page_order as usize, region_index as u64);
         self.log_since_commit
             .lock()
             .unwrap()
@@ -1250,14 +1361,15 @@ impl TransactionalMemory {
             let layout = self.layout.lock().unwrap();
             let (mut region_allocator, mut regions) = metadata.allocators_mut(&layout)?;
             // Free in the regional allocator
-            let mem = regions.get_regional_allocator_mut(page.region as usize);
+            let mut region = regions.get_region_mut(page.region as usize);
+            let mem = region.allocator_state_mut();
             self.regional_allocators.lock().unwrap().as_ref().unwrap()[page.region as usize].free(
                 mem,
                 page.page_index as u64,
                 page.page_order as usize,
             );
             // Ensure that the region is marked as having free space
-            region_allocator.clear(page.region as usize);
+            region_allocator.mark_free(page.page_order as usize, page.region as u64);
 
             self.log_since_commit
                 .lock()
@@ -1281,39 +1393,33 @@ impl TransactionalMemory {
         layout: &DatabaseLayout,
         required_order: usize,
     ) -> Result<PageNumber> {
+        if required_order > metadata.get_max_page_order() as usize {
+            return Err(Error::OutOfSpace);
+        }
         let regional_guard = self.regional_allocators.lock().unwrap();
         let (mut region_allocator, mut regions) = metadata.allocators_mut(layout)?;
-        let mut allocated_page = None;
-        let mut allocated_region = 0;
-        for region in 0..region_allocator.len() {
-            allocated_region = region;
-            if !region_allocator.get(region) {
-                let mem = regions.get_regional_allocator_mut(region);
-                match regional_guard.as_ref().unwrap()[region].alloc(mem, required_order) {
-                    Ok(page) => {
-                        allocated_page = Some(page);
-                        break;
-                    }
-                    Err(err) => {
-                        if matches!(err, Error::OutOfSpace) {
-                            // Mark the region, if it's full
-                            if required_order == 0 {
-                                region_allocator.set(region);
-                            }
-                        } else {
-                            return Err(err);
-                        }
+        loop {
+            let candidate_region = region_allocator.find_free(required_order)? as usize;
+            let mut region = regions.get_region_mut(candidate_region);
+            let mem = region.allocator_state_mut();
+            match regional_guard.as_ref().unwrap()[candidate_region].alloc(mem, required_order) {
+                Ok(page) => {
+                    return Ok(PageNumber::new(
+                        candidate_region as u32,
+                        page as u32,
+                        required_order as u8,
+                    ));
+                }
+                Err(err) => {
+                    if matches!(err, Error::OutOfSpace) {
+                        // Mark the region, if it's full
+                        region_allocator.mark_full(required_order, candidate_region as u64);
+                    } else {
+                        return Err(err);
                     }
                 }
             }
         }
-
-        let page = allocated_page.ok_or(Error::OutOfSpace)?;
-        Ok(PageNumber::new(
-            allocated_region as u32,
-            page as u32,
-            required_order as u8,
-        ))
     }
 
     // Safety: caller must guarantee that no references to free pages at the end of the last region exist
@@ -1326,7 +1432,8 @@ impl TransactionalMemory {
         let last_region_index = layout.num_regions() - 1;
         let last_allocator = allocators.as_ref().unwrap()[last_region_index].clone();
         let last_region = layout.region_layout(last_region_index);
-        let allocator_data = metadata.get_regional_allocator(last_region_index, layout);
+        let region = metadata.get_region(last_region_index, layout);
+        let allocator_data = region.allocator_state();
         let trailing_free = last_allocator.trailing_free_pages(allocator_data);
         // TODO: is this the right shrinkage heuristic?
         if trailing_free < last_allocator.len() / 2 {
@@ -1340,7 +1447,7 @@ impl TransactionalMemory {
 
         let (mut region_allocator, mut regions) = metadata.allocators_mut(layout)?;
         let new_usable_bytes = if reduce_to_pages == 0 {
-            region_allocator.set(last_region_index);
+            region_allocator.mark_full(0, last_region_index as u64);
             allocators
                 .as_mut()
                 .unwrap()
@@ -1349,7 +1456,8 @@ impl TransactionalMemory {
             // drop the whole region
             layout.usable_bytes() - last_region.usable_bytes()
         } else {
-            let mem = regions.get_regional_allocator_mut(last_region_index);
+            let mut region = regions.get_region_mut(last_region_index);
+            let mem = region.allocator_state_mut();
             allocators.as_mut().unwrap()[last_region_index].resize(mem, reduce_to_pages);
             layout.usable_bytes()
                 - (last_allocator.len() - reduce_to_pages) * metadata.get_page_size()
@@ -1430,22 +1538,32 @@ impl TransactionalMemory {
                 if new_region.len() != old_region.len() {
                     let (mut region_allocator, mut regions) =
                         metadata.allocators_mut(&new_layout)?;
-                    region_allocator.clear(i);
-                    let mem = regions.get_regional_allocator_mut(i);
+                    let mut region = regions.get_region_mut(i);
+                    let mem = region.allocator_state_mut();
                     allocator.resize(mem, new_region.num_pages());
+                    let highest_free = allocator.highest_free_order(mem).unwrap();
+                    region_allocator.mark_free(highest_free, i as u64);
                 }
                 allocator
             } else {
                 // brand new region
                 let (mut region_allocator, mut regions) = metadata.allocators_mut(&new_layout)?;
-                region_allocator.clear(i);
-                let mem = regions.get_regional_allocator_mut(i);
-                BuddyAllocator::init_new(
-                    mem,
+                let mut region = regions.get_region_mut(i);
+                region.initialize(
                     new_region.num_pages(),
                     new_layout.full_region_layout().num_pages(),
                     new_region.max_order(),
-                )
+                );
+                let allocator = BuddyAllocator::new(
+                    new_region.num_pages(),
+                    new_layout.full_region_layout().num_pages(),
+                    new_region.max_order(),
+                );
+                let mem = region.allocator_state_mut();
+                let highest_free = allocator.highest_free_order(mem).unwrap();
+                region_allocator.mark_free(highest_free, i as u64);
+
+                allocator
             };
             new_allocators.push(new_allocator);
         }
@@ -1517,7 +1635,8 @@ impl TransactionalMemory {
         let layout = self.layout.lock().unwrap();
         let mut count = 0;
         for i in 0..layout.num_regions() {
-            let mem = metadata.get_regional_allocator(i, &layout);
+            let region = metadata.get_region(i, &layout);
+            let mem = region.allocator_state();
             count += regional_guard.as_ref().unwrap()[i].count_free_pages(mem);
         }
 
diff --git a/src/tree_store/page_store/region.rs b/src/tree_store/page_store/region.rs
new file mode 100644
index 0000000..4162731
--- /dev/null
+++ b/src/tree_store/page_store/region.rs
@@ -0,0 +1,75 @@
+use crate::tree_store::page_store::buddy_allocator::BuddyAllocator;
+use std::mem::size_of;
+
+const REGION_FORMAT_VERSION: u8 = 1;
+const ALLOCATOR_LENGTH_OFFSET: usize = 4;
+const ALLOCATOR_OFFSET: usize = ALLOCATOR_LENGTH_OFFSET + size_of::<u32>();
+
+// Region header
+// 1 byte: region format version
+// 3 bytes: padding
+// 4 bytes: length of the allocator state in bytes
+// n bytes: the allocator state
+pub(super) struct RegionHeaderMutator<'a> {
+    mem: &'a mut [u8],
+}
+
+impl<'a> RegionHeaderMutator<'a> {
+    pub(crate) fn new(data: &'a mut [u8]) -> Self {
+        Self { mem: data }
+    }
+
+    pub(crate) fn initialize(
+        &mut self,
+        num_pages: usize,
+        max_page_capacity: usize,
+        max_order: usize,
+    ) {
+        self.mem[0] = REGION_FORMAT_VERSION;
+        let allocator_len = BuddyAllocator::required_space(max_page_capacity, max_order);
+        self.mem[ALLOCATOR_LENGTH_OFFSET..(ALLOCATOR_LENGTH_OFFSET + size_of::<u32>())]
+            .copy_from_slice(&(allocator_len as u32).to_le_bytes());
+        BuddyAllocator::init_new(
+            &mut self.mem[ALLOCATOR_OFFSET..(ALLOCATOR_OFFSET + allocator_len)],
+            num_pages,
+            max_page_capacity,
+            max_order,
+        );
+    }
+
+    fn get_allocator_len(&self) -> usize {
+        u32::from_le_bytes(
+            self.mem[ALLOCATOR_LENGTH_OFFSET..(ALLOCATOR_LENGTH_OFFSET + size_of::<u32>())]
+                .try_into()
+                .unwrap(),
+        ) as usize
+    }
+
+    pub(crate) fn allocator_state_mut(&mut self) -> &mut [u8] {
+        let len = self.get_allocator_len();
+        &mut self.mem[ALLOCATOR_OFFSET..(ALLOCATOR_OFFSET + len)]
+    }
+}
+
+pub(super) struct RegionHeaderAccessor<'a> {
+    mem: &'a [u8],
+}
+
+impl<'a> RegionHeaderAccessor<'a> {
+    pub(crate) fn new(data: &'a [u8]) -> Self {
+        assert_eq!(data[0], REGION_FORMAT_VERSION);
+        Self { mem: data }
+    }
+
+    fn get_allocator_len(&self) -> usize {
+        u32::from_le_bytes(
+            self.mem[ALLOCATOR_LENGTH_OFFSET..(ALLOCATOR_LENGTH_OFFSET + size_of::<u32>())]
+                .try_into()
+                .unwrap(),
+        ) as usize
+    }
+
+    pub(crate) fn allocator_state(&self) -> &[u8] {
+        &self.mem[ALLOCATOR_OFFSET..(ALLOCATOR_OFFSET + self.get_allocator_len())]
+    }
+}


Test patch:
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index ddeddea..7165393 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -810,6 +810,20 @@ fn regression14() {
     tx.abort().unwrap();
 }
 
+#[test]
+fn regression15() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db_size = 400998358365;
+    unsafe { Database::create(tmpfile.path(), db_size).unwrap() };
+}
+
+#[test]
+fn regression16() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+    let db_size = 4294985472;
+    unsafe { Database::create(tmpfile.path(), db_size).unwrap() };
+}
+
 #[test]
 fn non_durable_read_isolation() {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();