Scale-out and elasticity
An embedded engine is fast and bounded by one machine. The scale-out layer is what makes it a distributed database, and it is the part of FluxDB that is genuinely ours rather than inherited.
The shape is simple to state and detailed to implement: a shard is a full engine, shards are placed across cells by topology, and a query planner decides which shards to touch and how much work to push into each.
Sharding
| Strategy | When it fits |
|---|---|
| Hash | Even distribution when there is no natural range |
| Range | Ordered data where range scans should stay local |
| Composite | A resolver combining more than one key dimension |
| Directory | An explicit lookup, when placement is a decision rather than a function of the key |
Around the strategies sits a full shard lifecycle: split, merge, migration, recovery, and a file manager and data mover that physically relocate a shard's contents.
Placement is topology-aware
Shards are not scattered. A placement engine resolves where a shard and its replicas live against the real region, zone, rack and cell topology, with failure-domain awareness, so a replica set can be required to span things that do not fail together.
Placement consults the same topology that data residency is pinned on, which is why a residency constraint is a constraint on placement rather than a filter applied afterward, and why a required residency constraint no live cell can satisfy refuses rather than quietly degrading.
Query distribution
Figure: a query is routed by workload class and shard resolution, planned against a cost model with predicates and projections pushed down, executed in parallel as per-shard fragments, and gathered. Cross-cell fragment results are signature-verified before being merged.
Table equivalent
| Stage | What happens |
|---|---|
| Route | Workload-classified between the transactional and analytical engines, and resolved to the shards that hold the data |
| Plan | A cost model drives the plan, with predicate and projection pushdown so a shard returns the smallest useful result rather than raw rows |
| Execute | A parallel executor runs per-shard fragments concurrently |
| Gather | Results merge, and cross-cell fragments are signature-verified rather than trusted because they arrived |
Around that core: a join planner and cross-shard traversal for work that spans shards, recursive CTE coordination, a scatter-gather federation path, a result cache, a buffer pool, and materialized views with incremental refresh.
It gets faster without you tuning it
Three components close the loop between what the workload does and how it is executed:
- An adaptive optimizer and query optimizer that adjust plans against observed behaviour.
- An auto index manager and an index advisor, so indexing is recommended and maintained rather than guessed at during an incident.
- Cache warming, so a cold shard is not cold on first use.
Transactions across shards and cells
A multi-shard write is a two-phase commit with a coordinator on the initiating cell and participants on the others.
- Begin. The coordinator assigns a transaction id and identifies participants.
- Prepare. Each participant writes a prepare record, acquires locks, validates, and votes prepared or abort.
- Commit or abort. All-or-nothing on the vote.
- Cleanup. Final state recorded, resources released.
Crash recovery is specified rather than hoped for. A coordinator that dies after prepare but before commit reads its own log on restart and aborts stale preparing transactions, and a participant that dies after voting prepared resolves against the coordinator's recorded outcome.
For long-running work that cannot hold locks, a saga orchestrator provides the compensating-transaction path instead.
Alongside both: a lock manager, optimistic locking, a transaction manager, and consistency levels so a caller can choose its guarantee rather than taking one setting for the whole system.
Elastic in and out
Scaling is predictive, not reactive. A scaler runs trend analysis over observed shard statistics to forecast when a threshold will be breached, and triggers split or merge preemptively rather than after the shard is already hot.
That matters because a split is a data movement operation. Starting one at the moment a shard becomes overloaded means moving data while it is under maximum pressure, which is the worst possible time. Forecasting the breach moves the work earlier, into headroom.
Supporting it:
- A rebalancer driven by observability feedback, including hot-shard detection and load distribution.
- Shard migration with a verifier and a finalizer, so a move is checked rather than assumed.
- Replica promotion and routing, so reads spread and a failed primary is replaced.
- A failure detector.
- A resource governor and quota enforcer, so elasticity has bounds.
- Graceful degradation, so the system has a defined behaviour under pressure rather than an undefined one.
- Spill backends, so work that exceeds memory spills to disk rather than failing.
What this means for capacity
The scaling argument is architectural, and it is the reason the building-block choice is a strength rather than a compromise.
A shard is a full embedded engine running in-process against local storage. So per-shard capability is the capability of SQLite or DuckDB on your hardware, and both are independently benchmarked by people with no stake in this platform. The starting point is not a vendor claim; it is a number you can look up and then verify yourself.
Capacity is then a function of shard count and placement, not of one server's ceiling. A conventional deployment inherits its engine's single-node limit. Adding a shard here adds an entire execution core, and adding a cell adds its compute, its networking and its storage to the pool.
That is the whole argument in one line:
We did not pick engines that are fast enough. We picked engines that are already fast, and then removed the ceiling.
The placement engine, router and rebalancer are what make adding capacity an operation rather than a migration project, and pushdown is what keeps a scatter-gather from spending the gain on moving rows.
Characterise one shard on your hardware for your workload, which is an ordinary SQLite or DuckDB measurement with published third-party work already available. Multiply by the shard plan. Then subtract only the coordination you actually need: none for a single-shard query, and only the rows that survive pushdown for a fan-out.
That model tells you what a headline benchmark never does, which is what changes when you add hardware.
Related
- Storage and performance, where shards live and how they flush.
- Scale-out & distribution, the Data Platform view of the same plane.
- Deployment topology, the topology placement resolves against.
- Data residency, the constraint that can refuse a placement.