Implement RDB serialization and deserialization for graph data

Cargo.toml

@@ -28,7 +28,9 @@ graph = { path = "graph", version = "0.1.0" }
 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"
 ryu = "1.0.23"
+thin-vec = "0.2.14"
 orx-tree = "2.2.0"
 
 [build-dependencies]

flow.sh

@@ -42,7 +42,7 @@ fi
 # TEST="tests/flow/test_function_calls" FAIL_FAST=1 ./flow.sh
 
 if [[ ${#TEST_FILTER[@]} -eq 0 ]]; then
-    RLTest -f "$TESTS_FILE" --module "$TARGET_DIR/$TARGET" --no-progress $PARALLELISM $STOP_ON_FAILURE --clear-logs --log-dir tests/flow/logs $V
+    RLTest -f "$TESTS_FILE" --module "$TARGET_DIR/$TARGET" --no-progress $PARALLELISM $STOP_ON_FAILURE --clear-logs --log-dir tests/flow/logs --enable-debug-command --enable-protected-configs $V
 else
-    RLTest "${TEST_FILTER[@]}" --module "$TARGET_DIR/$TARGET" --no-progress $PARALLELISM $STOP_ON_FAILURE --clear-logs --log-dir tests/flow/logs $V
+    RLTest "${TEST_FILTER[@]}" --module "$TARGET_DIR/$TARGET" --no-progress $PARALLELISM $STOP_ON_FAILURE --clear-logs --log-dir tests/flow/logs --enable-debug-command --enable-protected-configs $V
 fi

flow_tests_done.txt

@@ -15,6 +15,7 @@ tests/flow/test_config.py
 tests/flow/test_create_clause
 tests/flow/test_distinct
 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
@@ -52,6 +53,7 @@ tests/flow/test_path_algorithms.py
 tests/flow/test_path_filter
 tests/flow/test_path_projections.py
 tests/flow/test_pending_queries_limit.py
+tests/flow/test_persistency.py
 tests/flow/test_point
 tests/flow/test_query_validation
 tests/flow/test_reduce.py

flow_tests_todo.txt

@@ -20,8 +20,6 @@ tests/flow/test_profile.py
 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

graph/src/graph/attribute_store.rs

@@ -83,6 +83,7 @@
 //! Each attribute is stored as a separate fjall entry:
 //! `entity_id (8 bytes big-endian) + attr_idx (2 bytes big-endian)`
 
+use std::cell::RefCell;
 use std::{collections::HashMap, sync::Arc};
 
 use fjall::{
@@ -92,6 +93,7 @@ use once_cell::sync::OnceCell;
 use roaring::RoaringTreemap;
 
 use super::attribute_cache::AttributeCache;
+use super::graphblas::serialization::{Decode, Encode, Reader, Writer};
 use crate::runtime::{ordermap::OrderMap, orderset::OrderSet, value::Value};
 
 /// Create a composite key from entity ID and attribute index.
@@ -134,6 +136,10 @@ pub struct AttributeStore {
     dirty_entities: RoaringTreemap,
     /// Entity IDs pending full deletion (all attributes) — applied on commit, cleared on rollback.
     pending_deletes: RoaringTreemap,
+    /// Encoding context: deleted entity IDs (set before serialization).
+    encode_deleted: RefCell<Option<RoaringTreemap>>,
+    /// Encoding context: maximum entity ID (set before serialization).
+    encode_max_id: RefCell<u64>,
 }
 
 impl Clone for AttributeStore {
@@ -148,6 +154,8 @@ impl Clone for AttributeStore {
             version: self.version,
             dirty_entities: self.dirty_entities.clone(),
             pending_deletes: self.pending_deletes.clone(),
+            encode_deleted: RefCell::new(None),
+            encode_max_id: RefCell::new(0),
         }
     }
 }
@@ -172,6 +180,8 @@ impl AttributeStore {
             version,
             dirty_entities: RoaringTreemap::new(),
             pending_deletes: RoaringTreemap::new(),
+            encode_deleted: RefCell::new(None),
+            encode_max_id: RefCell::new(0),
         }
     }
 
@@ -235,6 +245,8 @@ impl AttributeStore {
             version,
             dirty_entities: RoaringTreemap::new(),
             pending_deletes: RoaringTreemap::new(),
+            encode_deleted: RefCell::new(None),
+            encode_max_id: RefCell::new(0),
         }
     }
 
@@ -369,10 +381,10 @@ impl AttributeStore {
     pub fn get_all_attrs_by_id(
         &self,
         key: u64,
-    ) -> impl Iterator<Item = (u16, Value)> + '_ {
-        let cached = self.cache.get_entity(key, self.version);
-        let attrs = cached.unwrap_or_else(|| self.populate_cache_from_fjall(key));
-        attrs.into_iter()
+    ) -> Vec<(u16, Value)> {
+        self.cache
+            .get_entity(key, self.version)
+            .unwrap_or_else(|| self.populate_cache_from_fjall(key))
     }
 
     // ---- write path (cache only) ----------------------------------------
@@ -619,6 +631,20 @@ impl AttributeStore {
     pub const fn cache(&self) -> &Arc<AttributeCache> {
         &self.cache
     }
+
+    /// Set encoding context needed by `Encode::encode_with_range`.
+    ///
+    /// `node_attrs` and `rel_attrs` are the attribute name sets from the node and
+    /// relationship stores respectively, used to build the canonical global
+    /// attribute ordering (nodes first, then relationships, deduplicated).
+    pub fn set_encode_context(
+        &self,
+        deleted: &RoaringTreemap,
+        max_id: u64,
+    ) {
+        *self.encode_deleted.borrow_mut() = Some(deleted.clone());
+        *self.encode_max_id.borrow_mut() = max_id;
+    }
 }
 
 // SAFETY: AttributeStore is Send+Sync because:
@@ -628,3 +654,86 @@ impl AttributeStore {
 // - All other fields (`RoaringTreemap`, `OrderSet`, etc.) are owned and not shared
 unsafe impl Send for AttributeStore {}
 unsafe impl Sync for AttributeStore {}
+
+impl Encode<19> for AttributeStore {
+    fn encode(
+        &self,
+        _w: &mut dyn Writer,
+    ) {
+        unimplemented!("use encode_with_range for AttributeStore")
+    }
+
+    fn encode_with_range(
+        &self,
+        w: &mut dyn Writer,
+        count: u64,
+        offset: u64,
+    ) {
+        let binding = self.encode_deleted.borrow();
+        let deleted = binding.as_ref().expect("encode context not set");
+        let max_id = *self.encode_max_id.borrow();
+
+        let mut skipped = 0u64;
+        let mut encoded = 0u64;
+
+        for id in 0..=max_id {
+            if deleted.contains(id) {
+                continue;
+            }
+            if skipped < offset {
+                skipped += 1;
+                continue;
+            }
+
+            w.write_unsigned(id);
+
+            let props: Vec<(u16, Value)> = self.get_all_attrs_by_id(id);
+            w.write_unsigned(props.len() as u64);
+
+            for (attr_id, value) in props {
+                w.write_unsigned(attr_id as u64);
+                value.encode(w);
+            }
+
+            encoded += 1;
+            if encoded >= count {
+                break;
+            }
+        }
+    }
+}
+
+impl Decode<19> for AttributeStore {
+    fn decode(_r: &mut dyn Reader) -> Result<Self, String> {
+        unimplemented!("use decode_with_count for AttributeStore")
+    }
+
+    fn decode_with_count(
+        &mut self,
+        r: &mut dyn Reader,
+        count: u64,
+    ) -> Result<(), String> {
+        for _ in 0..count {
+            let entity_id = r.read_unsigned()?;
+            let attr_count = r.read_unsigned()?;
+
+            let mut entity_attrs = OrderMap::default();
+            for _ in 0..attr_count {
+                let attr_id = r.read_unsigned()? as u16;
+                let value = Value::decode(r)?;
+
+                if (attr_id as usize) < self.attrs_name.len() {
+                    let attr_name = self.attrs_name[attr_id as usize].clone();
+                    entity_attrs.insert(attr_name, value);
+                }
+            }
+
+            if !entity_attrs.is_empty() {
+                let mut batch = HashMap::new();
+                batch.insert(entity_id, entity_attrs);
+                self.insert_attrs(&batch)?;
+            }
+        }
+        Ok(())
+    }
+}