FalkorDB connector

Write to FalkorDB (a Redis-backed graph database) with support for node tables, relationship tables (edges), per-graph multitenancy, and vector indexes with cosine / euclidean / inner-product distances.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

The falkordb connector writes records to FalkorDB, a Cypher-compatible graph database that runs as a Redis module. It supports node tables (labels), relationship tables (edge types), per-graph multitenancy (one Redis instance, many isolated graphs), and vector indexes.

use cocoindex::connectors::falkordb::{self, TableSchema};

Connection setup

Connect a falkordb::Graph and provide it under a context key declared with cocoindex::context_key!. The graph holds the FalkorDB URI plus the target graph name, over one multiplexed Redis connection shared by all tables and batches.

use cocoindex::connectors::falkordb;
use cocoindex::prelude::*;

// The graph connection, provided to the environment in `main`. Its name is the
// graph's stable identity in the target states.
cocoindex::context_key!(static GRAPH: falkordb::Graph);

#[tokio::main]
async fn main() -> Result<()> {
    let graph = falkordb::Graph::connect(&uri, &graph_name).await?;
    let app = Environment::builder()
        .provide_key(&GRAPH, graph)
        .build()
        .await?
        .app("meeting_notes_graph_falkordb")
        .await?;
    // ...
    Ok(())
}

falkordb::Graph::connect(uri, graph) opens one multiplexed Redis connection. The URI is falkor://host:6379 or a redis:// URL; a password goes in the URI (falkor://:password@host:6379). The example provides the graph in main; a #[cocoindex::lifespan] function can do the same with builder.provide_key(&GRAPH, graph), and builder.on_shutdown(..) is the place for teardown. A Graph has no close() method.

Multitenancy

A single Redis instance can host many fully isolated graphs. Pair each graph with its own context key and falkordb::Graph::connect(uri, graph):

cocoindex::context_key!(static KG_DB: falkordb::Graph, key = "kg_db");
cocoindex::context_key!(static APIS_DB: falkordb::Graph, key = "apis_db");

let kg = falkordb::Graph::connect(&uri, "knowledge_graph").await?;
let apis = falkordb::Graph::connect(&uri, "apis_graph").await?;
let app = Environment::builder()
    .provide_key(&KG_DB, kg)
    .provide_key(&APIS_DB, apis)
    .build()
    .await?
    .app("kg_app")
    .await?;

context_key! names the key after the static by default; key = "..." sets the name the target states are tracked under.

Different context keys with different graph names produce fully separate target-state trees — changes to one never spill into the other.

As target

The falkordb connector provides target state APIs for writing nodes to node tables and edges to relationship tables. CocoIndex tracks what records should exist and automatically handles upserts and deletions.

Each graph.query call against FalkorDB is its own atomic unit (FalkorDB does not expose multi-statement transactions); the connector orders writes within a batch as node upserts → relation upserts → relation deletes → node deletes so dependent edges always see their endpoints.

Declaring target states

Node tables (parent state)

Declares a node label as a target state. Returns a TableTarget for declaring nodes.

pub fn declare_table_target(
    ctx: &Ctx,
    graph: &ContextKey<falkordb::Graph>,
    table_name: impl Into<String>,
    schema: TableSchema,
) -> Result<TableTarget>

pub fn declare_table_target_with_options(
    ctx: &Ctx,
    graph: &ContextKey<falkordb::Graph>,
    table_name: impl Into<String>,
    schema: TableSchema,
    options: ManagedTargetOptions,
) -> Result<TableTarget>

mount_table_target and mount_table_target_with_options take the same arguments, are async, and return a handle whose records resolve immediately. The schema is required. It carries the primary key: TableSchema::from_row::<Row>("id")? derives the columns from a row struct, and TableSchema::schemaless("id")? declares no columns. ManagedTargetOptions::user_managed() makes the table user-managed; the default is system-managed.

Parameters:

• graph — The context key of the falkordb::Graph.
• table_name — The Cypher node label (e.g. "Meeting").
• schema — Schema definition (see Table schema), which names the single primary-key property. FalkorDB does not enforce per-property types server-side, so the schema participates in CocoIndex’s fingerprint (so two flows declaring the same label must agree) but no per-column DDL is emitted. Compound primary keys are not supported.
• options — ManagedTargetOptions::default() (system-managed) or ManagedTargetOptions::user_managed().

Returns: A TableTarget. It is cheap to clone and serializes to its stable key, the graph’s context key name and the label, so it can be a mount_each! / use_mount! argument or a memoized function argument.

Nodes (child states)

Once a TableTarget is resolved, declare nodes to be upserted (translated to MERGE (n:Label {pk: $key_0}) SET n += $props):

pub fn TableTarget::declare_record<R: serde::Serialize>(
    &self,
    ctx: &Ctx,
    id: impl IntoStableKey,
    row: &R,
) -> Result<()>

The node’s primary-key value is passed as id (a &str, String, i64 or u64) and the row serializes to the node’s properties with serde.

A row that also carries the primary-key field must hold the same value as id (a null counts as absent), or the declaration fails. A property value must be a primitive or a list of primitives: a map, a list holding a null or a map, an integer beyond i64, or a NaN or infinite float fails the declaration before anything is written.

FalkorDB leaves empty strings out of its property indexes, so a MERGE on an empty-string key would never find the node it wrote: an empty-string id, or an empty-string relationship endpoint, fails the batch before it writes anything.

Relationship tables (parent state)

Declares a relationship type as a target state. Returns a RelationTarget for declaring edges.

pub fn declare_relation_target(
    ctx: &Ctx,
    graph: &ContextKey<falkordb::Graph>,
    relation_name: impl Into<String>,
    from_table: &TableTarget,
    to_table: &TableTarget,
) -> Result<RelationTarget>

pub fn declare_relation_target_with_schema(
    ctx: &Ctx,
    graph: &ContextKey<falkordb::Graph>,
    relation_name: impl Into<String>,
    from_table: &TableTarget,
    to_table: &TableTarget,
    schema: TableSchema,
) -> Result<RelationTarget>

pub fn declare_relation_target_with_options(
    ctx: &Ctx,
    graph: &ContextKey<falkordb::Graph>,
    relation_name: impl Into<String>,
    from_table: &TableTarget,
    to_table: &TableTarget,
    schema: TableSchema,
    options: ManagedTargetOptions,
) -> Result<RelationTarget>

Without a schema the relationship is schemaless and keyed by its id property. A schema names the primary key: TableSchema::from_row::<Edge>("edge_id")? or TableSchema::schemaless("edge_id")?. The mount_relation_target, mount_relation_target_with_schema and mount_relation_target_with_options variants take the same arguments and are async.

Parameters:

• graph — The context key of the falkordb::Graph.
• relation_name — The Cypher relationship type (e.g. "ATTENDED"). A node label and a relationship type may share a name.
• from_table — The node target whose nodes are the source endpoints of edges in this relationship.
• to_table — The node target whose nodes are the target endpoints of edges in this relationship.
• schema — Schema for the relationship’s own properties (see Table schema). It names the primary-key property that uniquely identifies each edge.
• options — ManagedTargetOptions::default() (system-managed) or ManagedTargetOptions::user_managed().

Returns: A RelationTarget. Like a TableTarget, it is cheap to clone and serializes to its stable key (the graph’s context key name and the relationship type), so it can be a mount_each! / use_mount! or memoized function argument.

Relationships (child states)

Once a RelationTarget is resolved, declare edges. Each declaration MERGEs the source endpoint, the target endpoint, then the relationship.

pub fn RelationTarget::declare_relation(
    &self,
    ctx: &Ctx,
    from_id: impl IntoStableKey,
    to_id: impl IntoStableKey,
) -> Result<()>

pub fn RelationTarget::declare_relation_record<R: serde::Serialize>(
    &self,
    ctx: &Ctx,
    from_id: impl IntoStableKey,
    to_id: impl IntoStableKey,
    record: &R,
) -> Result<()>

declare_relation declares an edge without properties. declare_relation_record adds the properties of record; a record that carries the primary key is keyed by it (a string or an integer), otherwise the id is derived from the endpoints.

Parameters:

• from_id — The source node’s primary-key value. The connector MERGEs (s:FromLabel {pk: $from_id}) so endpoints are auto-created if absent.
• to_id — The target node’s primary-key value. Same MERGE behavior.
• record — A Serialize value (with declare_relation_record) whose fields populate the relationship’s properties. Its primary-key field, if present, is the edge’s id.

If record is omitted or has no primary key value, the connector derives the edge id from its endpoints: the source label, source id, target label and target id, each written as <UTF-8 byte length>:<value>;. For example, an edge from Person "alice" to Task 42 gets the id 6:Person;5:alice;4:Task;2:42;. A string id that reads as an integer, or that starts with s:, is written with an s: prefix, so the ids 1 and "1" never share an edge. Endpoint ids are strings or integers.

An edge id names one edge. When the endpoints of an edge with an explicit id change, the edge moves: the upsert deletes the relationship of this type with the same id between the old endpoints. To find it, every upsert looks the edge up by its primary key property. A system-managed relationship table has an index on it; for a user-managed table, create one yourself.

Vector indexes (attachment)

Declares a vector index on a column of a node table. Vector indexes are an attachment to a TableTarget:

pub fn TableTarget::declare_vector_index(
    &self,
    ctx: &Ctx,
    field: &str,
    dimension: u32,
    metric: VectorMetric,
) -> Result<()>

pub fn TableTarget::declare_vector_index_named(
    &self,
    ctx: &Ctx,
    name: &str,
    field: &str,
    dimension: u32,
    metric: VectorMetric,
) -> Result<()>

For example, from the graph vector index test:

table.declare_vector_index(&ctx, "embedding", 3, VectorMetric::Cosine)?;
table.declare_node_index(&ctx, &["name"])?;

VectorMetric::Cosine, VectorMetric::Euclidean and VectorMetric::InnerProduct are the FalkorDB metrics, translated to FalkorDB’s similarityFunction option. declare_vector_index names the declaration vec_{table}__{field}; declare_vector_index_named takes the name. declare_node_index(ctx, &["name"]) declares a plain secondary index on the listed properties.

The connector emits CREATE VECTOR INDEX FOR (e:Label) ON (e.field) OPTIONS {dimension: N, similarityFunction: '...'}.

FalkorDB identifies a vector index by its label and field, not by name, so a field has at most one vector index. Renaming name keeps the index. Declaring a second name for the same field is an error.

Vectors are float32 only — wider vector dtypes are not supported.

Table schema: from a Rust struct

Build a TableSchema from a struct that derives SchemaFields:

pub fn TableSchema::from_row<T: SchemaFields>(primary_key: impl Into<String>) -> Result<Self>

The row type derives cocoindex::SchemaFields:

#[derive(serde::Serialize, cocoindex::SchemaFields)]
struct Person {
    id: String,
    name: Option<String>,
    #[coco(vector)]
    embedding: Vec<f32>,
}
let schema = cocoindex::connectors::falkordb::TableSchema::from_row::<Person>("id")?;

primary_key has no default and must name a field of the struct. An Option<T> field is a nullable column.

Default Rust → FalkorDB type mapping

The FalkorDB and Neo4j connectors share one type table, which records Cypher type names: BOOLEAN, INTEGER, FLOAT, STRING, BYTES, DATE, LOCAL_TIME, ZONED_DATETIME, DURATION, LIST<FLOAT> for #[coco(vector)], and MAP for collections and nested structs. The type is metadata only, so no DDL is emitted from it; a value is written as it serializes with serde.

Custom column types

Override the default mapping for a single column with a field attribute:

#[derive(SchemaFields)]
struct Row {
    id: String,
    #[coco(type = "POINT")]
    location: String,
    #[coco(json)]
    tags: Vec<String>,
    #[coco(rename = "label")]
    name: String,
}

#[coco(type = "…")] records the type string verbatim, #[coco(json)] records MAP for a field that would map to another type, and #[coco(rename = "…")] names the property.

To override a column for one schema only, call with_column_type on the derived schema. It replaces the column’s type and keeps its nullability:

let schema = TableSchema::from_row::<Row>("id")?.with_column_type("location", "POINT")?;

The type string is metadata-only — it participates in the schema fingerprint (so two flows declaring the same table must agree) but no DDL is emitted from it.

Vector fields

Mark a vector field #[coco(vector)] (or #[coco(vector = N)]). The property is recorded as LIST<FLOAT>, and the dimension belongs to the vector index declaration, not the property.

Table schema: explicit column definitions

Build a TableSchema directly from a list of column definitions when the row type is dynamic:

use cocoindex::connectors::falkordb::{ColumnDef, TableSchema};

let schema = TableSchema::new(
    [
        ("filename", ColumnDef::new("STRING")),
        ("title", ColumnDef::new("STRING")),
        ("summary", ColumnDef::new("STRING").nullable()),
    ],
    "filename",
)?;

A ColumnDef is NOT NULL unless .nullable() is called.

ColumnDef fields:

• cypher_type — The type string (metadata only; see above).
• nullable — Whether the column may be null. Defaults to false; .nullable() sets it.

DDL: indexes and constraints

For each managed table, the connector creates the supporting Cypher index on the primary key field on first run:

• For node tables: CREATE INDEX FOR (e:Label) ON (e.<pk>).
• For relationship tables: CREATE INDEX FOR ()-[e:RelType]-() ON (e.<pk>).

It then attempts a uniqueness constraint via the GRAPH.CONSTRAINT CREATE Redis command (best-effort — failures are logged but do not abort).

A system-managed table owns its label or relationship type, the way a Postgres table owns its rows. When the table is no longer declared, or on cocoindex drop, the connector destroys it: every node carrying the label is deleted (DETACH DELETE, so the relationships attached to those nodes go with them) or every relationship of the type is deleted, and then the constraint and index are dropped. A node table’s vector indexes (every vector index on its label) are dropped with it. This includes nodes and relationships written under that label or type by anything other than CocoIndex — to share a label with data CocoIndex doesn’t own, declare the table with ManagedTargetOptions::user_managed().

The same teardown happens when a table’s identity changes — its primary key field, or a relationship table’s endpoint tables or their primary keys: the table is destroyed and recreated, and the components that declared its records are reprocessed. Its vector indexes are dropped and created again. Attaching a schema or changing a property type does not rebuild the table; at most the records are re-upserted.

When a table is user-managed, the connector skips DDL entirely (you’re responsible for creating and dropping the index and constraint) and never bulk-deletes: when the table is no longer declared or the app is dropped, its nodes or relationships are left as-is (see Destroy vs. Abandon). Record-level upserts and deletes still work while the table is declared. Pass ManagedTargetOptions::user_managed() to mount_table_target_with_options or mount_relation_target_with_options.

Example: Node tables

use std::path::PathBuf;

use cocoindex::connectors::falkordb::{self, TableSchema};
use cocoindex::prelude::*;
use serde::{Deserialize, Serialize};

cocoindex::context_key!(static GRAPH: falkordb::Graph);

#[derive(Clone, Debug, Serialize, Deserialize, cocoindex::SchemaFields)]
struct Meeting {
    id: i64,
    note_file: String,
    time: String,
    note: String,
}

async fn build_graph(ctx: &Ctx, input_dir: &str) -> Result<()> {
    // Node table schemas are derived from the row structs.
    let meeting_table = falkordb::mount_table_target(
        ctx,
        &GRAPH,
        "Meeting",
        TableSchema::from_row::<Meeting>("id")?,
    )
    .await?;
    let meeting_id = 1;
    meeting_table.declare_record(
        ctx,
        meeting_id,
        &Meeting {
            id: meeting_id,
            note_file: "2024-01-10.md".to_string(),
            time: "2024-01-10".to_string(),
            note: "Kickoff".to_string(),
        },
    )?;
    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    let graph = falkordb::Graph::connect(&uri, &graph_name).await?;
    let app = Environment::builder()
        .db_path(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".cocoindex_db"))
        .provide_key(&GRAPH, graph)
        .build()
        .await?
        .app("meeting_notes_graph_falkordb")
        .await?;
    app.update(move |ctx| {
        let input_dir = input_dir.clone();
        async move { build_graph(&ctx, &input_dir).await }
    })
    .await?;
    Ok(())
}

The full program is examples/rust/meeting_notes_graph_falkordb.

Example: Relationship tables (knowledge graph)

#[derive(Clone, Debug, Serialize, Deserialize, cocoindex::SchemaFields)]
struct Person {
    name: String,
}

#[derive(Clone, Debug, Serialize, Deserialize, cocoindex::SchemaFields)]
struct Task {
    description: String,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
struct AttendedRel {
    is_organizer: bool,
}

async fn build_graph(ctx: &Ctx, input_dir: &str) -> Result<()> {
    let meeting_table = falkordb::mount_table_target(
        ctx,
        &GRAPH,
        "Meeting",
        TableSchema::from_row::<Meeting>("id")?,
    )
    .await?;
    let person_table = falkordb::mount_table_target(
        ctx,
        &GRAPH,
        "Person",
        TableSchema::from_row::<Person>("name")?,
    )
    .await?;
    let task_table = falkordb::mount_table_target(
        ctx,
        &GRAPH,
        "Task",
        TableSchema::from_row::<Task>("description")?,
    )
    .await?;
    // Relations are schemaless: their ids are derived from the endpoints.
    let attended_rel =
        falkordb::mount_relation_target(ctx, &GRAPH, "ATTENDED", &person_table, &meeting_table)
            .await?;
    let assigned_rel =
        falkordb::mount_relation_target(ctx, &GRAPH, "ASSIGNED_TO", &person_table, &task_table)
            .await?;

    // populate ...
    person_table.declare_record(ctx, "Alice Chen", &Person { name: "Alice Chen".to_string() })?;
    task_table.declare_record(
        ctx,
        "Write the roadmap",
        &Task { description: "Write the roadmap".to_string() },
    )?;
    attended_rel.declare_relation_record(
        ctx,
        "Alice Chen",
        meeting_id,
        &AttendedRel { is_organizer: true },
    )?;
    assigned_rel.declare_relation(ctx, "Alice Chen", "Write the roadmap")?;
    Ok(())
}

An edge with properties of its own is declared with declare_relation_record; one without is declared with declare_relation. To key edges by a property, mount the relation with mount_relation_target_with_schema(ctx, &GRAPH, "KNOWS", &people, &people, TableSchema::from_row::<Knows>("edge_id")?) and put the edge_id in the record.

The Person, Meeting and Task tables are mounted up-front so their indexes and constraints are reconciled before any edge MERGEs their endpoints. The relationship’s three-MERGE pattern (source endpoint → target endpoint → edge) means missing endpoints are auto-created — but it’s good practice to declare them explicitly so deletion-cascade behavior stays predictable.