Merge 'core/incremental: Wrap ViewTransactionState in Arc' from Pekka Enberg

Make it Send.

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

Closes #3173

core/incremental/cursor.rs

@@ -9,7 +9,6 @@ use crate::{
     types::{IOResult, SeekKey, SeekOp, SeekResult, Value},
     LimboError, Pager, Result,
 };
-use std::rc::Rc;
 use std::sync::{Arc, Mutex};
 
 /// State machine for seek operations
@@ -44,7 +43,7 @@ pub struct MaterializedViewCursor {
     uncommitted: RowKeyZSet,
 
     // Reference to shared transaction state for this specific view - shared with Connection
-    tx_state: Rc<ViewTransactionState>,
+    tx_state: Arc<ViewTransactionState>,
 
     // The transaction state always grows. It never gets reduced. That is in the very nature of
     // DBSP, because deletions are just appends with weight < 0. So we will use the length of the
@@ -66,7 +65,7 @@ impl MaterializedViewCursor {
         btree_cursor: Box<BTreeCursor>,
         view: Arc<Mutex<IncrementalView>>,
         pager: Arc<Pager>,
-        tx_state: Rc<ViewTransactionState>,
+        tx_state: Arc<ViewTransactionState>,
     ) -> Result<Self> {
         Ok(Self {
             btree_cursor,
@@ -334,7 +333,11 @@ mod tests {
     /// Helper to create a test cursor for the materialized view
     fn create_test_cursor(
         conn: &Arc<Connection>,
-    ) -> Result<(MaterializedViewCursor, Rc<ViewTransactionState>, Arc<Pager>)> {
+    ) -> Result<(
+        MaterializedViewCursor,
+        Arc<ViewTransactionState>,
+        Arc<Pager>,
+    )> {
         // Get the schema and view
         let view_mutex = conn
             .schema

core/incremental/view.rs

@@ -108,7 +108,7 @@ impl ViewTransactionState {
 /// Provides interior mutability for the map of view states
 #[derive(Debug, Clone, Default)]
 pub struct AllViewsTxState {
-    states: Rc<RefCell<HashMap<String, Rc<ViewTransactionState>>>>,
+    states: Rc<RefCell<HashMap<String, Arc<ViewTransactionState>>>>,
 }
 
 impl AllViewsTxState {
@@ -120,16 +120,16 @@ impl AllViewsTxState {
     }
 
     /// Get or create a transaction state for a view
-    pub fn get_or_create(&self, view_name: &str) -> Rc<ViewTransactionState> {
+    pub fn get_or_create(&self, view_name: &str) -> Arc<ViewTransactionState> {
         let mut states = self.states.borrow_mut();
         states
             .entry(view_name.to_string())
-            .or_insert_with(|| Rc::new(ViewTransactionState::new()))
+            .or_insert_with(|| Arc::new(ViewTransactionState::new()))
             .clone()
     }
 
     /// Get a transaction state for a view if it exists
-    pub fn get(&self, view_name: &str) -> Option<Rc<ViewTransactionState>> {
+    pub fn get(&self, view_name: &str) -> Option<Arc<ViewTransactionState>> {
         self.states.borrow().get(view_name).cloned()
     }