feat: RDB persistence, virtual keys, and module wiring

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

graph/src/graph/graph.rs

@@ -1427,7 +1427,7 @@ impl Graph {
 
     pub fn delete_relationships(
         &mut self,
-        rels: HashMap<RelationshipId, (NodeId, NodeId)>,
+        rels: &HashMap<RelationshipId, (NodeId, NodeId)>,
     ) -> Result<(), String> {
         self.deleted_relationships
             .extend(rels.keys().map(|id| id.0));
@@ -1438,7 +1438,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)))
         {

graph/src/runtime/pending.rs

@@ -709,8 +709,8 @@ impl Pending {
         }
         if !self.deleted_relationships.is_empty() {
             stats.borrow_mut().relationships_deleted += self.deleted_relationships.len();
-            let rels = std::mem::take(&mut self.deleted_relationships);
-            g.borrow_mut().delete_relationships(rels)?;
+            g.borrow_mut()
+                .delete_relationships(&self.deleted_relationships)?;
         }
         // Commit attribute changes and indexes after all deletions have been
         // applied. This ensures relationship_attrs.remove() pending_deletes

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}"))),
+    }
+}

src/commands/effect.rs

@@ -230,7 +230,7 @@ fn apply_effects(
                 let rel = RelationshipId::from(rel_id);
                 let mut rels = HashMap::new();
                 rels.insert(rel, (NodeId::from(src_id), NodeId::from(dst_id)));
-                g.delete_relationships(rels)?;
+                g.delete_relationships(&rels)?;
             }
 
             EFFECT_ADD_SCHEMA => {

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;
@@ -31,6 +32,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;

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,13 +97,40 @@ 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);
     }
 
     let graph = Arc::new(RwLock::new(ThreadedGraph::new(
         *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

src/lib.rs

@@ -42,11 +42,12 @@ mod graph_core;
 mod module_init;
 mod redis_type;
 mod reply;
+mod serializers;
 
 use allocator::ThreadCountingAllocator;
 use commands::{
-    graph_config, graph_delete, graph_explain, graph_list, graph_memory, graph_query, graph_record,
-    graph_ro_query, graph_udf,
+    graph_config, graph_debug, graph_delete, graph_effect, graph_explain, graph_list, graph_memory,
+    graph_query, graph_record, graph_ro_query, graph_udf,
 };
 use config::{
     CONFIGURATION_CACHE_SIZE, CONFIGURATION_CMD_INFO, CONFIGURATION_DELAY_INDEXING,
@@ -56,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, ""],
@@ -74,6 +75,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: [

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(()) => {}