Implement RDB serialization and deserialization for graph data

.devcontainer/Dockerfile

@@ -49,7 +49,7 @@ RUN ln -sf /usr/bin/clang-22 /usr/local/bin/clang && \
 RUN python3 -m venv /data/venv
 
 # Install Python test dependencies (setuptools needed for pkg_resources)
-RUN /data/venv/bin/pip install setuptools behave falkordb hypothesis pytest pytest-benchmark RLTest
+RUN /data/venv/bin/pip install setuptools behave falkordb falkordb-bulk-loader hypothesis pytest pytest-benchmark RLTest
 
 # Install Rust (needed before running scripts)
 RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y

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]

build/Dockerfile

@@ -13,7 +13,7 @@ RUN apt install -y git make cmake curl zlib1g zlib1g-dev libpolly-22-dev libomp-
 
 RUN python3 -m venv venv
 
-RUN venv/bin/pip install behave falkordb hypothesis pytest pytest-benchmark RLTest
+RUN venv/bin/pip install behave falkordb falkordb-bulk-loader hypothesis pytest pytest-benchmark RLTest
 
 RUN git clone --branch dev2 --single-branch https://github.com/DrTimothyAldenDavis/GraphBLAS.git
 # RUN git clone --branch v10.3.1 --single-branch https://github.com/DrTimothyAldenDavis/GraphBLAS.git

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

@@ -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
@@ -52,8 +55,10 @@ 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_rdb_load.py
 tests/flow/test_reduce.py
 tests/flow/test_results.py
 tests/flow/test_reversed_patterns

flow_tests_todo.txt

@@ -20,16 +20,10 @@ 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
 tests/flow/test_replication_states.py
-tests/flow/test_graph_versioning.py
 
 ## Redis Integration & Server Features
-tests/flow/test_acl.py
 tests/flow/test_bolt.py
 tests/flow/test_bulk_insertion.py
 tests/flow/test_graph_copy.py
@@ -47,5 +41,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

graph/src/graph/attribute_store.rs

@@ -83,7 +83,13 @@
 //! 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, sync::Arc};
+use std::{
+    collections::HashMap,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicU64, Ordering},
+    },
+};
 
 use fjall::{
     Database, Keyspace, KeyspaceCreateOptions, Readable, Snapshot, config::HashRatioPolicy,
@@ -92,6 +98,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 +141,12 @@ 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: Mutex<Option<RoaringTreemap>>,
+    /// Encoding context: maximum entity ID (set before serialization).
+    encode_max_id: AtomicU64,
+    /// Encoding context: mapping from local attr IDs to global attr IDs (set before serialization).
+    encode_attr_remap: Mutex<Option<Vec<u16>>>,
 }
 
 impl Clone for AttributeStore {
@@ -148,6 +161,9 @@ impl Clone for AttributeStore {
             version: self.version,
             dirty_entities: self.dirty_entities.clone(),
             pending_deletes: self.pending_deletes.clone(),
+            encode_deleted: Mutex::new(None),
+            encode_max_id: AtomicU64::new(0),
+            encode_attr_remap: Mutex::new(None),
         }
     }
 }
@@ -172,6 +188,9 @@ impl AttributeStore {
             version,
             dirty_entities: RoaringTreemap::new(),
             pending_deletes: RoaringTreemap::new(),
+            encode_deleted: Mutex::new(None),
+            encode_max_id: AtomicU64::new(0),
+            encode_attr_remap: Mutex::new(None),
         }
     }
 
@@ -235,6 +254,9 @@ impl AttributeStore {
             version,
             dirty_entities: RoaringTreemap::new(),
             pending_deletes: RoaringTreemap::new(),
+            encode_deleted: Mutex::new(None),
+            encode_max_id: AtomicU64::new(0),
+            encode_attr_remap: Mutex::new(None),
         }
     }
 
@@ -369,10 +391,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 +641,29 @@ impl AttributeStore {
     pub const fn cache(&self) -> &Arc<AttributeCache> {
         &self.cache
     }
+
+    /// Set encoding context needed by `Encode::encode_with_range`.
+    ///
+    /// Builds a mapping from local attribute IDs (indices in this store's attrs_name)
+    /// to global attribute IDs (indices in the provided global_attrs list).
+    pub fn set_encode_context(
+        &self,
+        deleted: &RoaringTreemap,
+        max_id: u64,
+        global_attrs: &[Arc<String>],
+    ) {
+        *self.encode_deleted.lock().unwrap() = Some(deleted.clone());
+        self.encode_max_id.store(max_id, Ordering::Relaxed);
+
+        // Build mapping from local attr ID to global attr ID
+        let mut remap = vec![u16::MAX; self.attrs_name.len()];
+        for (local_id, local_name) in self.attrs_name.iter().enumerate() {
+            if let Some(global_id) = global_attrs.iter().position(|n| n == local_name) {
+                remap[local_id] = global_id as u16;
+            }
+        }
+        *self.encode_attr_remap.lock().unwrap() = Some(remap);
+    }
 }
 
 // SAFETY: AttributeStore is Send+Sync because:
@@ -628,3 +673,95 @@ 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.lock().unwrap();
+        let deleted = binding.as_ref().expect("encode context not set");
+        let max_id = self.encode_max_id.load(Ordering::Relaxed);
+
+        let remap_binding = self.encode_attr_remap.lock().unwrap();
+        let remap = remap_binding.as_ref().expect("encode attr remap not set");
+
+        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 (local_attr_id, value) in props {
+                // Remap local attribute ID to global attribute ID
+                let global_attr_id = if (local_attr_id as usize) < remap.len() {
+                    remap[local_attr_id as usize]
+                } else {
+                    local_attr_id
+                };
+                w.write_unsigned(global_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(())
+    }
+}