跪拜 Guibai
← Back to the summary

Sharding a 100-Million-Row Order Table Without Breaking Merchant Queries

Foreword

When a single table exceeds 50 million rows, it's time to start planning for sharding.

A certain e-commerce platform's MySQL order table reached 700 million rows, and a fatal problem emerged: a simple SELECT * FROM orders WHERE user_id = xxx LIMIT 10 query took 12 seconds. The B+ tree index depth reached 5 levels, causing a surge in disk I/O; a single table exceeded 200GB, and the backup window broke through 6 hours; write concurrency reached 8000 QPS, and master-slave replication lag hit 15 minutes.

If you are responsible for a system with a daily order volume in the millions and existing data about to exceed 100 million, this article should help you. I will explain the entire practical approach to sharding, from sharding strategy design, order number generation (gene method), multi-dimensional merchant queries, technology selection, to smooth migration.

1. When Should You Shard?

Many people, when asked in an interview "when to shard," blurt out: "The Alibaba development manual says to shard when a single table exceeds 5 million rows."

This is a bit dogmatic. With current hardware (SSD + large memory), a single table with 10 million rows can still be blazing fast if indexes are built properly.

What really forces you to shard is usually not 'storage capacity,' but 'connection count' and 'maintenance cost':

Practical advice: Usually, start planning for sharding in the 8 million to 10 million row range. Prioritize hardware upgrades and index optimization first; only split when the system truly can't handle it.

2. Core Design: Sharding Strategy

2.1 How to Choose a Shard Key?

The choice of shard key determines how data is distributed and is the cornerstone of the entire design. Choosing wrong leads to pitfalls everywhere.

Three Principles for Choosing a Shard Key:

Principle Description Counter-example
Discreteness Avoid data hotspots Sharding by status (order status), where most orders are "completed"
Business Relevance 80% of queries must carry this field Sharding by a field that is almost never used
Stability The value does not change with business operations Sharding by phone number (users may change numbers)

For an e-commerce order system, the preferred shard key is user_id. The reasons are simple:

2.2 Sharding Strategy: Hash Modulo

Adopt the hash modulo strategy, the most mature and commonly used approach.

// Database sharding: user_id modulo the number of databases
int dbIndex = hash(user_id) % numberOfDatabases;

// Table sharding: user_id modulo the total number of tables, then divided by the number of databases
int tableIndex = hash(user_id) / numberOfDatabases % numberOfTablesPerDatabase;

2.3 Capacity Planning: Don't Just Calculate for the Present

Don't just focus on the current 100 million records. You need to estimate data growth for the next 3-5 years and design the total number of shards accordingly.

Calculation formula:

Total data volume = Existing data + (Daily increment × Estimated days)
Total shards = Number of databases × Tables per database ≥ Total data volume / Recommended capacity per table

Recommended single-table capacity: Control it within 5 million to 10 million rows, or keep the data file size within 2GB to 10GB.

Practical case: An order system currently has 80 million records and is expected to reach 500 million in 3 years:

Key technique: Powers of 2. The number of databases and tables per database should ideally be designed as powers of 2 (e.g., 8, 16, 32, 64, 128). This allows for future expansion via doubling, minimizing the amount of data migration.

2.4 ShardingSphere Configuration Example

Using ShardingSphere-JDBC, the configuration is as follows:

Maven dependency:

<dependency>
    <groupId>org.apache.shardingsphere</groupId>
    <artifactId>shardingsphere-jdbc-core</artifactId>
    <version>5.3.0</version>
</dependency>

YAML configuration (4 databases × 16 tables):

spring:
  shardingsphere:
    datasource:
      names: ds0, ds1, ds2, ds3
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/order_0
        username: root
        password: 123456
      # ds1, ds2, ds3 similarly...
    sharding:
      tables:
        t_order:
          actual-data-nodes: ds$->{0..3}.t_order_$->{0..15}
          table-strategy:
            standard:
              sharding-column: user_id
              precise-algorithm-class-name: com.example.UserIdTableShardingAlgorithm
          key-generator:
            column: id
            type: SNOWFLAKE

Sharding algorithm implementation:

public class UserIdTableShardingAlgorithm implements PreciseShardingAlgorithm<Long> {
    @Override
    public String doSharding(Collection<String> availableTargetNames, 
                              PreciseShardingValue<Long> shardingValue) {
        Long userId = shardingValue.getValue();
        // 4 databases
        int dbIndex = (int) (userId % 4);
        // 16 tables per database
        int tableIndex = (int) (userId / 4 % 16);
        return "ds" + dbIndex + ".t_order_" + tableIndex;
    }
}

3. The Order Number Query Challenge: The "Gene Method"

3.1 What's the Problem?

Users query orders in two main ways:

  1. Query their own order list → uses user_id, can directly locate the shard ✅
  2. Query by order number → uses order_id, cannot locate the shard ❌

If you query directly by order_id, because the shard key is user_id, the system has no idea which shard this order is on. It can only broadcast to all shards for a parallel query (a full table scan across all databases and tables) and then aggregate the results. With large data volumes, this operation can crash the database in minutes.

3.2 Solution: The "Gene Method"

The "Gene Method" is the classic solution to this problem. The core idea is: When generating the order number, embed part of the user ID's information (the "gene") into it.

This way, when querying by order number, the system can parse the "gene" from the order number, deduce which user the order belongs to, and accurately calculate the shard location.

Implementation: Modify the Snowflake algorithm.

The standard Snowflake 64-bit ID structure is: sign bit(1) + timestamp(41) + machine ID(10) + sequence number(12).

We change it to: sign bit(1) + timestamp(41) + shard gene(12) + sequence number(10).

Code implementation:

public class OrderIdGenerator {
    // Gene occupies 12 bits, supporting 2^12=4096 shards
    private static final int GENE_BITS = 12;
    private static final long EPOCH = 1288834974657L;
    
    public static long generateId(long userId) {
        long timestamp = System.currentTimeMillis() - EPOCH;
        // Extract the last 12 bits of the user ID as the gene
        long gene = userId & ((1L << GENE_BITS) - 1);
        long sequence = getSequence(); // Get sequence number
        // Assembly: timestamp left-shifted 22 bits | gene left-shifted 10 bits | sequence number
        return (timestamp << 22) | (gene << 10) | sequence;
    }
    
    // Reverse-engineer the shard location from the order ID
    public static int getShardKey(long orderId) {
        return (int) ((orderId >> 10) & 0xFFF); // Extract the middle 12 bits
    }
}

Routing logic:

public class OrderShardingRouter {
    private static final int DB_COUNT = 4;
    private static final int TABLE_COUNT_PER_DB = 16;
    
    public static String route(long orderId) {
        int gene = OrderIdGenerator.getShardKey(orderId);
        int dbIndex = gene % DB_COUNT;
        int tableIndex = gene / DB_COUNT % TABLE_COUNT_PER_DB;
        return "order_db_" + dbIndex + ".t_order_" + tableIndex;
    }
}

Key breakthrough: Through gene embedding, orders from the same user always land on the same shard, while also supporting direct shard location via the order ID.

4. How Do Merchants Query Orders? (The Core Pain Point)

4.1 The Essence of the Problem

This is the most easily overlooked and most fatal problem in sharding design.

A common interview scenario:

Interviewer: "Your order table has 200 million records. How did you shard it?"

Candidate: "We used user_id for modulo sharding, splitting into 16 databases with 64 tables each."

Interviewer: "Then how does a merchant (Seller) query their own store's order list? The merchant doesn't have a user_id. With your sharding method, wouldn't a single merchant query have to scan all 1024 tables?"

This question cuts to the essence of sharding: Sharding is not just about 'splitting the data'; the difficulty always lies in 'how to aggregate after splitting'.

4.2 Solution 1: Data Heterogenization to Elasticsearch (Highly Recommended)

This is currently the standard practice in major tech companies.

Architecture diagram:

Order Write → MySQL (sharded by user_id) → Canal listens to binlog → Sync → Elasticsearch (indexed by seller_id)
                                                          ↓
Merchant Query → First query ES (by seller_id + various conditions) → Return order_id list → Then query MySQL for details

Core components:

  1. Canal: Alibaba's open-source MySQL binlog incremental subscription component. It disguises itself as a MySQL slave, receives the master's binlog in real-time, and parses data change events.
  2. Elasticsearch: A massive data full-text search engine. Index order data by seller_id to support arbitrarily complex combined queries.

ES index design:

{
  "order_index": {
    "mappings": {
      "properties": {
        "order_id": { "type": "keyword" },
        "seller_id": { "type": "keyword" },
        "buyer_id": { "type": "keyword" },
        "order_status": { "type": "integer" },
        "total_amount": { "type": "double" },
        "create_time": { "type": "date" },
        "sku_name": { "type": "text", "analyzer": "ik_max_word" }
      }
    }
  }
}

Advantages:

Disadvantages:

4.3 Solution 2: Heterogeneous Index Table (Lightweight Approach)

If you don't want to introduce ES, you can use a "space-for-time" approach.

Method: Create an order index table sharded by seller_id.

Write strategy: After a user places an order and writes to the primary database, asynchronously sync a copy of the data to the merchant index table.

⚠️ Consistency guarantee: Use RocketMQ transactional messages or the local transaction table + scheduled polling pattern:

Note: This table should only keep hot data (e.g., from the last 6 months). Cold data should be exported offline.

4.4 Solution 3: Offline Data Warehouse (For Report-style Queries)

If the merchant's queries are "statistical report" in nature (like monthly sales summaries, category proportions), these operations are extremely CPU-intensive. Never put them in the online transaction database.

Method: Sync order data to a columnar storage database (like ClickHouse) or an offline data warehouse (Hive).

Strategy:

4.5 Comparison of the Three Solutions

Solution Applicable Scenario Complexity Real-time Recommendation
ES Heterogenization Complex combined queries, fuzzy search Medium Millisecond delay ⭐⭐⭐⭐⭐
Heterogeneous Index Table Fixed-field queries, limited resources Low Near real-time ⭐⭐⭐
Offline Data Warehouse Report statistics, big data analysis High T+1 or minute-level ⭐⭐⭐⭐

5. Pitfall Avoidance Guide

5.1 No Deep Pagination

Whether in ES or MySQL, LIMIT 100000, 20 will cause a performance crash.

Solutions:

5.2 Distributed Transactions

After sharding, a previously single-database transaction can evolve into a distributed transaction.

Coping strategies:

@GlobalTransactional
public void processPayment(Order order) {
    // Debit buyer's balance
    accountService.debit(order.getBuyerId(), order.getAmount());
    // Credit seller's balance
    accountService.credit(order.getSellerId(), order.getAmount());
    // Update order status
    orderService.updateStatus(order.getId(), "PAID");
}

5.3 No Cross-Database JOINs

After sharding, cross-database JOINs are performance killers.

Before optimization (cross-database JOIN):

SELECT o.*, u.nick FROM t_order o JOIN t_user u ON o.buyer_id = u.id

After optimization (split into two queries, assemble at the application layer):

-- First: Query orders
SELECT * FROM t_order WHERE buyer_id = ?
-- Second: Batch query users based on user_id
SELECT * FROM t_user WHERE id IN (?, ?, ?)

6. Smooth Migration: From Single Database to Sharding

Migrating from a single database to a sharded architecture absolutely cannot involve downtime. The dual-write scheme is recommended.

Migration Steps

Phase 1: Dual Write

Phase 2: Data Verification

Phase 3: Gradual Traffic Cutover

Phase 4: Decommission the Old Database

7. Technology Selection Summary

Component Recommended Solution Description
Sharding Middleware Apache ShardingSphere Mature ecosystem, rich documentation, active community
Distributed ID Modified Snowflake Algorithm (with gene) High performance, trend-increasing, built-in routing info
Data Heterogenization Canal + Elasticsearch Standard in major companies, thorough decoupling
Distributed Transaction Seata Alibaba open-source, integrates well with ShardingSphere
Data Migration DataX Alibaba open-source, efficient batch migration

Final Words

Sharding is a complex architectural transformation. Here is a summary of the core points:

  1. Shard Key: Prioritize user_id, use hash modulo, and make the number of shards a power of 2.
  2. Order Number: Use the "gene method" to modify the Snowflake algorithm so the order number carries routing information.
  3. Merchant Queries: Never scan all tables! Use ES heterogenization or an index table solution for physical isolation.
  4. Transactions: Avoid distributed transactions by using the same shard key where possible; if unavoidable, use Seata.
  5. Migration: Dual-write + gradual traffic cutover for a seamless user experience.

A final reminder: Do not attempt to use IN queries on the sharded primary database to traverse all shards and aggregate data. With over 100 million records, once concurrency is slightly high, the database connection pool will be instantly overwhelmed, directly causing a service avalanche.

Let 'buyer real-time queries' go through the sharded primary database link, and 'merchant complex queries' go through the ES heterogenized data link—two types of storage, physically isolated, each performing its own role.

💡 If this article was helpful, please like, bookmark, and follow!

If you have any questions, feel free to leave a comment for discussion~