Function _ExecuteQuery

src/commands/cmd_query.c:286–435

_ExecuteQuery accepts a GraphQueryCtx as an argument it may be called directly by a reader thread or the Redis main thread, or dispatched as a worker thread job when used for writing.

// it may be called directly by a reader thread or the Redis main thread,
// or dispatched as a worker thread job when used for writing.
static void _ExecuteQuery(void *args) {
	ASSERT(args != NULL);

	GraphQueryCtx  *gq_ctx      = args;
	QueryCtx       *query_ctx   = gq_ctx->query_ctx;
	GraphContext   *gc          = gq_ctx->graph_ctx;
	RedisModuleCtx *rm_ctx      = gq_ctx->rm_ctx;
	ExecutionCtx   *exec_ctx    = gq_ctx->exec_ctx;
	CommandCtx     *command_ctx = gq_ctx->command_ctx;
	AST            *ast         = exec_ctx->ast;
	ExecutionPlan  *plan        = exec_ctx->plan;
	ExecutionType  exec_type    = exec_ctx->exec_type;
	const bool     profile      = (query_ctx->flags & QueryExecutionTypeFlag_PROFILE);
	const bool     readonly     = !(query_ctx->flags & QueryExecutionTypeFlag_WRITE);

	// if we have migrated to a writer thread,
	// update thread-local storage and track the CommandCtx
	if (command_ctx->thread == EXEC_THREAD_WRITER) {
		// transition the query from waiting to executing
		QueryCtx_AdvanceStage(query_ctx);
		QueryCtx_SetTLS(query_ctx);
		Globals_TrackCommandCtx(command_ctx);
	}

	// instantiate the query ResultSet
	bool compact = command_ctx->compact;
	// replicated command don't need to return result
	ResultSetFormatterType resultset_format =
		profile || command_ctx->replicated_command
		? FORMATTER_NOP
		: (compact)
			? FORMATTER_COMPACT
			: FORMATTER_VERBOSE;
	ResultSet *result_set = NewResultSet(rm_ctx, resultset_format);
	if(exec_ctx->cached) {
		ResultSet_CachedExecution(result_set); // indicate a cached execution
	}

	QueryCtx_SetResultSet(result_set);

	// acquire the appropriate lock
	if(readonly) {
		Graph_AcquireReadLock(gc->g);
	} else {
		// if this is a writer query `we need to re-open the graph key with write flag
		// this notifies Redis that the key is "dirty" any watcher on that key will
		// be notified
		CommandCtx_ThreadSafeContextLock(command_ctx);
		{
			GraphContext_MarkWriter(rm_ctx, gc);
		}
		CommandCtx_ThreadSafeContextUnlock(command_ctx);
	}

	if(exec_type == EXECUTION_TYPE_QUERY) {  // query operation
		// set policy after lock acquisition,
		// avoid resetting policies between readers and writers
		Graph_SetMatrixPolicy(gc->g, SYNC_POLICY_FLUSH_RESIZE);