DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
Status
Current state: Under discussion
Discussion thread: TBD
JIRA: TBA
Released: TBA
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Executive Summary
This CEP introduces Flexible Placements, a new ownership model where ranges are scoped to individual keyspace/table pairs and replica placement is managed explicitly rather than derived from ring position. Building on CEP-21's Transactional Cluster Metadata, this approach enables:
- Incremental operations: Breaking large bootstrap and streaming operations into smaller, resumable steps that reduce the window of over-replication and use zero-copy streaming.
- Improved cluster density: Maintaining near-optimal load balance at any cluster size without requiring explicit rebalancing phases, with placement decisions driven by capacity and load rather than ring topology.
- Per-keyspace flexibility: Allowing different keyspaces to occupy different subsets of nodes based on their actual resource requirements.
CEP-21 has introduced Transactional Cluster Metadata and created a linearizable log of Schema and Ownership changes, making cluster operations more reliable. This CEP builds on this work and allows us to materialize cost savings and operational improvements.
Problem Statement
Ownership in Apache Cassandra is currently derived from the Token Ring: each node gets one or more tokens assigned to it. However, internally most operations are expressed in terms of range ownership. Range ownership is derived from token ownership by creating ranges with bounds of consecutive tokens on the ring. The node that owns the token defining the upper bound of a range is the primary replica for that range. Additional replicas are the next RF-1 nodes encountered when walking the ring clockwise (for simplicity of this discussion we do not bring NTS rules such as rack diversity and DC distribution).
This model has multiple shortcomings:
- With token-per-node, to expand cluster with minimal relocation cost, one has to double it in size. With Gossip, concurrent bootstraps of contiguous ranges were inherently unsafe; with TCM the situation is better but even then, the best option you have is to stagger bootstraps by placing concurrently bootstrapping nodes at least RF nodes away from one another, further increasing relocation costs.
- During bootstrap, the replica group enters an expanded state where writes go to
RF+1replicas instead ofRF. The extra cost is paid by the coordinator (extra network round-trip), and the pending node (receiving writes for the range it is gaining while simultaneously streaming it). - With VNodes, token moves are not allowed, so bringing the cluster back to balance requires operator involvement (also, decom and re-bootstrap with new tokens, which seems untenable).
- While shrinking the cluster, the decommission inevitably increases the load on a neighbor until the balancing moves have been done.
There are several other problems, including:
- There is currently no mechanism to plan complex join operations. For example, expanding an 80 node cluster by 8 nodes requires 8 individual node joins, and there's no way to plan this operation as a single multi-step action.
- Long operations can't be easily broken down into smaller sub-steps, and operations such as bootstrap either require tooling-driven sequencing (increasing the cost of the overarching operation), or a single all-or-nothing non-retryable step (reducing fault-tolerance).
Goals
Cluster Density
Substantially improve cluster density by allowing higher utilization, achieved through:
- Collecting more granular and detailed load metrics: in order to stop expressing load only in terms of percentage of ring ownership, and introduce per-table / per range bytes read, written, coordinated and owned metrics
- Use of detailed metrics for making better placement decisions, making the cluster balanced not only in terms of percentage of the ring, but also in terms of bytes-on-disk and read/write load
- More stable utilization of each node in the cluster, compared to the traditional sawtooth pattern when we double the size of a cluster, where the load on each node would go from 80% down to 40% and then up to 80% again as the load grows, we instead aim to stay within a couple of percent of the target utilisation at all times.
Incremental Operations
Reduce scope of individual in-progress operations and allow incremental/sub-range operations:
- Reduce the over-replication time window / overhead during ownership changes: allowing a change in node ownership to be broken down into series of smaller retry-able steps, without requiring additional tooling. In other words, a sub-range that has been fully streamed will be available for read from target replica, freeing the previous replica from responsibility to serve reads or writes for this range.
- Allow more granular sub-range operations to benefit from zero-copy streaming.
Dynamic Flexible Placements
Unlock (even) higher utilization through flexible placements:
- Decouple range placement across keyspaces and tables. In other words, a keyspace holding 100TB should be allowed to occupy a number of nodes in the cluster that is different from one that holds 20Gb, and this should be achievable without creating separate clusters for these keyspaces, for reasons such as (but not limited to):
- Independent Scaling: The 100TB dataset likely scales on capacity (disk), while the 20GB dataset likely scales on throughput (CPU/RAM). Decoupling prevents you from having to add unnecessary CPU just to get more disk space (or vice-versa).
- Cache & IO Isolation: Prevents the "elephant" (100TB) from thrashing caches/saturating IO during compaction/queries, and affecting latencies for the "mouse" (20GB). Especially because smaller dataset might be a source of critical metadata.
- Allow range-to-node assignment independent of token ring position, so placement decisions are driven by capacity and load rather than by which tokens a node's neighbors own.
Design Constraints
All of the above goals should be achieved while:
- Maintaining full backwards compatibility: allow operators to reap most benefits (i.e. incremental, sub-range operations, resumable bootstraps/moves) without requiring any operator to change tooling.
- Feature flagging / gradual rollout: allow operators to turn on flexible ranges on and off to try them out without any risks, with a path to revert the cluster to its previous state by creating a plan of compensating range movements, relocating the data back to where it would be in a token-based cluster.
- Delivering changes incrementally, making every individual change available without requiring delivery of the entire CEP in one go.
Many of the changes required to achieve the above goals are tightly coupled: making ownership more dynamic is only acceptable if changes in ownership do not create additional operational toil. Achieving balance requires reliable ownership changes (which was achieved by TCM), but is achieved not just by moving data around, but by efficient binpacking.
A nice side-effect of collecting per-subrange metrics, and reducing granularity of operations, is that it has the potential to restore the utility of Byte-Ordered Partitioner. BOP is important since having data written in key order rather than token order does open opportunities querying by partition key range, which is a desirable feature for many users. Unfortunately, one of the problems with BOP is that there is no generic way to split token ring into smaller pieces that are roughly equal in size, and flexible placements provide us with the opportunity to explore alternative ways to partition the token space, helping us to finally make BOP viable again.
Non-Goals
This CEP does not attempt to deprecate or forbid deriving ownership from token placements or vnodes. Au contraire:operators who can not or do not wish to move away from token-derived ownership (including vnodes) can continue using token ring model perpetually, while still benefitting from many (however not all) changes described in this CEP.
While this CEP is a necessary building block for auto-scaling, it will not implement auto-scaling itself. The implementation will expose the required primitives: per-range metrics as data sources and triggerable commands, but full auto-scaling support is out of scope and warrants its own CEP.
Approach
Incremental Operations
With TCM, we already have range placements materialized per replication factor. By adapting UCS and adding a custom shard manager, we can make flushes and compactions range-aware [1]. Having sub-ranges materialised in this way helps with streaming and breaking down large operations into series of smaller ones.
Range aware compaction helps make the load more even in the cluster as each compaction unit is much smaller, so the impact if one of the sub-ranges gets behind on compaction is reduced. Range aware compaction also allows us to zero copy stream data when moving sub-ranges and makes cleanup when the sub-range has moved away essentially free, since the sub-range that moved away has its own sstables and those can simply be deleted.
Aligning incremental repairs with these sub-ranges will almost entirely remove the need for anticompaction, so we will make sure that CEP-37/CASSANDRA-19918 is aware of these sub-ranges and repairs them one-by-one.
Routing
Most routing was refactored during Transient Replication implementation [7], with introduction of ReplicaPlan and ReplicaLayout. As of now, to get routing information, it is enough to know the replication params and a token.
An element of this proposal is to introduce the concept of a Routing Key, which enables further flexibility when identifying replicas for a given range. To enable the project to be delivered in managable pieces, we will start by introducing a token-based routing key which does not hold information about keyspace or table. Subsequent changes will introduce and wire up keyspace and table information required for routing in situations where token is insufficient (see Flexible Placements below for more details).
Flexible Placements
Introducing flexible placements shares several motivations of virtual nodes introduced in Cassandra 1.2. Like the vnode initiative, it aims to diversify streaming sources and reduce the amount of data that needs to be moved in order to add or remove nodes. The challenge is to achieve this without substantially increasing the degree of interconnectedness between the nodes in the cluster. This is important because as the number of peers a node shares ranges with increases, so does the probability that any two nodes becoming unavailable causes an outage.
While it is not impossible to achieve balance in load (where load is defined as real load, not only percentage of ring) with vnodes, the fact that position of the neighboring tokens on the ring also instructs replication unnecessarily complicates an already difficult binpacking problem.
The suggested approach is to allow associating any range with any node, eliminating the token assignments as the source of ownership. Instead, we propose using Tablets (ranges, scoped to a keyspace/table pair) become the primary primitive, with replica placement managed explicitly rather than derived from ring position.
Ownership should be flexible and placements should be derived from Tablets, where Tablet is a range for a particular keyspace/table pair, replicated to the number of nodes specified as a Replication Factor. A similar approach and name is used in Bigtable [2], Apache HBase [3], Apache Kudu [4], YugabyteDB [5], Vitess [6], as well as CockroachDB, TiDB, FoundationDB, Cosmos DB, and many other databases.
While placement strategy will be pluggable, this CEP will come with an efficient default implementation that combines best properties of TokenAllocator (for load distribution) and NetworkTopologyStrategy (accounting for failure domains), and has potential to deliver substantially better results (see Case Study in Appendix A), and give us a lot of flexibility in terms of placements while not overwhelming operators with a huge (i.e. thousands) of tablets in a small (i.e. 30-50 node) cluster.
There are two separate objectives:
- to build clusters incrementally by introducing and maintaining minimal imbalance in the ring, which can be exploited to avoid reassignment
- to improve the balance in existing, probably highly unbalanced clusters, built without the above algorithm in mind.
Placement allocation algorithms requirements
To avoid focusing on a specific algorithm here, since there are multiple contenders yielding similar results, for the purpose of this CEP we state that the final implementation will take into account (with operator-assigned weights) following optimization vectors:
- Failure domain anti-affinity: ensure we do not store range multiple times in the same rack
- Node interconnectedness (i.e. vnode problem, see quote below): make sure when the node is bounced, the number of nodes that should’t be bounced/fail at the same time is kept to minimum.
- Number of streaming sources: keep number of streaming sources fairly diversified.
- Load distribution: keep nodes as evenly loaded as possible.
- Minimal number of tablets: to reduce cognitive load on the operator, keep the number of ranges to minimum where possible.
- Number of relocations: how much streaming we can afford to pay for the above objectives.
Relocation should also take into consideration the following criteria:
- Hysteresis and change thresholds: avoid oscillation on changes close to boundary values.
- Stochastic evaluation: generate random candidate moves (not just "best" moves) to avoid settling on local minimum.
- Allow for inclusion of other metrics, such as load variance, move cost (for multi-cloud installations), read/write rates (for highly imbalanced clusters), etc.
- Maintenance domains / bounce groups: logical grouping of nodes that can be taken offline simultaneously while maintaining write availability and meeting the required replication quorum (
⌊RF/2⌋ + 1).
vnode problem, as per Joey Lynch:
For example on a cluster with three hosts, each node inter-depends with two others; with four hosts each node has three inter-dependents; and with five or more hosts each node has four inter-dependents. When a brand new host places 256 virtual nodes into the ring, each vnode it places inter-depends with 2-4 other nodes.
To highlight this once again, unlike vnodes, flexible placements avoids increasing the number of correlated failures by using an allocation algorithm that expressly optimizes for minimizing interconnectedness between nodes replicating neighboring ranges whilst ensuring failure domain anti-affinity.
Z3-based optimizations
Finding placements that balance competing objectives is an optimization problem. We explored several approaches, from greedy heuristics to constraint solvers, including Z3-based [8] optimization strategies. Lexicographic multi-objective optimization proved problematic: minimizing relocations first yields zero moves, while minimizing tablet count first moves everything. These objectives directly contradict each other, as each relocation costs exactly one move but reduces the tablet count by one.
Weighted and Pareto optimizers [9] produced more balanced results by allowing trade-offs between objectives rather than strict prioritization. The most effective approach was budget-constrained exploration: fixing a relocation budget (e.g., "at most 2 moves") and maximizing load balance within that constraint. Running this across multiple budgets lets the operator choose an acceptable cost/benefit trade-off.
For the scope of this CEP, placement allocation and optimizations will only be ran as a part of existing cluster workflows, such as bootstrap (expand) and decommission (shrink). Pending operational data, a separate proposal for load-triggered rebalancing can be considered.
Tablet Sizing
A natural intuition is to size tablets based on number of bytes, but decide placement/packing them together based on both size and load. However, capping the tablets by size is likely to lead to unnecessarily inflation of the number of tablets: let’s say we cap tablet size to 2GB, which would mean the number of tablets will grow proportionally with amount of data. If we can achieve balance with variable tablet size, making tablets uniform in size is not necessary. Instead, we can make nodes uniform in size, and use sub-range load to calculate a fraction of tablet for handoff.
Since optimal placement is an NP-hard problem, and clusters exhibit sufficient load uniformity despite variations, we can achieve balance without increasing tablet count with data size. The multi-objective optimization function above suggests that the best approach is allowing variable-sized tablets and sharding SSTables locally within tablets to avoid metadata overhead.
Tablets must be packed onto nodes such that resources are well-utilized, but no single tablet can overwhelm a host. There should always be sufficient headroom to relocate any tablet off a node without risking failure. As a guideline, a tablet should occupy no more than a defined fraction of the node's aggregate capacity (defined as a weighted aggregate of storage, CPU, memory, and network bandwidth). The exact threshold is TBD, likely 5-20% based on preliminary analysis. It must be low enough that relocations reliably succeed even under load; we expect to derive a concrete value (or formula) through testing.
Project Plan
- Phase 1: ML Discussions
- Phase 2: Range Aware Compaction prerequisites and minimal implementation, early refactoring, changes in ReplicaCollection, ReplicaPlan, and ReplicaLayout classes
- Phase 3: RoutingKey, with a single implementation for Token
- Phase 4: Metrics collection (likely to be broken down into even smaller patches and made available in earlier versions if community deems necessary)
- Phase 5: Flexible Placements with a simple “greedy” algorithm for choosing range allocations and planning operations.
- Phase 6: Introduce Flexible Placements with automatic cluster balancing based on greedy algorithm described above.
- Phase 7: Implement (implies that changes likely will be way in flight by that time) changes to Cassandra Analytics, sidecar, and drivers that allows interacting with a cluster that has flexible placements enabled.
Proposed Changes
New or Changed Public Interfaces:
- All existing configuration options will continue working for clusters without flexible placements
- Weights for allocation algorithm will be configurable (e.g. prioritize balance, number of streaming sources, failure domain anti-affinity, etc., and by how much)
Virtual Tables:
- Existing system tables (such as
peers) will continue working as before. - New virtual tables exposing range ownership will be added, serving as the basis for communication with external tooling (bulk readers/writers, clients, etc.). For token-based and vnode clusters, ranges will be derived from token ownership and exposed as range-to-replica mappings.
- New virtual tables for exposing per-range metrics will be added.
- Virtual tables for bounce groups will be added (likely available in earlier versions).
Nodetool commands:
- Existing commands (such as movetoken) will continue working but will error out for clusters that have no keyspaces with token ownership.
- Commands for changing a single keyspace to flexible placements and back will be added.
- Commands for managing tablets (splitting, merging, and moving) will be added
- Commands for reverting cluster back to placements derived from token ownership
- Commands for planning complex transitions (e.g. bootstrapping N nodes, or adding N and removing M nodes simultaneously, at maximum concurrency) will be added
Internal API changes:
- A concept of a Routing Key will be introduced to coordinator path.
Compatibility:
New code will be fully compatible with both token-based ownership and vnodes. We do not anticipate future deprecation, either. Some features, such as flexible placements, will only be possible for the users that have switched keyspace to tablets (see Migration Plan).
Migration Plan
We will allow enabling flexible placements on per-keyspace level. In other words, you can try this feature out for a single keyspace, which will be managed differently from the rest of the keyspaces.
In addition, we will provide tooling to roll the cluster back to the token ownership or vnode model. As a side-effect of this, it will also be possible, with same functionality, to "un-shuffle" a vnode cluster to single token, or change the number of vnodes/tokens. No new functionality will need to be developed for this.
Test Plan
CEP-21 introduced some test tooling to simulate cluster operations, and we will continue extending this tooling to support new features, exercising both metadata changes (in a tight loop, going through billions of seeds), but also performing large-scale testing (multi-hundred node clusters, under stress).
We will also continuing using Harry, our fuzz testing tool, for stressing clusters, and validating that data is where it needs to be even in a large cluster after a number of perturbations.
Individual components will be covered with unit tests; new functionality will be covered and tested with integration in-JVM and simulator tests.
Rejected Alternatives
- Achieving balance during expansion through "ripple": when bootstrapping new nodes in the cluster, we position new nodes as evenly placed away from one another as possible, and make them take amount of data proportional to number of nodes in an expanded ring. Then, we "ripple" effects by moving tokens of neighboring nodes, in both directions (clock- and counterclock- wise) across the ring. The equivalent is also possible for shrinks, and has similar shortcomings. The amount of data that needs to be moved in this situation is prohibitively large to make this solution viable.
- Improving VNodes: most of the work that needs to be done to finish this CEP is series of operational improvements making day-to-day operations more granular and reliable. Adding a new ownership model (tablet-derived) is a tiny fraction of the project. In other words, it will improve operations for both vnode and token-based clusters anyway, but reaping the envisioned utilization improvements does not seem possible without this addition (apart from "padding tokens" alternative rejected below). In trunk, placements are already "flexible": in that they only map a replication-factor / range pair to set of read and write replicas. We further extend this model by allowing placements to be derived from from tablet mappings.
- Padding Tokens: one of the ways to decouple ownership from position in the ring is by adding RF "padding" tokens. In other words, if we want nodes A, B, and C to own the range from 0 to 100. We can make them own tokens 0, 1, 2, but also 98, 99, 100, correspondingly. While this achieves a similar goal, it creates tiny ranges unnecessarily inflating metadata. In the amount of time we would spend making these padding tokens "virtual" (i.e. not creating a single-token ranges), we can implement a better solution
- Ring-per-keyspace: we can allow different keyspaces to be replicated to different number of nodes in the cluster. Unfortunately, this does not solve the load distribution problem, only gives some placement flexibility. Besides, achieving ring-per-keyspace will be possible after implementing this CEP anyway.
In short, most alternatives were rejected either because they simply can not take us far enough, or because they (or their very close sibling) will be implemented in the scope of this CEP anyways.
SOTA / Related Work
- Monarch: Google’s Planet-Scale In-Memory Time Series Database: https://www.vldb.org/pvldb/vol13/p3181-adams.pdf
- TiDB: A Raft-based HTAP Database: https://www.vldb.org/pvldb/vol13/p3072-huang.pdf
- https://www.yugabyte.com/blog/distributed-sql-sharding-how-many-tablets-size/
- https://jack-vanlightly.com/analyses/2023/11/21/serverless-cockroachdb-asds-chapter-4-part-3
- Cruise Control: https://github.com/linkedin/cruise-control/wiki/Overview
- Kora: A Cloud-Native Event Streaming Platform For Kafka: https://vldb.org/pvldb/vol16/p3822-povzner.pdf
- vnode allocation improvements: https://issues.apache.org/jira/browse/CASSANDRA-7032
- Sam Overton’s vnodes: https://www.scribd.com/document/253239514/Virtual-Nodes-Strategies-for-Apache-Cassandra
- ML discussion: https://www.mail-archive.com/dev@cassandra.apache.org/msg03837.html
- Simple Efficient Load Balancing Algorithms for Peer-to-Peer Systems: http://www.iptps.org/papers-2004/karger-load-balance.pdf
- Load Balancing in Structured P2P Systems: http://iptps03.cs.berkeley.edu/final-papers/load_balancing.ps
- Simple Load Balancing for Distributed Hash Tables: http://iptps03.cs.berkeley.edu/final-papers/simple_load_balancing.ps
Used Acronyms
- TCM - Transactional Cluster Metadata
- BOP - Byte Ordered Partitioner
- UCS - Unified Compaction Strategy
- RF - Replication Factor
- VNodes - Virtual Nodes
- NTS - Network Topology Strategy
Appendix A: Case Study, cost of keeping the cluster balanced
Placement decisions are inherently predictive (using workload forecasts to pre-position tablets), while migration decisions can be reactive (responding to observed utilization changes). This CEP focuses on reactive migrations: relocating tablets based on observed utilization changes, not forecasted load. Predictive placement strategies are out of scope, though operators will be able to collect utilization metrics that could inform such strategies in the future.
A common approach to expansion while preserving balance is to double the cluster by assigning each joining node a token midway between two existing tokens. This guarantees a balanced cluster post-expansion, but at significant data movement cost copying half the data to each joining node. For example, if the cluster stores 100GB of unreplicated data with RF=3, totaling 300GB replicated data, up to 150-200 GB (depending on bootstrap order) may need to be transferred, 1.5× to 2.0× of the total unreplicated dataset. The advantage is that the resulting layout is optimal, since it achieves ideal token distribution and minimizes the number of node pairs with overlapping ranges.
100 node cluster, before bootstrapping 101st node:
Min ownership: 1.0000% of ring
Max ownership: 1.0000% of ring
Difference (max-min): 0.0000% of ring
Std deviation: 0.0000% of ring
Ideal ownership per node: 1.0000% of ring
Adding one node to a perfectly balanced 100-node cluster creates immediate imbalance: the introduced node will receive a portion of the ring from its neighbor(s), but the rest of ring will remain unbalanced. With vnodes, the situation improves slightly: introducing a single node into the cluster does not substantially change the balance in the ring:
101 node cluster, after bootstrap of 101st node, but before executing balancing moves:
Min ownership: 0.4950% of ring
Max ownership: 1.0000% of ring
Difference (max-min): 0.5050% of ring
Std deviation: 0.0697% of ring
Ideal ownership per node: 0.9901% of ring
Balance is restorable, but requires explicit compensating moves after bootstrap:
After executing compensating moves:
Min ownership: 0.9901% of ring
Max ownership: 0.9901% of ring
Difference (max-min): 0.0000% of ring
Std deviation: 0.0000% of ring
Ideal ownership per node: 0.9901% of ring
Vnodes trade perfect balance for flexibility. The baseline is already imperfect, and adding a node only slightly changes it. Token allocation in this example was done using TokenAllocator, which also employs a greedy allocation algorithm similar to the one described above, available in Cassandra:
101 node cluster with 16 vnodes, before bootstrapping 101st node:
Min ownership: 0.6521% of ring
Max ownership: 1.2354% of ring
Difference (max-min): 0.5833% of ring
Std deviation: 0.1494% of ring
Ideal ownership per node: 1.0000% of ring
The new node slots into the existing distribution without improving or worsening balance. No compensating moves are required here, but also no convergence toward ideal.
101 node cluster with 16 vnodes, after bootstrap of 101st node:
Min ownership: 0.6521% of ring
Max ownership: 1.2354% of ring
Difference (max-min): 0.5833% of ring
Std deviation: 0.1488% of ring
Ideal ownership per node: 0.9901% of ring
Tablets offer finer-grained rebalancing, keeping the ring of any size close to optimal allocation. Bootstrapping a single node in a 100-node cluster using tablets/greedy allocation algorithm:
100 node cluster with tablets, before bootstrap of 101st node:
Min ownership: 0.9110% of ring
Max ownership: 1.0750% of ring
Difference (max-min): 0.1640% of ring
Std deviation: 0.0429%
Total tablets: 358
Here, adding a node tightens the distribution. The allocator opportunistically improves balance as part of the bootstrap, without requiring costly rebalancing steps.
101 node cluster with tablets, after bootstrap of 101st node:
Min ownership: 0.9170%
Max ownership: 1.0527%
Difference (max-min): 0.1357%
Std deviation: 0.0379% of ring
Total tablets: 359
To summarize, flexible placements combine the best properties: near-optimal balance at steady state, graceful single-node expansion, and self-correcting behavior without explicit rebalancing phases. For clusters requiring incremental scaling, they avoid both the rigidity of doubling and the entropy accumulation of vnodes.
References
[1] https://issues.apache.org/jira/browse/CASSANDRA-10540
[2] https://docs.cloud.google.com/bigtable/docs/overview
[3] https://svn.apache.org/repos/asf/hbase/hbase.apache.org/trunk/0.94/book/regions.arch.html
[4] https://kudu.apache.org/docs/
[5] https://docs.yugabyte.com/stable/architecture/
[6] https://vitess.io/docs/23.0/concepts/tablet/
[7] https://issues.apache.org/jira/browse/CASSANDRA-14404
[8] https://microsoft.github.io/z3guide/
[9] https://microsoft.github.io/z3guide/docs/optimization/combiningobjectives/