When One Database Is No Longer Enough: Sharding and Read/Write Splitting

October 31, 2026 · 28 min read

An overview of database and table sharding, online migration, dynamic scaling, global primary keys, and MySQL read/write splitting, source-replica replication, and replication lag handling.

If you have any questions, feel free to comment below. Click the block can copy the code.
And if you think it's helpful to you, just click on the ads which can support this site. Thanks!

Database and table sharding splits data originally concentrated in one database and one table across multiple tables and database servers, allowing them to share the storage and access load.

1. Why Shard Databases and Tables? #

1. What Problems Arise as Data Volume and Traffic Grow? #

When a business is just starting out, it has few users and little data, and a single MySQL server can handle most requests. As orders, users, and transaction records continue to grow, two types of pressure typically emerge:

• The data no longer fits, or maintenance becomes increasingly slow. Tables and indexes grow larger, disk usage keeps increasing, and backups, restores, index creation, and schema changes take longer.
• A single machine cannot handle the load. Large numbers of requests converge on the same database, making CPU, disk I/O, connection counts, or lock contention a bottleneck. Even after more application servers are added, they all still access the same database server.

For example, adding 100,000 records per day means 3 million additional records in a 30-day month; adding 500,000 per day means 15 million per month. In addition to storing these records, you need to store indexes and leave room for future growth and backups.

Table sharding and database sharding address different aspects of the problem. They can be used separately or together.

Table Sharding: Splitting Records from One Table Across Multiple Tables #

Suppose an orders table contains 30 million records. It can be split by user ID into multiple orders tables with identical schemas. When querying a user’s orders, the user ID identifies the corresponding table, so only a portion of the data needs to be accessed.

orders → orders_0, orders_1, orders_2, orders_3

After the split, each table’s data and indexes are smaller, and queries using the sharding key access a smaller range. If these tables remain in the same MySQL instance, they still share the same CPU, memory, and disks; the physical machine’s processing capacity has not increased.

A table with 2 million or 5 million rows can serve as an example for capacity planning, but reaching a certain row count does not automatically require sharding. Looking up one row by primary key has a very different processing cost from sorting or aggregating a large range of data. Check indexes and SQL first, then determine the scale of sharding based on data growth, query latency, and maintenance time.

Database Sharding: Distributing Data Across Different Database Instances #

If a single database server can no longer handle the total access load, different shards can be placed on different MySQL instances, allowing each to process requests and use its own disks.

Distinguish between a logical database and a database instance: creating multiple databases within the same MySQL process does not create additional CPU or disk capacity. Distributing them across different machines or instances with independent resources is what adds processing resources.

2. Database and Table Sharding Middleware #

Middleware^[1] primarily handles SQL parsing, shard routing, SQL rewriting, execution, and result merging. For example, when an application queries the logical table orders, the middleware calculates the actual database and table from user_id, then sends the request to the corresponding database.

Common deployment approaches fall into the client layer and the proxy layer.

JDBC sharding is embedded in the application and accesses databases directly; proxy sharding forwards requests through a standalone proxy

Figure: The sharding logic in the upper half runs within the application process; the proxy in the lower half is a separately deployed service. Both approaches require correctly configured actual data sources and routing rules.

3. Horizontal and Vertical Partitioning #

Horizontal Partitioning: Splitting by Row Without Changing the Schema #

Distribute different records from the same table across different tables, each retaining the same fields. For example, split orders using user_id % 2:

Even user_id → db_0.orders
Odd user_id  → db_1.orders

All of a user’s orders reside in the same shard, which a query containing the user ID can locate directly. If the query filters only by amount or status and does not include the sharding key, all shards may need to be queried.

Six order records are distributed across two database servers by even or odd user_id, with identical fields but different records

Figure: Horizontal partitioning changes which rows each shard stores. The two orders tables have identical schemas and together still contain all the original records; sharding should not cause the same record to be stored twice across two shards.

Vertical Partitioning: Splitting by Column While Retaining Linking Fields #

If a users table contains fields such as username, status, avatar, a long biography, and additional profile information, it can be split by access frequency into:

• user_basic: Stores frequently accessed fields such as user_id, username, and status.
• user_detail: Stores user_id and fields that are accessed less often or occupy more space, such as the biography.

Common queries read only basic information without loading large fields every time. Both tables retain the same user_id, allowing them to be joined when complete information is needed.

The users table is split by field into basic information and details, with all three users’ primary keys retained in both tables

Figure: Vertical partitioning does not divide users into two groups; it stores each user’s different fields separately. user_id is retained in both tables so they can be joined again.

At the business level, tables for different domains, such as users, orders, and payments, can also be placed in separate databases. This is vertical database partitioning along business boundaries. Once they are split across different instances, cross-domain queries and transactions need to be coordinated by the application or appropriate middleware.

Database and Table Routing #

The routing field should appear in common query conditions whenever possible. For example, if orders are mainly queried by user, consider user_id; if they are mainly queried by merchant, evaluate the merchant dimension. Large merchants or highly active users may create hotspots, so distributing by ID alone does not necessarily distribute requests evenly.

For larger datasets, you can plan 32 logical databases with 32 tables each, for a total of 1024 tables. At an estimated 5 million records per table, they can hold 5.12 billion records, roughly 5 billion in total; for many businesses, this scale can accommodate growth for a substantial period. The scaling section below expands on this scheme’s routing and deployment.

If data is split into 2 databases with 4 tables each, the following can be used for nonnegative integer user IDs:

db_index    = user_id % 2
table_index = (user_id / 2) % 4    # / denotes integer division

For user_id=13, the database index is 1 and the table index is 2, corresponding to db_1.orders_2. You cannot use user_id % 2 for both indexes, or only some database-table combinations will be reachable.

Range and Hash #

Time-based sharding suits workloads that query and archive data by month; user-based hash sharding suits workloads that locate data by user. For example, with id % 2, ID 2 is in shard 0; after changing to id % 4, it belongs to shard 2. Existing data must move accordingly; changing the configuration alone is not enough.

Range sharding routes by time period; changing the modulus from two to four requires some records to move

Figure: The modulo example shows only IDs 0–7. IDs 2 and 6 from shard 0, and IDs 3 and 7 from shard 1, must move to the new shards. Changing the rule without migrating the data means existing records will not be found at their new locations.

2. How Can You Migrate Smoothly to Sharded Databases and Tables? #

1. Migration with Downtime #

When a maintenance window is acceptable, stop writes first, then import the static data into the new databases and tables. The process is relatively straightforward:

1. Prepare the target databases, tables, and migration tools. Create shard tables, indexes, and routing rules, verify migration speed in a test environment, and estimate downtime.
2. Stop business writes. In addition to user-facing APIs, pause scheduled jobs, asynchronous consumers, and other write paths to ensure the old data stops changing during migration.
3. Export and import in batches. Read the old table by primary key range and write to the target databases and tables according to the sharding rules. Save progress so migration can resume from the last completed position after a failure.
4. Verify the data. Compare record counts, total amounts, and similar aggregates by range, then sample or compare key fields record by record; check indexes, primary keys, and relationships.
5. Switch connections and routing. Start the code that uses the new databases and tables, verify that queries, inserts, updates, and deletes work, and then resume business operations.

The larger the dataset and the slower the target writes, the longer the maintenance window. This method requires downtime and is not commonly used.

2. Dual-Write Migration #

If prolonged downtime is not acceptable, the business can continue using the old database while new changes are also written to the sharded databases, followed by backfilling historical data to complete a smooth migration.

Step 1: Deploy Dual Writes While Keeping the Old Database as the Source for Queries #

Add logic to write to the new databases for inserts, updates, and deletes that previously affected only the old database. Queries still read the old database at this stage; do not switch immediately.

For example, if an order is successfully written to the old database but the write to the new database fails, record the failure and retry it. The order and a pending synchronization task can be saved in the same local transaction in the old database, with a background process retrying the write to the new database. Deletes must also be synchronized; inserts alone are not enough.

Each synchronization operation carries a primary key and a version number. The version can be a monotonically increasing value maintained by the application; if gmt_modified is used, its precision, comparison rules, and update timestamps for the same record must be sufficient to distinguish ordering.

Step 2: Backfill Historical Data #

The migration tool scans the old table by primary key range and writes to the new databases in batches. Keep batch sizes within what the business can tolerate, save progress for each batch, and prevent backfilling from monopolizing database connections or I/O for extended periods.

Dual writes and backfilling update the new databases concurrently. For example, a backfill reads order version 9, but before it writes that version, the application has already dual-written version 10 to the new database. Version 9 must not overwrite version 10:

Record absent from the new database       → Insert
Backfill version > new database version   → Update
Backfill version <= new database version  → Keep the new database record

The comparison and write must be performed through a conditional database update or another atomic mechanism. The application must not read the version first and then overwrite unconditionally.

Deletes have the same problem: after the application deletes an order, backfilling may reinsert a previously read record. During migration, retain versioned deletion markers so an older backfill cannot “resurrect” the record; clean up the markers after backfilling and verification are complete.

Dual writes and historical backfilling run concurrently during migration; newer versions block older ones, and reads and writes switch only after verification

Figure: Dual writes synchronize new changes, while backfilling fills in historical records. Once version 10 reaches the new database, a late version 9 no longer overwrites it. Switch only after both paths converge and verification passes.

Step 3: Verify and Reconcile #

First confirm that failed dual-write tasks have been completed, then compare old and new data by primary key range. Check not only record counts but also primary key sets, key fields, and deletion status; equal counts do not mean the same records are stored.

When differences are found, reconcile against the old database, which remains the source of truth at this stage, and check again. For continuously changing data, combine version numbers, change progress, and brief write isolation to ensure the comparison reflects the same point in time.

Step 4: Switch Over and Retain a Rollback Window #

1. Route a small number of queries to the new databases first, comparing results, latency, and error rates.
2. Wait for historical backfilling to finish, incremental synchronization to catch up, and difference checks to pass.
3. Briefly pause writes or use a controlled cutover barrier, finish in-flight writes, and switch read/write routing together.
4. Observe the new databases before deciding to stop synchronizing the old database and reclaim the old data.

If the new databases are already accepting writes independently while the old database lacks those changes, traffic cannot simply be switched back. During the rollback window, continue synchronizing new changes to the old database, or reconcile the differences before rolling back.

3. How Do You Design Dynamic Scaling? #

1. Scaling Out with Downtime #

Scaling out with downtime resembles migration with downtime: stop writes, redistribute data from existing shards according to the new rules, then switch the configuration. With hundreds of millions of existing records, moving data, verifying it, and rebuilding indexes take considerable time and are difficult to complete within a short maintenance window.

Changing from id % 4 to id % 8 changes the destination of many records. Even if the new databases have already been created, records in the old shards will not automatically appear at their new locations; they still need to be migrated.

2. Define Logical Shards in Advance, Then Adjust Their Physical Locations #

Plan a larger number of logical databases and tables in advance and initially deploy them on a small number of database instances. Later, add machines by moving entire groups of logical shards rather than changing the ID modulo rule each time.

For 32 logical databases with 32 tables each, the total is:

32 × 32 = 1024 tables

For nonnegative integer order IDs, routing can be written as:

db_index    = order_id % 32
table_index = (order_id / 32) % 32    # Integer division

The logical database and table for a given ID remain unchanged; configuration then identifies the instance hosting that database:

order_id       → Logical database and table → Physical database instance
259            → db_3.orders_8 → MySQL-A
After scaling  → db_3.orders_8 → MySQL-C

Four logical shards move from two machines to four; the logical shard for each ID remains unchanged, while only the physical mapping changes

Figure: Only four logical shards are shown to make the mapping changes clear. Before scaling, A hosts S0 and S1, and B hosts S2 and S3; afterward, S1 moves to C and S3 moves to D. The identities and sharding rules of S1 and S3 remain unchanged; only the instance addresses hosting them change.

Capacity Estimation #

If each table is planned for 5 million records, 1024 tables provide a total of:

1024 × 5 million = 5.12 billion records

This estimates row counts. Disk planning must also account for row size, indexes, logs, and reserved space. For example, if each record including its indexes averages 1KB, 5.12 billion records occupy about 5.12TB in decimal units, before replicas and backups.

If the 32 logical databases are eventually deployed on 32 independent instances, each sustaining 1000 write requests per second, the total write capacity can be estimated at 32000 per second, assuming evenly distributed load and no application or network bottleneck. This multiplication does not apply if all 32 logical databases run in the same instance.

Scaling-Out Steps #

1. Select the logical shards to move. Use disk, CPU, I/O, and request distribution to select some databases and tables from heavily loaded instances.
2. Prepare new instances. Create databases, tables, indexes, permissions, and connection configurations, leaving room for migration and future growth.
3. Copy existing data and keep up with incremental changes. Use database migration tools or controlled backup and log synchronization to copy shard data to the new instances. Continue synchronizing changes generated during the move.
4. Catch up and verify. Check the data and complete pending synchronization. Briefly freeze writes to these shards during cutover so both locations do not become write entry points simultaneously.
5. Update the physical mapping. Change the data source addresses for the affected logical shards to the new instances, handle old connections and in-flight transactions, and then resume writes.
6. Observe before reclaiming old copies. Retain the data needed for rollback, and release the old locations after confirming that everything is correct.

This avoids recalculating each record’s logical shard, not moving the data. A fixed mapping of 32 databases also supports only database-level moves. To place different tables from one database on different instances, the middleware must support table-level data source mappings; changing a single database connection address is not enough.

3. Scaling In #

Scaling in reverses the process: move logical shards from lightly loaded instances to other instances, catch up on changes, verify the data, switch mappings, and finally retire the empty instances. Before consolidating, check that the target instances have enough storage and processing capacity.

When reducing only physical machines, logical shard identifiers can remain unchanged. If the number of logical shards is also reduced, such as changing the modulus from 32 to 16, data must be regrouped according to the new rule.

4. How Should Primary Key IDs Be Handled After Sharding? #

1. Database-Based Generation #

If each shard starts auto-incrementing from 1, different shards will generate id=1. If order IDs must be unique across all shards, use centralized allocation or a generation rule that prevents conflicts across nodes.

Auto-Increment in a Dedicated Database #

Create a table dedicated to generating IDs. When the application needs a new ID, first insert a record into this table, retrieve the auto-incremented ID, and then write to the actual orders shard.

The advantage is simplicity, with the database ensuring uniqueness; the drawback is that all generation requests pass through this database, making the generation service a shared dependency. This suits large datasets where the volume of new ID requests is not particularly high.

If accessing the database for every ID is too expensive, allocate a range at a time:

Generator node A receives: 10001–11000
Generator node B receives: 11001–12000

The database stores the upper bound already allocated. Transactional locking or conditional updates ensure that two allocations cannot receive overlapping ranges. Each node consumes its range locally and requests another when it runs out. A process crash may leave unused IDs, but an allocated range must not be allocated again.

Different Starting Points with a Fixed Increment #

For example, 8 independent ID generator nodes can each use a different starting point with an increment of 8:

Each node uses a different remainder, preventing duplicates. In a MySQL auto-increment scheme, configure auto_increment_increment and auto_increment_offset^[2] accordingly.

When scaling out, you cannot simply change every node’s increment from 8 to 16 and let new nodes start generating from small values, because the new sequences may collide with historical data. New starting points must be planned consistently, the old rules retired, or a new ID generation scheme adopted.

2. UUID #

UUIDs can be generated locally without accessing a central database each time. In Java, use:

String id = java.util.UUID.randomUUID().toString();

A UUID^[3] is 128 bits long and is usually represented as a 36-character string with hyphens. Removing the hyphens only reduces it to 32 characters; it does not turn it into a 64-bit integer. Using BINARY(16) reduces the overhead of string storage.

The main drawbacks of random UUIDs are that they are long, take up space, and have no increasing order. InnoDB’s clustered index^[4] uses a B+ tree, with leaf pages storing complete rows in primary key order:

• Increasing IDs: New records are typically inserted into the rightmost leaf page of the B+ tree, concentrating accesses and making it easier to reuse data pages in memory.
• Random UUIDs: New records may be inserted into leaf pages at different positions. If the target page is not in memory, it must first be read from disk before insertion. Insufficient page space may also trigger a page split, moving some records to a new page and updating the index structure.

Modified pages remain in memory first and are later flushed to disk in the background; inserting a record does not immediately write the entire node back to disk. Random UUIDs spread accesses and modifications across more data pages, increasing random I/O, page splits, and cache pressure. The larger each complete row, the fewer records fit in a page, making these costs more apparent.

In addition, InnoDB secondary index leaf records also store the primary key, so wider UUIDs increase secondary index space usage. High-write-volume orders tables therefore often use more compact integer IDs that generally increase over time. This discussion concerns random UUIDs; time-ordered UUIDs are different.

3. Using the Current Time #

Using System.currentTimeMillis() alone does not guarantee uniqueness: multiple threads or machines can obtain the same value within one millisecond, and a clock rollback may repeat an earlier timestamp.

Time can be combined with a node identifier and a sequence number, but the scheme must distinguish both “different nodes at the same time” and “multiple generations on the same node at the same time.” Snowflake^[5] uses this combination.

4. The Snowflake Algorithm #

How the 64 Bits Are Allocated #

Here, Java’s long, a signed 64-bit integer, is used; it can be stored as BIGINT in MySQL. Snowflake fixes the highest sign bit at 0, so generated IDs are nonnegative. Java has no long long type; in C/C++, int64_t explicitly represents a 64-bit integer.

The classic layout consists of:

Snowflake’s 64-bit layout: 1 sign bit, 41 time bits, 10 node bits, and 12 sequence bits

Figure: The colored segments are drawn strictly in proportion to their bit counts; the sign field is only one bit wide, with field names below. The time comes from the difference between the current time and the epoch, not from inserting a complete Unix timestamp in milliseconds.

A 41-bit time difference covers approximately 69.7 years. A 12-bit sequence ranges from 0 to 4095, allowing up to 4096 distinct sequence values per node per millisecond. Once exhausted, generation must wait until the next millisecond; the sequence cannot restart at 0 within the same millisecond.

ID = (time_difference << 22) | (node_id << 12) | sequence_within_millisecond

Node identifiers must be allocated uniquely. Two processes using the same node identifier and starting with the same sequence in the same millisecond will generate the same ID.

Generation Process #

1. Read the current time in milliseconds and compare it with the time of the previous generation.
2. If the clock has moved backward, stop this generation attempt and report an error; retry after the issue is handled or the clock recovers.
3. If still within the same millisecond, increment the sequence; when it reaches 4096, wait for the next millisecond.
4. If a new millisecond has begun, reset the sequence to 0.
5. Shift the time difference, node identifier, and sequence into their respective positions and combine them with bitwise OR to form the ID.

A node identifier should correspond to only one active generator instance. After a restart, a node must also avoid reusing sequence values within the millisecond of its last generation. This can be managed by persisting time progress or using mechanisms such as node leases and restart delays.

Snowflake IDs generally increase over time, which helps index writes, but different nodes can generate IDs within the same millisecond; they do not represent a strict business ordering across the entire system. Unique node identifiers, clock handling, and sequence management together prevent generation conflicts.

5. How Do You Implement MySQL Read/Write Splitting? #

1. Write to the Source, Read from Replicas #

Many workloads have more reads than writes. For example, after creating an order, users query order lists and details repeatedly. One source database can handle writes while replication synchronizes data to multiple replicas, which share the query load, implementing read/write splitting^[6].

• Send writes to the source: Inserts, updates, and deletes all go to the source.
• Send ordinary queries to replicas: Assign queries to a healthy replica, for example through round-robin or load-based selection.
• Send queries that must immediately see a recent write to the source: For example, checking an order’s status immediately after creation.

Read/write routing can use separate data sources in the application or be handled by JDBC middleware or a database proxy. Writes and related queries within a transaction should normally use the same source connection. A query must not be sent to a replica merely because it is a SELECT; for example, SELECT ... FOR UPDATE also needs to go to the source.

Routing sends application writes to the source and ordinary reads to replicas, while the source replicates data to two replicas

Figure: Each replica stores a copy of the source’s data to share read load; unlike shards, they do not each store only a portion of the data. Replication takes time, so ordinary replica queries may not immediately see changes just committed on the source.

Sharding and read/write splitting can be combined: first locate the logical shard using the sharding key, then choose that shard’s source or replica according to whether the operation is a read or a write. Adding replicas primarily scales reads; it does not directly increase the source’s write capacity.

2. The MySQL Source-Replica Replication Process #

Source-replica replication relies on the source’s binlog (binary log). Replicas first receive the change log, then apply the changes to their own tables:

1. The source records changes. The source records transaction changes in the binlog. This differs from InnoDB’s redo log for crash recovery; replication primarily reads the binlog.
2. The replica receives the log. The replica’s I/O receiver thread connects to the source and requests transactions starting from a specified log position or GTID. The source sends the log events.
3. Write to the relay log. The receiver thread writes received events to the replica’s local relay log.
4. Apply the log. The replica’s SQL applier thread or parallel workers read the relay log, execute its changes, and commit transactions on the replica. Only then can new queries read the data.

The binlog can record SQL statements or row changes. With row-based replication, replicas modify data according to row events in the log rather than simply reexecuting all the SQL originally sent by the application.

The source binlog is written to the replica relay log by the receiver thread, then applied to readable tables by the applier; semisynchronous acknowledgment follows relay log flushing

Figure: Receiving the log, flushing it to disk, and completing execution are separate stages. The green return path represents receipt acknowledgment in semisynchronous mode; the applier thread may not yet have executed the transaction.

Why Lag Occurs #

Source writes and replica execution do not complete simultaneously. Network transmission, slower replica disks, long transactions, lock waits, and time-consuming queries on replicas can all cause replicas to fall behind.

For example, the source may have committed a new order while the replica, despite receiving the log, is still processing earlier transactions. A query on the replica may therefore still fail to find the order. Lag may be only tens of milliseconds, or it may continue growing under load or during failures.

With default asynchronous replication, the source does not need to wait for replicas to receive the log before reporting a successful commit. If the source fails before the replica taking over has received the latest transactions, those changes may be missing after failover.

Semisynchronous Replication: First Confirm That at Least One Replica Has Received the Log #

Semisynchronous replication^[7] adds an acknowledgment wait to asynchronous replication:

1. The source writes to the binlog and sends the transaction log to replicas.
2. A replica receives the log, writes and flushes it to its local relay log, then returns an acknowledgment.
3. The source reports a successful commit to the client only after receiving acknowledgments from the required number of replicas; the default normally requires at least one.

This means a copy of the transaction log has been retained on a replica when the source reports success. However, the acknowledgment does not require the replica to have executed and committed the transaction, so semisynchronous replication does not directly guarantee that a write is immediately readable from a replica. If the acknowledgment wait times out, replication may fall back to asynchronous mode, so its actual operating state needs to be monitored.

Parallel Replication: Helping Replicas Apply the Log Faster #

When one applier thread executes transactions sequentially, a replica may not keep up with the source. Parallel replication uses a coordinator thread to distribute transactions that can run concurrently across multiple workers, increasing log application speed.

Early database-based parallelism assigned transactions from different databases to separate threads; if all writes were concentrated in one database, the improvement was limited. Modern MySQL can also schedule by transaction dependencies, so independent transactions within the same database may execute in parallel. Parallelism is not permanently limited to “multiple databases.”

Use replica_parallel_workers^[8] to configure the number of replica applier workers. Adding workers helps when the replica has spare resources and transactions can run in parallel; if many transactions update the same hot rows, lock contention still limits speed.

3. How to Handle Source-Replica Replication Lag #

Why Querying Immediately After a Write Can Cause Problems #

For example, code first creates an order, then queries it, and finally updates its status:

Insert an order on the source → Query the order on a replica → Order not found → Skip the subsequent status update

The order may already have been saved successfully; the replica simply has not applied the corresponding log yet. Treating “not found on the replica” as “the order does not exist” causes subsequent business processing to be skipped.

Checking Replication Status #

For MySQL 8.4 LTS, execute SHOW REPLICA STATUS^[9] on the replica:

SHOW REPLICA STATUS\G

Focus on the following fields:

Seconds_Behind_Source is measured in seconds, not milliseconds. A value of 0 does not prove that a particular just-committed transaction is readable, especially when the receiver thread itself is behind. For finer-grained checks, use transaction GTIDs or application heartbeat records.

Reducing Lag and Avoiding Incorrect Reads #

• Distribute writes through database sharding. Spread writes across multiple sources, each with its own replicas, to reduce pressure on each replication chain. The logical shard migration discussed earlier can be used to expand instances gradually.
• Increase replica application speed. Check parallel replication settings and address long transactions, lock waits, and disk I/O issues. Move heavy reporting queries to separate instances so they do not compete with replication appliers for resources.
• Remove unnecessary “query after insert” operations. If order creation has already returned the order ID and business fields, use those results directly instead of immediately querying a replica. Query only when database-computed results are actually needed.
• Route critical queries explicitly to the source. For workflows such as reading immediately after creation or checking a balance after a debit, explicitly route this small subset of queries to the source. Ordinary lists and historical records can still use replicas; there is no need to abandon read/write splitting entirely.
• When reading a replica is necessary, wait for the specified transaction to catch up. A system using GTIDs can obtain the GTID of the just-committed transaction, wait for it to execute on the selected replica, and then issue the query. Set a timeout; on timeout, read from the source or return a response asking the caller to retry later rather than waiting indefinitely. Do not reuse a previously established old snapshot on the replica to read new results.

Waiting a fixed few tens of milliseconds only reduces the probability of intermittent failures; once replication slows down, the data may still not be found. Workflows that genuinely require reading a just-written result should explicitly select the source or confirm that the target replica has executed the corresponding transaction.

The diagrams in this article were created with logic2image^[10].

Related readings

• 11 Multithreading and Locks
• 10 Annotations and Reflection
• 09 Exception Handling
• 08 Common Containers
• 07 Classes and Interfaces