Neo4j Distributed: Architecture & How It Works

Neo4j distributes a database by replicating it. Each database in a cluster has primary copies that agree on every write through the Raft protocol, plus optional secondary copies that follow behind for read scaling. Every copy holds the whole graph, so a traversal never crosses a network boundary. Splitting one graph across machines is a separate, more constrained feature, and knowing where the line between the two sits is most of what it takes to reason about Neo4j at scale.

This guide covers what the term refers to, how a cluster handles writes and reads, its components, the features layered on clustering, the older causal clustering term, transactions, query processing, benefits, best practices, deployment, and enterprise fit.

What is Neo4j distributed?

Neo4j does not ship a product called “Neo4j Distributed”. The phrase refers to running Neo4j as a distributed database, which means Enterprise Edition clustering: replicating databases across servers for fault tolerance and read scaling. Two further features, composite databases and property sharding, spread data across several databases. The Community Edition runs as a single instance. The current clustering model arrived in Neo4j 5, and Neo4j moved to calendar versioning in 2025, so releases now carry YYYY.MM numbers; this guide describes the current model and notes where 4.x differed.

The distinction to hold onto is the axis being distributed. Replication distributes copies: the same data on several servers, for availability and read throughput. Partitioning distributes data: different subsets on different servers, for volume beyond one machine. Neo4j’s clustering is entirely on the first axis; the second exists only through the two add-on features, in constrained forms.

How does Neo4j distributed work?

The operations manual states the starting point directly: “Servers and databases are decoupled: servers provide computation and storage power for databases to use.” A cluster is a pool of servers; each database is allocated to some of them as primary or secondary copies.

 Primaries process writes. One primary acts as leader for the database, orders write transactions, and replicates them synchronously to the other primaries. A commit needs a majority of them, which is why tolerating F primary failures needs M = 2F + 1 primaries.

Secondaries receive transactions asynchronously and serve reads. They do not vote, so adding them raises read capacity without changing fault tolerance.

Because secondaries lag, a client that writes and then immediately reads could see stale data. Bookmarks close that gap: after a transaction the client receives a token, and passing it with the next transaction guarantees that “only servers which have processed the client’s bookmarked transaction will run its next transaction.” That read-after-write guarantee is what the manual calls causal consistency.

Neo4j distributed architecture

Servers. Every server runs the same software and can host any database in either role; the upgrade guide summarizes the model as “All servers are created equal.” An operator can constrain a server to primary-only or secondary-only duty, or restrict which databases it may host, but roles belong to database copies, not to machines.

Databases and allocation. A database is created with a topology, for example CREATE DATABASE orders TOPOLOGY 3 PRIMARIES 2 SECONDARIES, and, per the same upgrade guide, “the cluster decides which databases are allocated to which servers.” Topology can be altered later, and databases can be reallocated as servers join or leave. Neo4j calls this automated placement autonomous clustering.

The system database. Cluster membership, database definitions, users, and roles live in the system database, itself replicated across its designated primaries. Forming a cluster means forming that replication group first.

Discovery and routing. Servers find each other through a configured discovery mechanism: a static endpoint list, DNS, or Kubernetes services. Clients connect with the neo4j:// URI scheme, fetch a routing table listing the writers, readers, and routers for each database, and refresh it periodically. Server-side routing forwards a query that arrives at a server unable to run it, such as a write at a non-leader, to one that can, and applies only to neo4j:// connections. Drivers using the plain bolt:// scheme perform no routing and receive none from the server.

Neo4j clustering and distributed databases

The baseline topology for high availability is three primaries and no secondaries, which tolerates one failure with the smallest write quorum. Adding secondaries turns the same cluster into a read-scaling deployment. Because topology is set per database, one cluster can hold a heavily written database with three primaries alongside a reference dataset with one primary and several secondaries.

Beyond replication sits the composite database, a layer of aliases over several ordinary databases, local or remote, that presents them as one queryable unit and stores nothing itself. A query can read across the constituents, and a write goes to one of them; the documentation is explicit that “relationships cannot span across graphs”, so a traversal cannot follow an edge from one constituent into another. The constituents stay ordinary databases, each with its own topology when clustered. The feature fits data that already splits along a boundary traversals rarely cross, such as tenant or region.

A newer option, sharded property databases, takes a different cut: nodes and relationships stay in a single graph shard, while their properties are hashed across property shards on other servers in the cluster. Introduced in Neo4j 2025.12, property sharding is not part of standard Enterprise Edition and requires an Infinigraph subscription. The rest of this guide concentrates on clustering.

Neo4j causal clustering

Much of the material on Neo4j’s distributed model describes causal clustering. That is the name Neo4j used through version 4.x, and the 4.4 operations manual still documents it: Core servers replicate “all transactions using the Raft protocol”, Read Replicas are “asynchronously replicated from Primary Servers” and handle read-only load, and causal consistency, carried by bookmarks, ensures that “client applications are guaranteed to read their own writes, regardless of which instance they communicate with.”

Neo4j 5 retired the name along with the implementation. The upgrade guide states that “the Causal Cluster is replaced by a new clustering implementation”: Core servers became primary database copies and Read Replicas became secondary database copies. The current manual talks about clustering and causal consistency, not causal clustering.

What carried over is the substance. Primaries still commit through Raft, secondaries still replicate asynchronously, and bookmarks still deliver causal consistency. The practical difference is placement: a 4.x Core server hosted every database on the cluster, whereas a 5.x server hosts whichever copies the cluster allocates to it.

Neo4j distributed transactions and consistency

A write transaction runs on the leader of its database. The manual states the commit rule: “The database writer synchronously pushes writes to other primaries and does not allow a commit to be completed until it receives confirmation that the data has been written to enough members.” In Raft terms, the leader appends the commit to its log, replicates the entry to the other primaries, and acknowledges once a majority has durably accepted it. If the leader fails, the remaining primaries elect a new one from those holding the up-to-date log, and any write that had not reached a majority is not committed.

The resulting consistency has two parts. Within one database, committed writes are totally ordered by the Raft log and every primary applies them in the same order. Across the cluster, reads are causally consistent rather than linearizable: a secondary may be behind, but a client carrying a bookmark sees at least the state that bookmark represents.

Write transactions do not span databases. Each database has its own Raft group and log, and a transaction commits in exactly one of them. Composite databases relax this only for reads: their documentation allows only transactions that read from multiple graphs, or read from multiple graphs and write to a single graph. There is no two-phase commit across databases and no atomic multi-database write.

Neo4j distributed query processing

A query against a clustered database executes on one server, against a complete local copy. Nothing about the query is distributed: no partial plans, no shuffles, no cross-server joins. A multi-hop traversal runs entirely inside one store, which is why clustering costs traversal nothing. What the cluster adds is the choice of server.

A driver connected through neo4j:// fetches a routing table for the database, naming its writer, readers, and routers. Write transactions go to the writer, the database’s current leader, because “only primaries are eligible to act as the writer for a database.” Read transactions go to a reader; “by default, read queries are routed away from the writer”, and in a single-primary topology with secondaries “the secondaries typically handle all of the read queries.” The application declares which kind it is running through the driver’s transaction functions, execute_read and execute_write in the Python driver. The same functions retry a transaction whose failure “is deemed to be transient”.

with driver.session(database="orders") as session:
    session.execute_write(
        lambda tx: tx.run(
            "MATCH (c:Customer {id: $id}) "
            "MERGE (c)-[:PLACED]->(o:Order {id: $orderId}) "
            "SET o.total = $total",
            id=customer_id, orderId=order_id, total=total,
        ).consume()
    )
    big_orders = session.execute_read(
        lambda tx: tx.run(
            "MATCH (c:Customer {id: $id})-[:PLACED]->(o:Order) "
            "WHERE o.total > 1000 RETURN count(o) AS n",
            id=customer_id,
        ).single()["n"]
    )

The write commits on the leader once a majority of primaries have accepted it. The read that follows is routed to a reader, possibly a secondary that has not yet applied that write, and still counts the new order, because “queries inside the same session are causally chained”: the session carries the bookmark from the write, and the reader waits until it has applied that transaction. When the chain crosses sessions or services, the application collects bookmarks with last_bookmarks() and passes them to the next session. A read with no bookmark is served from whatever state its reader has reached, which is the price of reads that scale with secondaries.

Cross-database query processing exists only in composite databases. Cypher’s USE clause selects a constituent, a CALL subquery with its own USE runs part of the query on another, and the composite database evaluates the outer part over the rows the constituents return. Each subquery is still a local traversal on one constituent.

Benefits of Neo4j distributed architecture

Fault tolerance without application logic. Raft commit and automatic leader election mean a three-primary cluster keeps accepting writes through the loss of a server, and drivers re-fetch routing tables to find the new leader.

Read scaling. Secondaries add read capacity, and bookmarks keep reads after writes correct.

Whole-graph locality. Because every copy holds the full graph, multi-hop traversals run at single-instance speed, with no network round trip per hop.

Automated placement. Topology is declared per database and maintained by the cluster as servers come and go.

The common thread is that Neo4j’s model keeps the semantics of a single database, replicated, for as long as the graph fits on one server.

Neo4j distributed architecture best practices

Sizing the topology comes first.

Size the primary count for quorum. Three primaries is the baseline; five buys tolerance for two concurrent failures. Each commit must be acknowledged by a majority of primaries, and more primaries means more servers to contact per write, so write latency grows with the primary count.

Treat secondaries as capacity, not safety. Secondaries do not vote. One primary with four secondaries has read scale and no write fault tolerance.

Client configuration then decides whether the cluster’s failover reaches the application.

Connect through neo4j:// and let the driver route. Pinning an application to one server’s bolt:// address bypasses routing, so a leader change becomes an application failure.

Use managed transactions. Transaction functions retry automatically when a failure is transient; auto-commit queries offer only limited retry guarantees.

The last guards the ceiling that every copy shares.

Plan storage for a full copy per server. Every primary and secondary holds the entire database. Reaching the ceiling of one server is the signal to consider composite databases, if the model splits along a boundary traversals rarely cross, or a different data placement.

How to deploy Neo4j as a distributed database

A minimal self-managed cluster is three Enterprise Edition servers. Each server’s configuration needs an advertised address other members can reach, a discovery setting, and defaults for how many primary and secondary copies new databases receive. The deployment guide names the settings: server.default_advertised_address, dbms.cluster.endpoints, dbms.cluster.discovery.resolver_type, and initial.dbms.default_primaries_count.

Once the cluster has formed, user databases are created with a topology:

CREATE DATABASE orders TOPOLOGY 3 PRIMARIES 2 SECONDARIES;
SHOW DATABASES;

Topology is changed later with ALTER DATABASE orders SET TOPOLOGY ...; both commands fail if the cluster lacks enough servers to satisfy the request.

On Kubernetes, Neo4j publishes a Helm chart and a quickstart that installs one release per cluster member, with Kubernetes service discovery replacing the static endpoint list. For teams that do not want to operate the cluster, Neo4j Aura provides managed instances.

Whichever path is chosen, the checklist is the same: an Enterprise license, an odd number of primaries, a discovery mechanism, drivers using the routing scheme, and backups taken from a server that currently holds the database, which under autonomous placement is no longer a fixed address.

Neo4j distributed for enterprise workloads

Neo4j’s distributed model fits transactional graph applications well: identity and access graphs, recommendation serving, fraud checks at transaction time, and other workloads where the graph is the system of record, fits on one well-provisioned server, and must stay writable through failures.

The model is under more strain for analytical workloads over data that is already distributed elsewhere. Neo4j’s own scaling guidance lists data volume for a clustered database as “limited to single server size”, and partitioning means adopting composite databases’ modeling constraints or a separate subscription. Before either applies, the data has to be loaded into Neo4j: a pipeline from the warehouse or lakehouse into the graph store, kept in sync, and duplicated in full on every copy. The scaling levers themselves are covered in more depth in Neo4j scalability.

That is the scenario PuppyGraph addresses. It queries existing SQL databases, data warehouses, and data lakes and lakehouses, including direct reads of open table formats like Iceberg and Delta Lake, as a graph, with no graph-specific ETL and no second store to replicate; the tables stay where they are. It compiles an openCypher or Gremlin query into a plan of node and edge operators that runs in its own distributed engine, issuing only simple projection and filter SQL to the source. Because the query is represented as graph operators end to end, the engine optimizes specifically for multi-hop traversals, which is where its traversal performance comes from. Capacity comes from adding executor nodes, and a single graph schema can span thousands of source tables, with deployments running at terabyte to petabyte volume. Because it speaks openCypher over the Bolt protocol, existing Neo4j drivers and applications can be repointed at it without rewriting queries. The data stays distributed by the platform that already holds it; only the traversal compute is added.

Conclusion

Neo4j distributes by replication. Each database has primaries that commit through Raft and secondaries that follow asynchronously; bookmarks give applications causal consistency across lagging copies; drivers discover the cluster through routing tables; and the cluster decides where each copy lives. Partitioning is separate, with real limits: composite reads may span constituents, writes target one, and relationships never cross the boundary.

The question for an enterprise evaluation is which axis the workload needs. A highly available, writable graph that fits on one server is what Neo4j’s clustering delivers, with little ceremony. Traversals over data that already lives, distributed, in a warehouse or lakehouse put the cost in the copy rather than the cluster.

Try the forever-free PuppyGraph Developer Edition and book a demo with the team to see how openCypher and Gremlin queries traverse warehouse and lakehouse tables, with no graph-specific ETL, and no graph store to cluster or keep in sync.