Building a Custom Database Sharding Component with MyBatis: Migration Strategies & SQL Rewriting

Background: Large Table Governance

In large systems, high traffic and data volume lead to large tables. A table is considered large when it exceeds 10 million rows or 10 GB, shows performance degradation on CRUD operations, requires over one hour for DDL operations, or impacts core business flows. For example, an inventory flow table generates 34 million rows per day (31 GB). Large tables cause read/write performance drops, long service latency, slow DDL/DML, difficult backup/recovery, and inaccurate statistics.

Governance Solutions

Sharding (分库分表): Horizontal split by business dimension (warehouse ID, SKU, primary key).

Table Partitioning: Partition by time, hash, or range to improve query efficiency.

Cold/Hot Separation: Archive cold data after a time threshold.

Index Optimization: Remove redundant/inefficient indexes to reduce maintenance cost.

Read/Write Splitting: Primary handles writes, replicas handle reads.

The system already used sharding by warehouse ID at the database level, but the flow table remained unsharded within each database (40-50 million rows per table in 20 days). Table partitioning was not adopted; other strategies were applied. Horizontal sharding is the most direct and effective approach and is the focus of this article.

Component Selection: Why Build Custom?

Common sharding middleware includes Sharding-JDBC, Mycat, Cobar, TDDL, Atlas. All have SQL syntax limitations due to the complexity of SQL parsing and routing in sharding scenarios, and architectural differences. Examples:

Sharding-JDBC: No support for OR, UNION/UNION ALL, HAVING, subqueries, DISTINCT aggregation, SELECT LAST_INSERT_ID(), CASE WHEN, multi-row INSERT, dual table.

Mycat: No cross-database multi-table joins/subqueries, CREATE TABLE LIKE/SELECT, SELECT FOR UPDATE/LOCK IN SHARE MODE, cross-shard UPDATE/DELETE, multi-table UPDATE or updating shard key, cross-shard ORDER BY LIMIT.

Cobar: No cross-database JOIN, pagination, sorting, subqueries; SAVEPOINT ignored; SET statements ignored (except transaction/charset); INSERT must include shard column; UPDATE cannot modify shard key; limited SQL node support.

TDDL: No JOIN, multi-table queries, cross-database JOIN, stored procedures/functions.

Atlas: Long-unmaintained community, limited features, no cross-database JOIN.

Because of these limitations, the department previously built a custom sharding plugin for database-level sharding. Following the KISS principle (keep it simple, sufficient), they chose to build a custom table-sharding component as well.

Component Design

Implemented as a MyBatis plugin interceptor using Strategy Pattern + Chain of Responsibility . The plugin chain includes:

┌─────────────────────────────────────────────────────────────┐
│ MyBatis Plugin Chain │
├─────────────────────┬───────────────────────────────────────┤
│ SQLMarkingPlugin    │ SQLBasedDatabaseProtectorPlugin │
├─────────────────────┼───────────────────────────────────────┤
│ DataSourceRouting   │ TableShardingInterceptor (分表拦截器) │
│ Interceptor         │                                     │
└─────────────────────┴───────────────────────────────────────┘

Module Division

Interception Module: intercept() method intercepts SQL execution.

Strategy Module: Policy enum defines read/write strategies.

SQL Processing Module: Parses, modifies, and rewrites SQL.

Routing Calculation Module: Interacts with ShardingSuffixCalculator to compute table suffix.

Parameter Handling Module: Manages SQL parameter mappings.

Component Implementation

4.1 Sharding Strategy

Sharding key: warehouseNo (warehouse number).

Supports multiple read/write policies: single-write/single-read, single-write/double-read, double-write/single-read, double-write/double-read.

Shard count configurable, e.g., st_stock_stream:128 means 128 tables.

Shard Suffix Calculation Flow

// Shard suffix calculation flow
1. Extract warehouseNo from SQL parameters
2. Query config table for existing shard assignment
3. If exists, use it
4. If not, select a shard via load balancing algorithm
5. Use distributed lock to ensure assignment uniqueness

4.2 Read/Write Policies

Five policy modes (illustrated in diagrams):

doubleWriteAndReadOld (双写读旧表): Write to both old and new tables; read only from old table.

doubleWriteAndReadNew (双写读新表): Write to both; read only from new (sharded) table.

singleWriteAndReadOld (单写读旧表): Write only to new table; read only from old table.

singleWriteAndReadNew (单写读新表): Write only to new table; read only from new table.

singleWriteAndDoubleRead (单写双读): Write only to new table; read from both old and new, merge results.

These policies enable smooth migration from single-table to sharded architecture with rollback capability.

4.3 SQL Processing Mechanism

Steps:

Identify table name in SQL.

Determine if sharding processing is needed.

Extract sharding key ( warehouseNo).

Compute shard suffix.

Rewrite SQL, replace table name (e.g., st_stock_stream → st_stock_stream_5).

Handle parameter mapping.

Special SQL Handling

COUNT query: SELECT COUNT(*) FROM table →

SELECT SUM(countNumForSharding) FROM (SELECT COUNT(*) AS countNumForSharding FROM table_0 UNION ALL ...)

SUM query: SELECT SUM(column) FROM table →

SELECT SUM(sumNumForSharding) FROM (SELECT SUM(column) AS sumNumForSharding FROM table_0 UNION ALL ...)

ORDER BY/LIMIT: In double-read mode, require special handling (see 4.5).

4.4 Double-Write Handling

Core Function

Convert a single write SQL into a double-write SQL that inserts into both original and sharded tables.

Implementation Logic

Pre-check : Get original SQL and log. Check if table is in ignore list; if yes, return without double-write.

Sharding Processing : Check if sharding configured. Call changeSql to get sharded SQL. If SQL changed (needs sharding), build double-write SQL using DOUBLE_WRITE_JOINER ("; "). Update BoundSql via reflection (common in MyBatis plugin development). Duplicate parameter mappings for both statements.

Double-Write SQL Example : Original:

INSERT INTO st_stock_stream (id, warehouse_no, sku, quantity) VALUES (?, ?, ?, ?)

→ Double-write:

INSERT INTO st_stock_stream_5 (id, warehouse_no, sku, quantity) VALUES (?, ?, ?, ?) ; INSERT INTO st_stock_stream (id, warehouse_no, sku, quantity) VALUES (?, ?, ?, ?)

Key Points

SQL Joiner: Semicolon ("; ") as statement separator for multi-statement execution.

Parameter Mapping:

boundSql.getParameterMappings().addAll(boundSql.getParameterMappings())

duplicates parameters for both statements.

Reflection Update:

ReflectionUtils.setFieldValue(BoundSql.class, boundSql, "sql", sql)

updates the immutable BoundSql SQL field.

The changeStatementForDoubleWrite method is the key component for smooth migration, allowing transition from single-table to sharded architecture without service interruption.

4.5 Double-Read Handling

Used in double-read strategy: rewrite single-table query to query both original and sharded tables, merge results via UNION ALL.

Core Function

Convert single-table query SQL into double-read SQL that queries both tables and handles result merging.

Implementation Logic

Pre-check : Get original SQL and log. Check if table is in ignore list; if yes, return without double-read.

Sharding Processing & SQL Rewriting : Check if sharding configured. Call changeSql to get sharded SQL. If SQL changed, merge using mergeSqlForDoubleRead. Handle ORDER BY clause: remove table alias prefix to avoid "Unknown column" error after UNION ALL. Handle COUNT query: wrap with outer SUM. Handle SUM query: similar outer SUM. Update BoundSql via reflection and append parameter mappings.

Key Technical Points

SQL Merging Mechanism :

private static String mergeSqlForDoubleRead(String changedSql, String originalSql) {
    String newSql = StringUtils.removeEnd(originalSql, SEMICOLON);
    newSql = newSql.trim().replaceAll(RegExpConstants.SQL_ORDER_BY, StringUtils.SPACE);
    newSql = newSql.trim().replaceAll(RegExpConstants.SQL_LIMIT, StringUtils.SPACE);
    return appendSelectColumns(newSql, changedSql);
}

Steps: remove trailing semicolon, strip ORDER BY and LIMIT (re-added later), merge SELECT columns via appendSelectColumns.

ORDER BY Handling :

if (StringUtils.containsIgnoreCase(sql, ORDER_BY)) {
    int orderByIndex = StringUtils.lastIndexOfIgnoreCase(sql, ORDER_BY);
    String orderByAndLimitSql = StringUtils.substring(sql, orderByIndex);
    String sqlBeforeOrderBy = StringUtils.substring(sql, 0, orderByIndex);
    // Critical: remove table alias prefix to avoid "Unknown column" error
    orderByAndLimitSql = orderByAndLimitSql.replaceAll("\\w+\\.", StringUtils.EMPTY);
    sql = StringTools.getJointCodeWithSpace(sqlBeforeOrderBy, orderByAndLimitSql);
}

Problem: In UNION ALL, if ORDER BY contains table alias (e.g., m.update_time), the merged result set loses the alias, causing "SQL Unknown column" error. Solution: strip alias prefix.

COUNT Query Handling :

if (StringUtils.containsIgnoreCase(sql, COUNT_NUM_FOR_SHARDING)) {
    sql = "SELECT SUM( " + COUNT_NUM_FOR_SHARDING + " ) FROM (" + sql + ") ".concat(TEMP_TABLE_NAME);
}

Transformation example: Original: SELECT COUNT(*) FROM table → Double-read:

SELECT COUNT(*) AS countNumForSharding FROM table_0 UNION ALL SELECT COUNT(*) AS countNumForSharding FROM table

→ Final:

SELECT SUM(countNumForSharding) FROM (SELECT COUNT(*) AS countNumForSharding FROM table_0 UNION ALL SELECT COUNT(*) AS countNumForSharding FROM table) tempForSharding

SUM Query Handling : Similar to COUNT: each shard's SUM becomes a subquery, outer query sums them.

Parameter Mapping Handling :

private static void appendParameterMappings(BoundSql boundSql) {
    String sql = boundSql.getSql().trim();
    int limitIndex = StringUtils.lastIndexOfIgnoreCase(sql, LIMIT);
    if (limitIndex < 0) {
        // No LIMIT: duplicate all parameters
        boundSql.getParameterMappings().addAll(boundSql.getParameterMappings());
        return;
    }
    // Handle LIMIT parameters
    String limitOffset = StringUtils.substring(sql, limitIndex);
    int countMatches = StringUtils.countMatches(limitOffset, SQL_PLACEHOLDER);
    if (countMatches == 2) {
        // LIMIT ?, ? (offset, pageSize)
        List<ParameterMapping> originalParameterMappings = Lists.newArrayList(boundSql.getParameterMappings());
        // Remove last two parameters (LIMIT params)
        boundSql.getParameterMappings().remove(boundSql.getParameterMappings().size() - 1);
        boundSql.getParameterMappings().remove(boundSql.getParameterMappings().size() - 1);
        // Add full parameter list
        boundSql.getParameterMappings().addAll(originalParameterMappings);
    }
    // ... other cases
}

For double-read, parameters must be duplicated for both sides of UNION ALL. If LIMIT exists, its parameters are handled specially.

SQL Transformation Examples

Simple Query : Original:

SELECT * FROM st_stock_stream WHERE warehouse_no = ? AND deleted = 0

→ Double-read:

SELECT * FROM st_stock_stream_5 WHERE warehouse_no = ? AND deleted = 0 UNION ALL SELECT * FROM st_stock_stream WHERE warehouse_no = ? AND deleted = 0

Query with ORDER BY and LIMIT : Original:

SELECT * FROM st_stock_stream WHERE warehouse_no = ? ORDER BY update_time DESC LIMIT 10

→ Double-read:

SELECT * FROM st_stock_stream_5 WHERE warehouse_no = ? UNION ALL SELECT * FROM st_stock_stream WHERE warehouse_no = ? ORDER BY update_time DESC LIMIT 10

COUNT Query : Original: SELECT COUNT(*) FROM st_stock_stream WHERE warehouse_no = ? → Double-read:

SELECT SUM(countNumForSharding) FROM (SELECT COUNT(*) AS countNumForSharding FROM st_stock_stream_5 WHERE warehouse_no = ? UNION ALL SELECT COUNT(*) AS countNumForSharding FROM st_stock_stream WHERE warehouse_no = ?) tempForSharding

Double-read ensures complete data retrieval (old + new tables) during migration, supports zero-downtime migration, and provides rollback capability.

4.6 Routing Calculation

Implementation of ShardingSuffixCalculator (extensible). Algorithm:

Config First: Check if warehouse already has assigned shard.

Load Balancing: Choose shard with fewest assigned warehouses.

Distributed Lock: Ensure assignment uniqueness under concurrency.

Fault Tolerance: Retry mechanism on lock failure.

4.7 Parameter Handling

Double-write: parameter list duplicated.

Double-read: parameters adjusted based on LIMIT clause.

Maintain consistency between parameters and SQL placeholders.

4.8 Performance Optimization

Caching: ThreadLocal caches shard suffix.

Batch Processing: Support batch operations with sharding.

Index Optimization: Sharding key indexed for query efficiency.

Connection Pool: Each shard uses independent connection pool.

4.9 Extensibility

Custom Sharding Strategy: Implement ShardingSuffixCalculator interface.

Flexible Configuration: Sharding rules via mybatis-config.xml.

Strategy Pattern: Multiple read/write policies, dynamically switchable.

4.10 Monitoring & Logging

Metrics: SQL rewrite latency, routing calculation latency, shard hit rate, distributed lock contention.

Logs: Shard key extraction, SQL before/after rewrite, routing result, exceptions.

5. Development Issues & Solutions

5.1 Double-Read: ORDER BY Field Missing Error

Error:

ERROR 1250 (42000): Table 'm' from one of the SELECTs cannot be used in field list

Cause: In UNION ALL, ORDER BY m.update_time references alias m which disappears after union.

Solution: Include sort column in subquery SELECT list:

select 1, m.update_time from wms_stock.st_stock_stream m UNION ALL select 1, m.update_time from wms_stock.st_stock_stream_0 m order by m.update_time asc limit 1;

5.2 Double-Read: SELECT COUNT Error

Case:

select count(1) from wms_stock.st_stock_stream UNION ALL select count(1) from wms_stock.st_stock_stream_1

fails.

Solution: Wrap each COUNT in subquery, then outer SUM:

5.3 SUM() with UNION ALL Requires Outer SUM

Case:

select sum(stock_qty) from wms_stock.st_stock UNION ALL select sum(stock_qty) from wms_stock.st_stock_1

returns multiple rows.

Solution: Each SUM as subquery, outer SUM aggregates: