Sharding and Replication
When a Go service's data stops fitting on one machine, or one machine stops holding the write flow, the data is spread across several instances — and all the interest begins after the word "spread." First you must distinguish partitioning (cutting one table into chunks within a single DB instance) from sharding (splitting data across different instances), then pick the shard key that decides which machine a row lands on, find a way to route a query to the right shard, survive rebalancing as the cluster grows, and somehow answer queries that need data from many shards at once. Running alongside all of this is replication: a shard almost never lives alone — it has replicas for durability and for read scaling.
The central trap of this topic is assuming sharding "just spreads data across servers" and that is that. In reality the cost hides in the details: a low-cardinality shard key (like a user's gender) creates a hot shard and load skew; a popular user or a viral post produces a hot key inside a shard; asynchronous replication brings lag and stale reads, while synchronous replication is slower; cross-shard JOIN, aggregation, and key-less search are expensive and eat the gain. A separate layer is the choice of storage (SQL versus NoSQL for the real access patterns) and the decoupling of the production OLTP load from the analytical OLAP load, so a heavy report does not take prod down. This topic breaks data distribution into layers — from the difference between partitions and shards to the pipeline from OLTP to OLAP.
Topic Map
- Partitioning versus sharding — partitioning slices one table into chunks (range/list/hash) inside a single DB instance so the planner skips irrelevant partitions, while sharding spreads data across different instances.
- Why shard — to raise write throughput, spread CPU load, geo-distribute data closer to users, and store a volume one server cannot hold; a shard always runs with replicas.
- Shard-key selection — high cardinality (user_id good, gender bad), even load distribution, and stable access patterns; a bad key creates a hot shard or a cross-shard query on every request.
- Hot keys — one key or shard taking disproportionate traffic (a celebrity user, a viral item); mitigated by splitting the hot range, salting the key, caching it, or giving it a dedicated shard.
- Shard routing — how a query finds its shard: a DSN string with connection info, a proxy that knows the map, or a coordinator node that plans and forwards (e.g. Citus for Postgres), with trade-offs in latency and the risk of a bottleneck.
- Shard rebalancing — triggered when a shard grows too large, runs hot, the cluster grows or shrinks, or the shard key changes; ranges move with minimal downtime, and consistent hashing minimizes how much data moves.
- Cross-shard operations — cross-shard JOIN, aggregation, and key-less row search are expensive; you avoid them with denormalization, fan-out then merge, or a search index (Elasticsearch).
- Replication — a primary/master takes writes, replicas serve reads; async replication is fast but causes lag and stale reads, sync replication is consistent but slower, and failover promotes a replica to primary.
- SQL versus NoSQL — SQL (relational, ACID, joins, rigid schema, strong consistency) versus NoSQL (document, key-value, wide-column, flexible schema, easy horizontal scale, often eventual consistency); choose by access patterns.
- From OLTP to OLAP — keep transactional (OLTP, Postgres) and analytical (OLAP) load apart, streaming changes via CDC or domain events and Kafka into OLAP (ClickHouse), a datalake (S3), and a warehouse (Snowflake).
Common Mistakes and Traps
| Mistake | Consequence |
|---|---|
| Confusing partitioning with sharding | You expect write scale from partitions inside one instance — the single-machine ceiling is still there |
| Sharding without replicas | A single node failure loses its data and part of the cluster — neither durability nor read scaling |
| Picking a low-cardinality shard key (gender, status) | A few overloaded shards and the rest idle — load skew |
| Choosing a key without regard for access patterns | Every query goes cross-shard because the needed row is not found by the key |
| Ignoring a hot key inside a shard | One celebrity user or a viral post takes a whole shard down |
| Routing through a single coordinator with no backup | The routing node becomes a bottleneck and a single point of failure |
| Rebalancing with a plain hash over the node count | Adding a node moves almost the whole cluster instead of a small fraction |
| Building the product on cross-shard JOIN and key-less search | Every such query fans out across all shards — expensive and slow |
| Treating an async replica as a source of fresh data | Replication lag gives stale reads and a read-after-write anomaly |
| Reaching for NoSQL "for scale" with no real reason | You lose joins and strong consistency where a simple Postgres would do |
| Running analytics with heavy queries on the production OLTP | The report loads the transactional DB and tanks the main service's latency |
Interview Relevance
Data distribution is a mandatory topic at the senior level of a Go interview in the system-design part, and the question is not whether you know the word "sharding" but whether you understand its cost. The interviewer checks whether you distinguish partitioning from sharding, whether you pick a shard key by cardinality and access patterns, whether you remember that a shard runs with replicas, and whether you understand that async replication brings lag and cross-shard operations are expensive.
What interviewers usually check:
- The difference between partitioning (one table, one instance) and sharding (data across different instances) and why shard at all.
- How to pick a shard key (high cardinality, even load, stable access patterns) and what a bad key costs you.
- What a hot key and a hot shard are and how to fight them (split the range, salt the key, cache it, dedicate a shard).
- How a query finds its shard (DSN, proxy, a coordinator like Citus) and where routing's bottleneck and point of failure are.
- When and how to rebalance shards and why consistent hashing minimizes data movement.
- Why cross-shard JOIN, aggregation, and key-less search are expensive and how to avoid them (denormalization, fan-out + merge, a search index).
- How replication works (a primary takes writes, replicas serve reads), how async differs from sync, and what replication lag and failover are.
- How to choose between SQL and NoSQL by access patterns and consistency and why it is sensible to start with a simple Postgres.
- Why to separate OLTP and OLAP and how to stream changes via CDC and Kafka into ClickHouse, S3, and Snowflake.
A typical wrong answer: "sharding is just spreading data across several servers." This triggers a discussion that without replicas a node failure loses data, that a low-cardinality key creates a hot shard, that routing and rebalancing are a separate engineering problem, that a cross-shard JOIN fans out across all shards, that async replication gives stale reads, and that analytics must not run on the production OLTP.