Replication, Partitioning, and Sharding Questions

Direct answer

A good shard key has four properties: high cardinality (many distinct values, so there's enough range to spread the data), even distribution (no single value, or small set of values, dominates the traffic), alignment with the dominant query pattern (queries can be answered by hitting one or a few shards, not all of them), and immutability (the value never changes after a row is created, so a row never needs to move shards just because a field it was sharded on was updated). A field like email hash or user_id typically has all four; a field like status (few distinct values, changes over time, and doesn't align with most queries) typically has none of them.

Structured elaboration

Terminology note: primary key vs. partition/shard key are not always the same thing, and conflating them causes real bugs. A traditional relational primary key's only job is uniquely identifying a row; it says nothing about physical placement. A partition (shard) key's job is exactly the opposite concern: where the row physically lives. In some systems they can be the same column (sharding by a table's own primary key), but in many real schemas they should not be: a messages table's primary key might reasonably be message_id (unique, but a poor shard key: queries almost never filter by message_id alone, they ask for a conversation's history, so sharding on message_id would scatter one conversation's messages across every shard), while its shard key should be conversation_id (great alignment, but not unique across rows, many messages share one conversation_id, which is exactly what makes it a good partition key and a poor primary key). Systems like Cassandra make this split explicit in their terminology: the "partition key" (placement) and the rest of the "primary key" (uniqueness and clustering order) are named and reasoned about separately.

Three common mistakes when picking a partition/shard key:

1. Choosing a low-cardinality field (a status, region, or boolean flag) because it seems like a natural grouping, without checking that it actually spreads load: a handful of distinct values can never spread data across more shards than there are values, and real-world skew across those values (most rows being status = 'active') makes it worse.
2. Choosing a monotonically increasing field (an auto-incrementing id, a created_at timestamp) under range-based sharding, which concentrates all new writes onto whichever shard currently owns the highest range, a classic write-hotspot failure mode: every new row's key sorts higher than the last, so under range-based sharding every write lands on whichever shard currently owns the top of the range, and added shards never pick up a share of new writes.
3. Choosing a field that doesn't match the dominant query pattern, purely because it has good cardinality and distribution: a key that spreads data beautifully but forces every real query to search every shard trades one problem (hotspots) for another (permanently expensive scatter-gather reads) that is just as damaging to performance.

Worked example

import hashlib, statistics

def hash_bucket(value: str, n: int) -> int:
    return int.from_bytes(hashlib.md5(value.encode()).digest()[:8], "big") % n

N = 8
# Candidate 1: user_id (high cardinality, evenly distributed, aligns with per-user queries)
user_ids = [f"user-{i}" for i in range(200_000)]
c1 = [0]*N
for u in user_ids: c1[hash_bucket(u, N)] += 1
print("user_id buckets:", c1, "stdev/mean:", round(statistics.pstdev(c1)/statistics.mean(c1), 4))

# Candidate 2: status (low cardinality, heavily skewed toward one value)
statuses = ["active"]*180_000 + ["suspended"]*15_000 + ["deleted"]*5_000
c2 = [0]*N
for s in statuses: c2[hash_bucket(s, N)] += 1
print("status buckets:  ", c2, "stdev/mean:", round(statistics.pstdev(c2)/statistics.mean(c2), 4))

user_id buckets: [25375, 25130, 25087, 24578, 24870, 25007, 24872, 25081] stdev/mean: 0.0088
status buckets:   [15000, 0, 180000, 0, 0, 0, 0, 5000] stdev/mean: 2.3516

user_id, with 200,000 distinct values, spreads within about 1% of perfectly even across 8 buckets. status, with only 3 distinct values, cannot spread at all in a meaningful sense, its bucket assignment is really just "which of 3 fixed positions each status value happens to land on," and 180,000 rows (90% of the data) all land in a single bucket regardless of how good the hash function is. This is what "low cardinality defeats even distribution no matter what" looks like as a number, not just a claim.

Trade-offs & pitfalls

• A key can satisfy three of the four properties and still be wrong. A perfectly high-cardinality, evenly-distributed, immutable key that doesn't align with any real query still forces scatter-gather reads on every request; all four properties matter together, not as a checklist where 3-out-of-4 passes.
• Cardinality and distribution are not the same thing, and conflating them is a common error: a field can have millions of distinct values and still be badly distributed if a small number of those values account for most of the traffic (a power-law key, like a small number of extremely popular products among millions of product ids).
• "Immutable" is about the key's value, not about whether the row can change at all. Plenty of fields on a row can be freely mutable, updated status, updated preferences, as long as the field chosen as the shard key specifically stays fixed once set.

Direct answer

Replication lag is the gap between when a write is committed on the leader and when a follower (also called a read replica: a secondary copy of the database that continuously applies the leader's writes and can serve read-only queries) has applied that same write, and it is caused by three things stacking up: network transit time to ship the change, follower disk I/O falling behind under write load, and the follower applying changes single-threaded or serially even when the leader wrote them with more parallelism. The user-visible impact is a stale read: a client reads from a replica and does not see a write it (or someone else) just made. Detecting it in production means collecting a lag metric per replica, not just a cluster-wide average, and percentiles matter far more than the average, because the average can look fine while a handful of replicas or a handful of moments are badly behind.

Causes, in the order they usually show up

1. Network: cross-region or congested links add fixed and variable latency to shipping the write-ahead log (WAL, the durable ordered record of every database change) or a logical change stream to the follower.
2. Apply-side bottleneck: the follower's own disk I/O, or lock contention on the follower, slows down how fast it can apply the incoming stream, even though it already received the bytes. This is the most common cause of large lag spikes under load, since receiving is usually network-bound and fast, while applying is disk- and CPU-bound and can queue up.
3. Long-running operations: a large batch job, a schema migration, or a long transaction on the leader produces a burst of changes the follower must apply serially, and single-writer replay (most replication designs apply changes from one source in the original commit order, which limits how parallel the apply side can be) means that burst takes time to drain even after it finishes on the leader.
4. Follower resource starvation: a follower doing double duty (also serving heavy read traffic, or running backups) has less I/O and CPU headroom to keep up with apply, so lag rises under its own read load, not the leader's write load.

Metrics and thresholds, concretely

For PostgreSQL, pg_stat_replication on the primary exposes sent_lsn, write_lsn, flush_lsn, and replay_lsn (LSN: log sequence number, a monotonically increasing position in the WAL), plus derived lag-time columns write_lag, flush_lag, and replay_lag; replay_lag is the one that matches "how stale is a read from this replica," since it measures the delay until the follower has actually applied the change, not just received it. For MySQL 8.0.22 and later, SHOW REPLICA STATUS (the modern name for the older SHOW SLAVE STATUS) exposes Seconds_Behind_Source (renamed from Seconds_Behind_Master) as the equivalent single-number lag metric. Collect the 50th, 90th, and 99th percentile of this lag across all replicas, sampled frequently (every few seconds), rather than a single cluster average, because a p99 spike on one replica behind a load balancer is invisible in an average across five otherwise-healthy replicas.
Reasonable alert thresholds differ sharply by workload, which is exactly why a single global threshold is a common mistake:

Worked example: advising a client on their alerting setup

A Solutions Architect reviewing a client's PostgreSQL fleet finds one alert: "page if replication lag > 30 seconds," applied identically to a user-facing OLTP replica and a nightly-reporting OLAP replica. The fix is not to change the number, it is to split it: page at 10 seconds on the OLTP replica's replay_lag (since that replica backs user-visible reads), and only page at 300 seconds on the OLAP replica while keeping a lower, non-paging warning threshold (60 seconds) to catch a slow degradation before it becomes a problem, plus adding per-replica dashboards rather than a single fleet-wide average so an individual replica falling behind under, say, a backup job is visible before it breaches threshold.

Trade-offs and pitfalls

Alerting on sent_lsn minus the follower's current position (a byte-distance metric) instead of a time-based lag metric is a common mistake: byte distance does not translate linearly to user-visible staleness, since apply speed varies with what kind of writes are queued. The other common mistake is a single lag threshold across a fleet with mixed workloads, as above, which either pages too often on the tolerant workload or not soon enough on the sensitive one. Watch for the "lag looks fine but reads are still stale" trap: replay_lag measures how far behind in time the apply process is, but a connection pooler or client that cached a stale read result outside the database entirely can show staleness even when replication itself is healthy.

Approach

Partition events by range on occurred_at, one monthly partition per range, and choose occurred_at as the partition key because at 100M events/day almost every real query on an events table is time-scoped ("events from today", "events from last month"), so monthly range partitions let the planner prune to just the partitions that overlap a query's time filter instead of scanning years of history. Composite the primary key as (id, occurred_at), not just id: PostgreSQL requires any unique constraint on a partitioned table to include the partition key, since uniqueness has to be enforced per-partition, not globally across the whole partitioned table in one structure. Index (user_id, occurred_at) for per-user lookups within a time window, and add a GIN index on payload (a JSONB column) since the schema implies it will be queried by its contents, not just fetched whole. For future partitions, create the next 1 to 3 months' worth ahead of need (via a scheduled job) rather than reactively, since a missing partition means inserts for that range either fail or, without a DEFAULT partition, error out entirely.

SQL

CREATE TABLE events (
  id BIGSERIAL,
  user_id BIGINT NOT NULL,
  occurred_at TIMESTAMP NOT NULL,
  event_type TEXT,
  payload JSONB,
  PRIMARY KEY (id, occurred_at)
) PARTITION BY RANGE (occurred_at);

CREATE TABLE events_2026_08 PARTITION OF events
  FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');
CREATE TABLE events_2026_09 PARTITION OF events
  FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');
CREATE TABLE events_2026_10 PARTITION OF events
  FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');

-- indexes are created on the parent and PostgreSQL propagates one per partition
CREATE INDEX ON events (user_id, occurred_at);
CREATE INDEX ON events USING gin (payload);

-- plan to add future partitions: run this monthly, well ahead of the month it covers,
-- e.g. as a scheduled job on the 1st of each month creating 2 months out
CREATE TABLE events_2026_11 PARTITION OF events
  FOR VALUES FROM ('2026-11-01') TO ('2026-12-01');

Executed against a real Postgres 16 instance with 300,000 rows seeded across Aug to Oct 2026 (INSERT ... SELECT ... FROM generate_series(1, 300000), then ANALYZE events), with the events_2026_11 partition already created per the future-partition step above but still empty:

-- a query scoped to one month prunes to exactly that partition:
EXPLAIN (COSTS OFF)
SELECT count(*) FROM events WHERE occurred_at >= '2026-09-01' AND occurred_at < '2026-10-01';

 Aggregate
   ->  Seq Scan on events_2026_09 events
         Filter: ((occurred_at >= '2026-09-01') AND (occurred_at < '2026-10-01'))

-- a query NOT scoped by the partition key (user_id alone) has to touch every partition,
-- including the still-empty events_2026_11 partition created ahead of need:
EXPLAIN (COSTS OFF)
SELECT * FROM events WHERE user_id = 42;

 Append
   ->  Bitmap Heap Scan on events_2026_08 events_1
         Recheck Cond: (user_id = 42)
         ->  Bitmap Index Scan on events_2026_08_user_id_occurred_at_idx
               Index Cond: (user_id = 42)
   ->  Bitmap Heap Scan on events_2026_09 events_2
         Recheck Cond: (user_id = 42)
         ->  Bitmap Index Scan on events_2026_09_user_id_occurred_at_idx
               Index Cond: (user_id = 42)
   ->  Bitmap Heap Scan on events_2026_10 events_3
         Recheck Cond: (user_id = 42)
         ->  Bitmap Index Scan on events_2026_10_user_id_occurred_at_idx
               Index Cond: (user_id = 42)
   ->  Seq Scan on events_2026_11 events_4
         Filter: (user_id = 42)

-- a month-boundary query correctly prunes to exactly the two overlapping partitions:
EXPLAIN (COSTS OFF)
SELECT * FROM events WHERE occurred_at >= '2026-08-20' AND occurred_at < '2026-09-05';

 Append
   ->  Seq Scan on events_2026_08 events_1
         Filter: ((occurred_at >= '2026-08-20') AND (occurred_at < '2026-09-05'))
   ->  Seq Scan on events_2026_09 events_2
         Filter: ((occurred_at >= '2026-08-20') AND (occurred_at < '2026-09-05'))

Key points

• occurred_at as the partition key, not id or user_id. It matches the dominant real-world access pattern for an events table (time-scoped queries) and lets range partitioning prune cleanly; id (monotonic) or user_id (no natural ranges) would not give partition pruning any query pattern to key off.
• The primary key had to become (id, occurred_at). This is a genuine PostgreSQL constraint, not a style choice: partitioned tables enforce uniqueness per-partition, so any unique or primary key must include every column of the partition key. If a caller genuinely needs id alone to be globally unique and independently queryable, that's a real trade-off this design makes, id uniqueness is only enforced within its own month's partition, not across the whole table (in practice, BIGSERIAL values are still globally unique in the sequence's own ordering, but the constraint enforcing it is now per-partition).
• The second EXPLAIN shows the honest cost of this design. A query filtered only by user_id (not occurred_at) gets no pruning benefit at all and has to scan every partition that currently exists, including the still-empty events_2026_11 created ahead of need, exactly what the (user_id, occurred_at) index and the partition structure trade off: fast time-scoped queries, at the cost of full fan-out for queries that ignore time entirely.
• Adding future partitions ahead of need, not on demand. Creating a partition is a fast metadata operation, but forgetting to create the next month's partition before the first event in it arrives causes that insert to fail outright (no partition matches its occurred_at), so this has to be a proactive scheduled job, not a reactive one.

Complexity

Partition pruning turns a query scoped to one month into effectively O(rows in that one partition) work instead of O(total rows across all history), the entire performance justification for partitioning at this scale. A query not scoped by the partition key (like the user_id-only example above) is still O(number of partitions) x O(index lookup per partition), better than a full unindexed scan of the whole table, but not the full benefit of pruning; that is the honest cost this schema pays for queries that don't filter on time. Creating a new monthly partition is a fast, roughly constant-time catalog operation, it does not require touching existing data.

Edge cases

• An event arriving with no future partition created yet fails the insert outright under this schema (no DEFAULT partition is defined). Adding a DEFAULT partition would catch stray rows instead of erroring, at the cost of that default partition potentially growing unbounded if the scheduled job silently stops running, worth an explicit choice, not a default assumption.
• An event with a slightly wrong or backdated occurred_at (clock skew, a backfill job inserting historical data) routes correctly to whichever partition actually covers that timestamp, including one that may already be near retention/detachment; retention logic (detaching old partitions) needs to account for late-arriving backdated data if that's a realistic scenario for this system.
• Month-boundary queries (a range like occurred_at >= '2026-08-20' AND occurred_at < '2026-09-05') correctly prune to exactly the two overlapping partitions, verified in the third EXPLAIN output on the events table above, not to every partition or to none.

Direct answer

A Snowflake-style 64-bit ID (Twitter's original scheme, still the reference design) packs a millisecond timestamp, a machine/worker identifier, and a per-machine, per-millisecond sequence counter into one integer, which is naturally sortable by time (the timestamp is the high-order bits), compact (a single 64-bit integer, not a 128-bit or string UUID), and collision-free as long as no two generators ever reuse the same (timestamp, machine-id, sequence) triple. At the required 100k IDs/sec cluster-wide, sequence-bit exhaustion is nowhere close to the binding constraint; the real design work is machine-id assignment and clock-skew handling, not throughput.

Structured elaboration

The bit layout

$N_{\text{ids/sec/node}} = 2^{\text{seq\_bits}} \times 1000$

Why this meets every stated requirement

• Sortable by time: because the timestamp occupies the highest-order bits (after the unused sign bit), IDs generated later numerically sort after IDs generated earlier, globally, without needing a separate created_at column for ordering.
• Space efficient: one 64-bit integer, versus 128 bits for a UUID, meaning smaller indexes and less storage per row, which matters at the scale of a large, heavily sharded table with billions of rows.
• Collision-free: a collision requires two generators to produce the same triple of (timestamp millisecond, machine ID, sequence number); since each machine ID's sequence is independently maintained and machine IDs are unique by assignment, this cannot happen as long as machine-ID assignment itself is correct and no generator's clock moves backward without the generator noticing.
• 100k IDs/sec cluster-wide: as the worked example below shows, this target is met with enormous headroom even by a single generator instance, so the interesting design question is not "can we hit the throughput," it is "how do we assign machine IDs safely" and "how do we handle clock anomalies."

Machine-ID assignment

10 bits allows 1,024 distinct concurrent generator instances. In practice this is managed one of two ways: statically, baked into each instance's configuration at deploy time (simple, but requires careful bookkeeping to avoid ever reusing an ID for two live instances), or dynamically, via a coordination service (etcd or ZooKeeper-style leader election / lease system) that hands out a machine ID on startup and releases it on clean shutdown. Dynamic assignment is the safer default at any real scale, since static assignment's failure mode, two instances accidentally configured with the same ID, is a silent collision generator, not a loud failure.

Clock-skew and NTP handling

The timestamp field assumes the generator's clock only moves forward. An NTP correction that steps the clock backward (rare, but real) can cause a generator to produce an ID with an EARLIER timestamp than one it already produced, breaking strict sortability and, in a worst case, risking a collision if the sequence counter also happens to repeat. The standard mitigation: each generator remembers the last timestamp it used; if the system clock ever reports an earlier time than that, the generator refuses to mint new IDs (or busy-waits) until the clock catches back up past the last-used timestamp, trading brief unavailability of that one generator for the sortability and collision-free guarantees.

Alternatives compared

Worked example

SIGN_BITS, TS_BITS, MACHINE_BITS, SEQ_BITS = 1, 41, 10, 12
assert SIGN_BITS + TS_BITS + MACHINE_BITS + SEQ_BITS == 64

max_ms = 2**TS_BITS - 1
years = max_ms / 1000 / 60 / 60 / 24 / 365.25
print(f"timestamp field: {TS_BITS} bits -> max {max_ms:,} ms = {years:.1f} years before rollover")

max_machines = 2**MACHINE_BITS
print(f"machine-id field: {MACHINE_BITS} bits -> {max_machines:,} distinct generator instances")

max_seq_per_ms = 2**SEQ_BITS
per_machine_ids_per_sec = max_seq_per_ms * 1000
print(f"sequence field: {SEQ_BITS} bits -> {max_seq_per_ms:,} IDs/ms/machine -> {per_machine_ids_per_sec:,} IDs/sec/machine ceiling")

REQUIRED = 100_000
print(f"required: {REQUIRED:,} IDs/sec cluster-wide; a SINGLE generator's ceiling covers it "
      f"{per_machine_ids_per_sec/REQUIRED:.0f}x over")
for n in (4, 16, 64):
    per_node_rate = REQUIRED / n
    print(f"  spread across {n:>3} generators: {per_node_rate:,.0f} IDs/sec/instance -> "
          f"{per_machine_ids_per_sec/per_node_rate:.0f}x headroom under the per-machine ceiling")

Output:

timestamp field: 41 bits -> max 2,199,023,255,551 ms = 69.7 years before rollover
machine-id field: 10 bits -> 1,024 distinct generator instances
sequence field: 12 bits -> 4,096 IDs/ms/machine -> 4,096,000 IDs/sec/machine ceiling
required: 100,000 IDs/sec cluster-wide; a SINGLE generator's ceiling covers it 41x over
  spread across   4 generators: 25,000 IDs/sec/instance -> 164x headroom under the per-machine ceiling
  spread across  16 generators: 6,250 IDs/sec/instance -> 655x headroom under the per-machine ceiling
  spread across  64 generators: 1,562 IDs/sec/instance -> 2621x headroom under the per-machine ceiling

The 69.7-year rollover confirms the well-known property of this exact bit layout, and the headroom numbers confirm that even a single generator meets the 100k/sec requirement 41 times over; running several generators (4 to 64, in this illustration) is therefore a choice made for availability and fault isolation (no single generator instance is a single point of failure), not because throughput demands it.

Trade-offs and pitfalls

• Sortability here is approximate, not exact: two IDs generated within the same millisecond on two different machines sort only by their machine-ID bits relative to each other, which is arbitrary, not causally meaningful; do not rely on Snowflake-style ID ordering for fine-grained, sub-millisecond causal ordering between machines.
• The machine-ID space (1,024 with 10 bits) is the practical ceiling on how many concurrent generator instances you can ever run; exhausting it means either widening the field (a breaking schema change to every consumer) or moving to dynamic, leased assignment to reuse retired instances' IDs safely.
• A generator that keeps minting IDs during a detected backward clock jump, instead of stalling, is the single most common way this scheme's collision-free guarantee quietly breaks in production; the stall-until-caught-up behavior is not optional hardening, it is the mechanism that makes the "collision-free" claim actually true.
• UUIDv7 is now a real, standardized alternative that gets you time-ordering without building or operating a machine-ID coordination scheme at all, at the cost of double the storage (128 vs 64 bits); for a new system with no existing Snowflake-style infrastructure, that trade is worth weighing before building a custom generator.

Direct answer

Three practical strategies for a cross-shard join, users sharded by user_id and transactions by account_id: broadcast the smaller side to every shard so the join runs locally, denormalize (co-locate) the needed fields so no join is required at query time at all, or scatter-gather (query every relevant shard and join in the application or coordinator layer). Pick broadcast when one side is small and rarely changes, denormalization when the query pattern is known ahead of time and you can tolerate write-side duplication, and scatter-gather when neither is true and the query pattern is genuinely ad hoc; for heavy analytical cross-shard aggregation specifically, none of the three is usually right, stream the data via change-data-capture into a separate analytical store instead and run the join there.

Structured elaboration

Strategy 1: Broadcast / replicate the smaller side

Copy the small table (here, users) to every shard that holds transactions, so a join against account_id can execute entirely locally on each shard without a network round trip to fetch the other side. A back-of-envelope check on whether "small" actually applies: 5 million users at roughly 200 bytes/row is about 1 GB, comfortably replicable to every shard's local cache; the same approach for 5 billion users, a 1,000x larger population at the same row size, is about 1 TB per shard (5,000,000,000 x 200 bytes = 1e12 bytes), no longer a "just replicate it" decision. Latency: lowest of the three once replicated (fully local join, no fan-out at query time). Complexity: the replication pipeline itself (keeping every shard's copy current as users changes) is the real cost, not the join logic. Pick this when one side is genuinely small and changes infrequently relative to query volume.

Strategy 2: Co-location / denormalization

Duplicate the fields you actually need from the small side directly onto the large side at write time (for example, store a user_region or user_tier column directly on each transactions row), so the query that used to be a join becomes a single-table filter. Latency: lowest possible, no join and no fan-out at all. Complexity: the write path now owns keeping the duplicated fields current whenever the source changes (an update to a user's tier has to fan out to every transaction row that denormalized it, or you accept eventual consistency on the duplicated copy). Pick this when the query pattern needing those specific fields is known in advance and stable, and the write-side cost of keeping duplicates current is smaller than the query-time cost of joining on every read.

Strategy 3: Scatter-gather with an application- or coordinator-side join

Query every shard that could hold matching transactions rows for a given set of user_ids, fetch each shard's local matches, and join the results in the application layer or a coordinator process rather than in the database. Latency: highest of the three, bounded by the slowest of however many shards are fanned out to, and that number grows with cluster size (a scatter-gather query gets slower, not faster, as you shard more finely for scale). Complexity: no schema change and no replication pipeline needed, the tax is paid entirely at query time, in fan-out cost and in application code that now does the join. Pick this for genuinely ad hoc queries where the access pattern is not known well enough in advance to justify broadcasting or denormalizing, accepting the latency cost as the price of flexibility.

A fourth option for analytics specifically: CDC to a separate analytical store

When the real need is aggregation or analytics across shards (dashboards, reports) rather than a single row's lookup, none of the three strategies above, all built for OLTP (online transaction processing: the day-to-day, single-row reads and writes a live application performs) workloads, is the right tool: they either bloat the OLTP write path (denormalization) or put unpredictable load on the OLTP shards at query time (scatter-gather). Stream both users and transactions changes via change-data-capture into a separate analytical store (a columnar warehouse or similar), and run the cross-shard join and aggregation there, off the OLTP critical path entirely. This keeps the transactional shards answering only the queries they were sized for, and gives the analytical workload a store actually built for scanning and joining at scale.

Worked example

Query: "for each region, total transaction amount in the last 30 days," users sharded by user_id, transactions by account_id.

• Broadcast fits if users (region lookup) is small: replicate users(user_id, region) to every transaction shard, then each shard runs SELECT region, SUM(amount) FROM transactions t JOIN users u ON t.user_id = u.user_id WHERE occurred_at > now() - interval '30 days' GROUP BY region locally, and the coordinator only has to sum the per-shard, per-region partials (an associative merge: summing each shard's partial sum and then summing those partials again gives the same total as summing every row at once), never re-scanning base rows.
• Denormalize fits if this exact query runs constantly and users rarely change region: add region directly to each transactions row at write time, and the query becomes a single-table GROUP BY with no join at all, on any single shard.
• Scatter-gather fits if this is a one-off, ad hoc report requested occasionally with varying filters: query all transaction shards for the raw rows in range, fetch the small set of (user_id, region) pairs needed to resolve them, and join in the application layer, accepting the fan-out cost since it is infrequent.
• CDC-to-analytics fits if this kind of report runs routinely across many dimensions, not just region: stream both tables into a warehouse once, and every future report variant is a single query there, none of them touching the OLTP shards.

Trade-offs and pitfalls

• Broadcasting a table that is not actually small (or that grows past "small" without anyone revisiting the decision) silently turns every shard's local storage and cache budget into a scaling bottleneck; put a size or growth-rate alert on any broadcast table, not just a one-time judgment call at design time.
• Denormalization's failure mode is staleness: if the fan-out update on change is missed or delayed, queries silently read a stale duplicated value with no error to signal it; decide explicitly whether that duplicated field needs strong consistency (synchronous fan-out update) or can tolerate eventual consistency, rather than leaving it undefined.
• Scatter-gather's fan-out cost is not fixed, it grows with shard count (more shards fanned out to, more chances one of them is slow), which means a query pattern that was fine at 8 shards can become a tail-latency problem purely from adding shards for unrelated scaling reasons, with no change to the query itself.
• Reaching for CDC-to-an-analytical-store for what is actually a single, simple, frequently-run OLTP-path query is over-engineering; that pattern earns its complexity specifically for genuinely analytical, multi-dimensional, or exploratory workloads, not as a default first choice for every cross-shard read.