ShardingSphere Sharding: Production Patterns, Pitfalls & Source-Code Deep Dive

Two Classic Incidents

Incident 1: An order table with 60 million rows; a fuzzy search by product name in the admin backend saturated the primary database CPU, causing site-wide order timeouts. The root cause: queries without a sharding key trigger full-database broadcast, turning a single large-table scan into a full-table scan multiplied by the number of databases.

Incident 2: After sharding, an admin export used LIMIT 1000000,10 for deep pagination. ShardingSphere pulled the first 1 million rows from each shard into memory for merging, causing an application OOM. Deep pagination is a classic sharding killer.

Common Root Cause: Sharding changes physical data distribution, but SQL writing habits remain unchanged. Middleware can route, rewrite, and merge, but it cannot redesign your query patterns — the sharding key is the anchor of the entire system; any SQL that departs from the anchor becomes a broadcast.

Technical Analysis: When to Shard, How to Shard, Tool Selection

2.1 When to Shard

InnoDB B+ trees with three levels hold roughly 20 million rows (row width 100–200 bytes). Beyond that, leaf splits become frequent, cache hit rates drop, and DDL turns into hour-long operations. Empirical thresholds:

Single table 5 million rows → consider index optimization.

20–50 million rows → consider sharding tables.

Single-instance write QPS hits ceiling (typically 5,000–10,000) → consider sharding databases.

Sharding databases solves write bottlenecks and connection limits; sharding tables solves data volume.

2.2 Vertical vs. Horizontal

Vertical database sharding: Split by business domain (order DB, product DB, user DB) — essentially the data boundary of microservices. Do this first.

Vertical table sharding: Move large fields (details, extension JSON) out; keep the main table narrow so the B+ tree stays shallow.

Horizontal database/table sharding: Distribute a single table across N databases and M tables by a sharding key (e.g., 2 databases × 8 tables = 16 shards). This is the main focus of the article.

2.3 ShardingSphere-JDBC vs. Proxy

JDBC (Sharding-JDBC): Embedded JAR; SQL routing happens inside the application. Zero extra deployment, best performance; Java-only, config upgrades require a new release.

Proxy (Sharding-Proxy): Standalone process masquerading as MySQL. Language-agnostic, centralized Ops control; adds one network hop, ~20%+ performance overhead. Suits polyglot environments or DBA-centric governance.

Sharding Key Selection Mnemonic: The most frequently queried equality-condition column + uniform hash distribution. Order scenario typically uses user_id for database sharding (keeps one user's data in one database) + order_id for table sharding. Warning: Once the sharding key is set, changing it is extremely hard — changing the sharding algorithm equals full data migration.

Core Process: A SQL's Five-Step Journey

Parse: SQL text → syntax tree; extract tables, columns, conditions, ordering, pagination.

Route: Compute target databases/tables per sharding strategy — exact routing (sharding value present) or full broadcast (no sharding value).

Rewrite: Logical table names replaced with physical names; pagination/aggregation SQL rewritten per "push-down" principle.

Execute: One real JDBC connection per routing unit, dispatched in parallel.

Merge: Multi-shard result sets sorted, aggregated, paginated in memory, combined into a single logical result set.

3.1 Production-Grade Configuration (2 Databases, 8 Tables)

spring:
  shardingsphere:
    datasource:
      names: ds0,ds1
      ds0: # Hikari pool config omitted
        jdbc-url: jdbc:mysql://db0:3306/order
        maximum-pool-size: 20
    rules:
      sharding:
        tables:
          t_order:
            actual-data-nodes: ds${0..1}.t_order_${0..3}
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: dbm
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: tbm
        sharding-algorithms:
          dbm:
            type: MOD
            props:
              sharding-count: 2
          tbm:
            type: MOD
            props:
              sharding-count: 4
        key-generate-strategy:
          column: order_id
          key-generator-name: snowflake

Division of labor: user_id shards databases to keep a single user's data in one database (user-dimension queries stay single-database); order_id shards tables to spread data within a database. Note: JDBC URL parameter is jdbc-url not url (Hikari); missing it causes NoSuchBeanDefinitionException.

3.2 Custom Sharding Algorithm

public class OrderDbAlgorithm implements
StandardShardingAlgorithm<Long> {

  // Exact query: precise routing to single database
  @Override
  public String doSharding(
      Collection<String> availableNames,
      PreciseShardingValue<Long> v) {
    return "ds" + (v.getValue() % availableNames.size());
  }

  // Range query (between/>/<): cannot route precisely
  // Must broadcast, return all candidate databases
  @Override
  public Collection<String> doSharding(&
      Collection<String> availableNames,
      RangeShardingValue<Long> v) {
    return availableNames;
  }
}

Core Source Code: Routing, Rewrite, Merge Engines

4.1 Routing Engine: From Logical SQL to RouteUnit

// ShardingStandardRoutingEngine
// Input: SQL syntax tree + sharding conditions
// Output: RouteUnit collection (database.table)

// 1. Database routing: shardingValue % databaseCount
//     where user_id=88 -> ds0
//     no user_id condition -> ds0,ds1 broadcast
// 2. Table routing: order_id % tableCount
//     order_id=1001 -> t_order_1
// 3. Combine into RouteUnit:
//     [ds0.t_order_1, ds0.t_order_5]
// Each RouteUnit = one real connection

Routing result count directly determines performance: exact routing = 1 connection, full broadcast = databaseCount × tableCount connections . Hence code standards must enforce "order queries must carry user_id or order_id " — every SQL missing the sharding key amplifies the attack surface.

4.2 Rewrite Engine: Table Replacement and Pagination Trap

// SQLRewriteEngine two rewrite types:
// 1. Logical table -> physical table
//     select * from t_order where...
//     -> select * from t_order_3 where...
// 2. Pagination rewrite (key): pagination must push down
//     limit 1000000, 10
//     -> limit 0, 1000010
//     Each shard pulls back 1,000,010 rows,
//     merge discards first 1,000,000 rows in memory
//     Deep pagination avalanche originates here

4.3 Merge Engine: K-Way Merge Trade-offs

// DQLMergeEngine dispatches by query shape:
// Single shard -> direct pass-through, zero merge overhead
// Queries with sharding key enjoy this
// Multi-shard -> StreamMergeEngine:
//   1. ORDER BY pushed down to each shard
//      each result set internally ordered
//   2. Priority queue for K-way merge
//      only compare heap tops each time
//   3. Aggregate function push-down rewrite:
//      sum -> partial sums then add
//      avg -> split into sum and count

Source-level insight: avg must not be pushed down directly — averaging each shard's average is mathematically wrong. ShardingSphere rewrites avg into sum and count pushed down, then recomputes at merge. Similarly max / min push down and take extreme values; sum / count merge by addition. Understanding this "push-down + secondary aggregation" model removes the merge black box.

4.4 Distributed ID: Snowflake and Clock Drift

// 64 bits = 1 sign + 41 timestamp
//         + 10 machine bits + 12 sequence bits
// Monotonic increasing -> index-friendly, shardable
// Clock drift is the only hard flaw:
synchronized (this) {
  long ts = System.currentTimeMillis();
  if (ts < lastTimestamp) {
    long delta = lastTimestamp - ts;
    if (delta <= 5) {
      // Drift within 5ms: wait a bit
      Thread.sleep(delta << 1);
    } else {
      // Large drift: refuse to generate
      throw new ClockException();
    }
  }
}

ShardingSphere bundles a Snowflake generator, but production often uses a dedicated ID service (Leaf segment mode / TinyId) because Snowflake has two hidden dangers:

Machine bits require manual planning (duplicate worker IDs produce duplicate IDs).

Clock drift needs a strategy (wait / reject / fallback bits). Segment mode centralizes ID issuance to a DB batch fetch, completely bypassing clock issues, at the cost of IDs lacking time semantics.

Three Advanced Topics: Pagination, Joins, Scaling

5.1 Correct Deep Pagination Pattern

-- Wrong: jump to page 100,000
SELECT * FROM t_order
ORDER BY order_id DESC
LIMIT 1000000, 10;

-- Correct: cursor method, always fetch next page only
-- Use last row's order_id as anchor
SELECT * FROM t_order
WHERE order_id < #{lastId}
ORDER BY order_id DESC
LIMIT 10;
-- With sharding key, executes on single shard, naturally efficient

5.2 Cross-Shard Joins and Binding Tables

# Binding tables: paired tables with identical sharding rules
binding-tables:
  - t_order,t_order_item
# Both sharded by user_id (database) and order_id (table)
# Join routes to same shards, executed locally,
# avoids Cartesian broadcast

# Broadcast tables: dictionary/config small tables
broadcast-tables:
  - t_dict
# Full copy in every database, writes sync to all

5.3 Scaling: Dual-Write Migration in Three Steps

// 2 databases 8 tables -> 4 databases 16 tables:
// Modulo scaling inevitably full reshuffle, no incremental
// 1. Create new shards + application dual-write
//     Old DB primary, new DB write failures only alert
// 2. Migrate existing data + verify
//     Recalculate destination per new algorithm, spot-check / full MD5
// 3. Gray-read cutover to new -> observe one week
//     -> stop old writes, decommission old DB
// Idempotency during dual-write guaranteed by unique keys
// (recalls idempotency article and distributed lock article)

Eight High-Frequency Pitfalls

Query missing sharding key. Full broadcast, connections = databases × tables, all in-memory aggregation. Code review must catch this.

Deep pagination with large OFFSET. Rewrite pulls full prefix per shard into memory. Use cursor method or business-enforced max page limit.

Cross-shard join without binding tables. t_order join t_order_item without binding becomes Cartesian broadcast. Must declare in rules.

Unique index lacks sharding key. Uniqueness enforced only per physical table; global uniqueness relies on distributed ID. Ordinary unique indexes become ineffective.

Cross-shard transaction running naked. One SQL broadcasts to multiple databases but uses local transactions; some commit, some fail. Use Seata AT/XA (see previous article) or idempotent compensation.

Sharding algorithm based on time. Monthly sharding appears uniform, but querying "entire last year" triggers full broadcast; old data is read-only, hotspot stays on newest table. Cold/hot separation must be separate.

Snowflake machine-bit collision. Manual workerId assignment; container IP drift causes duplicate IDs. Use ZooKeeper/DB for allocation, or switch to segment mode.

No scaling strategy before sharding. Pure modulo scaling forces full reshuffle. Start with power-of-two database/table counts + consistent hashing / gene method (embed sharding bits of user_id into order_id) so scaling can halve reshuffle.

Troubleshooting Three-Step Method

Step 1: Enable SQL show ( props sql-show=true), check logs for Logic SQL vs Actual SQL one-to-one mapping — Actual count equals routing width.

Step 2: Slow query analysis on Actual SQL: identify if a specific physical table is slow or all are slow, distinguishing index issues from broadcast issues.

Step 3: Merge overhead on application side: multi-shard result set size, whether sort fields have index push-down, decide to fix SQL or change sharding key.

Learning Path and Interview Questions

7.1 Four-Stage Roadmap

Beginner (1 week): Local 2-db 8-table CRUD; understand inline expressions, modulo sharding, sql-show logs.

Intermediate (2 weeks): Custom StandardShardingAlgorithm; verify binding/broadcast table behavior; compare logical vs physical execution plans with explain.

Advanced (1 month): Read routing/rewrite/merge engine source code; benchmark deep pagination OOM boundary; compare Snowflake, segment, UUID index performance.

Practice (ongoing): Design a real scaling drill (dual-write + verification + gray cut); establish "SQL must carry sharding key" standard with monitoring (broadcast SQL count alert).

7.2 Twelve Interview Questions

At what single-table data volume should you shard? What is the basis?

What problems do vertical database/table sharding and horizontal database/table sharding solve respectively?

How to choose between ShardingSphere-JDBC and Proxy?

What are the five stages a SQL goes through from entry to return?

How to pick a sharding key? Can database sharding key and table sharding key differ?

What happens when a query lacks a sharding key? How is connection count calculated?

Why does deep pagination cause OOM? What does the rewrite engine do?

How is multi-shard avg aggregation merged? Why can't you just average the averages?

What problem do binding tables solve? What data suits broadcast tables?

Describe Snowflake structure. How to handle clock drift?

How to migrate from 2-db 8-table to 4-db 16-table without downtime?

How to handle cross-shard transactions after sharding? How to integrate with Seata?

7.3 One-Page Cheat Sheet

Five-step chain: Parse -> Route -> Rewrite -> Execute -> Merge
Routing iron law: SQL must carry sharding key
Otherwise databases*tables full broadcast
Pagination iron law: cursor method replaces limit offset
Deep pagination = each shard pulls full prefix
Merge model: sort/aggregate push down to shards,
K-way priority queue secondary aggregation
ID two camps: Snowflake (clock drift must guard)
Segment (DB batch fetch)
Scaling three steps: Dual-write -> Migrate verify -> Gray cut

Summary: Sharding is not "just configure" but "system rebuild": sharding key is the anchor, SQL is the contract, routing/rewrite/merge three engines are the executors. Real engineering difficulty often lies not in middleware, but in SQL standards that enforce sharding keys, cursor pagination habits, and pre-designed scaling paths. Next series will open the microservices chapter.