Merge 'Request load page on insert_into_page' from Pere Diaz Bou

We assumed page was loaded because before inserting we would move there.
`NewRowId` unfortunately moves cursor to the rightmost page causing
eviction of root page -- this arose the issue with `insert_into_page`
not loading the page we were supposed to have loaded so I added
`return_if_locked_maybe_load` which is a utility macro to check if the
page is locked and if not, load it if needed.

Closes #1138

core/storage/btree.rs

@@ -14,7 +14,6 @@ use crate::{return_corrupt, LimboError, Result};
 use std::cell::{Cell, Ref, RefCell};
 use std::pin::Pin;
 use std::rc::Rc;
-use std::sync::Arc;
 
 use super::pager::PageRef;
 use super::sqlite3_ondisk::{
@@ -85,6 +84,18 @@ macro_rules! debug_validate_cells {
         }
     };
 }
+/// Check if the page is unlocked, if not return IO. If the page is not locked but not loaded, then try to load it.
+macro_rules! return_if_locked_maybe_load {
+    ($pager:expr, $expr:expr) => {{
+        if $expr.is_locked() {
+            return Ok(CursorResult::IO);
+        }
+        if !$expr.is_loaded() {
+            $pager.load_page($expr.clone())?;
+            return Ok(CursorResult::IO);
+        }
+    }};
+}
 
 /// State machine of destroy operations
 /// Keep track of traversal so that it can be resumed when IO is encountered
@@ -853,6 +864,7 @@ impl BTreeCursor {
 
     /// Move the cursor to the root page of the btree.
     fn move_to_root(&mut self) {
+        tracing::trace!("move_to_root({})", self.root_page);
         let mem_page = self.pager.read_page(self.root_page).unwrap();
         self.stack.clear();
         self.stack.push(mem_page);
@@ -1056,6 +1068,7 @@ impl BTreeCursor {
             match write_state {
                 WriteState::Start => {
                     let page = self.stack.top();
+                    return_if_locked_maybe_load!(self.pager, page);
                     let int_key = match key {
                         OwnedValue::Integer(i) => *i as u64,
                         _ => unreachable!("btree tables are indexed by integers!"),
@@ -1219,11 +1232,7 @@ impl BTreeCursor {
                 if parent_page.is_locked() {
                     return Ok(CursorResult::IO);
                 }
-                return_if_locked!(parent_page);
-                if !parent_page.is_loaded() {
-                    self.pager.load_page(parent_page.clone())?;
-                    return Ok(CursorResult::IO);
-                }
+                return_if_locked_maybe_load!(self.pager, parent_page);
                 parent_page.set_dirty();
                 self.pager.add_dirty(parent_page.get().id);
                 let parent_contents = parent_page.get().contents.as_ref().unwrap();
@@ -2095,12 +2104,7 @@ impl BTreeCursor {
 
                 DeleteState::LoadPage => {
                     let page = self.stack.top();
-                    return_if_locked!(page);
-
-                    if !page.is_loaded() {
-                        self.pager.load_page(page.clone())?;
-                        return Ok(CursorResult::IO);
-                    }
+                    return_if_locked_maybe_load!(self.pager, page);
 
                     let delete_info = self.state.mut_delete_info().unwrap();
                     delete_info.state = DeleteState::FindCell;
@@ -2499,12 +2503,7 @@ impl BTreeCursor {
                 }
                 DestroyState::LoadPage => {
                     let page = self.stack.top();
-                    return_if_locked!(page);
-
-                    if !page.is_loaded() {
-                        self.pager.load_page(Arc::clone(&page))?;
-                        return Ok(CursorResult::IO);
-                    }
+                    return_if_locked_maybe_load!(self.pager, page);
 
                     let destroy_info = self
                         .state

core/storage/page_cache.rs

@@ -188,6 +188,7 @@ impl DumbLruPageCache {
             // TODO: drop from another clean entry?
             return;
         }
+        tracing::debug!("pop_if_not_dirty(key={:?})", tail_entry.key);
         self.detach(tail, true);
         assert!(self.map.borrow_mut().remove(&tail_entry.key).is_some());
     }

core/storage/pager.rs

@@ -102,10 +102,12 @@ impl Page {
     }
 
     pub fn set_dirty(&self) {
+        tracing::debug!("set_dirty(page={})", self.get().id);
         self.get().flags.fetch_or(PAGE_DIRTY, Ordering::SeqCst);
     }
 
     pub fn clear_dirty(&self) {
+        tracing::debug!("clear_dirty(page={})", self.get().id);
         self.get().flags.fetch_and(!PAGE_DIRTY, Ordering::SeqCst);
     }