Merge 'More WAL work' from Pekka Enberg

Turn the WAL into a trait and open + parse WAL header.

Closes #267

bindings/wasm/lib.rs

@@ -14,7 +14,8 @@ impl Database {
     pub fn new(_path: &str) -> Database {
         let io = Arc::new(IO {});
         let page_io = Rc::new(DatabaseStorage {});
-        let inner = limbo_core::Database::open(io, page_io).unwrap();
+        let wal = Rc::new(Wal {});
+        let inner = limbo_core::Database::open(io, page_io, wal).unwrap();
         Database { _inner: inner }
     }
 
@@ -50,3 +51,27 @@ impl limbo_core::DatabaseStorage for DatabaseStorage {
         todo!()
     }
 }
+
+pub struct Wal {}
+
+impl limbo_core::Wal for Wal {
+    fn begin_read_tx(&self) -> Result<()> {
+        todo!()
+    }
+
+    fn end_read_tx(&self) -> Result<()> {
+        todo!()
+    }
+
+    fn find_frame(&self, _page_id: u64) -> Result<Option<u64>> {
+        todo!()
+    }
+
+    fn read_frame(
+        &self,
+        _frame_id: u64,
+        _page: Rc<std::cell::RefCell<limbo_core::Page>>,
+    ) -> Result<()> {
+        todo!()
+    }
+}

core/lib.rs

@@ -23,6 +23,8 @@ use std::{cell::RefCell, rc::Rc};
 use storage::database::FileStorage;
 use storage::pager::Pager;
 use storage::sqlite3_ondisk::DatabaseHeader;
+#[cfg(feature = "fs")]
+use storage::wal::WalFile;
 
 pub use error::LimboError;
 pub type Result<T> = std::result::Result<T, error::LimboError>;
@@ -31,6 +33,8 @@ pub type Result<T> = std::result::Result<T, error::LimboError>;
 pub use io::PlatformIO;
 pub use io::{Buffer, Completion, File, WriteCompletion, IO};
 pub use storage::database::DatabaseStorage;
+pub use storage::pager::Page;
+pub use storage::wal::Wal;
 pub use types::Value;
 
 pub struct Database {
@@ -44,13 +48,24 @@ impl Database {
     pub fn open_file(io: Arc<dyn crate::io::IO>, path: &str) -> Result<Database> {
         let file = io.open_file(path)?;
         let page_io = Rc::new(FileStorage::new(file));
-        Self::open(io, page_io)
+        let wal_path = format!("{}-wal", path);
+        let wal = Rc::new(WalFile::new(io.clone(), wal_path));
+        Self::open(io, page_io, wal)
     }
 
-    pub fn open(io: Arc<dyn crate::io::IO>, page_io: Rc<dyn DatabaseStorage>) -> Result<Database> {
+    pub fn open(
+        io: Arc<dyn crate::io::IO>,
+        page_io: Rc<dyn DatabaseStorage>,
+        wal: Rc<dyn Wal>,
+    ) -> Result<Database> {
         let db_header = Pager::begin_open(page_io.clone())?;
         io.run_once()?;
-        let pager = Rc::new(Pager::finish_open(db_header.clone(), page_io, io.clone())?);
+        let pager = Rc::new(Pager::finish_open(
+            db_header.clone(),
+            page_io,
+            wal,
+            io.clone(),
+        )?);
         let bootstrap_schema = Rc::new(Schema::new());
         let conn = Connection {
             pager: pager.clone(),

core/storage/pager.rs

@@ -265,7 +265,7 @@ pub struct Pager {
     /// Source of the database pages.
     pub page_io: Rc<dyn DatabaseStorage>,
     /// The write-ahead log (WAL) for the database.
-    wal: Option<Wal>,
+    wal: Rc<dyn Wal>,
     /// A page cache for the database.
     page_cache: RefCell<DumbLruPageCache>,
     /// Buffer pool for temporary data storage.
@@ -286,6 +286,7 @@ impl Pager {
     pub fn finish_open(
         db_header_ref: Rc<RefCell<DatabaseHeader>>,
         page_io: Rc<dyn DatabaseStorage>,
+        wal: Rc<dyn Wal>,
         io: Arc<dyn crate::io::IO>,
     ) -> Result<Self> {
         let db_header = RefCell::borrow(&db_header_ref);
@@ -294,7 +295,7 @@ impl Pager {
         let page_cache = RefCell::new(DumbLruPageCache::new(10));
         Ok(Self {
             page_io,
-            wal: None,
+            wal,
             buffer_pool,
             page_cache,
             io,
@@ -304,16 +305,12 @@ impl Pager {
     }
 
     pub fn begin_read_tx(&self) -> Result<()> {
-        if let Some(wal) = &self.wal {
-            wal.begin_read_tx()?;
-        }
+        self.wal.begin_read_tx()?;
         Ok(())
     }
 
     pub fn end_read_tx(&self) -> Result<()> {
-        if let Some(wal) = &self.wal {
-            wal.end_read_tx()?;
-        }
+        self.wal.end_read_tx()?;
         Ok(())
     }
 
@@ -326,16 +323,14 @@ impl Pager {
         }
         let page = Rc::new(RefCell::new(Page::new(page_idx)));
         RefCell::borrow(&page).set_locked();
-        if let Some(wal) = &self.wal {
-            if let Some(frame_id) = wal.find_frame(page_idx as u64)? {
-                wal.read_frame(frame_id, page.clone())?;
-                {
-                    let page = page.borrow_mut();
-                    page.set_uptodate();
-                }
-                page_cache.insert(page_idx, page.clone());
-                return Ok(page);
+        if let Some(frame_id) = self.wal.find_frame(page_idx as u64)? {
+            self.wal.read_frame(frame_id, page.clone())?;
+            {
+                let page = page.borrow_mut();
+                page.set_uptodate();
             }
+            page_cache.insert(page_idx, page.clone());
+            return Ok(page);
         }
         sqlite3_ondisk::begin_read_page(
             self.page_io.clone(),

core/storage/sqlite3_ondisk.rs

@@ -830,7 +830,7 @@ pub fn write_varint(buf: &mut [u8], value: u64) -> usize {
     return n;
 }
 
-pub fn begin_read_wal_header(io: &Box<dyn File>) -> Result<Rc<RefCell<WalHeader>>> {
+pub fn begin_read_wal_header(io: Rc<dyn File>) -> Result<Rc<RefCell<WalHeader>>> {
     let drop_fn = Rc::new(|_buf| {});
     let buf = Rc::new(RefCell::new(Buffer::allocate(32, drop_fn)));
     let result = Rc::new(RefCell::new(WalHeader::default()));

core/storage/wal.rs

@@ -1,32 +1,78 @@
-use std::{cell::RefCell, rc::Rc};
+use std::{cell::RefCell, rc::Rc, sync::Arc};
 
+use crate::io::{File, IO};
 use crate::{storage::pager::Page, Result};
 
+use super::sqlite3_ondisk;
+
 /// Write-ahead log (WAL).
-pub struct Wal {}
-
-impl Wal {
-    pub fn new() -> Self {
-        Self {}
-    }
-
+pub trait Wal {
     /// Begin a write transaction.
-    pub fn begin_read_tx(&self) -> Result<()> {
+    fn begin_read_tx(&self) -> Result<()>;
+
+    /// End a write transaction.
+    fn end_read_tx(&self) -> Result<()>;
+
+    /// Find the latest frame containing a page.
+    fn find_frame(&self, page_id: u64) -> Result<Option<u64>>;
+
+    /// Read a frame from the WAL.
+    fn read_frame(&self, frame_id: u64, page: Rc<RefCell<Page>>) -> Result<()>;
+}
+
+#[cfg(feature = "fs")]
+pub struct WalFile {
+    io: Arc<dyn crate::io::IO>,
+    wal_path: String,
+    file: RefCell<Option<Rc<dyn File>>>,
+    wal_header: RefCell<Option<Rc<RefCell<sqlite3_ondisk::WalHeader>>>>,
+}
+
+#[cfg(feature = "fs")]
+impl Wal for WalFile {
+    /// Begin a write transaction.
+    fn begin_read_tx(&self) -> Result<()> {
         Ok(())
     }
 
     /// End a write transaction.
-    pub fn end_read_tx(&self) -> Result<()> {
+    fn end_read_tx(&self) -> Result<()> {
         Ok(())
     }
 
     /// Find the latest frame containing a page.
-    pub fn find_frame(&self, _page_id: u64) -> Result<Option<u64>> {
+    fn find_frame(&self, _page_id: u64) -> Result<Option<u64>> {
+        self.ensure_init()?;
         Ok(None)
     }
 
     /// Read a frame from the WAL.
-    pub fn read_frame(&self, _frame_id: u64, _page: Rc<RefCell<Page>>) -> Result<()> {
+    fn read_frame(&self, _frame_id: u64, _page: Rc<RefCell<Page>>) -> Result<()> {
         todo!();
     }
 }
+
+#[cfg(feature = "fs")]
+impl WalFile {
+    pub fn new(io: Arc<dyn IO>, wal_path: String) -> Self {
+        Self {
+            io,
+            wal_path,
+            file: RefCell::new(None),
+            wal_header: RefCell::new(None),
+        }
+    }
+
+    fn ensure_init(&self) -> Result<()> {
+        if self.file.borrow().is_none() {
+            if let Ok(file) = self.io.open_file(&self.wal_path) {
+                *self.file.borrow_mut() = Some(file.clone());
+                let wal_header = sqlite3_ondisk::begin_read_wal_header(file)?;
+                // TODO: Return a completion instead.
+                self.io.run_once()?;
+                self.wal_header.replace(Some(wal_header));
+            }
+        }
+        Ok(())
+    }
+}