Byte-range shard partitioning divides a keyspace into contiguous byte ranges, each assigned to a specific node or replica group. Unlike hash-based partitioning, range partitioning preserves key ordering, enabling efficient range scans, prefix queries, and sequential access patterns that are critical for search indexes and ordered storage engines.
This approach is used by systems like Google Bigtable, HBase, CockroachDB, and TiKV. In a typical distributed search system, byte-range partitioning distributes search index segments across nodes, allowing queries to target only the shards that contain relevant key ranges rather than broadcasting to every node.
The tradeoff is that range partitioning requires active management of split and merge operations to handle hotspots. When write traffic concentrates on a narrow key range, that shard becomes a bottleneck. Effective implementations monitor shard sizes and access patterns to trigger automatic rebalancing.
[]byte{} (empty, representing the minimum) to nil (representing positive infinity).Range partitioning preserves key ordering across shards, making range scans and prefix queries efficient because they only touch contiguous shards rather than requiring scatter-gather across all nodes.
The shard mapping table is the single source of truth for data placement. Its consistency and availability are critical—an inconsistent mapping leads to misrouted requests and data loss. This is why it is typically replicated via Raft or stored in a strongly consistent coordination service.
Hotspot mitigation in range partitioning requires monitoring both shard size and request rate. A shard may be small but receive disproportionate traffic (e.g., a popular key prefix), requiring a split based on load rather than size alone.
The entire keyspace is treated as a continuous byte sequence from 0x00...00 to 0xFF...FF. Initially, the system starts with a single shard covering the entire range. As data is ingested, shards are split when they exceed a configured size threshold (e.g., 256 MB). Each split produces two shards with roughly equal data volume.
The split point is chosen by sampling keys in the shard and selecting the median. Some implementations use approximate techniques—for example, scanning the shard’s index to find the key closest to the midpoint by byte count rather than key count, which better handles variable-length values.
When a client needs to read or write a key, it consults the shard mapping table. The table is a sorted list of (startKey, shardID, nodeAddr) tuples. A binary search over the start keys finds the shard that owns the target key. Clients typically cache this table and refresh it when they receive a “not my shard” error from a node, indicating the mapping has changed due to a split or rebalance.
When a new node joins the cluster, the system identifies overloaded nodes and transfers some of their shards to the newcomer. Unlike hash-based schemes that rebalance by rehashing, range-based rebalancing transfers entire shards intact. The shard mapping table is updated atomically (via a Raft proposal or distributed transaction) to redirect traffic to the new owner.
In a distributed search index, byte-range partitioning maps document ID ranges to index shards. When a search query includes a document ID filter or range predicate, the query router uses the shard mapping to fan out only to the relevant shards, significantly reducing query latency compared to a broadcast approach.
In a distributed storage layer, range partitioning enables efficient compaction. Because each shard owns a contiguous key range, compaction within a shard is a local operation that does not interfere with adjacent shards. This allows the system to compact shards independently and in parallel.
Range partitioning also simplifies backup and restore operations. Each shard can be snapshot independently, and restoring a specific key range only requires loading the corresponding shard snapshots rather than reconstructing the full dataset.
Production systems often use a two-level routing scheme. A lightweight routing layer (or client library) caches the shard mapping table and routes requests directly to the owning node. If the mapping is stale, the node returns a redirect error with the updated mapping, and the client refreshes its cache. This avoids making the metadata service a bottleneck on every request.
If you know the key distribution in advance (e.g., keys are UUIDs with uniform distribution, or timestamps that will grow monotonically), you can pre-split the keyspace at initialization time. This avoids the cascading split storms that occur when all writes initially land on a single shard.
Each node periodically reports shard statistics (size in bytes, key count, read/write QPS) to the metadata service. The rebalancing controller uses these statistics to decide when to split, merge, or transfer shards. Implementing hysteresis (e.g., only split when a shard exceeds 256 MB and only merge when both neighbors are below 64 MB) prevents oscillation.
package sharding
import (
"bytes"
"fmt"
"sort"
"sync"
)
type Shard struct {
ID string
StartKey []byte // inclusive
EndKey []byte // exclusive; nil means +infinity
NodeAddr string
}
type ShardRouter struct {
mu sync.RWMutex
shards []Shard // sorted by StartKey
}
func NewShardRouter(shards []Shard) *ShardRouter {
sorted := make([]Shard, len(shards))
copy(sorted, shards)
sort.Slice(sorted, func(i, j int) bool {
return bytes.Compare(sorted[i].StartKey, sorted[j].StartKey) < 0
})
return &ShardRouter{shards: sorted}
}
// Route finds the shard that owns the given key using binary search.
func (r *ShardRouter) Route(key []byte) (*Shard, error) {
r.mu.RLock()
defer r.mu.RUnlock()
idx := sort.Search(len(r.shards), func(i int) bool {
return r.shards[i].StartKey != nil &&
bytes.Compare(r.shards[i].StartKey, key) > 0
}) - 1
if idx < 0 {
idx = 0
}
shard := &r.shards[idx]
if shard.EndKey != nil && bytes.Compare(key, shard.EndKey) >= 0 {
return nil, fmt.Errorf("key %x not covered by any shard", key)
}
return shard, nil
}
// RouteRange returns all shards that overlap with [startKey, endKey).
func (r *ShardRouter) RouteRange(startKey, endKey []byte) []Shard {
r.mu.RLock()
defer r.mu.RUnlock()
var result []Shard
for _, s := range r.shards {
// Skip shards entirely before the range
if s.EndKey != nil && bytes.Compare(s.EndKey, startKey) <= 0 {
continue
}
// Stop at shards entirely after the range
if endKey != nil && bytes.Compare(s.StartKey, endKey) >= 0 {
break
}
result = append(result, s)
}
return result
}package sharding
import "bytes"
type ShardStats struct {
ShardID string
SizeBytes uint64
KeyCount uint64
ReadQPS float64
WriteQPS float64
}
const (
MaxShardSize = 256 << 20 // 256 MB
MinShardSize = 64 << 20 // 64 MB
HotShardQPS = 10000 // QPS threshold for load-based split
)
type SplitDecision struct {
ShardID string
SplitKey []byte
Reason string
}
func EvaluateSplits(stats []ShardStats, getSplitKey func(string) []byte) []SplitDecision {
var decisions []SplitDecision
for _, s := range stats {
var reason string
switch {
case s.SizeBytes > MaxShardSize:
reason = "size_exceeded"
case s.ReadQPS+s.WriteQPS > HotShardQPS && s.KeyCount > 1000:
reason = "hot_shard"
default:
continue
}
splitKey := getSplitKey(s.ShardID)
if splitKey != nil {
decisions = append(decisions, SplitDecision{
ShardID: s.ShardID,
SplitKey: splitKey,
Reason: reason,
})
}
}
return decisions
}
type MergeDecision struct {
ShardA string
ShardB string
}
func EvaluateMerges(shards []Shard, stats map[string]ShardStats) []MergeDecision {
var decisions []MergeDecision
for i := 0; i < len(shards)-1; i++ {
a, b := shards[i], shards[i+1]
sa, sb := stats[a.ID], stats[b.ID]
// Only merge adjacent shards that are both small
if sa.SizeBytes < MinShardSize && sb.SizeBytes < MinShardSize {
combined := sa.SizeBytes + sb.SizeBytes
if combined < MaxShardSize {
if bytes.Equal(a.EndKey, b.StartKey) {
decisions = append(decisions, MergeDecision{
ShardA: a.ID, ShardB: b.ID,
})
i++ // skip b in next iteration
}
}
}
}
return decisions
}