Merge 'clear page cache on transaction failure' from Pere Diaz Bou

This is the first step towards rollback, since we still don't spill
pages with WAL, we can simply invalidate page cache in case of failure.

Reviewed-by: Jussi Saurio <jussi.saurio@gmail.com>

Closes #1599

core/lib.rs

@@ -531,7 +531,7 @@ impl Connection {
     }
 
     pub fn checkpoint(&self) -> Result<CheckpointResult> {
-        let checkpoint_result = self.pager.clear_page_cache();
+        let checkpoint_result = self.pager.wal_checkpoint();
         Ok(checkpoint_result)
     }
 

core/storage/page_cache.rs

@@ -474,8 +474,6 @@ impl DumbLruPageCache {
         );
     }
 
-    #[cfg(test)]
-    /// For testing purposes, in case we use cursor api directly, we might want to unmark pages as dirty because we bypass the WAL transaction layer
     pub fn unset_dirty_all_pages(&mut self) {
         for node in self.map.borrow_mut().iter_mut() {
             unsafe {
@@ -572,7 +570,6 @@ impl PageHashMap {
         self.buckets.iter().flat_map(|bucket| bucket.iter())
     }
 
-    #[cfg(test)]
     pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut HashMapNode> {
         self.buckets.iter_mut().flat_map(|bucket| bucket.iter_mut())
     }

core/storage/pager.rs

@@ -547,8 +547,19 @@ impl Pager {
         }
     }
 
-    // WARN: used for testing purposes
-    pub fn clear_page_cache(&self) -> CheckpointResult {
+    /// Invalidates entire page cache by removing all dirty and clean pages. Usually used in case
+    /// of a rollback or in case we want to invalidate page cache after starting a read transaction
+    /// right after new writes happened which would invalidate current page cache.
+    pub fn clear_page_cache(&self) {
+        self.dirty_pages.borrow_mut().clear();
+        self.page_cache.write().unset_dirty_all_pages();
+        self.page_cache
+            .write()
+            .clear()
+            .expect("Failed to clear page cache");
+    }
+
+    pub fn wal_checkpoint(&self) -> CheckpointResult {
         let checkpoint_result: CheckpointResult;
         loop {
             match self.wal.borrow_mut().checkpoint(

core/vdbe/execute.rs

@@ -1641,6 +1641,10 @@ pub fn op_halt(
     else {
         unreachable!("unexpected Insn {:?}", insn)
     };
+    if *err_code > 0 {
+        // invalidate page cache in case of error
+        pager.clear_page_cache();
+    }
     match *err_code {
         0 => {}
         SQLITE_CONSTRAINT_PRIMARYKEY => {

tests/integration/query_processing/test_write_path.rs

@@ -269,7 +269,6 @@ fn test_wal_checkpoint() -> anyhow::Result<()> {
     }
 
     do_flush(&conn, &tmp_db)?;
-    conn.clear_page_cache()?;
     let list_query = "SELECT * FROM test LIMIT 1";
     let mut current_index = 0;
     run_query_on_row(&tmp_db, &conn, list_query, |row: &Row| {