Merge 'fix raw read frame WAL API' from Nikita Sivukhin

This PR fixes `wal_read_frame_raw` API
Before, implementation of raw read API read only page content - which is
not enough as we also need page_no and size_after fields from the
header. This PR fixes that and also make few adjustments in the
signatures.

Reviewed-by: Preston Thorpe (@PThorpe92)

Closes #2229

core/lib.rs

@@ -809,15 +809,8 @@ impl Connection {
         self.pager.borrow().wal_frame_count()
     }
 
-    pub fn wal_get_frame(
-        &self,
-        frame_no: u32,
-        p_frame: *mut u8,
-        frame_len: u32,
-    ) -> Result<Arc<Completion>> {
-        self.pager
-            .borrow()
-            .wal_get_frame(frame_no, p_frame, frame_len)
+    pub fn wal_get_frame(&self, frame_no: u32, frame: &mut [u8]) -> Result<Arc<Completion>> {
+        self.pager.borrow().wal_get_frame(frame_no, frame)
     }
 
     /// Flush dirty pages to disk.

core/storage/pager.rs

@@ -959,19 +959,9 @@ impl Pager {
     }
 
     #[instrument(skip_all, level = Level::DEBUG)]
-    pub fn wal_get_frame(
-        &self,
-        frame_no: u32,
-        p_frame: *mut u8,
-        frame_len: u32,
-    ) -> Result<Arc<Completion>> {
+    pub fn wal_get_frame(&self, frame_no: u32, frame: &mut [u8]) -> Result<Arc<Completion>> {
         let wal = self.wal.borrow();
-        wal.read_frame_raw(
-            frame_no.into(),
-            self.buffer_pool.clone(),
-            p_frame,
-            frame_len,
-        )
+        wal.read_frame_raw(frame_no.into(), frame)
     }
 
     #[instrument(skip_all, level = Level::DEBUG, name = "pager_checkpoint",)]

core/storage/sqlite3_ondisk.rs

@@ -1509,6 +1509,24 @@ pub fn read_entire_wal_dumb(file: &Arc<dyn File>) -> Result<Arc<UnsafeCell<WalFi
     Ok(wal_file_shared_ret)
 }
 
+pub fn begin_read_wal_frame_raw(
+    io: &Arc<dyn File>,
+    offset: usize,
+    page_size: u32,
+    complete: Box<dyn Fn(Arc<RefCell<Buffer>>, i32)>,
+) -> Result<Arc<Completion>> {
+    tracing::trace!("begin_read_wal_frame_raw(offset={})", offset);
+    let drop_fn = Rc::new(|_buf| {});
+    let buf = Arc::new(RefCell::new(Buffer::allocate(
+        page_size as usize + WAL_FRAME_HEADER_SIZE,
+        drop_fn,
+    )));
+    #[allow(clippy::arc_with_non_send_sync)]
+    let c = Completion::new_read(buf, complete);
+    let c = io.pread(offset, c.into())?;
+    Ok(c)
+}
+
 pub fn begin_read_wal_frame(
     io: &Arc<dyn File>,
     offset: usize,

core/storage/wal.rs

@@ -20,8 +20,8 @@ use crate::fast_lock::SpinLock;
 use crate::io::{File, IO};
 use crate::result::LimboResult;
 use crate::storage::sqlite3_ondisk::{
-    begin_read_wal_frame, begin_write_wal_frame, finish_read_page, WAL_FRAME_HEADER_SIZE,
-    WAL_HEADER_SIZE,
+    begin_read_wal_frame, begin_read_wal_frame_raw, begin_write_wal_frame, finish_read_page,
+    WAL_FRAME_HEADER_SIZE, WAL_HEADER_SIZE,
 };
 use crate::types::IOResult;
 use crate::{turso_assert, Buffer, LimboError, Result};
@@ -214,14 +214,8 @@ pub trait Wal {
     /// Read a frame from the WAL.
     fn read_frame(&self, frame_id: u64, page: PageRef, buffer_pool: Arc<BufferPool>) -> Result<()>;
 
-    /// Read a frame from the WAL.
-    fn read_frame_raw(
-        &self,
-        frame_id: u64,
-        buffer_pool: Arc<BufferPool>,
-        frame: *mut u8,
-        frame_len: u32,
-    ) -> Result<Arc<Completion>>;
+    /// Read a raw frame (header included) from the WAL.
+    fn read_frame_raw(&self, frame_id: u64, frame: &mut [u8]) -> Result<Arc<Completion>>;
 
     /// Write a frame to the WAL.
     /// db_size is the database size in pages after the transaction finishes.
@@ -289,13 +283,7 @@ impl Wal for DummyWAL {
         Ok(())
     }
 
-    fn read_frame_raw(
-        &self,
-        _frame_id: u64,
-        _buffer_pool: Arc<BufferPool>,
-        _frame: *mut u8,
-        _frame_len: u32,
-    ) -> Result<Arc<Completion>> {
+    fn read_frame_raw(&self, _frame_id: u64, _frame: &mut [u8]) -> Result<Arc<Completion>> {
         todo!();
     }
 
@@ -633,15 +621,10 @@ impl Wal for WalFile {
     }
 
     #[instrument(skip_all, level = Level::DEBUG)]
-    fn read_frame_raw(
-        &self,
-        frame_id: u64,
-        buffer_pool: Arc<BufferPool>,
-        frame: *mut u8,
-        frame_len: u32,
-    ) -> Result<Arc<Completion>> {
+    fn read_frame_raw(&self, frame_id: u64, frame: &mut [u8]) -> Result<Arc<Completion>> {
         tracing::debug!("read_frame({})", frame_id);
         let offset = self.frame_offset(frame_id);
+        let (frame_ptr, frame_len) = (frame.as_mut_ptr(), frame.len());
         let complete = Box::new(move |buf: Arc<RefCell<Buffer>>, bytes_read: i32| {
             let buf = buf.borrow();
             let buf_len = buf.len();
@@ -651,15 +634,11 @@ impl Wal for WalFile {
             );
             let buf_ptr = buf.as_ptr();
             unsafe {
-                std::ptr::copy_nonoverlapping(buf_ptr, frame, frame_len as usize);
+                std::ptr::copy_nonoverlapping(buf_ptr, frame_ptr, frame_len);
             }
         });
-        let c = begin_read_wal_frame(
-            &self.get_shared().file,
-            offset + WAL_FRAME_HEADER_SIZE,
-            buffer_pool,
-            complete,
-        )?;
+        let c =
+            begin_read_wal_frame_raw(&self.get_shared().file, offset, self.page_size(), complete)?;
         Ok(c)
     }
 

sqlite3/src/lib.rs

@@ -1185,7 +1185,8 @@ pub unsafe extern "C" fn libsql_wal_get_frame(
     }
     let db: &mut sqlite3 = &mut *db;
     let db = db.inner.lock().unwrap();
-    match db.conn.wal_get_frame(frame_no, p_frame, frame_len) {
+    let frame = std::slice::from_raw_parts_mut(p_frame, frame_len as usize);
+    match db.conn.wal_get_frame(frame_no, frame) {
         Ok(c) => match db.io.wait_for_completion(c) {
             Ok(_) => SQLITE_OK,
             Err(_) => SQLITE_ERROR,