KlustronDB Horizontal Elastic Scaling (ScaleOut) Principles and Implementation

1. The Basic Core Capability of Distributed Databases - Horizontal Scaling (ScaleOut)

The evolution of database system architecture reflects the basic fact that the scale of data processing continuously increases with the ongoing development of the information society.

Nowadays, distributed database products have become an increasingly widely used technology choice in the storage services of information systems across various industries. The reason lies in their overwhelming advantage over standalone databases in terms of scalability and cost performance in massive data storage management.

Before we officially start introducing the scaling features of KlustronDB (Kunlun Database), let's first review the common patterns of database management system scaling.

Database expansion can generally be divided into the following two modes: vertical scaling (Vertically Scale Up) and horizontal scaling (Horizontally Scale Up).

- Vertically Scale Up

Vertical scaling refers to the expansion of physical resources, such as storage capacity, computing power, network bandwidth, and other resources.

This type of scaling is triggered by the bottleneck of a single machine's physical resources, with the solution being to expand the physical resources of the single machine to meet the database system's demand for physical resources. However, the drawbacks of this scaling method are also obvious: a single machine's physical resources always have an upper limit, and they are generally very expensive.

- Horizontally Scale Up

The basic idea of horizontal scaling is to distribute data across different physical devices according to certain rules, while the entire system still presents a logically single database service externally.

The system's method of expansion increases the number of physical devices to enhance the overall external service capability of the database. This expansion model can be said to achieve theoretically unlimited scalability.

Among them, if the extended physical node includes storage capability, it is called a share-nothing architecture; otherwise, it is called a share-storage architecture.

The Oracle RAC we are familiar with is a typical shared-storage architecture, where all Kluscomp instances share a single storage service, while KlustronDB (Kunlun Database) is a typical shared-nothing architecture.

If we evaluate the advantages and disadvantages of these two architectures from the perspective of read-write conflicts caused by physical device expansion, in a share-nothing architecture, data shards are stored in multiple storage shards, and there is no overlap between shards, so there is no issue of multi-point write conflicts. Therefore, linear scalability has an advantage over share-storage. However, core functions such as distributed transaction processing, distributed query processing, and automatic, application-transparent horizontal elastic scaling must be implemented under this architecture. Only when all these functions are available can it be considered a true distributed database system.

This is also why middleware for database sharding and application-level database sharding are no longer suitable for current technological requirements — these technical solutions lack the aforementioned features, which imposes a heavy burden and workload on application developers and DBAs. It is equivalent to having to implement distributed transaction processing and distributed query processing of a distributed database system case by case within application code, as well as requiring DBAs to manually take the service offline to complete scaling. These heavy burdens pose a significant risk to the stability and reliability of the application system and could negatively impact customer business operations and end-user experience.

When choosing the architecture, KlustronDB (Kunlun Database) fully took into account all of the above issues. Based on our extensive design and implementation experience with distributed database core functionalities such as distributed transaction processing, distributed query processing, and automatic horizontal scaling, as well as our deep understanding of related user needs and our ultimate pursuit of linear scalability, we have implemented a business-transparent horizontal elastic scaling mechanism based on a share-nothing architecture in KlustronDB (Kunlun Database) to better meet the service requirements of database systems under rapid business growth.

2. Introduction to KlustronDB (Kunlun Database) ScaleOut Feature

2.1 Horizontal Scaling of the Storage Layer

KlustronDB (Kunlun Database)'s storage service is logically composed of multiple independent MySQL clusters, each cluster being referred to as a shard.

Each shard is an independent disaster recovery service unit of the KlustronDB (Kunlun Database) storage layer, and data between different shards is mutually independent. Therefore, when scaling the storage layer, we only need to add this independent disaster recovery storage unit, the shard.

After completing the addition of a new shard to the cluster, another more important issue that needs to be addressed is how to migrate data from the existing shards to the new shard, thereby improving the overall service capability of the storage layer in terms of data read and write requests and data storage load.

KlustronDB (Kunlun Database) is based on the above business requirements and has implemented a lock-free data shuffle (migration) service. On the basis of effectively distributing hot data, it reliably ensures the continuity of storage services and is non-intrusive and transparent to application systems. This migration service can be used for both scaling up and scaling down. For example, temporary increases in computing and storage capacity may be required for a Double Eleven promotion, and after the promotion ends, the corresponding computing and storage resources are released and returned to the public cloud platform.

At the same time, this feature can also be used for hot and cold data separation. In a KlustronDB cluster, several special storage shards (cold shards) can be prepared for cold data (low-value data, infrequently used data, expired data, etc., such as bank account statements from three years ago, or user orders from an e-commerce system from one year ago). These shards use low-cost computer hardware and low-cost, high-capacity storage hardware (such as SATA disks). When the data expires, it can be moved from the hot shards to the cold shards. The data in these cold shards can still be read and written, but the performance is lower, and the storage cost is greatly reduced.

- lock-free shuffle service

When choosing which data to migrate, KlustronDB (Kunlun Database) selects the table shards that need to be moved based on certain rules (to be introduced in a later article), providing a set of tables to be migrated. The goal is to identify truly hot data, thereby achieving more balanced traffic distribution and more efficient distributed queries. At the same time, in the design of the data shuffle process, we have implemented a lock-free approach, ensuring that a table remains continuously readable and writable throughout the migration process, achieving zero business perception.

The entire shuffle process includes the following steps. Suppose the currently given shuffle-set (the database tables to be migrated) is being migrated from shard2 to a new shard.

Step One: The task that needs to be completed is the dump and load operations of the shuffle-set.

During the dump phase, a snapshot point will be retained, which serves as the starting position for subsequent binlog replication. The entire dump process will not block business requests. After the dump is completed, the dumped data files will be loaded into the new shard in parallel.

During the dump and load process, KlustronDB's (Kunlun Database) cluster management module cluster_mgr and node management module node_mgr work together to complete the data migration of tables in the shuffle-set.

As shown in the above figure, the entire process will be divided into three sub-tasks and assigned to the corresponding node_mgr:

- Dump_task

Cluster_mgr assigns a dump task to the standby machine of shard2, and the corresponding node_mgr uses the mydumper tool to start executing the dump task concurrently. After the task is completed, it responds to cluster_mgr that the task is finished.

- Transfer_task

After the first phase of the dump task is completed, cluster_mgr will issue a transfer_task command to node_mgr on the shard-new host, and shard-new will pull the corresponding dump files from shard2. After the download is completed, node_mgr on shard-new will respond to cluster_mgr that the task is completed.

- Load_task

After completing the transfer_task, cluster_mgr begins issuing load_task to shard_new. During the execution of the task, node_mgr uses the myloader tool to concurrently load data into shard_new, and after completion, responds to cluster_mgr that the task was successful.

Step 2: Data synchronization between the new shard and the source shard will be established, applying all data changes in the source shard from the snapshot point to the current time's shuffle-set onto the new shard.

The establishment of the data synchronization link uses MySQL's native binlog-based synchronization mechanism. By creating a temporary synchronization channel that only contains shuffle-set synchronized tables and using a dump snapshot as the starting point of synchronization, it pulls incremental data logs and replays these logs concurrently on shard-new until the overall synchronization delay is within the predetermined time range (default 3 seconds). After that, the table switch operation begins.

The table switching operation will rename the table on the source shard, cutting off new requests from the business. Subsequently, if the Kluscomp instance accesses the table, it will find that the table does not exist, and the Kluscomp instance will find the new location of the table from its own metadata, which will be updated by cluster_mgr in subsequent steps.

After completing the rename operation on shard2, the rename operation will be confirmed to have been replayed on shard-new, after which the entire data synchronization channel will be cut off. Assuming table A is in the shuffle-set, at this moment, shard-2 contains a table A-renamed, and at the same time, shard-new also contains an identical table A-renamed.

Step 3: Notify all Kluscomp instances of the route update for the shuffle-set table. The shuffle-set table officially goes into service.

When the data synchronization delay is within a reasonably small range, a rename operation will be performed on the table in the source shard, cutting off new business requests. After completing the rename operation, the entire data synchronization chain will be disconnected. Assuming table A is in the shuffle-set, at this moment, shard-2 contains a table A-renamed, and shard-new also contains an identical table A-renamed (the rename operation will also be synchronized from the source shard to the target shard).

2.2 Horizontal Scaling of the Computing Layer

KlustronDB (Kunlun Database) computation nodes are designed as stateless services, and none of the computation nodes locally persist any cluster-related data.

Therefore, in the design and implementation of computational capacity expansion, KlustronDB (Kunlun Database) has a natural advantage, which is the rapid deployment of Kluscomp instance services to the cluster. After a Kluscomp instance service is started, it will automatically synchronize relevant information from the metashard, including routing information, Klustore instance-related information, and so on.

Since all metadata processing is done in memory, the time window from when a Kluscomp instance scales up to when it starts providing external services is very short, allowing agile horizontal scaling of computing power.

2.3 Fault Tolerance and Rollback Mechanism

From the above expansion process, we can see that the entire procedure includes multiple subtasks and involves multiple physical devices, so robust fault tolerance and rollback design are essential requirements to ensure the high availability of the system.

Based on the above expansion process design, the handling methods for failures occurring at different stages are as follows:

- dump file failed

If dumping the file fails, you can retry (the default number of retries is 3, configurable).

- Transfer file failed

If the transfer of the dumped data file between physical devices fails, it can be retried (the default number of retries is 3, and it is configurable).

- Failed to load data

If the load operation fails, it can be retried (the default number of retries is 3, configurable).

- Table catch up failed

Table catch up is the process of establishing a data synchronization channel between the new shard and the source shard regarding the shuffle-set, and replaying the incremental data change logs. If this process fails, error information needs to be returned for analysis. Afterwards, the process is terminated, and the shuffle-set related data on the new instance is cleaned up.

- Table switch failed

After entering the table switch process, in principle, the table switch must succeed as long as no disasters such as network isolation or physical device power failure occur. Therefore, in terms of implementation, before performing the rename operation on the source shard, we forcibly terminate sessions holding locks to ensure that the rename does not fail due to lock contention.

Within 3 seconds after completing the rename operation on the source shard, if the rename replay has indeed not been completed on the new shard (for example, in the event of the aforementioned physical disaster), the process needs to be terminated, and the tables that were renamed and included in the shuffle-set on the source instance should be quickly restored to resume external service.

3. Planning and Outlook

With the continuous iteration and development of KlustronDB (Kunlun Database) versions, the construction of horizontal scaling capabilities will gradually become more refined and intelligent. Currently, the related feature plans cover aspects such as the selection of shuffle sets and the optimization and efficiency improvement of the shuffle process.

Specifically as follows:

- A more efficient shuffle-set construction algorithm

The construction of shuffle sets is, on one hand, to better expand hot shards and enhance the overall database service's capability to manage hot data. On the other hand, flexible and effective shuffle sets are also an important means to efficiently execute distributed query optimization. Therefore, in subsequent version iterations, this capability will also be a key focus for us.

- Optimization of the shuffle process

A fast and efficient shuffle helps improve the overall performance of the database. Future versions will focus on the efficiency of shuffle, such as attempting to implement a pipeline-style data shuffle strategy.

-END-