diff --git a/Cargo.lock b/Cargo.lock
index 587f70a4..150e0332 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -395,6 +395,7 @@ dependencies = [
  "redis-module",
  "redis-module-macros",
  "roaring",
+ "rustc-hash",
  "ryu",
  "thin-vec",
 ]
@@ -532,6 +533,7 @@ dependencies = [
  "regex",
  "roaring",
  "rquickjs",
+ "rustc-hash",
  "thin-vec",
  "ureq",
 ]
diff --git a/Cargo.toml b/Cargo.toml
index 4c8c49c8..827481ec 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,6 +29,7 @@ lazy_static = "1.5.0"
 parking_lot = "0.12.5"
 redis-module = { git = "https://github.com/AviAvni/redismodule-rs", branch = "master" }
 roaring = "0.11.3"
+rustc-hash = "2"
 ryu = "1.0.23"
 thin-vec = "0.2.16"
 orx-tree = "2.2.0"
diff --git a/flow_tests_done.txt b/flow_tests_done.txt
index 5e4371e7..6cff3231 100644
--- a/flow_tests_done.txt
+++ b/flow_tests_done.txt
@@ -14,7 +14,9 @@ tests/flow/test_concurrent_query.py
 tests/flow/test_config.py
 tests/flow/test_create_clause
 tests/flow/test_distinct
+tests/flow/test_effects.py
 tests/flow/test_empty_query
+tests/flow/test_encode_decode.py
 tests/flow/test_entity_update
 tests/flow/test_execution_plan_print.py
 tests/flow/test_expand_into
@@ -24,6 +26,7 @@ tests/flow/test_function_calls
 tests/flow/test_graph_create
 tests/flow/test_graph_deletion
 tests/flow/test_graph_merge
+tests/flow/test_graph_versioning.py
 tests/flow/test_hashjoin.py
 tests/flow/test_imdb
 tests/flow/test_index_create
@@ -56,6 +59,7 @@ tests/flow/test_pending_queries_limit.py
 tests/flow/test_point
 tests/flow/test_profile.py
 tests/flow/test_query_validation
+tests/flow/test_rdb_load.py
 tests/flow/test_reduce.py
 tests/flow/test_results.py
 tests/flow/test_reversed_patterns
diff --git a/flow_tests_todo.txt b/flow_tests_todo.txt
index f32352d8..d7275dd8 100644
--- a/flow_tests_todo.txt
+++ b/flow_tests_todo.txt
@@ -16,12 +16,8 @@ tests/flow/test_stress.py
 
 ## Persistence & Replication
 tests/flow/test_persistency.py
-tests/flow/test_encode_decode.py
-tests/flow/test_rdb_load.py
-tests/flow/test_prev_rdb_decode.py
 tests/flow/test_replication.py
 tests/flow/test_replication_states.py
-tests/flow/test_graph_versioning.py
 
 ## Redis Integration & Server Features
 tests/flow/test_acl.py
@@ -42,5 +38,4 @@ tests/flow/test_undo_log.py
 
 ## Metadata & Internals
 tests/flow/test_multi_label.py
-tests/flow/test_effects.py
-tests/flow/test_intern_string.py
+tests/flow/test_intern_string.py
\ No newline at end of file
diff --git a/graph/Cargo.toml b/graph/Cargo.toml
index d1c57fcc..f8c7cf27 100644
--- a/graph/Cargo.toml
+++ b/graph/Cargo.toml
@@ -35,4 +35,5 @@ regex = "1.12.3"
 rquickjs = { version = "0.11", features = ["bindgen", "classes", "macro", "parallel"] }
 ureq = { version = "3.3.0", default-features = false, features = ["rustls"] }
 roaring = "0.11.3"
+rustc-hash = "2"
 thin-vec = "0.2.16"
diff --git a/graph/src/graph/attribute_store.rs b/graph/src/graph/attribute_store.rs
index 718099b5..2e02e433 100644
--- a/graph/src/graph/attribute_store.rs
+++ b/graph/src/graph/attribute_store.rs
@@ -83,10 +83,11 @@
 //! Each attribute is stored as a separate fjall entry:
 //! `entity_id (8 bytes big-endian) + attr_idx (2 bytes big-endian)`
 
-use std::{collections::HashMap, process, sync::Arc};
+use std::{process, sync::Arc};
 
 use std::cmp::Ordering;
-use std::collections::HashMap as StdHashMap;
+
+use rustc_hash::FxHashMap;
 
 use fjall::{
     Database, Keyspace, KeyspaceCreateOptions, Readable, Snapshot, config::HashRatioPolicy,
@@ -98,6 +99,11 @@ use super::attribute_cache::AttributeCache;
 use super::graphblas::serialization::{Decode, Encode, Reader, Writer};
 use crate::runtime::{ordermap::OrderMap, orderset::OrderSet, value::Value};
 
+/// Shared empty attribute vector to avoid per-entity allocations when an
+/// entity has no properties.
+static EMPTY_ATTRS: once_cell::sync::Lazy<Arc<Vec<(u16, Value)>>> =
+    once_cell::sync::Lazy::new(|| Arc::new(Vec::new()));
+
 /// Create a composite key from entity ID and attribute index.
 fn make_key(
     entity_id: u64,
@@ -268,14 +274,13 @@ impl AttributeStore {
     ) -> Arc<Vec<(u16, Value)>> {
         // If this entity is pending full deletion, return empty regardless of fjall state.
         if self.pending_deletes.contains(entity_id) {
-            return Arc::new(Vec::new());
+            return EMPTY_ATTRS.clone();
         }
-        // Fast path: if the fjall keyspace was never initialized, no data has
-        // ever been flushed from cache to fjall — all entity data is either in
-        // the cache (which already returned None to the caller) or doesn't
-        // exist. Skip the expensive keyspace creation + prefix scan.
+        // If the fjall keyspace was never initialized, no data was ever flushed
+        // to persistent storage. All live data is in the cache. Return empty
+        // without triggering expensive keyspace creation or cache writes.
         if self.keyspace.get().is_none() {
-            return Arc::new(Vec::new());
+            return EMPTY_ATTRS.clone();
         }
         let prefix = entity_id.to_be_bytes();
         let attrs: Vec<(u16, Value)> = self
@@ -296,6 +301,77 @@ impl AttributeStore {
         Arc::new(attrs)
     }
 
+    /// Returns `true` if this store has a fjall keyspace that might contain
+    /// cold data not present in cache.  When `false`, all data is in cache
+    /// and the fork child can safely read from cache without touching fjall.
+    pub fn has_fjall_data(&self) -> bool {
+        self.keyspace.get().is_some()
+    }
+
+    /// Build a complete snapshot of all entity attributes by merging cache and
+    /// fjall data.  Returns a map from entity-ID to its attribute list.
+    ///
+    /// Called **before** Redis forks for BGSAVE so the fork child can encode
+    /// entities without touching fjall (which is unsafe after fork).
+    /// The returned map is passed to [`encode_with_range`] as the data source.
+    pub fn build_rdb_snapshot(
+        &self,
+        deleted: &RoaringTreemap,
+        max_id: u64,
+    ) -> FxHashMap<u64, Arc<Vec<(u16, Value)>>> {
+        let mut snap: FxHashMap<u64, Arc<Vec<(u16, Value)>>> = FxHashMap::default();
+
+        // 1. Collect everything from fjall (cold store) in a single sequential scan.
+        if self.keyspace.get().is_some() {
+            let mut current_id: Option<u64> = None;
+            let mut current_attrs: Vec<(u16, Value)> = Vec::new();
+            for entry in self.snapshot().iter(self.keyspace()) {
+                let Ok((key, data)) = entry.into_inner() else {
+                    continue;
+                };
+                let Some(attr_idx) = extract_attr_idx(&key) else {
+                    continue;
+                };
+                let eid = u64::from_be_bytes(key[..8].try_into().unwrap());
+                let Some((value, _)) = Value::from_bytes(&data) else {
+                    continue;
+                };
+                if current_id != Some(eid) {
+                    if let Some(prev_id) = current_id {
+                        if !deleted.contains(prev_id) && !self.pending_deletes.contains(prev_id) {
+                            current_attrs.sort_by_key(|item| item.0);
+                            snap.insert(prev_id, Arc::new(std::mem::take(&mut current_attrs)));
+                        } else {
+                            current_attrs.clear();
+                        }
+                    }
+                    current_id = Some(eid);
+                }
+                current_attrs.push((attr_idx, value));
+            }
+            if let Some(prev_id) = current_id
+                && !deleted.contains(prev_id)
+                && !self.pending_deletes.contains(prev_id)
+            {
+                current_attrs.sort_by_key(|item| item.0);
+                snap.insert(prev_id, Arc::new(current_attrs));
+            }
+        }
+
+        // 2. Overlay cache entries (hot store) — cache wins over fjall because
+        //    it may contain newer dirty writes not yet flushed.
+        for id in 0..=max_id {
+            if deleted.contains(id) || self.pending_deletes.contains(id) {
+                continue;
+            }
+            if let Some(cached) = self.cache.get_entity(id, self.version) {
+                snap.insert(id, cached);
+            }
+        }
+
+        snap
+    }
+
     // ---- read path (cache → fjall) --------------------------------------
 
     pub fn remove(
@@ -470,24 +546,23 @@ impl AttributeStore {
     /// the number of attributes *replaced* and the number of non-null attributes *set*.
     pub fn insert_attrs(
         &mut self,
-        attrs: &HashMap<u64, OrderMap<Arc<String>, Value>>,
+        attrs: &FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     ) -> Result<(usize, usize), String> {
         let mut nremoved = 0;
         let mut nset = 0;
 
         // Pre-resolve all unique attribute names → indices ONCE.
         // Uses Arc pointer identity as key to avoid rehashing strings.
-        let mut name_to_idx: StdHashMap<*const String, u16> =
-            StdHashMap::with_capacity(attrs.values().next().map_or(0, |v| v.len()));
+        let mut name_to_idx: FxHashMap<*const String, u16> = FxHashMap::default();
         for entity_attrs in attrs.values() {
             for (attr, _) in entity_attrs.iter() {
                 let ptr = Arc::as_ptr(attr);
-                if !name_to_idx.contains_key(&ptr) {
+                if let std::collections::hash_map::Entry::Vacant(e) = name_to_idx.entry(ptr) {
                     let idx = self.attrs_name.get_index_of(attr).unwrap_or_else(|| {
                         self.attrs_name.insert(attr.clone());
                         self.attrs_name.len() - 1
                     }) as u16;
-                    name_to_idx.insert(ptr, idx);
+                    e.insert(idx);
                 }
             }
         }
@@ -528,7 +603,7 @@ impl AttributeStore {
                 merged.clear();
                 merged.reserve(current.len() + new_entries.len());
                 merged.extend_from_slice(&current);
-                merged.extend(new_entries.drain(..));
+                merged.append(&mut new_entries);
             } else {
                 null_indices.sort_unstable();
 
@@ -600,20 +675,19 @@ impl AttributeStore {
     /// Returns the number of non-null attributes imported.
     pub fn import_attrs(
         &mut self,
-        attrs: &HashMap<u64, OrderMap<Arc<String>, Value>>,
+        attrs: &FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     ) -> usize {
         // Pre-resolve all unique attribute names → indices ONCE.
-        let mut name_to_idx: StdHashMap<*const String, u16> =
-            StdHashMap::with_capacity(attrs.values().next().map_or(0, |v| v.len()));
+        let mut name_to_idx: FxHashMap<*const String, u16> = FxHashMap::default();
         for entity_attrs in attrs.values() {
             for (attr, _) in entity_attrs.iter() {
                 let ptr = Arc::as_ptr(attr);
-                if !name_to_idx.contains_key(&ptr) {
+                if let std::collections::hash_map::Entry::Vacant(e) = name_to_idx.entry(ptr) {
                     let idx = self.attrs_name.get_index_of(attr).unwrap_or_else(|| {
                         self.attrs_name.insert(attr.clone());
                         self.attrs_name.len() - 1
                     }) as u16;
-                    name_to_idx.insert(ptr, idx);
+                    e.insert(idx);
                 }
             }
         }
@@ -745,39 +819,6 @@ impl AttributeStore {
         Ok(())
     }
 
-    /// Flush an entity's pending dirty attributes to fjall, then invalidate from cache.
-    ///
-    /// This ensures that any unflushed writes to the cache are persisted to fjall
-    /// before the cache entry is removed, preventing data loss when the entry is
-    /// about to be deleted from fjall.
-    ///
-    /// However, if the entity was modified by the current transaction
-    /// (`dirty_entities`), the flush is skipped — those writes are uncommitted
-    /// and must not be persisted to fjall until `commit()`.  This prevents
-    /// rollback from leaving current-tx inserts in the durable store.
-    fn flush_and_invalidate(
-        &self,
-        entity_id: u64,
-    ) -> Result<(), String> {
-        if !self.dirty_entities.contains(entity_id)
-            && let Some((cached, dirty)) = self.cache.get_entity_with_dirty(entity_id, self.version)
-            && dirty
-            && !cached.is_empty()
-        {
-            // Write dirty cached attributes to fjall before losing the cache entry.
-            // Safe to flush: these are pre-existing dirty entries from prior
-            // transactions, not from the active one.
-            let mut batch = get_database().batch();
-            for &(attr_idx, ref value) in cached.iter() {
-                let composite_key = make_key(entity_id, attr_idx);
-                batch.insert(self.keyspace(), composite_key, value.to_bytes());
-            }
-            batch.durability(None).commit().map_err(|e| e.to_string())?;
-        }
-        self.cache.invalidate(entity_id);
-        Ok(())
-    }
-
     /// Access the shared cache (for background flush scheduling).
     #[must_use]
     pub const fn cache(&self) -> &Arc<AttributeCache> {
@@ -820,14 +861,14 @@ impl AttributeStore {
             if current_id != Some(eid) {
                 // Flush previous entity.
                 if let Some(prev_id) = current_id {
-                    if !self.pending_deletes.contains(prev_id) {
+                    if self.pending_deletes.contains(prev_id) {
+                        current_attrs.clear();
+                    } else {
                         let _ = self.cache.insert_entity_if_older(
                             prev_id,
                             std::mem::take(&mut current_attrs),
                             self.version,
                         );
-                    } else {
-                        current_attrs.clear();
                     }
                 }
                 current_id = Some(eid);
@@ -835,16 +876,18 @@ impl AttributeStore {
             current_attrs.push((idx, value));
         }
         // Flush the last entity.
-        if let Some(prev_id) = current_id {
-            if !self.pending_deletes.contains(prev_id) {
-                let _ = self
-                    .cache
-                    .insert_entity_if_older(prev_id, current_attrs, self.version);
-            }
+        if let Some(prev_id) = current_id
+            && !self.pending_deletes.contains(prev_id)
+        {
+            let _ = self
+                .cache
+                .insert_entity_if_older(prev_id, current_attrs, self.version);
         }
     }
 
-    /// Encode a range of entities, borrowing the deleted bitmap directly.
+    /// Encode a range of entities, using the pre-built RDB snapshot when
+    /// provided, falling back to cache → fjall otherwise.
+    #[allow(clippy::too_many_arguments)]
     pub fn encode_with_range(
         &self,
         w: &mut dyn Writer,
@@ -853,9 +896,10 @@ impl AttributeStore {
         global_attrs: &[Arc<String>],
         count: u64,
         offset: u64,
+        rdb_snapshot: Option<&FxHashMap<u64, Arc<Vec<(u16, Value)>>>>,
     ) {
         // Build attr remap inline.
-        let global_index: std::collections::HashMap<&Arc<String>, usize> = global_attrs
+        let global_index: FxHashMap<&Arc<String>, usize> = global_attrs
             .iter()
             .enumerate()
             .map(|(i, n)| (n, i))
@@ -888,7 +932,14 @@ impl AttributeStore {
 
             w.write_unsigned(id);
 
-            let props = self.get_all_attrs_by_id(id);
+            let props = rdb_snapshot.map_or_else(
+                || self.get_all_attrs_by_id(id),
+                |snap| {
+                    snap.get(&id)
+                        .cloned()
+                        .unwrap_or_else(|| EMPTY_ATTRS.clone())
+                },
+            );
             w.write_unsigned(props.len() as u64);
 
             for &(local_attr_id, ref value) in props.iter() {
diff --git a/graph/src/graph/graph.rs b/graph/src/graph/graph.rs
index d32ca62d..97e83359 100644
--- a/graph/src/graph/graph.rs
+++ b/graph/src/graph/graph.rs
@@ -72,6 +72,8 @@ use std::{
     time::{Duration, Instant},
 };
 
+use rustc_hash::FxHashMap;
+
 use atomic_refcell::AtomicRefCell;
 use itertools::Itertools;
 use lru::LruCache;
@@ -207,6 +209,13 @@ pub struct MemoryUsageReport {
     pub indices_sz: usize,
 }
 
+/// Pre-built attribute snapshots for RDB save.
+/// Built before Redis forks so the child never accesses fjall.
+pub struct RdbSnapshots {
+    pub nodes: FxHashMap<u64, Arc<Vec<(u16, Value)>>>,
+    pub relationships: FxHashMap<u64, Arc<Vec<(u16, Value)>>>,
+}
+
 /// The main graph data structure.
 ///
 /// Stores nodes, relationships, labels, and properties using sparse matrices
@@ -981,8 +990,8 @@ impl Graph {
 
     pub fn set_nodes_attributes(
         &mut self,
-        attrs: &HashMap<u64, OrderMap<Arc<String>, Value>>,
-        index_add_docs: &mut HashMap<u64, RoaringTreemap>,
+        attrs: &FxHashMap<u64, OrderMap<Arc<String>, Value>>,
+        index_add_docs: &mut FxHashMap<u64, RoaringTreemap>,
     ) -> Result<(usize, usize), String> {
         let (nremoved, nset) = self.node_attrs.insert_attrs(attrs)?;
 
@@ -1003,8 +1012,8 @@ impl Graph {
 
     pub fn import_node_attrs(
         &mut self,
-        attrs: &HashMap<u64, OrderMap<Arc<String>, Value>>,
-        index_add_docs: &mut HashMap<u64, RoaringTreemap>,
+        attrs: &FxHashMap<u64, OrderMap<Arc<String>, Value>>,
+        index_add_docs: &mut FxHashMap<u64, RoaringTreemap>,
     ) -> usize {
         let nset = self.node_attrs.import_attrs(attrs);
 
@@ -1025,7 +1034,7 @@ impl Graph {
 
     pub fn import_relationship_attrs(
         &mut self,
-        attrs: &HashMap<u64, OrderMap<Arc<String>, Value>>,
+        attrs: &FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     ) -> usize {
         self.relationship_attrs.import_attrs(attrs)
     }
@@ -1033,7 +1042,7 @@ impl Graph {
     pub fn set_nodes_labels(
         &mut self,
         nodes_labels: &mut Matrix,
-        index_add_docs: &mut HashMap<u64, RoaringTreemap>,
+        index_add_docs: &mut FxHashMap<u64, RoaringTreemap>,
     ) {
         self.resize();
 
@@ -1064,7 +1073,7 @@ impl Graph {
     pub fn remove_nodes_labels(
         &mut self,
         nodes_labels: &mut Matrix,
-        remove_docs: &mut HashMap<u64, RoaringTreemap>,
+        remove_docs: &mut FxHashMap<u64, RoaringTreemap>,
     ) {
         self.resize();
 
@@ -1081,7 +1090,7 @@ impl Graph {
     pub fn delete_nodes(
         &mut self,
         deleted_nodes: &RoaringTreemap,
-        remove_docs: &mut HashMap<u64, RoaringTreemap>,
+        remove_docs: &mut FxHashMap<u64, RoaringTreemap>,
     ) -> Result<(), String> {
         self.deleted_nodes |= deleted_nodes;
         self.node_count -= deleted_nodes.len();
@@ -1375,7 +1384,7 @@ impl Graph {
 
     pub fn create_relationships(
         &mut self,
-        relationships: &HashMap<RelationshipId, PendingRelationship>,
+        relationships: &FxHashMap<RelationshipId, PendingRelationship>,
     ) {
         self.relationship_count += relationships.len() as u64;
         self.reserved_relationship_count -= relationships.len() as u64;
@@ -1400,11 +1409,10 @@ impl Graph {
         }
 
         // Pre-resolve type names → (matrix index, type_id) ONCE.
-        let mut type_cache: std::collections::HashMap<*const String, (usize, u64)> =
-            std::collections::HashMap::new();
+        let mut type_cache: FxHashMap<*const String, (usize, u64)> = FxHashMap::default();
         for rel in relationships.values() {
             let ptr = Arc::as_ptr(&rel.type_name);
-            if !type_cache.contains_key(&ptr) {
+            if let std::collections::hash_map::Entry::Vacant(e) = type_cache.entry(ptr) {
                 // Ensure the type + matrix exist
                 self.get_relationship_matrix_mut(&rel.type_name);
                 let type_idx = self
@@ -1412,15 +1420,14 @@ impl Graph {
                     .iter()
                     .position(|t| t.as_str() == rel.type_name.as_str())
                     .unwrap();
-                type_cache.insert(ptr, (type_idx, type_idx as u64));
+                e.insert((type_idx, type_idx as u64));
             }
         }
 
         self.resize();
 
         // Collect entries per-tensor, plus adjacency and type matrix entries
-        let mut by_tensor: std::collections::HashMap<usize, Vec<(u64, u64, u64)>> =
-            std::collections::HashMap::new();
+        let mut by_tensor: FxHashMap<usize, Vec<(u64, u64, u64)>> = FxHashMap::default();
         let mut adj_entries: Vec<(u64, u64)> = Vec::with_capacity(relationships.len());
         let mut type_entries: Vec<(u64, u64)> = Vec::with_capacity(relationships.len());
 
@@ -1445,7 +1452,7 @@ impl Graph {
 
     pub fn set_relationships_attributes(
         &mut self,
-        attrs: &HashMap<u64, OrderMap<Arc<String>, Value>>,
+        attrs: &FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     ) -> Result<(usize, usize), String> {
         let (nremoved, nset) = self.relationship_attrs.insert_attrs(attrs)?;
         Ok((nremoved, nset))
@@ -1499,7 +1506,7 @@ impl Graph {
 
     pub fn delete_relationships(
         &mut self,
-        rels: HashMap<RelationshipId, (NodeId, NodeId)>,
+        rels: &FxHashMap<RelationshipId, (NodeId, NodeId)>,
     ) -> Result<(), String> {
         self.deleted_relationships
             .extend(rels.keys().map(|id| id.0));
@@ -1510,7 +1517,7 @@ impl Graph {
             rels.values().map(|(src, dst)| (src.0, dst.0)).collect();
 
         for (type_id, rels) in &rels
-            .into_iter()
+            .iter()
             .map(|(id, (src, dst))| (id.0, src.0, dst.0))
             .into_group_map_by(|(id, _, _)| self.get_relationship_type_id(RelationshipId(*id)))
         {
@@ -1551,7 +1558,7 @@ impl Graph {
     pub fn delete_implicit_edges(
         &mut self,
         deleted_nodes: &RoaringTreemap,
-        explicit_rels: &HashMap<RelationshipId, (NodeId, NodeId)>,
+        explicit_rels: &FxHashMap<RelationshipId, (NodeId, NodeId)>,
     ) -> Result<Vec<(RelationshipId, NodeId, NodeId)>, String> {
         if self.relationship_matrices.is_empty() {
             return Ok(Vec::new());
@@ -1559,7 +1566,8 @@ impl Graph {
 
         let mut all_implicit: Vec<(RelationshipId, NodeId, NodeId)> = Vec::new();
         // Pairs where only one endpoint is deleted — need adjacency check
-        let mut check_adj_pairs: std::collections::HashSet<(u64, u64)> = Default::default();
+        let mut check_adj_pairs: std::collections::HashSet<(u64, u64)> =
+            std::collections::HashSet::default();
 
         for type_idx in 0..self.relationship_matrices.len() {
             let mut rels: Vec<(u64, u64, u64)> = Vec::new();
@@ -1989,10 +1997,30 @@ impl Graph {
         Ok(())
     }
 
+    /// Returns `true` if any attribute store has cold data in fjall that
+    /// would be unsafe to read from a fork child.
+    pub fn needs_rdb_snapshot(&self) -> bool {
+        self.node_attrs.has_fjall_data() || self.relationship_attrs.has_fjall_data()
+    }
+
+    /// Pre-populate attribute caches from fjall for RDB save.
+    pub fn build_rdb_snapshots(&self) -> RdbSnapshots {
+        let node_snap = self
+            .node_attrs
+            .build_rdb_snapshot(&self.deleted_nodes, self.max_node_id());
+        let rel_snap = self
+            .relationship_attrs
+            .build_rdb_snapshot(&self.deleted_relationships, self.max_relationship_id());
+        RdbSnapshots {
+            nodes: node_snap,
+            relationships: rel_snap,
+        }
+    }
+
     pub fn commit_index(
         &mut self,
-        index_add_docs: &mut HashMap<u64, RoaringTreemap>,
-        remove_docs: &mut HashMap<u64, RoaringTreemap>,
+        index_add_docs: &mut FxHashMap<u64, RoaringTreemap>,
+        remove_docs: &mut FxHashMap<u64, RoaringTreemap>,
     ) {
         let lock = self.node_indexer.write_lock();
         let _guard = lock.lock();
@@ -2347,6 +2375,7 @@ impl Graph {
         w: &mut dyn Writer,
         p: &PayloadEntry,
         global_attrs: &[Arc<String>],
+        snapshots: Option<&RdbSnapshots>,
     ) {
         match p.state {
             EncodeState::Nodes => {
@@ -2357,6 +2386,7 @@ impl Graph {
                     global_attrs,
                     p.count,
                     p.offset,
+                    snapshots.map(|s| &s.nodes),
                 );
             }
             EncodeState::DeletedNodes => {
@@ -2370,6 +2400,7 @@ impl Graph {
                     global_attrs,
                     p.count,
                     p.offset,
+                    snapshots.map(|s| &s.relationships),
                 );
             }
             EncodeState::DeletedEdges => {
diff --git a/graph/src/graph/graphblas/matrix.rs b/graph/src/graph/graphblas/matrix.rs
index b0afdbb8..4521db8c 100644
--- a/graph/src/graph/graphblas/matrix.rs
+++ b/graph/src/graph/graphblas/matrix.rs
@@ -80,17 +80,18 @@ use super::{
     GrB_DESC_RT0T1, GrB_DESC_RT1, GrB_DESC_S, GrB_DESC_SC, GrB_DESC_SCT0, GrB_DESC_SCT0T1,
     GrB_DESC_SCT1, GrB_DESC_ST0, GrB_DESC_ST0T1, GrB_DESC_ST1, GrB_DESC_T0, GrB_DESC_T0T1,
     GrB_DESC_T1, GrB_Descriptor, GrB_GLOBAL, GrB_Global_set_INT32, GrB_Info, GrB_Matrix,
-    GrB_Matrix_build_BOOL, GrB_Matrix_clear, GrB_Matrix_dup, GrB_Matrix_eWiseAdd_Semiring,
-    GrB_Matrix_eWiseMult_Semiring, GrB_Matrix_extractElement_BOOL,
-    GrB_Matrix_extractElement_UINT64, GrB_Matrix_free, GrB_Matrix_get_INT32, GrB_Matrix_ncols,
-    GrB_Matrix_new, GrB_Matrix_nrows, GrB_Matrix_nvals, GrB_Matrix_removeElement,
-    GrB_Matrix_resize, GrB_Matrix_setElement_BOOL, GrB_Matrix_setElement_UINT64, GrB_Matrix_wait,
-    GrB_Mode, GrB_UINT64, GrB_WaitMode, GrB_finalize, GrB_mxm, GrB_transpose, GxB_ANY_BOOL,
-    GxB_ANY_PAIR_BOOL, GxB_Container_free, GxB_Container_new, GxB_Iterator, GxB_Iterator_free,
-    GxB_Iterator_new, GxB_Matrix_fprint, GxB_Matrix_memoryUsage, GxB_Matrix_type, GxB_Option_Field,
-    GxB_Print_Level, GxB_init, GxB_load_Matrix_from_Container, GxB_rowIterator_attach,
-    GxB_rowIterator_getColIndex, GxB_rowIterator_getRowIndex, GxB_rowIterator_nextCol,
-    GxB_rowIterator_nextRow, GxB_rowIterator_seekRow, GxB_unload_Matrix_into_Container,
+    GrB_Matrix_build_BOOL, GrB_Matrix_build_UINT64, GrB_Matrix_clear, GrB_Matrix_dup,
+    GrB_Matrix_eWiseAdd_Semiring, GrB_Matrix_eWiseMult_Semiring, GrB_Matrix_extractElement_BOOL,
+    GrB_Matrix_extractElement_UINT64, GrB_Matrix_extractTuples_BOOL, GrB_Matrix_free,
+    GrB_Matrix_get_INT32, GrB_Matrix_ncols, GrB_Matrix_new, GrB_Matrix_nrows, GrB_Matrix_nvals,
+    GrB_Matrix_removeElement, GrB_Matrix_resize, GrB_Matrix_setElement_BOOL,
+    GrB_Matrix_setElement_UINT64, GrB_Matrix_wait, GrB_Mode, GrB_UINT64, GrB_WaitMode,
+    GrB_finalize, GrB_mxm, GrB_transpose, GxB_ANY_BOOL, GxB_ANY_PAIR_BOOL, GxB_ANY_UINT64,
+    GxB_Container_free, GxB_Container_new, GxB_Iterator, GxB_Iterator_free, GxB_Iterator_new,
+    GxB_Matrix_fprint, GxB_Matrix_memoryUsage, GxB_Matrix_type, GxB_Option_Field, GxB_Print_Level,
+    GxB_init, GxB_load_Matrix_from_Container, GxB_rowIterator_attach, GxB_rowIterator_getColIndex,
+    GxB_rowIterator_getRowIndex, GxB_rowIterator_nextCol, GxB_rowIterator_nextRow,
+    GxB_rowIterator_seekRow, GxB_unload_Matrix_into_Container,
 };
 
 /// Initializes the GraphBLAS library in non-blocking mode.
@@ -840,6 +841,57 @@ impl Matrix {
             debug_assert_eq!(info, GrB_Info::GrB_SUCCESS);
         }
     }
+
+    /// Bulk-insert UINT64 entries from (row, col, val) arrays. Matrix must be empty and UINT64 typed.
+    pub fn build_uint64(
+        &mut self,
+        rows: &[u64],
+        cols: &[u64],
+        vals: &[u64],
+    ) {
+        debug_assert_eq!(rows.len(), cols.len());
+        debug_assert_eq!(rows.len(), vals.len());
+        if rows.is_empty() {
+            return;
+        }
+        let nvals = rows.len() as u64;
+        unsafe {
+            let info = GrB_Matrix_build_UINT64(
+                *self.m,
+                rows.as_ptr(),
+                cols.as_ptr(),
+                vals.as_ptr(),
+                nvals,
+                GxB_ANY_UINT64,
+            );
+            debug_assert_eq!(info, GrB_Info::GrB_SUCCESS);
+        }
+    }
+
+    /// Bulk-extract all (row, col) entries from a boolean matrix.
+    #[must_use]
+    pub fn extract_tuples_bool(&self) -> (Vec<u64>, Vec<u64>) {
+        let mut nvals = self.nvals();
+        if nvals == 0 {
+            return (Vec::new(), Vec::new());
+        }
+        let mut rows = vec![0u64; nvals as usize];
+        let mut cols = vec![0u64; nvals as usize];
+        let mut vals = vec![false; nvals as usize];
+        unsafe {
+            let info = GrB_Matrix_extractTuples_BOOL(
+                rows.as_mut_ptr(),
+                cols.as_mut_ptr(),
+                vals.as_mut_ptr(),
+                &raw mut nvals,
+                *self.m,
+            );
+            debug_assert_eq!(info, GrB_Info::GrB_SUCCESS);
+        }
+        rows.truncate(nvals as usize);
+        cols.truncate(nvals as usize);
+        (rows, cols)
+    }
 }
 
 impl Transpose for Matrix
diff --git a/graph/src/graph/graphblas/tensor.rs b/graph/src/graph/graphblas/tensor.rs
index e8e7c38b..102fb06d 100644
--- a/graph/src/graph/graphblas/tensor.rs
+++ b/graph/src/graph/graphblas/tensor.rs
@@ -55,6 +55,8 @@
 //! For example, two "TRANSFERRED" relationships between the same bank accounts
 //! with different amounts and dates.
 
+use std::collections::HashMap;
+
 use super::{
     matrix::{Dup, Matrix, New, Remove, Set, Size, Transpose},
     serialization::{Decode, Encode, Reader, Writer},
@@ -246,72 +248,131 @@ impl Tensor {
 const MSB_MASK: u64 = 1u64 << 63;
 
 impl Encode<19> for Tensor {
+    #[allow(clippy::similar_names)]
     fn encode(
         &self,
         w: &mut dyn Writer,
     ) {
-        // Build a UINT64 forward matrix for C compatibility.
-        // Single-edge (src,dst): cell = edge_id
-        // Multi-edge (src,dst): cell = edge_count | MSB_MASK
-        let (m, dp) = self.m.extract_m_dp();
-
-        let mut uint64_m = Matrix::new_uint64(m.nrows(), m.ncols());
-        let mut uint64_dp = Matrix::new_uint64(dp.nrows(), dp.ncols());
-        // Track multi-edge (src, dst) pairs per sub-matrix for tensor section
-        let mut multi_edge_m: Vec<(u64, u64)> = Vec::new();
-        let mut multi_edge_dp: Vec<(u64, u64)> = Vec::new();
-
-        for (matrix, uint64_matrix, multi_edges) in [
-            (&m, &mut uint64_m, &mut multi_edge_m),
-            (&dp, &mut uint64_dp, &mut multi_edge_dp),
-        ] {
-            for (src, dst) in matrix.iter(0, u64::MAX) {
-                let compound_key = (src << 32) | dst;
-                let mut edge_ids: Vec<u64> = self
-                    .me
-                    .iter(compound_key, compound_key)
-                    .map(|(_, edge_id)| edge_id)
-                    .collect();
-
-                if edge_ids.len() == 1 {
-                    // Single edge: store edge ID directly
-                    uint64_matrix.set_uint64(src, dst, edge_ids[0]);
+        let nrows = self.m.nrows();
+        let ncols = self.m.ncols();
+
+        // Compute both nvals at once to avoid redundant wait()/nvals() FFI calls.
+        let m_nvals = self.m.nvals();
+        let me_nvals = self.me.nvals();
+        let has_multi = m_nvals != me_nvals;
+        let total = me_nvals;
+
+        // Reusable empty UINT64 matrix — shared across the 2 empty slots.
+        let empty = Matrix::new_uint64(nrows, ncols);
+
+        let (multi_edge_m, multi_edge_dp, edge_id_map) = if has_multi {
+            let (m, dp) = self.m.extract_m_dp();
+            let (me_rows, me_cols) = self.me.extract_all_tuples();
+            let mut edge_id_map: HashMap<u64, Vec<u64>> = HashMap::with_capacity(me_rows.len());
+            for i in 0..me_rows.len() {
+                edge_id_map.entry(me_rows[i]).or_default().push(me_cols[i]);
+            }
+
+            let mut multi_edge_m: Vec<(u64, u64)> = Vec::new();
+            let mut multi_edge_dp: Vec<(u64, u64)> = Vec::new();
+
+            for (matrix, multi_edges) in [(&m, &mut multi_edge_m), (&dp, &mut multi_edge_dp)] {
+                let (rows, cols) = matrix.extract_tuples_bool();
+                let mut u_rows = Vec::with_capacity(rows.len());
+                let mut u_cols = Vec::with_capacity(rows.len());
+                let mut u_vals = Vec::with_capacity(rows.len());
+
+                for i in 0..rows.len() {
+                    let src = rows[i];
+                    let dst = cols[i];
+                    let compound_key = (src << 32) | dst;
+                    let edge_ids = edge_id_map
+                        .get(&compound_key)
+                        .map_or(&[][..], |v| v.as_slice());
+
+                    u_rows.push(src);
+                    u_cols.push(dst);
+                    if edge_ids.len() == 1 {
+                        u_vals.push(edge_ids[0]);
+                    } else {
+                        u_vals.push(edge_ids.len() as u64 | MSB_MASK);
+                        multi_edges.push((src, dst));
+                    }
+                }
+
+                if u_rows.is_empty() {
+                    empty.encode(w);
                 } else {
-                    // Multi-edge: store count with MSB set
-                    uint64_matrix.set_uint64(src, dst, edge_ids.len() as u64 | MSB_MASK);
-                    multi_edges.push((src, dst));
+                    let mut uint64_mat = Matrix::new_uint64(nrows, ncols);
+                    uint64_mat.build_uint64(&u_rows, &u_cols, &u_vals);
+                    uint64_mat.encode(w);
                 }
             }
-        }
 
-        // Encode the UINT64 forward matrix (as a VersionedMatrix: m, dp, dm)
-        let dm = Matrix::new_uint64(m.nrows(), m.ncols()); // empty delta-minus
-        uint64_m.encode(w);
-        uint64_dp.encode(w);
-        dm.encode(w);
+            empty.encode(w);
+
+            (multi_edge_m, multi_edge_dp, Some(edge_id_map))
+        } else {
+            // Fast path: no multi-edges, no extract_m_dp() needed.
+            let ((me_m_rows, me_m_cols), (me_dp_rows, me_dp_cols)) = self.me.extract_m_dp_tuples();
+
+            // Encode uint64_m
+            if me_m_rows.is_empty() {
+                empty.encode(w);
+            } else {
+                let m_len = me_m_rows.len();
+                let mut m_src = Vec::with_capacity(m_len);
+                let mut m_dst = Vec::with_capacity(m_len);
+                for &row in &me_m_rows {
+                    m_src.push(row >> 32);
+                    m_dst.push(row & 0xFFFF_FFFF);
+                }
+                let mut uint64_m = Matrix::new_uint64(nrows, ncols);
+                uint64_m.build_uint64(&m_src, &m_dst, &me_m_cols);
+                uint64_m.encode(w);
+            }
+
+            // Encode uint64_dp
+            if me_dp_rows.is_empty() {
+                empty.encode(w);
+            } else {
+                let dp_len = me_dp_rows.len();
+                let mut dp_src = Vec::with_capacity(dp_len);
+                let mut dp_dst = Vec::with_capacity(dp_len);
+                for &row in &me_dp_rows {
+                    dp_src.push(row >> 32);
+                    dp_dst.push(row & 0xFFFF_FFFF);
+                }
+                let mut uint64_dp = Matrix::new_uint64(nrows, ncols);
+                uint64_dp.build_uint64(&dp_src, &dp_dst, &me_dp_cols);
+                uint64_dp.encode(w);
+            }
+
+            // Empty delta-minus
+            empty.encode(w);
+
+            (Vec::new(), Vec::new(), None)
+        };
 
-        let total = self.edge_count();
         w.write_unsigned(total);
 
         if total == 0 {
             return;
         }
 
-        // Tensor section: only multi-edge pairs
         let mut v = Vector::<u64>::new(GrB_INDEX_MAX);
-        for (multi_edges, _matrix) in [(&multi_edge_m, &m), (&multi_edge_dp, &dp)] {
+        for multi_edges in [&multi_edge_m, &multi_edge_dp] {
             w.write_unsigned(multi_edges.len() as u64);
             for &(src, dst) in multi_edges {
                 let compound_key = (src << 32) | dst;
                 v.clear();
 
-                for (idx, edge_id) in self
-                    .me
-                    .iter(compound_key, compound_key)
-                    .map(|(_, edge_id)| edge_id)
-                    .enumerate()
+                if let Some(ref map) = edge_id_map
+                    && let Some(ids) = map.get(&compound_key)
                 {
-                    v.set(idx as u64, edge_id);
+                    for (idx, &edge_id) in ids.iter().enumerate() {
+                        v.set(idx as u64, edge_id);
+                    }
                 }
 
                 w.write_unsigned(src);
diff --git a/graph/src/graph/graphblas/versioned_matrix.rs b/graph/src/graph/graphblas/versioned_matrix.rs
index f0d03783..4133c85e 100644
--- a/graph/src/graph/graphblas/versioned_matrix.rs
+++ b/graph/src/graph/graphblas/versioned_matrix.rs
@@ -186,13 +186,50 @@ impl VersionedMatrix {
 
     #[must_use]
     pub fn extract_m_dp(&self) -> (Matrix, Matrix) {
-        let mut m = Matrix::new(self.m.nrows(), self.m.ncols());
-        let mut dp = Matrix::new(self.dp.nrows(), self.dp.ncols());
+        if self.dm.nvals() == 0 {
+            // Fast path: no deletions, return dups of m and dp directly
+            (self.m.dup(), self.dp.dup())
+        } else {
+            let mut m = Matrix::new(self.m.nrows(), self.m.ncols());
+            let mut dp = Matrix::new(self.dp.nrows(), self.dp.ncols());
+            m.select(&self.dm, &self.m);
+            dp.select(&self.dm, &self.dp);
+            (m, dp)
+        }
+    }
 
-        m.select(&self.dm, &self.m);
-        dp.select(&self.dm, &self.dp);
+    /// Bulk-extract all effective entries as (row, col) arrays.
+    ///
+    /// Returns `(rows, cols)` from `(m - dm) ∪ dp`, avoiding iterator overhead
+    /// on matrices with huge dimensions (e.g., GrB_INDEX_MAX).
+    #[must_use]
+    pub fn extract_all_tuples(&self) -> (Vec<u64>, Vec<u64>) {
+        self.wait();
+        if self.dm.nvals() == 0 {
+            // Fast path: no deletions, just combine m and dp tuples
+            let (mut rows_m, mut cols_m) = self.m.extract_tuples_bool();
+            let (rows_dp, cols_dp) = self.dp.extract_tuples_bool();
+            rows_m.extend_from_slice(&rows_dp);
+            cols_m.extend_from_slice(&cols_dp);
+            (rows_m, cols_m)
+        } else {
+            // Slow path: materialize effective matrix then extract
+            let effective = self.to_matrix();
+            effective.extract_tuples_bool()
+        }
+    }
 
-        (m, dp)
+    /// Bulk-extract tuples from base `m` and delta-plus `dp` separately.
+    ///
+    /// Returns `((m_rows, m_cols), (dp_rows, dp_cols))`.
+    /// Only valid when `dm` is empty (asserted in debug builds).
+    #[must_use]
+    pub fn extract_m_dp_tuples(&self) -> ((Vec<u64>, Vec<u64>), (Vec<u64>, Vec<u64>)) {
+        self.wait();
+        debug_assert_eq!(self.dm.nvals(), 0, "extract_m_dp_tuples requires empty dm");
+        let m_tuples = self.m.extract_tuples_bool();
+        let dp_tuples = self.dp.extract_tuples_bool();
+        (m_tuples, dp_tuples)
     }
 
     /// Bulk-remove all entries matching a mask matrix.
diff --git a/graph/src/index/mod.rs b/graph/src/index/mod.rs
index a81fbcef..7369c4db 100644
--- a/graph/src/index/mod.rs
+++ b/graph/src/index/mod.rs
@@ -381,7 +381,7 @@ pub struct Document {
     id: u64,
     /// CStrings for array string elements. RediSearch stores raw pointers
     /// into these during `set`, so they must live until after `add_document`.
-    _string_arr_values: Vec<CString>,
+    string_arr_values: Vec<CString>,
 }
 
 impl Document {
@@ -389,7 +389,7 @@ impl Document {
     pub fn new(id: u64) -> Self {
         Self {
             id,
-            _string_arr_values: Vec::new(),
+            string_arr_values: Vec::new(),
             rs_doc: unsafe {
                 let doc = RediSearch_CreateDocument2(
                     (&raw const id).cast::<c_void>(),
@@ -536,7 +536,7 @@ impl Document {
                         // Keep string content CStrings alive — the pointer
                         // array in RediSearch references them. They'll be
                         // properly freed when the Document is dropped.
-                        self._string_arr_values.extend(string_cstrs);
+                        self.string_arr_values.extend(string_cstrs);
                     }
                 }
                 Value::VecF32(_) => {} // Only for vector fields
@@ -782,19 +782,19 @@ impl Index {
     fn build_numeric_range_node(
         &self,
         key: &Arc<String>,
-        min: Option<Value>,
-        max: Option<Value>,
+        min: Option<&Value>,
+        max: Option<&Value>,
         include_min: bool,
         include_max: bool,
     ) -> *mut redisearch::RSQNode {
-        let min_f = match &min {
+        let min_f = match min {
             Some(v) => match Self::value_to_numeric(v) {
                 Some(f) => f,
                 None => return std::ptr::null_mut(),
             },
             None => RSRANGE_NEG_INF,
         };
-        let max_f = match &max {
+        let max_f = match max {
             Some(v) => match Self::value_to_numeric(v) {
                 Some(f) => f,
                 None => return std::ptr::null_mut(),
@@ -926,8 +926,8 @@ impl Index {
                     // Numeric range (Int, Float, Bool)
                     self.build_numeric_range_node(
                         key,
-                        min.clone(),
-                        max.clone(),
+                        min.as_ref(),
+                        max.as_ref(),
                         include_min,
                         include_max,
                     )
diff --git a/graph/src/planner/optimizer/mod.rs b/graph/src/planner/optimizer/mod.rs
index b93cd203..cb8a0960 100644
--- a/graph/src/planner/optimizer/mod.rs
+++ b/graph/src/planner/optimizer/mod.rs
@@ -110,6 +110,7 @@ pub(crate) fn collect_subtree_variables(node: &orx_tree::DynNode<IR>) -> HashSet
 /// # Returns
 /// An optimized copy of the plan
 #[must_use]
+#[allow(clippy::implicit_hasher)]
 pub fn optimize(
     plan: &DynTree<IR>,
     graph: &Graph,
diff --git a/graph/src/planner/optimizer/push_filters_down.rs b/graph/src/planner/optimizer/push_filters_down.rs
index 7ae7057a..93a1b7e7 100644
--- a/graph/src/planner/optimizer/push_filters_down.rs
+++ b/graph/src/planner/optimizer/push_filters_down.rs
@@ -250,54 +250,54 @@ pub(super) fn push_filters_down(optimized_plan: &mut DynTree<IR>) {
             // When a conjunct references variables from a proper subset of CP
             // branches (>=2 but < all), extract those branches into an inner
             // CP wrapped with that conjunct as a filter.
-            if !remaining.is_empty() && child_conjuncts.iter().all(Vec::is_empty) {
-                if let Some(filter_child) = optimized_plan.node(idx).get_child(0)
-                    && matches!(filter_child.data(), IR::CartesianProduct)
-                    && filter_child.num_children() > 2
-                {
-                    let total_children = children.len();
-                    let mut split_idx = None;
-
-                    for (ci, conjunct) in remaining.iter().enumerate() {
-                        let conj_vars = collect_expr_variables(conjunct);
-                        // Find solving branches: children that contribute >=1
-                        // variable referenced by this conjunct.
-                        let solving: Vec<usize> = children
-                            .iter()
-                            .enumerate()
-                            .filter(|(_, (_, child_vars))| {
-                                conj_vars.iter().any(|v| child_vars.contains(v))
-                            })
-                            .map(|(i, _)| i)
-                            .collect();
-
-                        if solving.len() >= 2 && solving.len() < total_children {
-                            split_idx = Some((ci, solving));
-                            break;
-                        }
-                    }
-
-                    if let Some((ci, solving)) = split_idx {
-                        // Rebuild the entire plan using a recursive approach
-                        // to avoid in-place tree mutation issues with prune.
-                        let split_filter_idx = idx;
-                        let conjunct_to_split = remaining[ci].clone();
-                        let solving_set: HashSet<usize> = solving.iter().copied().collect();
-                        let mut remaining_conjuncts = remaining.clone();
-                        remaining_conjuncts.remove(ci);
-
-                        *optimized_plan = rebuild_with_cp_split(
-                            optimized_plan,
-                            split_filter_idx,
-                            &conjunct_to_split,
-                            &solving_set,
-                            &remaining_conjuncts,
-                        );
-
-                        changed = true;
+            if !remaining.is_empty()
+                && child_conjuncts.iter().all(Vec::is_empty)
+                && let Some(filter_child) = optimized_plan.node(idx).get_child(0)
+                && matches!(filter_child.data(), IR::CartesianProduct)
+                && filter_child.num_children() > 2
+            {
+                let total_children = children.len();
+                let mut split_idx = None;
+
+                for (ci, conjunct) in remaining.iter().enumerate() {
+                    let conj_vars = collect_expr_variables(conjunct);
+                    // Find solving branches: children that contribute >=1
+                    // variable referenced by this conjunct.
+                    let solving: Vec<usize> = children
+                        .iter()
+                        .enumerate()
+                        .filter(|(_, (_, child_vars))| {
+                            conj_vars.iter().any(|v| child_vars.contains(v))
+                        })
+                        .map(|(i, _)| i)
+                        .collect();
+
+                    if solving.len() >= 2 && solving.len() < total_children {
+                        split_idx = Some((ci, solving));
                         break;
                     }
                 }
+
+                if let Some((ci, solving)) = split_idx {
+                    // Rebuild the entire plan using a recursive approach
+                    // to avoid in-place tree mutation issues with prune.
+                    let split_filter_idx = idx;
+                    let conjunct_to_split = remaining[ci].clone();
+                    let solving_set: HashSet<usize> = solving.iter().copied().collect();
+                    let mut remaining_conjuncts = remaining.clone();
+                    remaining_conjuncts.remove(ci);
+
+                    *optimized_plan = rebuild_with_cp_split(
+                        optimized_plan,
+                        split_filter_idx,
+                        &conjunct_to_split,
+                        &solving_set,
+                        &remaining_conjuncts,
+                    );
+
+                    changed = true;
+                    break;
+                }
             }
 
             // Skip if nothing can be pushed down
@@ -390,7 +390,7 @@ fn rebuild_with_cp_split(
 
             let inner_cp = tree!(IR::CartesianProduct; extracted);
             let filter_expr = Arc::new(conjunct.clone());
-            let inner_filtered = tree!(IR::Filter(filter_expr); vec![inner_cp]);
+            let inner_filtered = tree!(IR::Filter(filter_expr); [inner_cp]);
 
             let mut new_children = others;
             new_children.push(inner_filtered);
@@ -401,6 +401,7 @@ fn rebuild_with_cp_split(
                 let remaining_filter = if remaining_conjuncts.len() == 1 {
                     Arc::new(remaining_conjuncts[0].clone())
                 } else {
+                    #[allow(clippy::unnecessary_to_owned)]
                     Arc::new(tree!(ExprIR::And; remaining_conjuncts.to_vec()))
                 };
                 let cp = tree!(IR::CartesianProduct; new_children);
diff --git a/graph/src/runtime/ops/cond_traverse.rs b/graph/src/runtime/ops/cond_traverse.rs
index 157e1e3e..11ccc78c 100644
--- a/graph/src/runtime/ops/cond_traverse.rs
+++ b/graph/src/runtime/ops/cond_traverse.rs
@@ -82,6 +82,7 @@ pub struct CondTraverseOp<'a> {
 }
 
 impl<'a> CondTraverseOp<'a> {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         runtime: &'a Runtime<'a>,
         child: Box<BatchOp<'a>>,
@@ -427,10 +428,10 @@ impl<'a> Iterator for CondTraverseOp<'a> {
 
     fn next(&mut self) -> Option<Self::Item> {
         // Check if record_cap already reached.
-        if let Some(cap) = self.record_cap {
-            if self.produced >= cap {
-                return None;
-            }
+        if let Some(cap) = self.record_cap
+            && self.produced >= cap
+        {
+            return None;
         }
 
         let mut envs = Vec::with_capacity(BATCH_SIZE);
diff --git a/graph/src/runtime/ops/create.rs b/graph/src/runtime/ops/create.rs
index 82fd0b62..bbb69a69 100644
--- a/graph/src/runtime/ops/create.rs
+++ b/graph/src/runtime/ops/create.rs
@@ -117,7 +117,7 @@ impl Runtime<'_> {
             // NOTE: eval() may borrow pending internally (e.g. property reads),
             // so we cannot hold pending.borrow_mut() across eval calls.
             let mut all_attrs: Vec<OrderMap<Arc<String>, Value>> = Vec::with_capacity(active_len);
-            for (_i, row) in batch.active_indices().enumerate() {
+            for row in batch.active_indices() {
                 let env = batch.env_ref(row);
                 let attrs = ExprEval::from_runtime(self).eval(
                     &node.attrs,
@@ -216,7 +216,7 @@ impl Runtime<'_> {
             // Same as nodes: eval() may borrow pending, so separate eval from insert.
             let mut all_rel_attrs: Vec<OrderMap<Arc<String>, Value>> =
                 Vec::with_capacity(rel_ids.len());
-            for (_i, row) in batch.active_indices().enumerate() {
+            for row in batch.active_indices() {
                 let env = batch.env_ref(row);
                 let attrs = ExprEval::from_runtime(self).eval(
                     &rel.attrs,
diff --git a/graph/src/runtime/ops/delete.rs b/graph/src/runtime/ops/delete.rs
index ee45e671..00072ebd 100644
--- a/graph/src/runtime/ops/delete.rs
+++ b/graph/src/runtime/ops/delete.rs
@@ -141,7 +141,8 @@ impl Runtime<'_> {
         for &id in node_ids {
             if self.pending.borrow().is_node_deleted(id) {
                 continue;
-            } else if self.pending.borrow().is_node_created(id) {
+            }
+            if self.pending.borrow().is_node_created(id) {
                 // Created in this txn — use existing per-node path
                 self.delete_entity(&Value::Node(id))?;
             } else if !self.g.borrow().is_node_deleted(id) {
diff --git a/graph/src/runtime/ops/expand_into.rs b/graph/src/runtime/ops/expand_into.rs
index 84c41c00..8839dbed 100644
--- a/graph/src/runtime/ops/expand_into.rs
+++ b/graph/src/runtime/ops/expand_into.rs
@@ -213,10 +213,10 @@ impl<'a> Iterator for ExpandIntoOp<'a> {
 
     fn next(&mut self) -> Option<Self::Item> {
         // Check if record_cap already reached.
-        if let Some(cap) = self.record_cap {
-            if self.produced >= cap {
-                return None;
-            }
+        if let Some(cap) = self.record_cap
+            && self.produced >= cap
+        {
+            return None;
         }
 
         let mut envs = Vec::with_capacity(BATCH_SIZE);
diff --git a/graph/src/runtime/ops/node_by_index_scan.rs b/graph/src/runtime/ops/node_by_index_scan.rs
index 2b82e2c6..a0628308 100644
--- a/graph/src/runtime/ops/node_by_index_scan.rs
+++ b/graph/src/runtime/ops/node_by_index_scan.rs
@@ -235,7 +235,7 @@ impl<'a> NodeByIndexScanOp<'a> {
         match q {
             IndexQuery::Equal { value, .. } => is_indexable(value),
             IndexQuery::Range { min, max, .. } => {
-                min.as_ref().is_none_or(is_indexable) && max.as_ref().map_or(true, is_indexable)
+                min.as_ref().is_none_or(is_indexable) && max.as_ref().is_none_or(is_indexable)
             }
             IndexQuery::And(children) | IndexQuery::Or(children) => {
                 children.iter().all(Self::can_utilize_index)
diff --git a/graph/src/runtime/ops/set.rs b/graph/src/runtime/ops/set.rs
index 99764d5f..59b86137 100644
--- a/graph/src/runtime/ops/set.rs
+++ b/graph/src/runtime/ops/set.rs
@@ -85,7 +85,7 @@ impl Runtime<'_> {
     ) -> Result<(), String> {
         // Pre-check: if no nodes/relationships have been deleted in this
         // transaction, we can skip the per-row deletion checks entirely.
-        let has_deleted_nodes = self.deleted_nodes.borrow().len() > 0;
+        let has_deleted_nodes = !self.deleted_nodes.borrow().is_empty();
         let has_pending_deleted_nodes = self.pending.borrow().has_deleted_nodes();
         let has_pending_deleted_rels = self.pending.borrow().has_deleted_relationships();
         let skip_delete_checks =
diff --git a/graph/src/runtime/pending.rs b/graph/src/runtime/pending.rs
index 3b66c257..d3af0dbc 100644
--- a/graph/src/runtime/pending.rs
+++ b/graph/src/runtime/pending.rs
@@ -24,7 +24,9 @@
 //!
 //! On error or ROLLBACK, the Pending is simply dropped without applying.
 
-use std::{cell::RefCell, collections::HashMap, sync::Arc};
+use std::{cell::RefCell, sync::Arc};
+
+use rustc_hash::FxHashMap;
 
 use atomic_refcell::AtomicRefCell;
 use roaring::RoaringTreemap;
@@ -107,27 +109,27 @@ pub struct Pending {
     /// Nodes created in this transaction
     created_nodes: RoaringTreemap,
     /// Relationships created (id → pending relationship data)
-    created_relationships: HashMap<RelationshipId, PendingRelationship>,
+    created_relationships: FxHashMap<RelationshipId, PendingRelationship>,
     /// Nodes to be deleted
     deleted_nodes: RoaringTreemap,
     /// Relationships to be deleted (edge_id, src, dst)
-    deleted_relationships: HashMap<RelationshipId, (NodeId, NodeId)>,
+    deleted_relationships: FxHashMap<RelationshipId, (NodeId, NodeId)>,
     /// Property updates for newly created nodes (fast path: skip fjall)
-    new_nodes_attrs: HashMap<u64, OrderMap<Arc<String>, Value>>,
+    new_nodes_attrs: FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     /// Property updates for existing nodes (full merge path)
-    existing_nodes_attrs: HashMap<u64, OrderMap<Arc<String>, Value>>,
+    existing_nodes_attrs: FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     /// Property updates for newly created relationships (fast path: skip fjall)
-    new_relationships_attrs: HashMap<u64, OrderMap<Arc<String>, Value>>,
+    new_relationships_attrs: FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     /// Property updates for existing relationships (full merge path)
-    existing_relationships_attrs: HashMap<u64, OrderMap<Arc<String>, Value>>,
+    existing_relationships_attrs: FxHashMap<u64, OrderMap<Arc<String>, Value>>,
     /// Labels to add (node_id × label_id matrix)
     set_node_labels: Matrix,
     /// Labels to remove
     remove_node_labels: Matrix,
     /// Documents to add to indexes (keyed by label id)
-    index_add_docs: HashMap<u64, RoaringTreemap>,
+    index_add_docs: FxHashMap<u64, RoaringTreemap>,
     /// Documents to remove from indexes (keyed by label id)
-    index_remove_docs: HashMap<u64, RoaringTreemap>,
+    index_remove_docs: FxHashMap<u64, RoaringTreemap>,
     /// Schema baseline: number of labels when the current commit window started.
     schema_label_count: usize,
     /// Schema baseline: number of relationship types when the current commit window started.
@@ -149,17 +151,17 @@ impl Pending {
     pub fn new() -> Self {
         Self {
             created_nodes: RoaringTreemap::new(),
-            created_relationships: HashMap::new(),
+            created_relationships: FxHashMap::default(),
             deleted_nodes: RoaringTreemap::new(),
-            deleted_relationships: HashMap::new(),
-            new_nodes_attrs: HashMap::new(),
-            existing_nodes_attrs: HashMap::new(),
-            new_relationships_attrs: HashMap::new(),
-            existing_relationships_attrs: HashMap::new(),
+            deleted_relationships: FxHashMap::default(),
+            new_nodes_attrs: FxHashMap::default(),
+            existing_nodes_attrs: FxHashMap::default(),
+            new_relationships_attrs: FxHashMap::default(),
+            existing_relationships_attrs: FxHashMap::default(),
             set_node_labels: Matrix::new(0, 0),
             remove_node_labels: Matrix::new(0, 0),
-            index_add_docs: HashMap::new(),
-            index_remove_docs: HashMap::new(),
+            index_add_docs: FxHashMap::default(),
+            index_remove_docs: FxHashMap::default(),
             schema_label_count: 0,
             schema_rel_type_count: 0,
             schema_node_attr_count: 0,
@@ -741,7 +743,10 @@ impl Pending {
         }
         if !explicit_rels.is_empty() {
             stats.borrow_mut().relationships_deleted += explicit_rels.len();
-            g.borrow_mut().delete_relationships(explicit_rels)?;
+            // Re-record explicit rels so the effects buffer can serialize them.
+            self.deleted_relationships
+                .extend(explicit_rels.iter().map(|(&k, &v)| (k, v)));
+            g.borrow_mut().delete_relationships(&explicit_rels)?;
         }
         // Commit attribute changes and indexes after all deletions have been
         // applied. This ensures relationship_attrs.remove() pending_deletes
@@ -944,7 +949,7 @@ impl Pending {
         {
             let nrows = self.set_node_labels.nrows();
             if nrows > 0 {
-                let mut label_map: HashMap<u64, Vec<u64>> = HashMap::new();
+                let mut label_map: FxHashMap<u64, Vec<u64>> = FxHashMap::default();
                 for (node_id, label_id) in self.set_node_labels.iter(0, nrows - 1) {
                     if !self.created_nodes.contains(node_id) {
                         label_map.entry(node_id).or_default().push(label_id);
@@ -968,7 +973,7 @@ impl Pending {
         {
             let nrows = self.remove_node_labels.nrows();
             if nrows > 0 {
-                let mut label_map: HashMap<u64, Vec<u64>> = HashMap::new();
+                let mut label_map: FxHashMap<u64, Vec<u64>> = FxHashMap::default();
                 for (node_id, label_id) in self.remove_node_labels.iter(0, nrows - 1) {
                     label_map.entry(node_id).or_default().push(label_id);
                 }
diff --git a/graph/src/runtime/value.rs b/graph/src/runtime/value.rs
index d5fbf8d6..521b1a33 100644
--- a/graph/src/runtime/value.rs
+++ b/graph/src/runtime/value.rs
@@ -1342,7 +1342,7 @@ impl DisplayJson for Value {
                     if i > 0 {
                         write!(f, ",")?;
                     }
-                    write_json_string(f, &k)?;
+                    write_json_string(f, k)?;
                     write!(f, ":")?;
                     v.fmt_json(f, runtime)?;
                 }
@@ -1448,7 +1448,7 @@ fn write_node_json(
         if i > 0 {
             write!(f, ",")?;
         }
-        write_json_string(f, &k)?;
+        write_json_string(f, k)?;
         write!(f, ":")?;
         v.fmt_json(f, runtime)?;
     }
diff --git a/src/commands/debug.rs b/src/commands/debug.rs
new file mode 100644
index 00000000..08539697
--- /dev/null
+++ b/src/commands/debug.rs
@@ -0,0 +1,41 @@
+use crate::redis_type::{create_virtual_keys, delete_stale_virtual_keys, finalize_pending_graphs};
+use crate::serializers::DECODE_STATE;
+use redis_module::{Context, NextArg, RedisError, RedisResult, RedisString, RedisValue};
+
+pub fn graph_debug(
+    ctx: &Context,
+    args: Vec<RedisString>,
+) -> RedisResult {
+    if args.len() < 3 {
+        return Err(RedisError::WrongArity);
+    }
+    let mut args_iter = args.into_iter().skip(1);
+    let subcmd = args_iter.next_str()?;
+
+    match subcmd.to_uppercase().as_str() {
+        "AUX" => debug_aux(ctx, args_iter),
+        _ => Err(RedisError::String(format!(
+            "Unknown DEBUG subcommand: {subcmd}"
+        ))),
+    }
+}
+
+fn debug_aux(
+    ctx: &Context,
+    mut args: impl Iterator<Item = RedisString>,
+) -> RedisResult {
+    let action = args.next_str()?;
+    match action.to_uppercase().as_str() {
+        "START" => {
+            DECODE_STATE.lock().clear();
+            unsafe { create_virtual_keys(ctx.ctx) };
+            Ok(RedisValue::Integer(1))
+        }
+        "END" => {
+            finalize_pending_graphs();
+            unsafe { delete_stale_virtual_keys(ctx.ctx) };
+            Ok(RedisValue::Integer(0))
+        }
+        _ => Err(RedisError::String(format!("Unknown AUX action: {action}"))),
+    }
+}
diff --git a/src/commands/effect.rs b/src/commands/effect.rs
index 45af3dcf..9e3466de 100644
--- a/src/commands/effect.rs
+++ b/src/commands/effect.rs
@@ -32,7 +32,8 @@ use graph::{
 use parking_lot::RwLock;
 use redis_module::{Context, NextArg, RedisResult, RedisString, RedisValue};
 use roaring::RoaringTreemap;
-use std::{collections::HashMap, sync::Arc};
+use rustc_hash::FxHashMap;
+use std::sync::Arc;
 
 pub fn graph_effect(
     ctx: &Context,
@@ -107,8 +108,8 @@ fn apply_effects(
         return Err(format!("unsupported effects version: {version}"));
     }
 
-    let mut index_add_docs: HashMap<u64, RoaringTreemap> = HashMap::new();
-    let mut index_remove_docs: HashMap<u64, RoaringTreemap> = HashMap::new();
+    let mut index_add_docs: FxHashMap<u64, RoaringTreemap> = FxHashMap::default();
+    let mut index_remove_docs: FxHashMap<u64, RoaringTreemap> = FxHashMap::default();
     let mut has_index_ops = false;
 
     while offset < buf.len() {
@@ -143,7 +144,7 @@ fn apply_effects(
                 let attr_count = read_u16(buf, &mut offset)?;
                 if attr_count > 0 {
                     let attrs = read_attrs(buf, &mut offset, attr_count)?;
-                    let mut attr_map = HashMap::new();
+                    let mut attr_map = FxHashMap::default();
                     attr_map.insert(node_id_raw, attrs);
                     g.set_nodes_attributes(&attr_map, &mut index_add_docs)?;
                 }
@@ -159,7 +160,7 @@ fn apply_effects(
 
                 let pending_rel =
                     PendingRelationship::new(NodeId::from(src_id), NodeId::from(dst_id), type_name);
-                let mut rels = HashMap::new();
+                let mut rels = FxHashMap::default();
                 rels.insert(RelationshipId::from(rel_id_raw), pending_rel);
                 g.create_relationships(&rels);
 
@@ -167,7 +168,7 @@ fn apply_effects(
                 let attr_count = read_u16(buf, &mut offset)?;
                 if attr_count > 0 {
                     let attrs = read_attrs(buf, &mut offset, attr_count)?;
-                    let mut attr_map = HashMap::new();
+                    let mut attr_map = FxHashMap::default();
                     attr_map.insert(rel_id_raw, attrs);
                     g.set_relationships_attributes(&attr_map)?;
                 }
@@ -177,7 +178,7 @@ fn apply_effects(
                 let node_id = read_u64(buf, &mut offset)?;
                 let attr_count = read_u16(buf, &mut offset)?;
                 let attrs = read_attrs(buf, &mut offset, attr_count)?;
-                let mut attr_map = HashMap::new();
+                let mut attr_map = FxHashMap::default();
                 attr_map.insert(node_id, attrs);
                 g.set_nodes_attributes(&attr_map, &mut index_add_docs)?;
             }
@@ -186,7 +187,7 @@ fn apply_effects(
                 let rel_id = read_u64(buf, &mut offset)?;
                 let attr_count = read_u16(buf, &mut offset)?;
                 let attrs = read_attrs(buf, &mut offset, attr_count)?;
-                let mut attr_map = HashMap::new();
+                let mut attr_map = FxHashMap::default();
                 attr_map.insert(rel_id, attrs);
                 g.set_relationships_attributes(&attr_map)?;
             }
@@ -228,9 +229,9 @@ fn apply_effects(
                 let src_id = read_u64(buf, &mut offset)?;
                 let dst_id = read_u64(buf, &mut offset)?;
                 let rel = RelationshipId::from(rel_id);
-                let mut rels = HashMap::new();
+                let mut rels = FxHashMap::default();
                 rels.insert(rel, (NodeId::from(src_id), NodeId::from(dst_id)));
-                g.delete_relationships(rels)?;
+                g.delete_relationships(&rels)?;
             }
 
             EFFECT_ADD_SCHEMA => {
diff --git a/src/commands/mod.rs b/src/commands/mod.rs
index 3a659a92..a375462c 100644
--- a/src/commands/mod.rs
+++ b/src/commands/mod.rs
@@ -20,6 +20,7 @@
 use redis_module::{RedisError, RedisResult};
 
 pub mod config_cmd;
+pub mod debug;
 pub mod delete;
 pub mod effect;
 pub mod explain;
@@ -32,6 +33,7 @@ pub mod ro_query;
 pub mod udf;
 
 pub use config_cmd::graph_config;
+pub use debug::graph_debug;
 pub use delete::graph_delete;
 pub use effect::graph_effect;
 pub use explain::graph_explain;
diff --git a/src/commands/profile.rs b/src/commands/profile.rs
index 5d09f686..e27be253 100644
--- a/src/commands/profile.rs
+++ b/src/commands/profile.rs
@@ -27,7 +27,7 @@ pub fn graph_profile(
     if let Some(graph) = read_key.get_value::<Arc<RwLock<ThreadedGraph>>>(&GRAPH_TYPE)? {
         let graph = graph.clone();
         drop(read_key);
-        return profile_mut(ctx, &graph, query, key_name);
+        return profile_mut(ctx, &graph, query, &key_name);
     }
 
     // Graph doesn't exist - open writable key to create it.
@@ -35,14 +35,14 @@ pub fn graph_profile(
     let key = ctx.open_key_writable(&key_str);
     if let Some(graph) = key.get_value::<Arc<RwLock<ThreadedGraph>>>(&GRAPH_TYPE)? {
         let graph = graph.clone();
-        return profile_mut(ctx, &graph, query, key_name);
+        return profile_mut(ctx, &graph, query, &key_name);
     }
 
     let graph = Arc::new(RwLock::new(ThreadedGraph::new(
         *CONFIGURATION_CACHE_SIZE.lock(ctx) as usize,
         &key_str.to_string(),
     )));
-    let result = profile_mut(ctx, &graph, query, key_name);
+    let result = profile_mut(ctx, &graph, query, &key_name);
     key.set_value(&GRAPH_TYPE, graph)?;
     result
 }
diff --git a/src/commands/query.rs b/src/commands/query.rs
index 92b0bc3f..5c7cbc42 100644
--- a/src/commands/query.rs
+++ b/src/commands/query.rs
@@ -24,7 +24,8 @@ use crate::{
     redis_type::GRAPH_TYPE,
 };
 use parking_lot::RwLock;
-use redis_module::{Context, NextArg, RedisResult, RedisString};
+use redis_module::{Context, NextArg, RedisResult, RedisString, raw};
+use std::ffi::CString;
 use std::sync::Arc;
 #[cfg(feature = "fuzz")]
 use std::sync::atomic::{AtomicI32, Ordering};
@@ -52,11 +53,15 @@ pub fn graph_query(
 
     let mut compact = false;
     let mut track_memory = false;
+    let mut version_check: Option<u64> = None;
     while let Ok(arg) = args.next_str() {
         if arg == "--compact" {
             compact = true;
         } else if arg == "--track-memory" {
             track_memory = true;
+        } else if arg == "version" {
+            let ver_str = args.next_str()?;
+            version_check = Some(ver_str.parse::<u64>()?);
         }
     }
 
@@ -66,6 +71,22 @@ pub fn graph_query(
 
     if let Some(graph) = read_key.get_value::<Arc<RwLock<ThreadedGraph>>>(&GRAPH_TYPE)? {
         let graph = graph.clone();
+
+        // Check version if provided
+        if let Some(provided_version) = version_check {
+            let current_schema_version = graph.read().graph.read().borrow().schema_version;
+            if provided_version != current_schema_version {
+                drop(read_key);
+                drop(graph);
+                // Return array with [error, version]
+                raw::reply_with_array(ctx.ctx, 2);
+                let err_msg = CString::new("ERR invalid graph version").unwrap();
+                raw::reply_with_error(ctx.ctx, err_msg.as_ptr());
+                raw::reply_with_long_long(ctx.ctx, current_schema_version as i64);
+                return Ok(redis_module::RedisValue::NoReply);
+            }
+        }
+
         drop(read_key);
         return query_mut(ctx, &graph, query, compact, true, track_memory, key_name);
     }
@@ -76,6 +97,20 @@ pub fn graph_query(
     // Re-check: another client may have created it between our read and write open.
     if let Some(graph) = key.get_value::<Arc<RwLock<ThreadedGraph>>>(&GRAPH_TYPE)? {
         let graph = graph.clone();
+
+        // Check version if provided
+        if let Some(provided_version) = version_check {
+            let current_schema_version = graph.read().graph.read().borrow().schema_version;
+            if provided_version != current_schema_version {
+                // Return array with [error, version]
+                raw::reply_with_array(ctx.ctx, 2);
+                let err_msg = CString::new("ERR invalid graph version").unwrap();
+                raw::reply_with_error(ctx.ctx, err_msg.as_ptr());
+                raw::reply_with_long_long(ctx.ctx, current_schema_version as i64);
+                return Ok(redis_module::RedisValue::NoReply);
+            }
+        }
+
         return query_mut(ctx, &graph, query, compact, true, track_memory, key_name);
     }
 
@@ -83,6 +118,19 @@ pub fn graph_query(
         *CONFIGURATION_CACHE_SIZE.lock(ctx) as usize,
         &key_str.to_string(),
     )));
+
+    // For a newly-created graph, the initial schema_version is 0
+    if let Some(provided_version) = version_check
+        && provided_version != 0
+    {
+        // Return array with [error, version]
+        raw::reply_with_array(ctx.ctx, 2);
+        let err_msg = CString::new("ERR invalid graph version").unwrap();
+        raw::reply_with_error(ctx.ctx, err_msg.as_ptr());
+        raw::reply_with_long_long(ctx.ctx, 0i64);
+        return Ok(redis_module::RedisValue::NoReply);
+    }
+
     let result = query_mut(ctx, &graph, query, compact, true, track_memory, key_name);
     key.set_value(&GRAPH_TYPE, graph)?;
     result
diff --git a/src/graph_core.rs b/src/graph_core.rs
index 362f4c1d..846014b4 100644
--- a/src/graph_core.rs
+++ b/src/graph_core.rs
@@ -528,11 +528,11 @@ pub fn profile_mut(
     ctx: &Context,
     graph: &Arc<RwLock<ThreadedGraph>>,
     query: &str,
-    key_name: Arc<String>,
+    key_name: &Arc<String>,
 ) -> RedisResult {
     // Inside MULTI/EXEC: execute synchronously.
     if ctx.get_flags().contains(ContextFlags::MULTI) {
-        return profile_sync(ctx, graph, query, &key_name);
+        return profile_sync(ctx, graph, query, key_name);
     }
 
     let bc = BlockedClient {
@@ -684,6 +684,9 @@ pub fn process_write_queued_query(graph: &Arc<RwLock<ThreadedGraph>>) {
                     graph.graph.rollback();
                 }
             }
+            // Yield between batched writes so other graph write loops
+            // (on different threadpool workers) can make progress.
+            std::thread::yield_now();
         }
         graph.write_loop.store(false, Ordering::Release);
     }
diff --git a/src/lib.rs b/src/lib.rs
index d178f6c3..e822854f 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -46,8 +46,8 @@ mod serializers;
 
 use allocator::ThreadCountingAllocator;
 use commands::{
-    graph_config, graph_delete, graph_explain, graph_list, graph_memory, graph_profile,
-    graph_query, graph_record, graph_ro_query, graph_udf,
+    graph_config, graph_debug, graph_delete, graph_effect, graph_explain, graph_list, graph_memory,
+    graph_profile, graph_query, graph_record, graph_ro_query, graph_udf,
 };
 use config::{
     CONFIGURATION_CACHE_SIZE, CONFIGURATION_CMD_INFO, CONFIGURATION_DELAY_INDEXING,
@@ -57,13 +57,13 @@ use config::{
 };
 use module_init::graph_init;
 use redis_module::{configuration::ConfigurationFlags, redis_module};
-use redis_type::GRAPH_TYPE;
+use redis_type::{GRAPH_TYPE, GRAPHMETA_TYPE};
 
 redis_module! {
     name: "graph",
     version: 1,
     allocator: (ThreadCountingAllocator, ThreadCountingAllocator),
-    data_types: [GRAPH_TYPE],
+    data_types: [GRAPH_TYPE, GRAPHMETA_TYPE],
     init: graph_init,
     commands: [
         ["graph.DELETE", graph_delete, "write deny-script", 1, 1, 1, ""],
@@ -76,6 +76,8 @@ redis_module! {
         ["graph.MEMORY", graph_memory, "readonly deny-script", 2, 2, 1, ""],
         ["graph.CONFIG", graph_config, "readonly deny-script allow-busy", 0, 0, 0, ""],
         ["graph.UDF", graph_udf, "write deny-script", 0, 0, 0, ""],
+        ["graph.DEBUG", graph_debug, "write deny-script", 0, 0, 0, ""],
+        ["graph.EFFECT", graph_effect, "write deny-script", 1, 1, 1, ""],
     ],
     configurations: [
         i64: [
diff --git a/src/module_init.rs b/src/module_init.rs
index f0c6bae7..bb78179f 100644
--- a/src/module_init.rs
+++ b/src/module_init.rs
@@ -26,6 +26,7 @@ use crate::config::{
     CONFIGURATION_JS_HEAP_SIZE, CONFIGURATION_JS_STACK_SIZE, CONFIGURATION_TEMP_FOLDER,
     OMP_THREAD_COUNT, get_thread_count,
 };
+use crate::redis_type::on_persistence;
 use graph::{
     graph::graphblas::matrix::init,
     index::redisearch::{REDISEARCH_INIT_LIBRARY, RediSearch_Init},
@@ -44,6 +45,10 @@ use std::{os::raw::c_int, os::raw::c_void, panic};
 #[allow(non_upper_case_globals)]
 static RedisModuleEvent_FlushDB: RedisModuleEvent = RedisModuleEvent { id: 2, dataver: 1 };
 
+/// Redis event ID for Persistence events (RDB save start/end).
+#[allow(non_upper_case_globals)]
+static RedisModuleEvent_Persistence: RedisModuleEvent = RedisModuleEvent { id: 1, dataver: 1 };
+
 pub fn graph_init(
     ctx: &Context,
     _: &Vec<redis_module::RedisString>,
@@ -72,6 +77,17 @@ pub fn graph_init(
             Some(on_flush),
         );
         debug_assert_eq!(res, REDISMODULE_OK as c_int);
+
+        // Subscribe to persistence events for virtual key management.
+        let res = RedisModule_SubscribeToServerEvent.unwrap()(
+            ctx.ctx,
+            RedisModuleEvent_Persistence,
+            Some(on_persistence),
+        );
+        if res != REDISMODULE_OK as c_int {
+            eprintln!("FalkorDB: failed to subscribe to persistence events: code {res}");
+            return Status::Err;
+        }
     }
     match init_functions() {
         Ok(()) => {}
diff --git a/src/redis_type.rs b/src/redis_type.rs
index 962c82f0..d3c89022 100644
--- a/src/redis_type.rs
+++ b/src/redis_type.rs
@@ -1,8 +1,13 @@
 //! Redis native type declaration for graph storage and UDF persistence.
 //!
 //! Registers `GRAPH_TYPE` -- a Redis module type named `"graphdata"` --
+//! and `GRAPHMETA_TYPE` -- a Redis module type named `"graphmeta"` --
 //! along with RDB and lifecycle callbacks that Redis invokes automatically.
 //!
+//! `GRAPHMETA_TYPE` is needed to load C FalkorDB RDB files, which use
+//! `"graphmeta"` for virtual keys and AUX data. Rust's own virtual keys
+//! use `"graphdata"` so that C FalkorDB can also load them.
+//!
 //! ## Callbacks
 //!
 //! ```text
@@ -33,125 +38,743 @@
 //!                        Redis invokes `free` callback -> graph_free()
 //! ```
 
+use crate::config::CONFIGURATION_VKEY_MAX_ENTITY_COUNT;
 use crate::graph_core::{ThreadedGraph, graph_free};
-use crate::serializers::decoder::rdb_load_graph;
-use crate::serializers::encoder::rdb_save_graph;
+use crate::serializers;
+use crate::serializers::encoder::build_multi_key_payloads;
+use crate::serializers::{DECODE_STATE, VKEY_STATE};
 use graph::graph::mvcc_graph::MvccGraph;
 use graph::runtime::functions::{GraphFn, register_udf};
 use graph::udf::get_udf_repo;
 use parking_lot::RwLock;
-use redis_module::raw::{load_string_buffer, load_unsigned, save_string, save_unsigned};
+use redis_module::logging::log_notice;
+use redis_module::raw::{
+    self, RedisModuleCtx, load_string_buffer, load_unsigned, save_string, save_unsigned,
+};
 use redis_module::{
     REDISMODULE_TYPE_METHOD_VERSION, RedisModuleIO, RedisModuleTypeMethods, native_types::RedisType,
 };
+use std::ffi::CString;
 use std::sync::Arc;
 use std::{os::raw::c_void, ptr::null_mut};
 
-/// Decode a graph from the RDB stream.
-///
-/// Called by Redis for each key of type `GRAPH_TYPE` during RDB load.
-/// Returns a heap-allocated `Arc<RwLock<ThreadedGraph>>` that Redis
-/// will associate with the key, or null on failure.
+/// Default cache size used when loading from RDB (no Redis context available).
+const DEFAULT_CACHE_SIZE: usize = 25;
+
+// ---------------------------------------------------------------------------
+// graphdata rdb_load / rdb_save
+// ---------------------------------------------------------------------------
+
 #[unsafe(no_mangle)]
 unsafe extern "C" fn graph_rdb_load(
     rdb: *mut RedisModuleIO,
     _encver: i32,
 ) -> *mut c_void {
-    // Default cache size for the query plan cache.
-    // During RDB load we don't have a Context to read the module config,
-    // so use the default value (25).
-    let cache_size = 25;
+    // Get the key name for looking up finalized graphs.
+    let key_name = unsafe {
+        let rm_key_name = raw::RedisModule_GetKeyNameFromIO.unwrap()(rdb);
+        if rm_key_name.is_null() {
+            "<unknown>".to_string()
+        } else {
+            let mut len: usize = 0;
+            let ptr = raw::RedisModule_StringPtrLen.unwrap()(rm_key_name, &raw mut len);
+            String::from_utf8_lossy(std::slice::from_raw_parts(ptr.cast(), len)).to_string()
+        }
+    };
 
-    match rdb_load_graph(rdb, cache_size) {
+    match serializers::decoder::rdb_load_graph(rdb, DEFAULT_CACHE_SIZE) {
         Ok(Some(graph)) => {
+            // Single-key load (key_count == 1) -- graph is fully loaded.
             let mvcc = MvccGraph::from_graph(graph);
-            let tg = Arc::new(RwLock::new(ThreadedGraph::from_mvcc(mvcc)));
-            Box::into_raw(Box::new(tg)).cast::<c_void>()
+            let graph_arc = mvcc.read();
+            graph_arc.borrow_mut().set_indexer_graph(graph_arc.clone());
+            let tg = ThreadedGraph::from_mvcc(mvcc);
+            let boxed: Box<Arc<RwLock<ThreadedGraph>>> = Box::new(Arc::new(RwLock::new(tg)));
+            Box::into_raw(boxed).cast()
         }
         Ok(None) => {
-            // Multi-key graph: data accumulated in DECODE_STATE.
-            // Return null for now; the graph will be finalized later.
-            null_mut()
+            // Multi-key load (key_count > 1) -- data stored in DECODE_STATE.
+            // Check if all keys have already been loaded (inline finalization),
+            // in which case we can return the real graph directly.
+            {
+                let mut decode_state = DECODE_STATE.lock();
+                if let Some(graph) = decode_state.finalized.remove(&key_name) {
+                    let mvcc = MvccGraph::from_graph(graph);
+                    let graph_arc = mvcc.read();
+                    graph_arc.borrow_mut().set_indexer_graph(graph_arc.clone());
+                    let tg = ThreadedGraph::from_mvcc(mvcc);
+                    let boxed: Box<Arc<RwLock<ThreadedGraph>>> =
+                        Box::new(Arc::new(RwLock::new(tg)));
+                    return Box::into_raw(boxed).cast();
+                }
+            }
+
+            // Graph not yet finalized - more keys still need to load.
+            // Return a placeholder that will be replaced later.
+            let tg = ThreadedGraph::new(DEFAULT_CACHE_SIZE, "__placeholder__");
+            let arc = Arc::new(RwLock::new(tg));
+
+            // Store an Arc clone keyed by graph name for later finalization.
+            {
+                let mut decode_state = DECODE_STATE.lock();
+                decode_state.placeholders.insert(key_name, arc.clone());
+            }
+
+            // Hand ownership of a Box<Arc<...>> to Redis.
+            let boxed: Box<Arc<RwLock<ThreadedGraph>>> = Box::new(arc);
+            Box::into_raw(boxed).cast()
         }
         Err(e) => {
-            eprintln!("FalkorDB: RDB load error: {e}");
+            eprintln!("graph rdb_load error: {e}");
             null_mut()
         }
     }
 }
 
-/// Encode a graph into the RDB stream.
-///
-/// Called by Redis for each key of type `GRAPH_TYPE` during RDB save
-/// (BGSAVE, SAVE, or replication). The `value` pointer is the
-/// `Arc<RwLock<ThreadedGraph>>` that was stored via `set_value`.
 #[unsafe(no_mangle)]
 unsafe extern "C" fn graph_rdb_save(
     rdb: *mut RedisModuleIO,
     value: *mut c_void,
 ) {
-    let tg = &*(value.cast::<Arc<RwLock<ThreadedGraph>>>());
-    let guard = tg.read();
-    let g_arc = guard.graph.read();
-    let g = g_arc.borrow();
-    rdb_save_graph(rdb, &g);
+    unsafe {
+        // Get the key name to determine if this is a main key or virtual key.
+        let rm_key_name = raw::RedisModule_GetKeyNameFromIO.unwrap()(rdb);
+        let key_name = if rm_key_name.is_null() {
+            String::new()
+        } else {
+            let mut len: usize = 0;
+            let ptr = raw::RedisModule_StringPtrLen.unwrap()(rm_key_name, &raw mut len);
+            String::from_utf8_lossy(std::slice::from_raw_parts(ptr.cast(), len)).to_string()
+        };
+
+        let vkey_state = VKEY_STATE.lock();
+
+        // Check if this is a virtual key by looking up in VKEY_STATE.
+        // Virtual keys have their graph ref stored separately because
+        // they hold a placeholder value, not the actual graph.
+        if let Some((graph_name, payloads)) = vkey_state.get_vkey_payloads(&key_name) {
+            // Virtual key: use the stored graph reference.
+            let graph_name = graph_name.to_string();
+            let payloads = payloads.to_vec();
+            let key_count = vkey_state
+                .graph_vkeys
+                .get(&graph_name)
+                .map_or(1, |vkeys| (vkeys.len() + 1) as u64);
+            let Some(graph_arc) = vkey_state.get_graph_ref(&graph_name).cloned() else {
+                return;
+            };
+            let snap = vkey_state.rdb_snapshots.get(&graph_name).cloned();
+            drop(vkey_state);
+
+            let tg = graph_arc.read();
+            let g = tg.graph.read();
+            let graph = g.borrow();
+            serializers::encoder::rdb_save_graph_key(
+                rdb,
+                &graph,
+                &payloads,
+                key_count,
+                snap.as_ref().map(AsRef::as_ref),
+            );
+        } else {
+            // Main key: use the value pointer directly.
+            let graph_arc = &*(value.cast::<Arc<RwLock<ThreadedGraph>>>());
+            let tg = graph_arc.read();
+            let g = tg.graph.read();
+            let graph = g.borrow();
+            let graph_name = graph.name().to_string();
+
+            if let Some((_gn, payloads)) = vkey_state.get_vkey_payloads(&graph_name) {
+                let payloads = payloads.to_vec();
+                let key_count = vkey_state
+                    .graph_vkeys
+                    .get(&graph_name)
+                    .map_or(1, |vkeys| (vkeys.len() + 1) as u64);
+                let snap = vkey_state.rdb_snapshots.get(&graph_name).cloned();
+                drop(vkey_state);
+                serializers::encoder::rdb_save_graph_key(
+                    rdb,
+                    &graph,
+                    &payloads,
+                    key_count,
+                    snap.as_ref().map(AsRef::as_ref),
+                );
+            } else {
+                let snap = vkey_state.rdb_snapshots.get(&graph_name).cloned();
+                drop(vkey_state);
+                serializers::encoder::rdb_save_graph(rdb, &graph, snap.as_ref().map(AsRef::as_ref));
+            }
+        }
+    }
 }
 
-/// Save UDF libraries to RDB.
+// ---------------------------------------------------------------------------
+// aux_save / aux_load
+// ---------------------------------------------------------------------------
+
 #[unsafe(no_mangle)]
 unsafe extern "C" fn graph_aux_save(
     rdb: *mut RedisModuleIO,
-    _when: i32,
+    when: i32,
 ) {
-    let repo = get_udf_repo();
-    let libs = repo.serialize();
-    save_unsigned(rdb, libs.len() as u64);
-    for (name, code) in &libs {
-        save_string(rdb, name);
-        save_string(rdb, code);
+    if when == raw::Aux::Before as i32 {
+        // BEFORE_RDB: Save UDF libraries.
+        let repo = get_udf_repo();
+        let libs = repo.serialize();
+        save_unsigned(rdb, libs.len() as u64);
+        for (name, code) in &libs {
+            save_string(rdb, name);
+            save_string(rdb, code);
+        }
+    } else {
+        // AFTER_RDB: Write placeholder so aux_load(AFTER_RDB) has something to read.
+        save_unsigned(rdb, 0);
     }
 }
 
-/// Load UDF libraries from RDB.
 #[unsafe(no_mangle)]
 unsafe extern "C" fn graph_aux_load(
     rdb: *mut RedisModuleIO,
     _encver: i32,
-    _when: i32,
+    when: i32,
 ) -> i32 {
-    let Ok(count) = load_unsigned(rdb) else {
-        return 1; // REDISMODULE_ERR
-    };
-
-    let repo = get_udf_repo();
-    let mut libs = Vec::with_capacity(count as usize);
-    for _ in 0..count {
-        let name = match load_string_buffer(rdb) {
-            Ok(buf) => String::from_utf8_lossy(buf.as_ref()).to_string(),
-            Err(_) => return 1,
-        };
-        let code = match load_string_buffer(rdb) {
-            Ok(buf) => String::from_utf8_lossy(buf.as_ref()).to_string(),
-            Err(_) => return 1,
+    if when == raw::Aux::Before as i32 {
+        // BEFORE_RDB: Load UDFs.
+        let Ok(count) = load_unsigned(rdb) else {
+            return 1;
         };
-        libs.push((name, code));
+
+        let repo = get_udf_repo();
+        let mut libs = Vec::with_capacity(count as usize);
+        for _ in 0..count {
+            let name = match load_string_buffer(rdb) {
+                Ok(buf) => String::from_utf8_lossy(buf.as_ref()).to_string(),
+                Err(_) => return 1,
+            };
+            let code = match load_string_buffer(rdb) {
+                Ok(buf) => String::from_utf8_lossy(buf.as_ref()).to_string(),
+                Err(_) => return 1,
+            };
+            libs.push((name, code));
+        }
+
+        repo.deserialize(&libs).map_or(1, |loaded_libs| {
+            graph::runtime::functions::flush_udfs();
+            for lib in &loaded_libs {
+                for qname in &lib.function_names {
+                    let graph_fn = Arc::new(GraphFn::new_udf(qname));
+                    register_udf(qname, graph_fn);
+                }
+            }
+            0
+        })
+    } else {
+        // AFTER_RDB: Read placeholder, finalize pending multi-key graphs.
+        let _ = load_unsigned(rdb);
+        finalize_pending_graphs();
+        0
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Persistence event handler -- creates/deletes virtual keys
+// ---------------------------------------------------------------------------
+
+/// Called by Redis persistence events. Creates virtual keys before RDB save,
+/// deletes them after save completes or fails.
+///
+/// # Safety
+/// Called by Redis internals with a valid module context.
+pub unsafe extern "C" fn on_persistence(
+    ctx: *mut RedisModuleCtx,
+    _eid: redis_module::RedisModuleEvent,
+    subevent: u64,
+    _data: *mut c_void,
+) {
+    unsafe {
+        match subevent {
+            raw::REDISMODULE_SUBEVENT_PERSISTENCE_RDB_START
+            | raw::REDISMODULE_SUBEVENT_PERSISTENCE_SYNC_RDB_START => {
+                create_virtual_keys(ctx);
+            }
+            raw::REDISMODULE_SUBEVENT_PERSISTENCE_ENDED
+            | raw::REDISMODULE_SUBEVENT_PERSISTENCE_FAILED => {
+                delete_virtual_keys(ctx);
+            }
+            _ => {}
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Virtual key management helpers
+// ---------------------------------------------------------------------------
+
+pub unsafe fn create_virtual_keys(ctx: *mut RedisModuleCtx) {
+    unsafe {
+        // Delete stale graphmeta keys (from C FalkorDB RDB loads).
+        delete_stale_graphmeta_keys(ctx);
+
+        // Single graphdata scan: collect real graphs and delete stale virtual keys.
+        let graphs = scan_and_clean_graphdata_keys(ctx);
+
+        let mut vkey_state = VKEY_STATE.lock();
+        vkey_state.clear();
+
+        let context = redis_module::Context::new(ctx);
+        let vkey_max = *CONFIGURATION_VKEY_MAX_ENTITY_COUNT.lock(&context);
+
+        for (graph_name, graph_ref) in &graphs {
+            let tg = graph_ref.read();
+            let g = tg.graph.read();
+            let graph = g.borrow();
+
+            // Build attribute snapshots (cache + fjall) before fork.
+            // The fork child will use these instead of accessing fjall.
+            // Skip if no fjall data exists — all data is in cache already.
+            if graph.needs_rdb_snapshot() {
+                let snapshots = Arc::new(graph.build_rdb_snapshots());
+                vkey_state
+                    .rdb_snapshots
+                    .insert(graph_name.clone(), snapshots);
+            }
+
+            let multi_payloads = build_multi_key_payloads(&graph, vkey_max as u64);
+            let key_count = multi_payloads.len();
+
+            if key_count <= 1 {
+                continue;
+            }
+
+            // Store graph reference for virtual key rdb_save to use.
+            vkey_state.store_graph_ref(graph_name, graph_ref.clone());
+
+            let virtual_key_count = key_count - 1;
+            let mut vkey_names = Vec::with_capacity(virtual_key_count);
+
+            // Store key 0's payloads under the graph name.
+            vkey_state.vkey_map.insert(
+                graph_name.clone(),
+                (graph_name.clone(), 0, multi_payloads[0].clone()),
+            );
+
+            // Create virtual keys for keys 1..N.
+            for (i, payloads) in multi_payloads.iter().enumerate().skip(1) {
+                let uuid = uuid_v4();
+                let vkey_name = if graph_name.contains('{') {
+                    format!("{graph_name}_{uuid}")
+                } else {
+                    format!("{{{graph_name}}}{graph_name}_{uuid}")
+                };
+
+                vkey_state
+                    .vkey_map
+                    .insert(vkey_name.clone(), (graph_name.clone(), i, payloads.clone()));
+
+                // Create the Redis key.
+                let rm_str = raw::RedisModule_CreateString.unwrap()(
+                    ctx,
+                    vkey_name.as_ptr().cast(),
+                    vkey_name.len(),
+                );
+                let key =
+                    raw::RedisModule_OpenKey.unwrap()(ctx, rm_str, raw::KeyMode::WRITE.bits());
+                // Must pass a non-null value; Redis skips keys with null values during RDB save.
+                // Create a placeholder ThreadedGraph so graph_free can handle it.
+                let tg = ThreadedGraph::new(DEFAULT_CACHE_SIZE, "__vkey_placeholder__");
+                let boxed: Box<Arc<RwLock<ThreadedGraph>>> = Box::new(Arc::new(RwLock::new(tg)));
+                let value = Box::into_raw(boxed).cast();
+                raw::RedisModule_ModuleTypeSetValue.unwrap()(
+                    key,
+                    *GRAPH_TYPE.raw_type.borrow(),
+                    value,
+                );
+                raw::RedisModule_CloseKey.unwrap()(key);
+                raw::RedisModule_FreeString.unwrap()(ctx, rm_str);
+
+                vkey_names.push(vkey_name);
+            }
+
+            log_notice(format!(
+                "Created {virtual_key_count} virtual keys for graph {graph_name}"
+            ));
+
+            vkey_state
+                .graph_vkeys
+                .insert(graph_name.clone(), vkey_names);
+        }
+    }
+}
+
+unsafe fn delete_virtual_keys(ctx: *mut RedisModuleCtx) {
+    unsafe {
+        let mut vkey_state = VKEY_STATE.lock();
+
+        for (graph_name, vkey_names) in &vkey_state.graph_vkeys {
+            for vkey_name in vkey_names {
+                let rm_str = raw::RedisModule_CreateString.unwrap()(
+                    ctx,
+                    vkey_name.as_ptr().cast(),
+                    vkey_name.len(),
+                );
+                let key =
+                    raw::RedisModule_OpenKey.unwrap()(ctx, rm_str, raw::KeyMode::WRITE.bits());
+                raw::RedisModule_DeleteKey.unwrap()(key);
+                raw::RedisModule_CloseKey.unwrap()(key);
+                raw::RedisModule_FreeString.unwrap()(ctx, rm_str);
+            }
+
+            log_notice(format!(
+                "Deleted {} virtual keys for graph {graph_name}",
+                vkey_names.len(),
+            ));
+        }
+
+        vkey_state.clear();
     }
+}
+
+/// Single-pass scan of graphdata keys: collects real graphs and deletes stale
+/// virtual/placeholder keys in one traversal (instead of scanning twice).
+unsafe fn scan_and_clean_graphdata_keys(
+    ctx: *mut RedisModuleCtx
+) -> Vec<(String, Arc<RwLock<ThreadedGraph>>)> {
+    unsafe {
+        let mut result = Vec::new();
+        let mut stale_keys = Vec::new();
+
+        let scan_cmd = CString::new("SCAN").unwrap();
+        let type_arg = CString::new("TYPE").unwrap();
+        let graphdata_arg = CString::new("graphdata").unwrap();
+        let fmt = CString::new("ccc").unwrap();
 
-    // Validate all libraries, then atomically swap the repo contents.
-    // On failure the live repo and function table remain unchanged.
-    repo.deserialize(&libs).map_or(1, |loaded_libs| {
-        // Re-register bridge functions for the new set of libraries.
-        graph::runtime::functions::flush_udfs();
-        for lib in &loaded_libs {
-            for qname in &lib.function_names {
-                let graph_fn = Arc::new(GraphFn::new_udf(qname));
-                register_udf(qname, graph_fn);
+        let mut cursor_val = CString::new("0").unwrap();
+
+        loop {
+            let reply = raw::RedisModule_Call.unwrap()(
+                ctx,
+                scan_cmd.as_ptr(),
+                fmt.as_ptr(),
+                cursor_val.as_ptr(),
+                type_arg.as_ptr(),
+                graphdata_arg.as_ptr(),
+            );
+            if reply.is_null() {
+                break;
+            }
+
+            let reply_type = raw::call_reply_type(reply);
+            if reply_type != raw::ReplyType::Array {
+                raw::free_call_reply(reply);
+                break;
             }
+
+            let len = raw::call_reply_length(reply);
+            if len < 2 {
+                raw::free_call_reply(reply);
+                break;
+            }
+
+            // Get new cursor.
+            let cursor_reply = raw::call_reply_array_element(reply, 0);
+            let mut cursor_len: usize = 0;
+            let cursor_ptr =
+                raw::RedisModule_CallReplyStringPtr.unwrap()(cursor_reply, &raw mut cursor_len);
+            let new_cursor = std::str::from_utf8_unchecked(std::slice::from_raw_parts(
+                cursor_ptr.cast(),
+                cursor_len,
+            ));
+            let done = new_cursor == "0";
+
+            // Get keys array.
+            let arr_reply = raw::call_reply_array_element(reply, 1);
+            let arr_len = raw::call_reply_length(arr_reply);
+
+            for i in 0..arr_len {
+                let elem = raw::call_reply_array_element(arr_reply, i);
+                let mut key_len: usize = 0;
+                let kptr = raw::RedisModule_CallReplyStringPtr.unwrap()(elem, &raw mut key_len);
+                let key_name =
+                    std::str::from_utf8_unchecked(std::slice::from_raw_parts(kptr.cast(), key_len))
+                        .to_string();
+
+                let rm_str = raw::RedisModule_CreateString.unwrap()(
+                    ctx,
+                    key_name.as_ptr().cast(),
+                    key_name.len(),
+                );
+                let key = raw::RedisModule_OpenKey.unwrap()(ctx, rm_str, raw::KeyMode::READ.bits());
+                let value = raw::RedisModule_ModuleTypeGetValue.unwrap()(key);
+
+                if !value.is_null() {
+                    let graph_arc_ref = &*(value.cast::<Arc<RwLock<ThreadedGraph>>>());
+                    let tg = graph_arc_ref.read();
+                    let name = tg.name();
+                    if name.starts_with("__placeholder") || name.starts_with("__vkey_placeholder") {
+                        // Stale virtual key — mark for deletion.
+                        stale_keys.push(key_name);
+                    } else {
+                        // Real graph — collect it.
+                        drop(tg);
+                        result.push((key_name, graph_arc_ref.clone()));
+                    }
+                }
+
+                raw::RedisModule_CloseKey.unwrap()(key);
+                raw::RedisModule_FreeString.unwrap()(ctx, rm_str);
+            }
+
+            cursor_val = CString::new(new_cursor).unwrap();
+            raw::free_call_reply(reply);
+
+            if done {
+                break;
+            }
+        }
+
+        // Delete stale virtual keys.
+        for key_name in &stale_keys {
+            let rm_str = raw::RedisModule_CreateString.unwrap()(
+                ctx,
+                key_name.as_ptr().cast(),
+                key_name.len(),
+            );
+            let key = raw::RedisModule_OpenKey.unwrap()(ctx, rm_str, raw::KeyMode::WRITE.bits());
+            raw::RedisModule_DeleteKey.unwrap()(key);
+            raw::RedisModule_CloseKey.unwrap()(key);
+            raw::RedisModule_FreeString.unwrap()(ctx, rm_str);
+        }
+
+        if !stale_keys.is_empty() {
+            log_notice(format!(
+                "Deleted {} stale graphdata virtual keys before save",
+                stale_keys.len()
+            ));
         }
-        0 // REDISMODULE_OK
-    })
+
+        result
+    }
 }
 
+/// Delete stale graphmeta keys (from C FalkorDB RDB loads).
+unsafe fn delete_stale_graphmeta_keys(ctx: *mut RedisModuleCtx) {
+    unsafe {
+        let scan_cmd = CString::new("SCAN").unwrap();
+        let type_arg = CString::new("TYPE").unwrap();
+        let fmt = CString::new("ccc").unwrap();
+
+        let mut keys_to_delete = Vec::new();
+        let graphmeta_arg = CString::new("graphmeta").unwrap();
+        scan_keys_by_type(
+            ctx,
+            &scan_cmd,
+            &type_arg,
+            &graphmeta_arg,
+            &fmt,
+            &mut keys_to_delete,
+        );
+
+        for key_name in &keys_to_delete {
+            let rm_str = raw::RedisModule_CreateString.unwrap()(
+                ctx,
+                key_name.as_ptr().cast(),
+                key_name.len(),
+            );
+            let key = raw::RedisModule_OpenKey.unwrap()(ctx, rm_str, raw::KeyMode::WRITE.bits());
+            raw::RedisModule_DeleteKey.unwrap()(key);
+            raw::RedisModule_CloseKey.unwrap()(key);
+            raw::RedisModule_FreeString.unwrap()(ctx, rm_str);
+        }
+
+        if !keys_to_delete.is_empty() {
+            log_notice(format!(
+                "Deleted {} stale graphmeta keys before save",
+                keys_to_delete.len()
+            ));
+        }
+    }
+}
+
+/// Delete any stale virtual keys left in the keyspace from a previous RDB load.
+/// Public entry point used by the debug command.
+pub unsafe fn delete_stale_virtual_keys(ctx: *mut RedisModuleCtx) {
+    unsafe {
+        delete_stale_graphmeta_keys(ctx);
+        // scan_and_clean_graphdata_keys deletes stale graphdata keys as a side effect.
+        let _ = scan_and_clean_graphdata_keys(ctx);
+    }
+}
+
+unsafe fn scan_keys_by_type(
+    ctx: *mut RedisModuleCtx,
+    scan_cmd: &CString,
+    type_arg: &CString,
+    type_name: &CString,
+    fmt: &CString,
+    out: &mut Vec<String>,
+) {
+    unsafe {
+        let mut cursor_val = CString::new("0").unwrap();
+
+        loop {
+            let reply = raw::RedisModule_Call.unwrap()(
+                ctx,
+                scan_cmd.as_ptr(),
+                fmt.as_ptr(),
+                cursor_val.as_ptr(),
+                type_arg.as_ptr(),
+                type_name.as_ptr(),
+            );
+            if reply.is_null() {
+                break;
+            }
+
+            let reply_type = raw::call_reply_type(reply);
+            if reply_type != raw::ReplyType::Array {
+                raw::free_call_reply(reply);
+                break;
+            }
+
+            let len = raw::call_reply_length(reply);
+            if len < 2 {
+                raw::free_call_reply(reply);
+                break;
+            }
+
+            let cursor_reply = raw::call_reply_array_element(reply, 0);
+            let mut cursor_len: usize = 0;
+            let cursor_ptr =
+                raw::RedisModule_CallReplyStringPtr.unwrap()(cursor_reply, &raw mut cursor_len);
+            let new_cursor = std::str::from_utf8_unchecked(std::slice::from_raw_parts(
+                cursor_ptr.cast(),
+                cursor_len,
+            ));
+            let done = new_cursor == "0";
+
+            let arr_reply = raw::call_reply_array_element(reply, 1);
+            let arr_len = raw::call_reply_length(arr_reply);
+
+            for i in 0..arr_len {
+                let elem = raw::call_reply_array_element(arr_reply, i);
+                let mut name_len: usize = 0;
+                let kptr = raw::RedisModule_CallReplyStringPtr.unwrap()(elem, &raw mut name_len);
+                let key_name = std::str::from_utf8_unchecked(std::slice::from_raw_parts(
+                    kptr.cast(),
+                    name_len,
+                ))
+                .to_string();
+                out.push(key_name);
+            }
+
+            cursor_val = CString::new(new_cursor).unwrap();
+            raw::free_call_reply(reply);
+
+            if done {
+                break;
+            }
+        }
+    }
+}
+
+/// Finalize any pending multi-key graph loads from DECODE_STATE.
+///
+/// This handles two scenarios:
+/// 1. Graphs already finalized inline (stored in decode_state.finalized)
+/// 2. Graphs with keys_remaining == 0 that haven't been finalized yet
+///
+/// In both cases, the placeholder ThreadedGraph's inner MvccGraph is replaced
+/// using the raw pointer stored during graph_rdb_load.
+pub fn finalize_pending_graphs() {
+    let mut decode_state = DECODE_STATE.lock();
+
+    // First, handle graphs that were already finalized inline during rdb_load_graph.
+    let finalized_names: Vec<String> = decode_state.finalized.keys().cloned().collect();
+    for graph_name in &finalized_names {
+        if let Some(graph) = decode_state.finalized.remove(graph_name) {
+            let placeholder = decode_state.placeholders.remove(graph_name);
+            install_graph(graph_name, graph, placeholder);
+        }
+    }
+
+    // Then, handle graphs with keys_remaining == 0 (finalized via the old path).
+    let pending_names: Vec<String> = decode_state
+        .pending
+        .iter()
+        .filter(|(_, pg)| pg.keys_remaining == 0)
+        .map(|(name, _)| name.clone())
+        .collect();
+
+    for graph_name in &pending_names {
+        let pg = decode_state.pending.remove(graph_name).unwrap();
+        let placeholder = decode_state.placeholders.remove(graph_name);
+
+        match serializers::decoder::finalize_pending_graph(pg) {
+            Ok(graph) => {
+                install_graph(graph_name, graph, placeholder);
+            }
+            Err(e) => {
+                eprintln!("FalkorDB: failed to finalize graph {graph_name}: {e}");
+            }
+        }
+    }
+
+    // Only clear if all pending graphs have been finalized.
+    if decode_state.pending.is_empty() && decode_state.finalized.is_empty() {
+        decode_state.placeholders.clear();
+    }
+}
+
+/// Install a finalized Graph into the placeholder ThreadedGraph.
+fn install_graph(
+    graph_name: &str,
+    graph: graph::graph::graph::Graph,
+    placeholder: Option<Arc<RwLock<ThreadedGraph>>>,
+) {
+    let mvcc = MvccGraph::from_graph(graph);
+    let graph_arc = mvcc.read();
+    graph_arc.borrow_mut().set_indexer_graph(graph_arc.clone());
+    let tg = ThreadedGraph::from_mvcc(mvcc);
+
+    if let Some(ph) = placeholder {
+        let mut placeholder_tg = ph.write();
+        // Replace entire ThreadedGraph (graph, sender, receiver, write_loop)
+        // to ensure the write queue is properly bound to the new graph
+        *placeholder_tg = tg;
+    } else {
+        eprintln!(
+            "FalkorDB: WARNING - no placeholder pointer for graph '{graph_name}', graph data will be lost"
+        );
+    }
+}
+
+/// Generate a simple UUID v4 string.
+fn uuid_v4() -> String {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::time::{SystemTime, UNIX_EPOCH};
+    static COUNTER: AtomicU64 = AtomicU64::new(0);
+    let t = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_nanos();
+    let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
+    let a = (t as u64) ^ seq;
+    let b = a
+        .wrapping_mul(6_364_136_223_846_793_005)
+        .wrapping_add(1_442_695_040_888_963_407);
+    format!(
+        "{:08x}-{:04x}-4{:03x}-{:04x}-{:012x}",
+        (a >> 32) as u32,
+        (a >> 16) as u16,
+        (a & 0xFFF) as u16,
+        (0x8000 | (b & 0x3FFF)) as u16,
+        b & 0xFFFF_FFFF_FFFF
+    )
+}
+
+// ---------------------------------------------------------------------------
+// Type statics
+// ---------------------------------------------------------------------------
+
 pub static GRAPH_TYPE: RedisType = RedisType::new(
     "graphdata",
     19,
@@ -168,7 +791,102 @@ pub static GRAPH_TYPE: RedisType = RedisType::new(
         aux_load: Some(graph_aux_load),
         aux_save: None,
         aux_save2: Some(graph_aux_save),
-        aux_save_triggers: 1, // REDISMODULE_AUX_BEFORE_RDB
+        aux_save_triggers: 3, // REDISMODULE_AUX_BEFORE_RDB | REDISMODULE_AUX_AFTER_RDB
+
+        free_effort: None,
+        unlink: None,
+        copy: None,
+        defrag: None,
+
+        copy2: None,
+        free_effort2: None,
+        mem_usage2: None,
+        unlink2: None,
+    },
+);
+
+// ---------------------------------------------------------------------------
+// graphmeta -- kept for loading C FalkorDB RDB files.
+//
+// C FalkorDB uses "graphmeta" for virtual keys and emits graphmeta AUX data.
+// We register this type with rdb_load + aux_load so Rust can consume C's RDB
+// stream. We intentionally omit aux_save so that Rust never emits graphmeta
+// AUX data (which C can't load since it doesn't register "graphmeta" either).
+// ---------------------------------------------------------------------------
+
+/// Load a C FalkorDB graphmeta virtual key.
+#[unsafe(no_mangle)]
+unsafe extern "C" fn graphmeta_rdb_load(
+    rdb: *mut RedisModuleIO,
+    _encver: i32,
+) -> *mut c_void {
+    match serializers::decoder::rdb_load_graph(rdb, DEFAULT_CACHE_SIZE) {
+        Ok(_) => {
+            // Return a non-null dummy value. Redis needs non-null for successful load.
+            Box::into_raw(Box::new(0u8)).cast()
+        }
+        Err(e) => {
+            eprintln!("graphmeta rdb_load error: {e}");
+            null_mut()
+        }
+    }
+}
+
+/// Save callback for graphmeta keys left over from a C RDB load.
+/// These should be cleaned up before save by `delete_stale_virtual_keys`,
+/// but this is kept as a safety net.
+#[allow(clippy::missing_const_for_fn)]
+#[unsafe(no_mangle)]
+unsafe extern "C" fn graphmeta_rdb_save(
+    _rdb: *mut RedisModuleIO,
+    _value: *mut c_void,
+) {
+    // Stale graphmeta keys should have been deleted before save.
+    // If we get here, write nothing — the key will be empty.
+}
+
+/// Free callback for graphmeta keys. These hold a dummy u8 value.
+#[unsafe(no_mangle)]
+unsafe extern "C" fn graphmeta_free(value: *mut c_void) {
+    if !value.is_null() {
+        unsafe {
+            drop(Box::from_raw(value.cast::<u8>()));
+        }
+    }
+}
+
+/// Consume C FalkorDB's graphmeta AUX data during RDB load.
+#[unsafe(no_mangle)]
+unsafe extern "C" fn graphmeta_aux_load(
+    rdb: *mut RedisModuleIO,
+    _encver: i32,
+    when: i32,
+) -> i32 {
+    let _ = load_unsigned(rdb);
+    if when == raw::Aux::After as i32 {
+        finalize_pending_graphs();
+    }
+    0
+}
+
+pub static GRAPHMETA_TYPE: RedisType = RedisType::new(
+    "graphmeta",
+    19,
+    RedisModuleTypeMethods {
+        version: REDISMODULE_TYPE_METHOD_VERSION as u64,
+        rdb_load: Some(graphmeta_rdb_load),
+        rdb_save: Some(graphmeta_rdb_save),
+        aof_rewrite: None,
+        free: Some(graphmeta_free),
+
+        mem_usage: None,
+        digest: None,
+
+        // aux_load only — consume C's graphmeta AUX data but never emit it.
+        aux_load: Some(graphmeta_aux_load),
+        aux_save: None,
+        aux_save2: None,
+        aux_save_triggers: 3,
 
         free_effort: None,
         unlink: None,
diff --git a/src/reply.rs b/src/reply.rs
index f675286f..597e4a20 100644
--- a/src/reply.rs
+++ b/src/reply.rs
@@ -365,7 +365,7 @@ pub fn reply_verbose_value(
 
                 let attrs = bg.get_node_all_attrs(*id);
                 raw::reply_with_array(ctx.ctx, attrs.len() as _);
-                for (key, value) in attrs.iter() {
+                for (key, value) in &attrs {
                     raw::reply_with_array(ctx.ctx, 2);
                     raw::reply_with_string_buffer(
                         ctx.ctx,
diff --git a/src/serializers/encoder/mod.rs b/src/serializers/encoder/mod.rs
index 5886b3f7..59e9d9f1 100644
--- a/src/serializers/encoder/mod.rs
+++ b/src/serializers/encoder/mod.rs
@@ -1,4 +1,4 @@
-use graph::graph::graph::Graph;
+use graph::graph::graph::{Graph, RdbSnapshots};
 use graph::graph::graphblas::serialization::{Encode, EncodeState, PayloadEntry};
 use redis_module::RedisModuleIO;
 
@@ -11,9 +11,10 @@ use super::{Header, Schema};
 pub fn rdb_save_graph(
     rdb: *mut RedisModuleIO,
     graph: &Graph,
+    snapshots: Option<&RdbSnapshots>,
 ) {
     let payloads = build_payloads(graph);
-    rdb_save_graph_key(rdb, graph, &payloads, 1);
+    rdb_save_graph_key(rdb, graph, &payloads, 1, snapshots);
 }
 
 /// Encode a single key's portion of the graph (used for both primary and virtual keys).
@@ -22,6 +23,7 @@ pub fn rdb_save_graph_key(
     graph: &Graph,
     payloads: &[PayloadEntry],
     key_count: u64,
+    snapshots: Option<&RdbSnapshots>,
 ) {
     let mut w = BufferedWriter::new(rdb);
 
@@ -43,7 +45,7 @@ pub fn rdb_save_graph_key(
 
     // --- Payload data ---
     for p in payloads {
-        graph.encode_payload(&mut w, p, &global_attrs);
+        graph.encode_payload(&mut w, p, &global_attrs, snapshots);
     }
 
     w.finish();
diff --git a/src/serializers/mod.rs b/src/serializers/mod.rs
index f5ced828..5b474a0c 100644
--- a/src/serializers/mod.rs
+++ b/src/serializers/mod.rs
@@ -32,6 +32,8 @@ pub struct VirtualKeyState {
     pub graph_vkeys: HashMap<String, Vec<String>>,
     /// Graph references keyed by graph_name for use by virtual key rdb_save.
     graph_refs: HashMap<String, Arc<RwLock<ThreadedGraph>>>,
+    /// Pre-built attribute snapshots (cache + fjall) built before fork.
+    pub rdb_snapshots: HashMap<String, Arc<graph::graph::graph::RdbSnapshots>>,
 }
 
 impl VirtualKeyState {
@@ -40,6 +42,7 @@ impl VirtualKeyState {
             vkey_map: HashMap::new(),
             graph_vkeys: HashMap::new(),
             graph_refs: HashMap::new(),
+            rdb_snapshots: HashMap::new(),
         }
     }
 
@@ -47,6 +50,7 @@ impl VirtualKeyState {
         self.vkey_map.clear();
         self.graph_vkeys.clear();
         self.graph_refs.clear();
+        self.rdb_snapshots.clear();
     }
 
     pub fn get_vkey_payloads(
diff --git a/tests/common.py b/tests/common.py
index b2a7457d..9f4672b8 100644
--- a/tests/common.py
+++ b/tests/common.py
@@ -33,10 +33,11 @@ def start_redis(release=None, moduleEnvs=[]):
         shutdown = True
         if os.path.exists("redis-test.log"):
             os.remove("redis-test.log")
-        redis_server = subprocess.Popen(executable="/usr/local/bin/redis-server",
-                                        args=["--save", "", "--port", port, "--logfile", "redis-test.log",
-                                              "--loadmodule", target] + moduleEnvs,
-                                        stdout=subprocess.PIPE)
+        redis_server = subprocess.Popen(
+            ["/usr/local/bin/redis-server",
+             "--save", "", "--port", port, "--logfile", "redis-test.log",
+             "--loadmodule", target] + moduleEnvs,
+            stdout=subprocess.PIPE)
     while True:
         try:
             r.ping()
diff --git a/tests/flow/dumps/10.dump b/tests/flow/dumps/10.dump
deleted file mode 100644
index 77bfb118..00000000
Binary files a/tests/flow/dumps/10.dump and /dev/null differ
diff --git a/tests/flow/dumps/11.dump b/tests/flow/dumps/11.dump
deleted file mode 100644
index 362e2517..00000000
Binary files a/tests/flow/dumps/11.dump and /dev/null differ
diff --git a/tests/flow/dumps/12.dump b/tests/flow/dumps/12.dump
deleted file mode 100644
index 2ae609e2..00000000
Binary files a/tests/flow/dumps/12.dump and /dev/null differ
diff --git a/tests/flow/dumps/13.dump b/tests/flow/dumps/13.dump
deleted file mode 100644
index 4ae28284..00000000
Binary files a/tests/flow/dumps/13.dump and /dev/null differ
diff --git a/tests/flow/dumps/14.dump b/tests/flow/dumps/14.dump
deleted file mode 100644
index f655a05b..00000000
Binary files a/tests/flow/dumps/14.dump and /dev/null differ
diff --git a/tests/flow/dumps/15.dump b/tests/flow/dumps/15.dump
deleted file mode 100644
index 86d64921..00000000
Binary files a/tests/flow/dumps/15.dump and /dev/null differ
diff --git a/tests/flow/redis.conf b/tests/flow/redis.conf
index fa01de10..e69de29b 100644
--- a/tests/flow/redis.conf
+++ b/tests/flow/redis.conf
@@ -1 +0,0 @@
-stop-writes-on-bgsave-error no
diff --git a/tests/flow/test_persistency.py b/tests/flow/test_persistency.py
index caa9c681..8f1fdc15 100644
--- a/tests/flow/test_persistency.py
+++ b/tests/flow/test_persistency.py
@@ -6,7 +6,7 @@
 from index_utils import *
 from collections import OrderedDict
 from click.testing import CliRunner
-from datetime import datetime, date, time
+from datetime import datetime, date, time, timezone
 from dateutil.relativedelta import relativedelta
 from falkordb_bulk_loader.bulk_insert import bulk_insert
 
@@ -76,7 +76,7 @@ def populate_graph(self, graph_name):
         graph.create_node_range_index("person", "name", "height")
         graph.create_node_range_index("country", "name", "population")
         graph.create_edge_range_index("visit", "purpose")
-        graph.query("CALL db.idx.fulltext.createNodeIndex({label: 'person', stopwords: ['A', 'B'], language: 'english'}, { field: 'text', nostem: true, weight: 2, phonetic: 'dm:en' })")
+        graph.query("CREATE FULLTEXT INDEX FOR (n:person) ON (n.text) OPTIONS {stopwords: ['A', 'B'], language: 'english', nostem: true, weight: 2, phonetic: true}")
         create_node_vector_index(graph, "person", 'embedding1', dim=128, m=64, efConstruction=10, efRuntime=10)
         create_node_vector_index(graph, "person", 'embedding2', dim=256, similarity_function='cosine', m=32, efConstruction=20, efRuntime=20)
         wait_for_indices_to_sync(graph)
@@ -106,6 +106,8 @@ def populate_dense_graph(self, graph_name):
 
         return dense_graph
 
+    # TODO: enable after indexes completed
+    @skip()
     def test_save_load(self):
         graph_names = ["G", "{tag}_G"]
         for graph_name in graph_names:
@@ -211,11 +213,11 @@ def test_restore_properties(self):
 
             # Verify that the properties are loaded correctly.
             expected_result = [[True, 5.5, 'str', [1, 2, 3], {"latitude": 5.5, "longitude": 6.0},
-                                [1, 0, 3],
-                                [[1, 8, 3], [1, -1, 4], [2, 2, 3]],
+                                [1.0, 0.0, 3.0],
+                                [[1.0, 8.0, 3.0], [1.0, -1.0, 4.0], [2.0, 2.0, 3.0]],
                                 date(year=1984, month=10, day=21),
                                 time(hour=10, minute=30, second=10),
-                                datetime(year=1984, month=10, day=21, hour=5, minute=30, second=10),
+                                datetime(year=1984, month=10, day=21, hour=5, minute=30, second=10, tzinfo=timezone.utc),
                                 relativedelta(years=1, months=1, days=1, hours=1, minutes=1, seconds=1)]]
 
             self.env.assertEqual(actual_result.result_set, expected_result)
@@ -270,6 +272,8 @@ def test_load_large_graph(self):
             self.env.assertEqual(actual_result.result_set, expected_result)
 
     # Verify that graphs created using the GRAPH.BULK endpoint are persisted correctly
+    # TODO: enable when bulk loader is implemented
+    @skip()
     def test_bulk_insert(self):
         port      = self.env.envRunner.port
         runner    = CliRunner()
diff --git a/tests/flow/test_prev_rdb_decode.py b/tests/flow/test_prev_rdb_decode.py
deleted file mode 100644
index 2b330dc5..00000000
--- a/tests/flow/test_prev_rdb_decode.py
+++ /dev/null
@@ -1,182 +0,0 @@
-import os
-import time
-from common import *
-from falkordb import FalkorDB
-
-# decoders versions to tests
-VERSIONS = [
-        {'decoder_version': 10, 'tag': 'redislabs/redisgraph:2.8.7'},
-        {'decoder_version': 11, 'tag': 'redislabs/redisgraph:2.8.12'},
-        {'decoder_version': 12, 'tag': 'redislabs/redisgraph:2.8.14'},
-        {'decoder_version': 13, 'tag': 'redislabs/redisgraph:2.12.8'},
-        {'decoder_version': 14, 'tag': 'falkordb/falkordb:v4.0.7'},
-        {'decoder_version': 15, 'tag': 'falkordb/falkordb:v4.2.2'},
-        {'decoder_version': 16, 'tag': 'falkordb/falkordb:v4.8.5'},
-        {'decoder_version': 17, 'tag': 'falkordb/falkordb:v4.10.3'}
-        ]
-
-QUERIES = [
-        "CREATE (:L1 {val:1, strval: 'str', numval: 5.5, nullval: NULL, boolval: true, array: [1,2,3], point: POINT({latitude: 32, longitude: 34})})-[:E{val:2}]->(:L2{val:3})",
-        "CREATE INDEX ON :L1(val)",
-        "CREATE INDEX ON :L1(none_existsing)",
-        "CREATE (:L3)-[:E2]->(:L4)",
-        "MATCH (n1:L3)-[r:E2]->(n2:L4) DELETE n1, r, n2"]
-
-def graph_id(v):
-    return f"v{v}_rdb_restore"
-
-def get_image_tag(v):
-    return [item['tag'] for item in VERSIONS if item ['decoder_version'] == v][0]
-
-# starts db using docker
-def run_db(image):
-    import docker
-    from random import randint
-
-    # Initialize the Docker client
-    client = docker.from_env()
-
-    random_port = randint(49152, 65535)
-
-    # Run the FalkorDB container
-    container = client.containers.run(
-        image,                            # Image
-        detach=True,                      # Run container in the background
-        ports={'6379/tcp': random_port},  # Map port 6379
-    )
-
-    return container, random_port
-
-# stop and remove docker container
-def stop_db(container):
-    container.stop()
-    container.remove()
-
-# generate a graph dump
-def generate_dump(key, port):
-    # Connect to FalkorDB
-    db = FalkorDB(port=port)
-
-    # Select the social graph
-    g = db.select_graph(key)
-    try:
-        g.delete()
-    except:
-        pass
-
-    # Populate graph
-    for q in QUERIES:
-        g.query(q)
-
-    # Dump key
-    return db.connection.dump(key)
-
-# get graph dump from a specified FalkorDB version
-# check if dump already exists locally, if not generates and saves dump
-# to "./dumps/{v}.dump"
-def get_dump(v):
-    path = f"./dumps/{v}.dump"
-
-    # get dump
-    if not os.path.exists(path):
-        # get decoder docker image tag
-        tag = get_image_tag(v)
-
-        # start Docker container
-        container, port = run_db(tag)
-
-        # wait for DB to accept connections
-        time.sleep(2)
-
-        # generate dump
-        dump = generate_dump(graph_id(v), port)
-        print(f"dump: {dump}")
-
-        # ensure the directory exists, create if missing
-        os.makedirs(os.path.dirname(path), exist_ok=True)
-
-        # save dump to file
-        with open(path, 'wb') as f:
-            f.write(dump)
-            f.flush()
-
-        # stop db
-        stop_db(container)
-
-    with open(path, 'rb') as f:
-        return f.read()
-
-class test_prev_rdb_decode():
-    def __init__(self):
-        self.env, self.db = Env()
-        self.redis_con = self.env.getConnection()
-
-    def _test_decode(self, decoder_id):
-        key = graph_id(decoder_id)
-        dump = get_dump(decoder_id)
-
-        # restore dump
-        self.redis_con.restore(key, 0, dump, True)
-
-        # select graph
-        graph = self.db.select_graph(key)
-
-        # expected entities
-        node0  = Node(node_id=0, labels='L1', properties={'val': 1, 'strval': 'str', 'numval': 5.5, 'boolval': True, 'array': [1,2,3], 'point': {'latitude': 32, 'longitude': 34}})
-        node1  = Node(node_id=1, labels='L2', properties={'val': 3})
-        edge01 = Edge(src_node=0, relation='E', dest_node=1, edge_id=0, properties={'val':2})
-
-        # validations
-        results = graph.query("MATCH (n)-[e]->(m) RETURN n, e, m")
-        self.env.assertEqual(results.result_set, [[node0, edge01, node1]])
-
-        plan = str(graph.explain("MATCH (n:L1 {val:1}) RETURN n"))
-        self.env.assertContains("Index Scan", plan)
-
-        results = graph.query("MATCH (n:L1 {val:1}) RETURN n")
-        self.env.assertEqual(results.result_set, [[node0]])
-
-    def test_v10_decode(self):
-        decoder_id = 10
-        self._test_decode(decoder_id)
-
-    def test_v11_decode(self):
-        decoder_id = 11
-        self._test_decode(decoder_id)
-
-    def test_v12_decode(self):
-        decoder_id = 12
-        self._test_decode(decoder_id)
-
-    def test_v13_decode(self):
-        decoder_id = 13
-        self._test_decode(decoder_id)
-
-    def test_v14_decode(self):
-        decoder_id = 14
-        self._test_decode(decoder_id)
-
-    def test_v15_decode(self):
-        decoder_id = 15
-        self._test_decode(decoder_id)
-
-    def test_v16_decode(self):
-        # under sanitizer we're seeing:
-        # Unhandled exception: DUMP payload version or checksum are wrong
-        if SANITIZER:
-            self.env.skip()
-            return
-
-        decoder_id = 16
-        self._test_decode(decoder_id)
-
-    def test_v17_decode(self):
-        # under sanitizer we're seeing:
-        # Unhandled exception: DUMP payload version or checksum are wrong
-        if SANITIZER:
-            self.env.skip()
-            return
-
-        decoder_id = 17
-        self._test_decode(decoder_id)
-
diff --git a/tests/flow/test_rdb_compat.py b/tests/flow/test_rdb_compat.py
new file mode 100644
index 00000000..9924630e
--- /dev/null
+++ b/tests/flow/test_rdb_compat.py
@@ -0,0 +1,275 @@
+"""
+RDB cross-compatibility tests between FalkorDB C (v4.18.1) and FalkorDB Rust.
+
+Both implementations use encoding version 19 and module type "graphdata".
+Uses Redis replication (REPLICAOF) to transfer RDB data between servers,
+avoiding DUMP/RESTORE version-checksum mismatches.
+"""
+
+import os
+import time
+from random import randint, seed
+from common import *
+
+FALKORDB_C_IMAGE = 'falkordb/falkordb:v4.18.1'
+
+# ──────────────────────────── Docker helpers ────────────────────────────
+
+def run_db(image):
+    """Start a FalkorDB container on a random port."""
+    import docker
+    client = docker.from_env()
+    port = randint(49152, 65535)
+    container = client.containers.run(
+        image,
+        detach=True,
+        ports={'6379/tcp': port},
+        extra_hosts={'host.docker.internal': 'host-gateway'},
+    )
+    return container, port
+
+def stop_db(container):
+    """Stop and remove a Docker container."""
+    container.stop()
+    container.remove()
+
+def wait_for_db(port, timeout=30):
+    """Poll until the Redis instance at *port* accepts connections."""
+    import redis as _redis
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        try:
+            r = _redis.Redis(host='localhost', port=port)
+            r.ping()
+            return
+        except Exception:
+            time.sleep(0.5)
+    raise RuntimeError(f"FalkorDB container on port {port} did not start in {timeout}s")
+
+def wait_for_replication(conn, timeout=30):
+    """Wait until a replica has completed initial sync."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        info = conn.info('replication')
+        if info.get('role') == 'slave':
+            if info.get('master_link_status') == 'up' and info.get('master_sync_in_progress', 0) == 0:
+                return
+        time.sleep(0.5)
+    raise RuntimeError("Replication did not complete in time")
+
+# ──────────────────────── Graph creation helpers ────────────────────────
+
+SIMPLE_QUERIES = [
+    "CREATE (:Person {name: 'Alice', age: 30, score: 9.5, active: true, tags: [1,2,3], loc: POINT({latitude: 32.0816, longitude: 34.7818})})-[:KNOWS {since: 2020}]->(:Person {name: 'Bob', age: 25})",
+    "CREATE (:City {name: 'TLV', population: 460613})",
+    "CREATE INDEX FOR (p:Person) ON (p.name)",
+    "CREATE INDEX FOR (p:Person) ON (p.age)",
+]
+
+SIMPLE_VERIFICATION = [
+    ("labels", "CALL db.labels() YIELD label RETURN label ORDER BY label"),
+    ("rel types", "CALL db.relationshiptypes() YIELD relationshipType RETURN relationshipType ORDER BY relationshipType"),
+    ("node count", "MATCH (n) RETURN count(n)"),
+    ("edge count", "MATCH ()-[e]->() RETURN count(e)"),
+    ("persons", "MATCH (p:Person) RETURN p.name, p.age, p.score, p.active, p.tags ORDER BY p.name"),
+    ("city", "MATCH (c:City) RETURN c.name, c.population"),
+    ("edges", "MATCH ()-[e]->() RETURN type(e), properties(e) ORDER BY e"),
+    ("index scan name", "MATCH (p:Person) WHERE p.name = 'Alice' RETURN p.name, p.age"),
+    ("index scan age", "MATCH (p:Person) WHERE p.age > 20 RETURN p.name ORDER BY p.name"),
+    ("point exists", "MATCH (p:Person {name: 'Alice'}) RETURN p.loc IS NOT NULL"),
+]
+
+def create_simple_graph(g):
+    """Populate *g* with the simple test graph and wait for indexes."""
+    for q in SIMPLE_QUERIES:
+        g.query(q)
+    _wait_for_indexes(g)
+
+def _wait_for_indexes(g, timeout=30):
+    """Wait until all indexes on *g* are OPERATIONAL."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        result = g.ro_query(
+            "CALL db.indexes() YIELD status WHERE status <> 'OPERATIONAL' RETURN count(1)"
+        )
+        if result.result_set[0][0] == 0:
+            return
+        time.sleep(0.2)
+    raise RuntimeError("Indexes did not become OPERATIONAL in time")
+
+def capture_state(g, queries):
+    """Run *queries* on graph *g* and return {label: result_set}."""
+    state = {}
+    for label, q in queries:
+        state[label] = g.ro_query(q).result_set
+    return state
+
+def assert_state_eq(env, expected, actual):
+    """Assert two captured states are identical."""
+    for label in expected:
+        if expected[label] != actual.get(label):
+            print(f"MISMATCH in '{label}':")
+            print(f"  expected: {expected[label]}")
+            print(f"  actual:   {actual.get(label)}")
+        env.assertEqual(expected[label], actual.get(label))
+
+# ───────────────────────── Random graph helpers ─────────────────────────
+
+RANDOM_LABELS = ['Alpha', 'Beta', 'Gamma', 'Delta']
+RANDOM_REL_TYPES = ['LINKS', 'CONNECTS', 'FOLLOWS']
+
+def create_random_graph(g, rng_seed=42):
+    """Create a deterministic random graph on *g*."""
+    seed(rng_seed)
+
+    # Nodes: deterministic properties per label
+    for label in RANDOM_LABELS:
+        count = randint(15, 25)
+        g.query(
+            f"UNWIND range(1, {count}) AS i "
+            f"CREATE (:{label} {{id: i, name: '{label}_' + toString(i), "
+            f"val: toFloat(i) * 1.5, flag: i % 2 = 0, nums: [i, i+1, i+2]}})"
+        )
+
+    # Edges: deterministic cross-label connections
+    for rel_type in RANDOM_REL_TYPES:
+        src = RANDOM_LABELS[randint(0, len(RANDOM_LABELS) - 1)]
+        dst = RANDOM_LABELS[randint(0, len(RANDOM_LABELS) - 1)]
+        g.query(
+            f"MATCH (a:{src}), (b:{dst}) "
+            f"WITH a, b LIMIT 20 "
+            f"CREATE (a)-[:{rel_type} {{weight: toFloat(a.id + b.id)}}]->(b)"
+        )
+
+    # Range indexes
+    for label in RANDOM_LABELS:
+        g.query(f"CREATE INDEX FOR (n:{label}) ON (n.id)")
+
+    _wait_for_indexes(g)
+
+RANDOM_VERIFICATION = [
+    ("labels", "CALL db.labels() YIELD label RETURN label ORDER BY label"),
+    ("rel types", "CALL db.relationshiptypes() YIELD relationshipType RETURN relationshipType ORDER BY relationshipType"),
+    ("node count", "MATCH (n) RETURN count(n)"),
+    ("edge count", "MATCH ()-[e]->() RETURN count(e)"),
+    ("nodes", "MATCH (n) RETURN labels(n), properties(n) ORDER BY n"),
+    ("edges", "MATCH ()-[e]->() RETURN type(e), properties(e) ORDER BY e"),
+    ("index count", "CALL db.indexes() YIELD label RETURN count(label)"),
+]
+
+# ═══════════════════════════ Test class ═══════════════════════════════
+
+class testRdbCompat():
+    def __init__(self):
+        self.env, self.db = Env(enableDebugCommand=True)
+        self.redis_con = self.env.getConnection()
+        self.rust_port = self.env.port
+        # Allow Docker containers to connect to the Rust server
+        self.redis_con.execute_command('CONFIG', 'SET', 'bind', '0.0.0.0')
+        self.redis_con.execute_command('CONFIG', 'SET', 'protected-mode', 'no')
+
+    # ── Test 1: C -> Rust (simple) ──
+
+    def test01_c_to_rust_simple(self):
+        """C produces RDB, Rust loads it via replication."""
+        key = 'G'
+
+        container, c_port = run_db(FALKORDB_C_IMAGE)
+        try:
+            wait_for_db(c_port)
+            c_db = FalkorDB(port=c_port)
+            c_graph = c_db.select_graph(key)
+            create_simple_graph(c_graph)
+            expected = capture_state(c_graph, SIMPLE_VERIFICATION)
+
+            # Rust replicates from C
+            self.redis_con.execute_command('REPLICAOF', 'localhost', str(c_port))
+            wait_for_replication(self.redis_con)
+            self.redis_con.execute_command('REPLICAOF', 'NO', 'ONE')
+        finally:
+            stop_db(container)
+
+        r_graph = self.db.select_graph(key)
+        actual = capture_state(r_graph, SIMPLE_VERIFICATION)
+        assert_state_eq(self.env, expected, actual)
+
+    # ── Test 2: C -> Rust (random) ──
+
+    def test02_c_to_rust_random(self):
+        """C produces random graph RDB, Rust loads it via replication."""
+        key = 'R'
+
+        container, c_port = run_db(FALKORDB_C_IMAGE)
+        try:
+            wait_for_db(c_port)
+            c_db = FalkorDB(port=c_port)
+            c_graph = c_db.select_graph(key)
+            create_random_graph(c_graph)
+            expected = capture_state(c_graph, RANDOM_VERIFICATION)
+
+            self.redis_con.execute_command('REPLICAOF', 'localhost', str(c_port))
+            wait_for_replication(self.redis_con)
+            self.redis_con.execute_command('REPLICAOF', 'NO', 'ONE')
+        finally:
+            stop_db(container)
+
+        r_graph = self.db.select_graph(key)
+        actual = capture_state(r_graph, RANDOM_VERIFICATION)
+        assert_state_eq(self.env, expected, actual)
+
+    # ── Test 3: Rust -> C (simple) ──
+
+    def test03_rust_to_c_simple(self):
+        """Rust produces RDB, C loads it via replication."""
+        key = 'G'
+        self.redis_con.flushall()
+
+        r_graph = self.db.select_graph(key)
+        create_simple_graph(r_graph)
+        expected = capture_state(r_graph, SIMPLE_VERIFICATION)
+
+        container, c_port = run_db(FALKORDB_C_IMAGE)
+        try:
+            wait_for_db(c_port)
+            import redis as _redis
+            c_conn = _redis.Redis(host='localhost', port=c_port)
+
+            # C replicates from Rust
+            c_conn.execute_command('REPLICAOF', 'host.docker.internal', str(self.rust_port))
+            wait_for_replication(c_conn)
+            c_conn.execute_command('REPLICAOF', 'NO', 'ONE')
+
+            c_db = FalkorDB(port=c_port)
+            c_graph = c_db.select_graph(key)
+            actual = capture_state(c_graph, SIMPLE_VERIFICATION)
+            assert_state_eq(self.env, expected, actual)
+        finally:
+            stop_db(container)
+
+    # ── Test 4: Rust -> C (random) ──
+
+    def test04_rust_to_c_random(self):
+        """Rust produces random graph RDB, C loads it via replication."""
+        key = 'R'
+        self.redis_con.flushall()
+
+        r_graph = self.db.select_graph(key)
+        create_random_graph(r_graph)
+        expected = capture_state(r_graph, RANDOM_VERIFICATION)
+
+        container, c_port = run_db(FALKORDB_C_IMAGE)
+        try:
+            wait_for_db(c_port)
+            import redis as _redis
+            c_conn = _redis.Redis(host='localhost', port=c_port)
+
+            c_conn.execute_command('REPLICAOF', 'host.docker.internal', str(self.rust_port))
+            wait_for_replication(c_conn)
+            c_conn.execute_command('REPLICAOF', 'NO', 'ONE')
+
+            c_db = FalkorDB(port=c_port)
+            c_graph = c_db.select_graph(key)
+            actual = capture_state(c_graph, RANDOM_VERIFICATION)
+            assert_state_eq(self.env, expected, actual)
+        finally:
+            stop_db(container)
diff --git a/tests/flow/test_rdb_load.py b/tests/flow/test_rdb_load.py
index a6fbf19d..9c957beb 100644
--- a/tests/flow/test_rdb_load.py
+++ b/tests/flow/test_rdb_load.py
@@ -1,13 +1,5 @@
 from common import *
 
-# TODO: when introducing new encoder/decoder this needs to be updated consider
-# using GRAPH.DEBUG command to be able to get this data
-keys = {
-    b'x': b'\x07\x81\x82\xb6\xa9\x85\xd6\xadh\n\x05\x02x\x00\x02\x1e\x02\x00\x02\x01\x02\x00\x02\x03\x02\x01\x05\x02v\x00\x02\x01\x02\x00\x05\x02N\x00\x02\x01\x02\x01\x05\x02v\x00\x02\x00\x02\x01\x02\x01\x02\n\x02\x00\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x01\x02\x01\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x02\x02\x02\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x03\x02\x03\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x04\x02\x04\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x05\x02\x05\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x06\x02\x06\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x07\x02\x07\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x08\x02\x08\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\t\x02\t\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\n\x00\t\x00\x84\xf96Z\xd1\x98\xec\xc0',
-    b'{x}x_a244836f-fe81-4f8d-8ee2-83fc3fbcf102': b'\x07\x81\x82\xb6\xa9\x86g\xadh\n\x05\x02x\x00\x02\x1e\x02\x00\x02\x01\x02\x00\x02\x03\x02\x01\x05\x02v\x00\x02\x01\x02\x00\x05\x02N\x00\x02\x01\x02\x01\x05\x02v\x00\x02\x00\x02\x01\x02\x01\x02\n\x02\n\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x0b\x02\x0b\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x0c\x02\x0c\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\r\x02\r\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x0e\x02\x0e\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x0f\x02\x0f\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x10\x02\x10\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x11\x02\x11\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x12\x02\x12\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x13\x02\x13\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x14\x00\t\x00\x13H\x11\xb8\x15\xd3\xdc~',
-    b'{x}x_53ab30bb-1dbb-47b2-a41d-cac3acd68b8c': b'\x07\x81\x82\xb6\xa9\x86g\xadh\n\x05\x02x\x00\x02\x1e\x02\x00\x02\x01\x02\x00\x02\x03\x02\x01\x05\x02v\x00\x02\x01\x02\x00\x05\x02N\x00\x02\x01\x02\x01\x05\x02v\x00\x02\x00\x02\x05\x02\x01\x02\n\x02\x02\x02\x00\x02\x03\x02\x00\x02\x04\x02\x00\x02\x05\x02\x01\x02\x14\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x15\x02\x15\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x16\x02\x16\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x17\x02\x17\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x18\x02\x18\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x19\x02\x19\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x1a\x02\x1a\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x1b\x02\x1b\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x1c\x02\x1c\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x1d\x02\x1d\x02\x01\x02\x00\x02\x01\x02\x00\x02`\x00\x02\x1e\x00\t\x00\x1b\xa64\xd6\xf5\x0bk\xa6'
-}
-
 
 class testRdbLoad():
     def __init__(self):
@@ -19,39 +11,66 @@ def validate_key_count(self, n):
         keys = self.conn.keys('*')
         self.env.assertEqual(len(keys), n)
 
-    # restore the key data
-    def restore_key(self, key):
-        self.conn.restore(key, '0', keys[key])
-
     # validate that the imported data exists
     def _test_data(self):
         expected = [[i] for i in range(1, 31)]
-        q = "MATCH (n:N) RETURN n.v"
+        q = "MATCH (n:N) RETURN n.v ORDER BY n.v"
         result = self.conn.execute_command("GRAPH.RO_QUERY", "x", q)
         self.env.assertEqual(result[1], expected)
-    
+
     def test_rdb_load(self):
+        # Create a graph with 30 nodes so virtual keys are generated
+        graph = self.db.select_graph("x")
+        graph.query("UNWIND range(1, 30) AS v CREATE (:N {v: v})")
+
+        # Verify data before save
+        self._test_data()
+
+        # Use GRAPH.DEBUG AUX START to create virtual keys
         aux = self.conn.execute_command("GRAPH.DEBUG", "AUX", "START")
         self.env.assertEqual(aux, 1)
 
-        self.restore_key(b'{x}x_a244836f-fe81-4f8d-8ee2-83fc3fbcf102')
-        self.restore_key(b'{x}x_53ab30bb-1dbb-47b2-a41d-cac3acd68b8c')
+        # Dump all keys (graphdata + graphmeta virtual keys)
+        all_keys = self.conn.keys('*')
+        self.env.assertEqual(len(all_keys), 3)  # 1 graphdata key + 2 graphmeta keys
+        dumps = {}
+        for key in all_keys:
+            dumps[key] = self.conn.dump(key)
 
-        self.conn.flushall()
+        # Separate graphdata key from graphmeta keys
+        graphdata_key = None
+        graphmeta_keys = []
+        for key in all_keys:
+            # The graphdata key is just the graph name 'x'
+            key_str = key.decode() if isinstance(key, bytes) else key
+            if key_str == 'x':
+                graphdata_key = key
+            else:
+                graphmeta_keys.append(key)
+
+        self.env.assertIsNotNone(graphdata_key)
 
+        # Flush and verify empty
+        self.conn.flushall()
         self.validate_key_count(0)
 
+        # Start AUX load simulation
         aux = self.conn.execute_command("GRAPH.DEBUG", "AUX", "START")
         self.env.assertEqual(aux, 1)
 
-        self.restore_key(b'{x}x_a244836f-fe81-4f8d-8ee2-83fc3fbcf102')
-        self.restore_key(b'{x}x_53ab30bb-1dbb-47b2-a41d-cac3acd68b8c')
-        self.restore_key(b'x')
+        # Restore graphmeta keys first, then the graphdata key
+        for key in graphmeta_keys:
+            self.conn.restore(key, '0', dumps[key])
+
+        self.conn.restore(graphdata_key, '0', dumps[graphdata_key])
 
+        # Finalize
         aux = self.conn.execute_command("GRAPH.DEBUG", "AUX", "END")
         self.env.assertEqual(aux, 0)
 
+        # Verify only the graphdata key remains (graphmeta keys cleaned up)
         self.validate_key_count(1)
         self._test_data()
 
+        # Verify save works after load
         self.conn.save()
diff --git a/tests/flow/test_replication.py b/tests/flow/test_replication.py
index 9ad0529b..6b39d58b 100644
--- a/tests/flow/test_replication.py
+++ b/tests/flow/test_replication.py
@@ -73,7 +73,7 @@ def test_CRUD_replication(self):
         create_node_fulltext_index(src, 'L', 'title', 'desc', sync=True)
 
         # create full-text index with index config
-        q = "CALL db.idx.fulltext.createNodeIndex({label: 'L1', language: 'german', stopwords: ['a', 'b'] }, 'title', 'desc')"
+        q = "CREATE FULLTEXT INDEX FOR (n:L1) ON (n.title, n.desc) OPTIONS {language: 'german', stopwords: ['a', 'b']}"
         src.query(q)
 
         #-----------------------------------------------------------------------