How does sharding change queries and transactions

• Single-shard query – the router computes the owning shard from the shard key and sends the query to that one node, which answers it locally against a smaller dataset, so cost equals an unsharded query.
• Scatter-gather – when the shard key is absent or all data is needed, the router fans the query to every shard, each executes locally, and partial results are merged, bounded by the slowest shard.
• Cross-shard transaction – single-node ACID no longer applies because each shard is an independent database, so atomicity across shards requires two-phase commit (2PC) or a saga with compensating actions.
• Co-location principle – related data that is queried or changed together is placed on the same shard so common operations stay single-shard and keep ACID.
• Shard key selection – the key must provide even distribution AND ensure the most frequent operations fall within one shard, not just one or the other.

What is sharding and why does it change queries?

Sharding horizontally partitions a dataset across multiple independent database nodes. Each node holds a distinct slice of the data determined by a shard key. The benefit is that both storage and write throughput scale with the number of shards. The cost is that queries and transactions that were trivial on a single node become fundamentally different depending on whether they touch one shard or many.

How does a single-shard query work?

When the query includes the shard key, the shard router computes which node owns that key and sends the query only there. That node executes it against its own smaller dataset and indexes and returns the result. The cost is the same as a query on an unsharded database, which is why single-shard queries are cheap and are the operations a good design optimizes for.

package com.bank.repository;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class ShardedAccountRepository {

    private final ShardRouter router;

    public ShardedAccountRepository(ShardRouter router) {
        this.router = router;
    }

    // Single-shard query: shard key (accountId) is present
    public BigDecimal findBalance(String accountId) throws Exception {
        Connection conn = router.connectionFor(accountId); // routes to one shard
        try (PreparedStatement ps = conn.prepareStatement(
                "SELECT balance FROM accounts WHERE account_id = ?")) {
            ps.setString(1, accountId);
            ResultSet rs = ps.executeQuery();
            return rs.next() ? rs.getBigDecimal("balance") : BigDecimal.ZERO;
        }
    }
}

Expected output (single-shard):
Query routed to shard-2 for accountId SAV2001
Balance: 150000.00

What is a scatter-gather and why is it expensive?

When the query does not include the shard key, or it inherently needs all data ("count all accounts", "sum every balance"), the router fans it out to every shard in parallel. Each shard runs it locally, and the router gathers all partial results and merges them. This is expensive for several reasons.

• It consumes resources on every shard simultaneously.
• Latency is bounded by the slowest shard, since the router waits for all of them.
• Global sorting, pagination, and aggregation need cross-shard merging and sometimes over-fetching.
• Adding shards makes each scatter-gather fan out wider, so the operation gets relatively more expensive as the cluster grows.

// Scatter-gather: no shard key, must fan out
public long countAllAccounts() throws Exception {
    long total = 0;
    for (Connection conn : router.allShardConnections()) {
        try (PreparedStatement ps = conn.prepareStatement(
                "SELECT COUNT(*) AS cnt FROM accounts")) {
            ResultSet rs = ps.executeQuery();
            if (rs.next()) total += rs.getLong("cnt");
        }
    }
    return total; // bounded by slowest shard
}

How do cross-shard transactions differ from single-node ACID?

On a single node, a transfer that debits Rajesh and credits Priya is one local ACID transaction that either commits or rolls back. When Rajesh and Priya are on different shards, that guarantee disappears because each shard is an independent database. Making it atomic requires a distributed-transaction protocol.

Two-phase commit (2PC)

A coordinator asks all shards to prepare, then tells them all to commit only if every one voted yes. This preserves atomicity but holds locks across a network round trip, is slow, and blocks if the coordinator fails at the wrong moment.

Saga pattern

A saga replaces one atomic transaction with a sequence of local transactions plus compensating actions. It keeps each shard's work local and avoids distributed locks but gives up isolation, so intermediate states are visible.

package com.bank.saga;

import java.math.BigDecimal;

public class TransferSaga {

    private final ShardedAccountRepository repo;

    public TransferSaga(ShardedAccountRepository repo) {
        this.repo = repo;
    }

    public void transfer(String fromId, String toId, BigDecimal amount) {
        // Step 1: local debit on source shard
        boolean debited = repo.debit(fromId, amount);
        if (!debited) throw new IllegalStateException("Insufficient funds");

        // Step 2: local credit on target shard
        try {
            repo.credit(toId, amount);
        } catch (Exception e) {
            // Compensating action: reverse the debit
            repo.credit(fromId, amount);
            throw new RuntimeException("Transfer failed, debit reversed", e);
        }
    }
}

Expected output (happy path):
Debit SAV2001 on shard-1: 50000.00
Credit SAV3002 on shard-3: 50000.00
Transfer complete.

How is 2PC different from a saga?

| Aspect | Two-Phase Commit | Saga |
|---|---|---|
| Atomicity | All-or-nothing across shards | Local per step, compensated on failure |
| Isolation | Full (locks held) | Relaxed (intermediate states visible) |
| Latency | High (multiple network round trips) | Lower per step |
| Availability | Low (blocks if coordinator fails) | Higher (no single coordinator lock) |
| Complexity | Coordinator is a single point of failure | Compensations must be idempotent |
| Best for | Short, rare, must-be-atomic operations | High-throughput, long-running flows |

Why must you co-locate related data?

Because single-shard operations are cheap and cross-shard operations are expensive, you place related data that is queried or changed together onto the same shard. In banking, you shard by customer ID so that all of Rajesh's accounts and transactions land on one shard. Viewing his profile is single-shard. Moving money between his own accounts is a local ACID transaction. Only a transfer between two customers on different shards requires a saga.

What are common mistakes with sharded queries?

The wrong approach is choosing a shard key that distributes evenly but forces scatter-gathers on the hot path.

// WRONG: sharding by random UUID distributes evenly but
// "find all accounts for customer X" becomes scatter-gather
// because the customer ID is not the shard key.
String shardKey = UUID.randomUUID().toString();

// CORRECT: shard by customer ID so all of a customer's
// accounts land on one shard and lookups are single-shard.
String shardKey = customer.getId(); // e.g. "CUST-1001"

What about denormalization in sharded systems?

Denormalize deliberately so related data sits together even at the cost of duplication, because a cross-shard join is far worse than some redundancy. For example, store the customer name alongside each transaction row on the same shard rather than joining back to a customer table on another shard.

What should a senior engineer know?

• Prefer sagas over 2PC for high-throughput flows; reserve 2PC for rare, short, must-be-atomic cases.
• Make every cross-shard step idempotent so retries after partial failure are safe.
• Accept eventual consistency for cross-shard views while keeping strong consistency within a shard.
• Push scatter-gather queries to an analytics replica or a search index rather than running them on the operational shards.
• Monitor per-shard latency so you detect data skew early.

What are common interview follow-up questions?

• How do you handle global unique constraints across shards? (Central sequence or UUID generation.)
• How do you paginate across shards? (Each shard returns its page; merge and re-sort on the router, or use a search index.)
• When is a scatter-gather acceptable? (For rare analytics, off-peak batch jobs, or when served from replicas.)