Merge 'Add MVCC checkpoint threshold pragma' from bit-aloo

closes: #3575

Closes #3604

core/lib.rs

@@ -2360,6 +2360,23 @@ impl Connection {
     pub(crate) fn get_mv_tx(&self) -> Option<(u64, TransactionMode)> {
         *self.mv_tx.read()
     }
+
+    pub(crate) fn set_mvcc_checkpoint_threshold(&self, threshold: u64) -> Result<()> {
+        match self.db.mv_store.as_ref() {
+            Some(mv_store) => {
+                mv_store.set_checkpoint_threshold(threshold);
+                Ok(())
+            }
+            None => Err(LimboError::InternalError("MVCC not enabled".into())),
+        }
+    }
+
+    pub(crate) fn mvcc_checkpoint_threshold(&self) -> Result<u64> {
+        match self.db.mv_store.as_ref() {
+            Some(mv_store) => Ok(mv_store.checkpoint_threshold()),
+            None => Err(LimboError::InternalError("MVCC not enabled".into())),
+        }
+    }
 }
 
 #[derive(Debug)]

core/mvcc/database/mod.rs

@@ -2035,6 +2035,10 @@ impl<Clock: LogicalClock> MvStore<Clock> {
     pub fn set_checkpoint_threshold(&self, threshold: u64) {
         self.storage.set_checkpoint_threshold(threshold)
     }
+
+    pub fn checkpoint_threshold(&self) -> u64 {
+        self.storage.checkpoint_threshold()
+    }
 }
 
 /// A write-write conflict happens when transaction T_current attempts to update a

core/mvcc/persistent_storage/logical_log.rs

@@ -235,6 +235,10 @@ impl LogicalLog {
     pub fn set_checkpoint_threshold(&mut self, threshold: u64) {
         self.checkpoint_threshold = threshold;
     }
+
+    pub fn checkpoint_threshold(&self) -> u64 {
+        self.checkpoint_threshold
+    }
 }
 
 pub enum StreamingResult {

core/mvcc/persistent_storage/mod.rs

@@ -49,6 +49,10 @@ impl Storage {
             .unwrap()
             .set_checkpoint_threshold(threshold)
     }
+
+    pub fn checkpoint_threshold(&self) -> u64 {
+        self.logical_log.read().unwrap().checkpoint_threshold()
+    }
 }
 
 impl Debug for Storage {

core/pragma.rs

@@ -127,6 +127,10 @@ pub fn pragma_for(pragma: &PragmaName) -> Pragma {
             PragmaFlags::Result0 | PragmaFlags::SchemaReq | PragmaFlags::NoColumns1,
             &["cipher"],
         ),
+        PragmaName::MvccCheckpointThreshold => Pragma::new(
+            PragmaFlags::NoColumns1 | PragmaFlags::Result0,
+            &["mvcc_checkpoint_threshold"],
+        ),
     }
 }
 

core/translate/pragma.rs

@@ -378,6 +378,15 @@ fn update_pragma(
             connection.set_data_sync_retry(retry_enabled);
             Ok((program, TransactionMode::None))
         }
+        PragmaName::MvccCheckpointThreshold => {
+            let threshold = match parse_signed_number(&value)? {
+                Value::Integer(size) if size > 0 => size as u64,
+                _ => bail_parse_error!("mvcc_checkpoint_threshold must be a positive integer"),
+            };
+
+            connection.set_mvcc_checkpoint_threshold(threshold)?;
+            Ok((program, TransactionMode::None))
+        }
     }
 }
 
@@ -687,6 +696,14 @@ fn query_pragma(
             program.add_pragma_result_column(pragma.to_string());
             Ok((program, TransactionMode::None))
         }
+        PragmaName::MvccCheckpointThreshold => {
+            let threshold = connection.mvcc_checkpoint_threshold()?;
+            let register = program.alloc_register();
+            program.emit_int(threshold as i64, register);
+            program.emit_result_row(register, 1);
+            program.add_pragma_result_column(pragma.to_string());
+            Ok((program, TransactionMode::None))
+        }
     }
 }
 

parser/src/ast.rs

@@ -1449,6 +1449,8 @@ pub enum PragmaName {
     UserVersion,
     /// trigger a checkpoint to run on database(s) if WAL is enabled
     WalCheckpoint,
+    /// Sets or queries the threshold (in bytes) at which MVCC triggers an automatic checkpoint.
+    MvccCheckpointThreshold,
 }
 
 /// `CREATE TRIGGER` time