Graph version (#1382)

* add graph version to graph context object

* change graph version from string to uint32

* testing graph version

* response with an error when graph version mismatch

* release graph context on version mismatch

* address PR comments

* increase command arity to accomadate graph version argument

* address PR comments

Co-authored-by: DvirDukhan <[email protected]>

Author: swilly22

src/commands/cmd_dispatcher.c

@@ -4,24 +4,29 @@
 * This file is available under the Redis Labs Source Available License Agreement
 */
 
+#include "../RG.h"
 #include "commands.h"
 #include "cmd_context.h"
 #include "../RG.h"
 #include <assert.h>
 #include <strings.h>
 
+#define GRAPH_VERSION_MISSING -1
+
 // Command handler function pointer.
 typedef void(*Command_Handler)(void *args);
 
 // Read configuration flags, returning REDIS_MODULE_ERR if flag parsing failed.
-static int _read_flags(RedisModuleString **argv, int argc, bool *compact, long long *timeout,
-					   char **errmsg) {
+static int _read_flags(RedisModuleString **argv, int argc, bool *compact,
+		long long *timeout, uint *graph_version, char **errmsg) {
+
 	ASSERT(compact);
 	ASSERT(timeout);
 
 	// set defaults
 	*timeout = 0;      // no timeout
 	*compact = false;  // verbose
+	*graph_version = GRAPH_VERSION_MISSING;
 
 	// GRAPH.QUERY <GRAPH_KEY> <QUERY>
 	// make sure we've got more than 3 arguments
@@ -37,6 +42,24 @@ static int _read_flags(RedisModuleString **argv, int argc, bool *compact, long l
 			continue;
 		}
 
+		if(!strcasecmp(arg, "version")) {
+			long long v = GRAPH_VERSION_MISSING;
+			int err = REDISMODULE_ERR;
+			if(i < argc - 1) {
+				i++; // Set the current argument to the version value.
+				err = RedisModule_StringToLongLong(argv[i], &v);
+				*graph_version = v;
+			}
+
+			// Emit error on missing, negative, or non-numeric version values.
+			if(err != REDISMODULE_OK || v < 0 || v > UINT_MAX) {
+				asprintf(errmsg, "Failed to parse graph version value");
+				return REDISMODULE_ERR;
+			}
+
+			continue;
+		}
+
 		// query timeout
 		if(!strcasecmp(arg, "timeout")) {
 			int err = REDISMODULE_ERR;
@@ -55,6 +78,20 @@ static int _read_flags(RedisModuleString **argv, int argc, bool *compact, long l
 	return REDISMODULE_OK;
 }
 
+// Returns false if client provided a graph version
+// which mismatch the current graph version
+static bool _verifyGraphVersion(GraphContext *gc, uint version) {
+	// caller did not specify graph version
+	if(version == GRAPH_VERSION_MISSING) return true;
+	return (GraphContext_GetVersion(gc) == version);
+}
+
+static void _rejectOnVersionMismatch(RedisModuleCtx *ctx, uint version) {
+	RedisModule_ReplyWithArray(ctx, 2);
+	RedisModule_ReplyWithError(ctx, "version mismatch");
+	RedisModule_ReplyWithLongLong(ctx, version);
+}
+
 // Return true if the command has a valid number of arguments.
 static inline bool _validate_command_arity(GRAPH_Commands cmd, int arity) {
 	switch(cmd) {
@@ -63,7 +100,7 @@ static inline bool _validate_command_arity(GRAPH_Commands cmd, int arity) {
 	case CMD_EXPLAIN:
 	case CMD_PROFILE:
 		// Expect a command, graph name, a query, and optional config flags.
-		return arity >= 3 && arity <= 6;
+		return arity >= 3 && arity <= 8;
 	case CMD_SLOWLOG:
 		// Expect just a command and graph name.
 		return arity == 2;
@@ -103,17 +140,20 @@ static GRAPH_Commands determine_command(const char *cmd_name) {
 }
 
 int CommandDispatch(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
-	CommandCtx *context;
+	char *errmsg;
+	bool compact;
+	uint version;
+	long long timeout;
+	CommandCtx *context = NULL;
 
 	RedisModuleString *graph_name = argv[1];
 	RedisModuleString *query = (argc > 2) ? argv[2] : NULL;
 	const char *command_name = RedisModule_StringPtrLen(argv[0], NULL);
 	GRAPH_Commands cmd = determine_command(command_name);
+
 	// Parse additional query arguments.
-	char *errmsg;
-	bool compact;
-	long long timeout;
-	int res = _read_flags(argv, argc, &compact, &timeout, &errmsg);
+	int res = _read_flags(argv, argc, &compact, &timeout, &version, &errmsg);
+
 	if(res == REDISMODULE_ERR) {
 		// Emit error and exit if argument parsing failed.
 		RedisModule_ReplyWithError(ctx, errmsg);
@@ -123,11 +163,19 @@ int CommandDispatch(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
 	}
 
 	if(_validate_command_arity(cmd, argc) == false) return RedisModule_WrongArity(ctx);
+
 	Command_Handler handler = get_command_handler(cmd);
 	GraphContext *gc = GraphContext_Retrieve(ctx, graph_name, true, true);
 	// If the GraphContext is null, key access failed and an error has been emitted.
 	if(!gc) return REDISMODULE_ERR;
 
+	// return incase caller provided a mismatched graph version
+	if(!_verifyGraphVersion(gc, version)) {
+		_rejectOnVersionMismatch(ctx, GraphContext_GetVersion(gc));
+		GraphContext_Release(gc);
+		return REDISMODULE_OK;
+	}
+
 	/* Determin query execution context
 	 * queries issued within a LUA script or multi exec block must
 	 * run on Redis main thread, others can run on different threads. */

src/graph/graphcontext.c

@@ -7,7 +7,9 @@
 #include <sys/param.h>
 #include <pthread.h>
 #include "graphcontext.h"
+#include "../RG.h"
 #include "../util/arr.h"
+#include "../util/uuid.h"
 #include "../query_ctx.h"
 #include "../redismodule.h"
 #include "../util/rmalloc.h"
@@ -25,6 +27,7 @@ extern RedisModuleType *GraphContextRedisModuleType;
 
 // Forward declarations.
 static void _GraphContext_Free(void *arg);
+static void _GraphContext_UpdateVersion(GraphContext *gc, const char *str);
 
 static inline void _GraphContext_IncreaseRefCount(GraphContext *gc) {
 	__atomic_fetch_add(&gc->ref_count, 1, __ATOMIC_RELAXED);
@@ -53,6 +56,7 @@ GraphContext *GraphContext_New(const char *graph_name, size_t node_cap, size_t e
 
 	gc->ref_count = 0;      // No refences.
 	gc->index_count = 0;    // No indicies.
+	gc->version = 0;        // Initial graph version.
 
 	// Initialize the graph's matrices and datablock storage
 	gc->g = Graph_New(node_cap, edge_cap);
@@ -186,6 +190,32 @@ void GraphContext_Rename(GraphContext *gc, const char *name) {
 	gc->graph_name = rm_strdup(name);
 }
 
+XXH32_hash_t GraphContext_GetVersion(const GraphContext *gc) {
+	ASSERT(gc != NULL);
+
+	return gc->version;
+}
+
+// Update graph context version
+static void _GraphContext_UpdateVersion(GraphContext *gc, const char *str) {
+	ASSERT(gc != NULL);
+	ASSERT(str != NULL);
+
+	/* Update graph version by hashing 'str' representing the current
+	 * addition to the graph schema: (Label, Relationship-type, Attribute)
+	 *
+	 * Using the current graph version as a seed, by doing so we avoid
+	 * hashing the entire graph schema on each change, while guaranteeing the
+	 * exact same version across a cluster: same graph version on both
+	 * primary and replica shards. */
+
+	XXH32_state_t *state = XXH32_createState();
+	XXH32_reset(state, gc->version);
+	XXH32_update(state, str, strlen(str));
+	gc->version = XXH32_digest(state);
+	XXH32_freeState(state);
+}
+
 //------------------------------------------------------------------------------
 // Schema API
 //------------------------------------------------------------------------------
@@ -232,6 +262,9 @@ Schema *GraphContext_AddSchema(GraphContext *gc, const char *label, SchemaType t
 		gc->relation_schemas = array_append(gc->relation_schemas, schema);
 	}
 
+	// new schema added, update graph version
+	_GraphContext_UpdateVersion(gc, label);
+
 	return schema;
 }
 
@@ -279,6 +312,9 @@ Attribute_ID GraphContext_FindOrAddAttribute(GraphContext *gc, const char *attri
 					  attribute_id,
 					  NULL);
 			gc->string_mapping = array_append(gc->string_mapping, rm_strdup(attribute));
+
+			// new attribute been added, update graph version
+			_GraphContext_UpdateVersion(gc, attribute);
 		}
 	}
 

src/graph/graphcontext.h

@@ -17,6 +17,16 @@
 #include "../serializers/decode_context.h"
 #include "../util/cache/cache.h"
 
+/* GraphContext holds refrences to various elements of a graph object
+ * It is the value sitting behind a Redis graph key
+ *
+ * the graph context is versioned, the version value itself is meaningless
+ * it is used as a "signature" for the graph schema: (labels, relationship-types
+ * and attribute set) client libraries which cache the mapping between graph
+ * schema elements and their internal IDs (see COMPACT reply formatter)
+ * can use the graph version to understand if the schema was modified
+ * and take action accordingly */
+
 typedef struct {
 	Graph *g;                               // Container for all matrices and entity properties
 	int ref_count;                          // Number of active references.
@@ -31,9 +41,13 @@ typedef struct {
 	GraphEncodeContext *encoding_context;   // Encode context of the graph.
 	GraphDecodeContext *decoding_context;   // Decode context of the graph.
 	Cache **cache_pool;                     // Pool of execution plan caches, one per thread.
+	XXH32_hash_t version;                   // Graph version.
 } GraphContext;
 
-/* GraphContext API */
+//------------------------------------------------------------------------------
+// GraphContext API
+//------------------------------------------------------------------------------
+
 // Creates and initializes a graph context struct.
 GraphContext *GraphContext_New(const char *graph_name, size_t node_cap, size_t edge_cap);
 /* Retrive the graph context according to the graph name
@@ -48,7 +62,16 @@ void GraphContext_MarkWriter(RedisModuleCtx *ctx, GraphContext *gc);
 // Mark graph as deleted, reduce graph reference count by 1.
 void GraphContext_Delete(GraphContext *gc);
 
-/* Schema API */
+// Rename a graph context.
+void GraphContext_Rename(GraphContext *gc, const char *name);
+
+// Get graph context version
+XXH32_hash_t GraphContext_GetVersion(const GraphContext *gc);
+
+//------------------------------------------------------------------------------
+// Schema API
+//------------------------------------------------------------------------------
+
 // Retrieve number of schemas created for given type.
 unsigned short GraphContext_SchemaCount(const GraphContext *gc, SchemaType t);
 // Retrieve the specific schema for the provided ID
@@ -70,7 +93,10 @@ const char *GraphContext_GetAttributeString(GraphContext *gc, Attribute_ID id);
 // Retrieve an attribute ID given a string, or ATTRIBUTE_NOTFOUND if attribute doesn't exist.
 Attribute_ID GraphContext_GetAttributeID(GraphContext *gc, const char *str);
 
-/* Index API */
+//------------------------------------------------------------------------------
+// Index API
+//------------------------------------------------------------------------------
+
 bool GraphContext_HasIndices(GraphContext *gc);
 // Attempt to retrieve an index on the given label and attribute
 Index *GraphContext_GetIndex(const GraphContext *gc, const char *label, Attribute_ID *attribute_id,
@@ -93,10 +119,10 @@ GraphContext *GraphContext_GetRegisteredGraphContext(const char *graph_name);
 // Remove GraphContext from global array
 void GraphContext_RemoveFromRegistry(GraphContext *gc);
 
-// Rename a graph context.
-void GraphContext_Rename(GraphContext *gc, const char *name);
+//------------------------------------------------------------------------------
+// Slowlog API
+//------------------------------------------------------------------------------
 
-/* Slowlog API */
 SlowLog *GraphContext_GetSlowLog(const GraphContext *gc);
 
 /* Cache API - Return cache associated with graph context and current thread id. */

src/query_ctx.c

@@ -208,45 +208,51 @@ bool QueryCtx_LockForCommit(void) {
 
 }
 
-void QueryCtx_UnlockCommit(OpBase *writer_op) {
-	QueryCtx *ctx = _QueryCtx_GetCtx();
-	// Check that the writer_op is entitled to release the lock.
-	if(ctx->internal_exec_ctx.last_writer != writer_op) return;
-	if(!ctx->internal_exec_ctx.locked_for_commit) return;
-	RedisModuleCtx *redis_ctx = ctx->global_exec_ctx.redis_ctx;
+static void _QueryCtx_UnlockCommit(QueryCtx *ctx) {
 	GraphContext *gc = ctx->gc;
-	if(ResultSetStat_IndicateModification(ctx->internal_exec_ctx.result_set->stats))
+	RedisModuleCtx *redis_ctx = ctx->global_exec_ctx.redis_ctx;
+
+	if(ResultSetStat_IndicateModification(ctx->internal_exec_ctx.result_set->stats)) {
 		// Replicate only in case of changes.
 		RedisModule_Replicate(redis_ctx, ctx->global_exec_ctx.command_name, "cc!", gc->graph_name,
 							  ctx->query_data.query);
+	}
+
 	ctx->internal_exec_ctx.locked_for_commit = false;
 	// Release graph R/W lock.
 	Graph_ReleaseLock(gc->g);
+
 	// Close Key.
 	RedisModule_CloseKey(ctx->internal_exec_ctx.key);
+
 	// Unlock GIL.
 	_QueryCtx_ThreadSafeContextUnlock(ctx);
 }
 
+void QueryCtx_UnlockCommit(OpBase *writer_op) {
+	QueryCtx *ctx = _QueryCtx_GetCtx();
+
+	// check that the writer_op is entitled to release the lock.
+	if(ctx->internal_exec_ctx.last_writer != writer_op) return;
+
+	// already unlocked?
+	if(!ctx->internal_exec_ctx.locked_for_commit) return;
+
+	_QueryCtx_UnlockCommit(ctx);
+}
+
 void QueryCtx_ForceUnlockCommit() {
 	QueryCtx *ctx = _QueryCtx_GetCtx();
+
+	// already unlocked?
 	if(!ctx->internal_exec_ctx.locked_for_commit) return;
+
 	RedisModuleCtx *redis_ctx = ctx->global_exec_ctx.redis_ctx;
-	GraphContext *gc = ctx->gc;
 	RedisModule_Log(redis_ctx, "warning",
 					"RedisGraph used forced unlocking commit flow for the query %s",
 					ctx->query_data.query);
-	if(ResultSetStat_IndicateModification(ctx->internal_exec_ctx.result_set->stats))
-		// Replicate only in case of changes.
-		RedisModule_Replicate(redis_ctx, ctx->global_exec_ctx.command_name, "cc!", gc->graph_name,
-							  ctx->query_data.query);
-	ctx->internal_exec_ctx.locked_for_commit = false;
-	// Release graph R/W lock.
-	Graph_ReleaseLock(gc->g);
-	// Close Key.
-	RedisModule_CloseKey(ctx->internal_exec_ctx.key);
-	// Unlock GIL.
-	_QueryCtx_ThreadSafeContextUnlock(ctx);
+
+	_QueryCtx_UnlockCommit(ctx);
 }
 
 inline bool QueryCtx_EncounteredError(void) {

src/resultset/resultset.c

@@ -14,7 +14,7 @@
 
 static void _ResultSet_ReplayStats(RedisModuleCtx *ctx, ResultSet *set) {
 	char buff[512] = {0};
-	size_t resultset_size = 2; /* query execution time and cached execution. */
+	size_t resultset_size = 2; // execution time, cached
 	int buflen;
 
 	if(set->stats.labels_added > 0) resultset_size++;

src/resultset/resultset_statistics.c

@@ -16,3 +16,4 @@ bool ResultSetStat_IndicateModification(ResultSetStatistics stats) {
 			|| stats.indices_created > 0
 			|| stats.indices_deleted > 0);
 }
+