Neo4j connector

Write to Neo4j (a property graph database) with support for node tables, relationship tables (edges), per-database multitenancy, atomic-batch writes via Bolt transactions, real CREATE CONSTRAINT uniqueness, and vector indexes with cosine / euclidean similarity.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

The neo4j connector writes records to Neo4j, a property graph database. It supports node tables (labels), relationship tables (edge types), per-database multitenancy (one Neo4j cluster, many isolated databases), real Cypher uniqueness constraints, and vector indexes via the CREATE VECTOR INDEX DDL form.

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

The module is compiled with the neo4j Cargo feature of the cocoindex crate.

Connection setup

Connect a neo4j::Graph and provide it under a context key. A Graph holds a connection pool and the name of the database its statements run on; all tables and batches share it.

use cocoindex::connectors::neo4j;
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: neo4j::Graph);

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

neo4j::Graph::connect(uri, user, password, database) opens a connection pool with basic authentication; the user, password and database are required arguments. For TLS, a custom pool size or other driver settings, build a neo4rs::Graph with neo4rs::ConfigBuilder and wrap it with neo4j::Graph::from_client(client, database). 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 Neo4j cluster can host many isolated databases. Pair each database with its own context key and neo4j::Graph:

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

let kg = neo4j::Graph::connect(&uri, &user, &password, "kg").await?;
let apis = neo4j::Graph::connect(&uri, &user, &password, "apis").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 database names produce fully separate target-state trees — changes to one never spill into the other.

As target

The neo4j 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.

The record writes of all tables of one database are applied together, in batches. Within a batch, writes are ordered as node upserts → relation upserts → relation deletes → node deletes so dependent edges always see their endpoints. A batch is applied in Neo4j transactions of at most 1,000 record writes each, one after the other. A transaction that fails rolls back, and the components of the batch are retried; earlier transactions of the batch stay committed, which is safe because the writes are idempotent (MERGE). A transaction that fails with a transient error (typically a deadlock between concurrent writers merging edges onto the same nodes) is retried with backoff, up to 8 attempts, before the failure is reported. A transaction that goes over the server’s transaction memory limit is split in two instead.

Within a transaction, the records of one statement shape (the node upserts of one label, the edge upserts of one relationship type and endpoint labels, and so on) are written by one UNWIND $rows AS row … statement.

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<neo4j::Graph>,
    table_name: impl Into<String>,
    schema: TableSchema,
) -> Result<TableTarget>

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

declare_table_target declares the table in the current component; its records resolve when the component commits. mount_table_target and mount_table_target_with_options take the same arguments, are async, and return a target whose records can be declared at once.

Parameters:

• graph — The context key of the neo4j::Graph.
• table_name — The Cypher node label (e.g. "Meeting"). It must not contain __ or end with _: constraint and index names join the label to a property name with __, so such a label could share a name with another table’s.
• schema — The table schema (see Table schema: from a row struct). It names the single primary-key property; compound primary keys are not supported. TableSchema::from_row::<Row>("id")? derives the columns from a row struct, and TableSchema::schemaless("id")? declares no columns. The schema participates in CocoIndex’s fingerprint (so two flows declaring the same label must agree); no per-property type DDL is emitted.
• options — ManagedTargetOptions::user_managed() assumes the table exists; the default (ManagedTargetOptions::default()) lets CocoIndex manage its lifecycle.

Returns: A TableTarget. A TableTarget serializes to its stable key (the graph’s context key and the label), so it can be an argument of mount!, use_mount!, mount_each! and memoized functions.

Nodes (child states)

Once a TableTarget is resolved, declare nodes to be upserted (translated to UNWIND $rows AS row MERGE (n:Label {pk: row.key}) SET n += row.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.

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<neo4j::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<neo4j::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<neo4j::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")?. Each function has an async mount_relation_target* twin with the same arguments, whose edges can be declared at once.

Parameters:

• graph — The context key of the neo4j::Graph.
• relation_name — The Cypher relationship type (e.g. "ATTENDED"). Same naming rule as a node label: no __, no trailing _. A node label and a relationship type may share a name.
• from_table — The node table whose nodes are the source endpoints of edges in this relationship.
• to_table — The node table whose nodes are the target endpoints of edges in this relationship.
• schema — The schema for the relationship’s own properties; its primary key uniquely identifies each edge. Without it, the edges are keyed by the id property.
• options — ManagedTargetOptions::user_managed() assumes the relationship type exists; the default lets CocoIndex manage its lifecycle.

Returns: A RelationTarget, which serializes to its stable key like a TableTarget.

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 — The row whose fields populate the relationship’s properties (declare_relation_record only). Its primary key field, if present, is the edge’s id.

If the edge is declared with declare_relation or the record 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.

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 target test:

people.declare_vector_index_named(ctx, "person_embedding", "embedding", 3, VectorMetric::Cosine)?;

VectorMetric::Cosine and VectorMetric::Euclidean are the Neo4j metrics; VectorMetric::InnerProduct is FalkorDB-only and fails on Neo4j. The Rust TableTarget also has declare_node_index(ctx, &["name"]), which declares a plain secondary index on the listed properties.

Parameters:

• name — Logical name for the index (declare_vector_index_named only). Defaults to vec_{table_name}__{field}.
• field — The node property holding the vector.
• metric — Similarity metric: VectorMetric::Cosine or VectorMetric::Euclidean. Translated to Neo4j’s vector.similarity_function option.
• dimension — The vector’s dimension. Required.

The connector emits:

CREATE VECTOR INDEX `coco_vec_<Label>__<field>` IF NOT EXISTS
FOR (n:`Label`) ON n.`field`
OPTIONS { indexConfig: {
  `vector.dimensions`: <N>,
  `vector.similarity_function`: '<metric>'
} }

The Neo4j index is named after the label and field, not after 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.

Table schema: from a row struct

Build a TableSchema from a row struct:

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::neo4j::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.

To override the type of one column, put a #[coco(..)] attribute on the field (see Column type overrides), or call with_column_type on the derived schema:

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

with_column_type(field, cypher_type) replaces the recorded Cypher type of one column and keeps its nullability. It fails if the schema has no column named field.

Default Rust → Neo4j type mapping

Most types pass through native Bolt encoding — no per-value transform applied:

Column type overrides

Override the default mapping for a single column with a #[coco(..)] 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. A value is written as it serializes with serde.

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

Vector fields

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

Table schema: explicit column definitions

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

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

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

ColumnDef::new(cypher_type) takes the Neo4j type string (metadata only; see the table above). A column is NOT NULL unless .nullable() is called.

DDL: indexes and constraints

For each managed table, the connector creates supporting Cypher artifacts on first run:

• For node tables: a uniqueness constraint on the primary key —

  Copy

  CREATE CONSTRAINT `coco_uniq_<Label>__<pk>` IF NOT EXISTS
  FOR (n:`<Label>`) REQUIRE n.`<pk>` IS UNIQUE

  Neo4j auto-creates a backing index for each constraint, so a separate CREATE INDEX is redundant on nodes.
• For relationship tables:

  Copy

  CREATE INDEX `coco_idx_rel_<RelType>__<pk>` IF NOT EXISTS
  FOR ()-[r:`<RelType>`]-() ON (r.`<pk>`)

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 or index is dropped. A node table’s vector indexes (the coco_vec_<Label>__<field> indexes 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 schema) 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::neo4j::{self, TableSchema};
use cocoindex::prelude::*;
use serde::{Deserialize, Serialize};

cocoindex::context_key!(static GRAPH: neo4j::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 = neo4j::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 = neo4j::Graph::connect(&uri, &user, &password, &database).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_neo4j")
        .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_neo4j.

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 = neo4j::mount_table_target(
        ctx,
        &GRAPH,
        "Meeting",
        TableSchema::from_row::<Meeting>("id")?,
    )
    .await?;
    let person_table = neo4j::mount_table_target(
        ctx,
        &GRAPH,
        "Person",
        TableSchema::from_row::<Person>("name")?,
    )
    .await?;
    let task_table = neo4j::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 =
        neo4j::mount_relation_target(ctx, &GRAPH, "ATTENDED", &person_table, &meeting_table)
            .await?;
    let assigned_rel =
        neo4j::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 and Task tables are declared up-front (via mount_table_target) so their uniqueness 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.