diff --git a/bindings/wasm/lib.rs b/bindings/wasm/lib.rs index 1d36966ca..00d67a814 100644 --- a/bindings/wasm/lib.rs +++ b/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> { + todo!() + } + + fn read_frame( + &self, + _frame_id: u64, + _page: Rc>, + ) -> Result<()> { + todo!() + } +} diff --git a/core/lib.rs b/core/lib.rs index 44abf6d5b..fb695a3aa 100644 --- a/core/lib.rs +++ b/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 = std::result::Result; @@ -31,6 +33,8 @@ pub type Result = std::result::Result; 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, path: &str) -> Result { 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, page_io: Rc) -> Result { + pub fn open( + io: Arc, + page_io: Rc, + wal: Rc, + ) -> Result { 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(), diff --git a/core/storage/pager.rs b/core/storage/pager.rs index 023bfb1aa..783687355 100644 --- a/core/storage/pager.rs +++ b/core/storage/pager.rs @@ -265,7 +265,7 @@ pub struct Pager { /// Source of the database pages. pub page_io: Rc, /// The write-ahead log (WAL) for the database. - wal: Option, + wal: Rc, /// A page cache for the database. page_cache: RefCell, /// Buffer pool for temporary data storage. @@ -286,6 +286,7 @@ impl Pager { pub fn finish_open( db_header_ref: Rc>, page_io: Rc, + wal: Rc, io: Arc, ) -> Result { 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(), diff --git a/core/storage/sqlite3_ondisk.rs b/core/storage/sqlite3_ondisk.rs index 885c8cd90..4c19577b7 100644 --- a/core/storage/sqlite3_ondisk.rs +++ b/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) -> Result>> { +pub fn begin_read_wal_header(io: Rc) -> Result>> { 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())); diff --git a/core/storage/wal.rs b/core/storage/wal.rs index b25a8acbd..10583e26b 100644 --- a/core/storage/wal.rs +++ b/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>; + + /// Read a frame from the WAL. + fn read_frame(&self, frame_id: u64, page: Rc>) -> Result<()>; +} + +#[cfg(feature = "fs")] +pub struct WalFile { + io: Arc, + wal_path: String, + file: RefCell>>, + wal_header: RefCell>>>, +} + +#[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> { + fn find_frame(&self, _page_id: u64) -> Result> { + self.ensure_init()?; Ok(None) } /// Read a frame from the WAL. - pub fn read_frame(&self, _frame_id: u64, _page: Rc>) -> Result<()> { + fn read_frame(&self, _frame_id: u64, _page: Rc>) -> Result<()> { todo!(); } } + +#[cfg(feature = "fs")] +impl WalFile { + pub fn new(io: Arc, 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(()) + } +} diff --git a/testing/wal/users.db b/testing/wal/users.db new file mode 100644 index 000000000..210e70546 Binary files /dev/null and b/testing/wal/users.db differ diff --git a/testing/wal/users.db-wal b/testing/wal/users.db-wal new file mode 100644 index 000000000..6e48a4013 Binary files /dev/null and b/testing/wal/users.db-wal differ