Elastic Scaling #

With rapid business growth, distributed database solutions will probably face the problem of data storage capacity again. Therefore, elastic scaling is an indispensable capability of distributed architecture.

Overview #

SphereEx-DBPlusEngine’s Elastic Scaling Plugin can achieve online expansion or contraction of homogeneous storage nodes and support data comparison between old and new environments.

Currently, the database products supported by SphereEx-DBPlusEngine Elastic Scaling plugin are MySQL, PostgreSQL, and openGauss, and the specific version, environment requirements, and permission requirements are as follows:

	Database	Version Supported	Environmental requirements	Permission requirements
1	MySQL	5.1.15 ~ 8.x	`my.cnf` Configuration log-bin=mysql-bin binlog-format=row binlog-row-image=full –The confirmation command is as follows show variables like ‘%log_bin%’; show variables like ‘%binlog%’;	GRANT REPLICATION SLAVE, REPLICATION CLIENT ON . TO ${username}@${host} –The confirmation command is as follows SHOW GRANTS FOR ‘user’;
2	PostgreSQL	9.4 Or higher	`postgresql.conf` Configuration: wal_level = logical max_wal_senders = 10 max_replication_slots = 10 max_connections = 600	`pg_hba.conf` Configuration: host replication repl_acct 0.0.0.0/0 md5
3	openGauss	2.0.1 ~ 3.0.0	`postgresql.conf` Configuration: wal_level = logical max_wal_senders = 10 max_replication_slots = 10 max_connections = 600	`pg_hba.conf` Configuration: host replication repl_acct 0.0.0.0/0 md5

During work execution, SphereEx-DBPlusEngine can reasonably schedule jobs according to the cluster resource situation to maximize the overall resource utilization and avoid resource contention, thus preserving job execution efficiency. It also supports automatic failover of tasks to provide necessary protection for long-running tasks.

Basic Concepts #

Node

Refers to an instance running a compute or storage layer process, which can be a physical machine, virtual machine, container, etc.

Cluster

Multiple nodes that are assembled together in order to provide a specific service.

Data Migration Job

The complete process of migrating data from one cluster to another.

Inventory Data

Data that already exists in a data node before the data migration operation begins.

Incremental Data

The new data generated by the business system during the execution of the data migration job.

Application Scenario #

Storage node expansion, to cope with the scenario of insufficient storage node capacity or connection number.
A scenario of storage node shrinkage, reducing storage nodes, and improving resource utilization.

Prerequisites for use #

SphereEx-DBPlusEngine and new database nodes have been deployed, the network is smooth and the service is normal.
You need to rely on Zookeeper for the expansion and shrinkage operation process and need to complete the deployment in advance.
Currently, SphereEx-DBPlusEngine data sharding architecture has been used, and the expansion and contraction nodes and the current storage nodes are the same structure and the same version of the database.
Since the storage node will keep two versions of sharded data during the expansion process, please make sure that the physical storage space is more than twice the current data file.
During the incremental data addition phase, please make sure that the database logs are not cleared.
Please make sure that the database logs are not cleared during the incremental data addition phase.

Usage Restrictions #

Support items #

Support tables with any index structure.

Unsupported items #

Does not support manipulating tables with no primary key and no unique key.
The target proxy is not supported to use the HINT sharding strategy;
Changes to the source table structure during the process are not supported.

Caution #

During the phase of stock and incremental data migration, data transfer can be performed online with no impact on business. The traffic switching phase, on the other hand, requires a certain window of time to ensure that the data at the source and target ends have caught up, and the phase requires a brief cessation of operations.

Principle Introduction #

If the sharding algorithm and usage method meet the conditions, it can achieve efficient expansion and contraction without moving data or moving only a small amount of data.

Similar to the principle of data migration plugin, SphereEx-DBPlusEngine’s Elastic Scaling plugin also goes through 4 phases: preparation phase, stock data migration phase, incremental data synchronization phase and traffic switching phase.

The difference is that elastic scaling is an in-place expansion, i.e., the phase before the traffic switch, and the original node will store two copies of data.

Preparation phase

In the preparation phase, the data migration module performs data source connectivity and permission verification, as well as stock data statistics and logging loci, and finally slices the tasks according to the data volume and user-set parallelism.

Stock data migration phase

Execute the stock data migration tasks split in the preparation phase. The stock migration phase adopts JDBC query to read data directly from the source side and write to the target side based on the configured slicing and other rules.

Incremental data synchronization phase

Since the time consumed for stock data migration is affected by data volume and parallelism, it is necessary to synchronize the data added to the business during this period. Different databases use different technical details, but in general, they are all based on replication protocols or WAL logs to achieve the change data capture function.

MySQL: subscribe and parse binlogs.
PostgreSQL: uses official logical replication test_decoding.

These captured incremental data are also written to the new data node by the data migration module. When incremental data is basically synchronized (the incremental data is continuous because the business system is not stopped), the traffic switching phase is entered.

Traffic switching phase

In this phase, there may be a certain time read-only window for the business, and the data in the source data node is made static for a short period of time to ensure that the incremental synchronization is fully completed by stopping the write function of SphereEx-DBPlusEngine or setting the database read-only or controlling the source write traffic.

The length of this read-only window period depends on whether the user needs to perform consistency checks on the data and the amount of data. Consistency verification is a standalone task that supports separate start/stop and breakpoint transfer.

After the confirmation is complete, the data migration is finished. Users can then switch the read traffic or write traffic to SphereEx-DBPlusEngine.

Usage guide #

Develop a sharding strategy

Before data migration, you need to make a reasonable sharding strategy with your business. Please refer to Data Sharding for related content.

Optimize data migration configuration (optional)

SphereEx-DBPlusEngine provides optimized configuration for data migration, which can control the threads of data extraction and data writing. With abundant resources, you can maximize the use of resources and complete migration operations efficiently. At the same time, the migration traffic can be limited to minimize the impact on the business system.

Data validation (optional)

To ensure the consistency of data, it can be confirmed by the data validation function, which is usually performed during the window time. Only one consistency verification job is allowed to run at the same time. Users can query, interrupt, restart and delete consistency verification.

Switching metadata

After confirming that the target environment has online conditions, the cluster traffic switch can be triggered to complete data migration.

Data movement description #

	Algorithm	Does the expansion move data
1	Range	No
2	List	Yes
3	Mod	Yes
4	Hash_mod	Yes

Operation Guide #

Please refer to Scaling DistSQL for specific operation commands.

Storage space confirmation

During the expansion and contraction process, there will be two versions of data in the database, so please make sure that the storage space of each node is more than twice of the current data file, so as to reserve space for two versions of data during the expansion and contraction.

Current shard confirmation

You need to confirm the information of the slice rule before the operation, you can use SHOW SHARDING TABLE NODES or PREVIEW command to verify.

Optimized data migration configuration (optional)

SphereEx-DBPlusEngine provides optimized configuration for data migration and control over the threads of data extraction and data writing. In case of abundant resources, you can maximize the use of resources and complete the job efficiently. At the same time, migration traffic can also be limited in order to minimize the impact on business systems.

Data validation (optional)

To ensure data consistency, this can be confirmed by the data validation function, a process that usually takes place during window times.

Switching traffic

After confirming that the target environment has online conditions, you can trigger the cluster traffic switch to complete the expansion and reduction of capacity.

MySQL User’s Manual #

Please refer to Scaling DistSQL for specific operation commands.

Environment Requirements #

Supported MySQL versions: 5.1.15 ~ 8.0.x.

Permission requirements #

Enables binlog

MySQL 5.7 my.cnf configuration example:

[mysqld]
server-id=1
log-bin=mysql-bin
binlog-format=row
binlog-row-image=full
max_connections=600

Execute the following command to confirm that binlog is enabled.

show variables like '%log_bin%';
show variables like '%binlog%';

If the following is displayed, then binlog is enabled

+-----------------------------------------+---------------------------------------+
| Variable_name                           | Value                                 |
+-----------------------------------------+---------------------------------------+
| log_bin                                 | ON                                    |
| binlog_format                           | ROW                                   |
| binlog_row_image                        | FULL                                  |
+-----------------------------------------+---------------------------------------+

Give the MySQL account Replication-related privileges.

Execute the following command to see if the user has migration privileges.

SHOW GRANTS FOR 'user';

Example results.

+------------------------------------------------------------------------------+
|Grants for ${username}@${host}                                                |
+------------------------------------------------------------------------------+
|GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ${username}@${host}     |
|.......                                                                       |
+------------------------------------------------------------------------------+

Complete Process Example #

Prerequisites #

Create the database in MySQL.

Example:

DROP DATABASE IF EXISTS resharding_ds_0;
CREATE DATABASE resharding_ds_0 DEFAULT CHARSET utf8;
    
DROP DATABASE IF EXISTS resharding_ds_1;
CREATE DATABASE resharding_ds_1 DEFAULT CHARSET utf8;
    
DROP DATABASE IF EXISTS resharding_ds_2;
CREATE DATABASE resharding_ds_2 DEFAULT CHARSET utf8;

Operation steps #

Create a new logical database in proxy and configure resources and rules, create tables and initialize some data.

CREATE DATABASE sharding_db;

USE sharding_db;

REGISTER STORAGE UNIT ds_0 (
    URL="jdbc:mysql://127.0.0.1:3306/resharding_ds_0?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="root",
    PASSWORD="root"
), ds_1 (
    URL="jdbc:mysql://127.0.0.1:3306/resharding_ds_1?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="root",
    PASSWORD="root"
);
    
CREATE SHARDING TABLE RULE t_order(
STORAGE_UNITS(ds_0,ds_1),
SHARDING_COLUMN=order_id,
TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="4")),
KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

CREATE TABLE t_order (order_id INT NOT NULL, user_id INT NOT NULL, status VARCHAR(45) NULL, PRIMARY KEY (order_id));

INSERT INTO t_order (order_id, user_id, status) VALUES (1,2,'ok'),(2,4,'ok'),(3,6,'ok'),(4,1,'ok'),(5,3,'ok'),(6,5,'ok');

Add a new data source to the Proxy.

Example:

REGISTER STORAGE UNIT ds_2 (
    URL="jdbc:mysql://127.0.0.1:3306/resharding_ds_2?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="root",
    PASSWORD="root"
);

Start the expansion and contraction.

Added ds_2 data source. Example:

    STORAGE_UNITS(ds_0, ds_1, ds_2),
    SHARDING_COLUMN=order_id,
    TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="6")),
    KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

View the list of expansion and contraction jobs.

SHOW RESHARDING LIST;

Example results:

+---------------------------------------+---------+----------------------+--------+---------------------+-----------+
| id                                    | tables  | sharding_total_count | active | create_time         | stop_time |
+---------------------------------------+---------+----------------------+--------+---------------------+-----------+
| j51017f973ac82cb1edea4f5238a258c25e89 | t_order | 2                    | true   | 2022-10-25 10:10:58 | NULL      |
+---------------------------------------+---------+----------------------+--------+---------------------+-----------+

View the expansion and contraction details.

SHOW RESHARDING STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

Example results:

+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| item | data_source | status                   | active | processed_records_count | inventory_finished_percentage | remaining_seconds | incremental_idle_seconds | error_message |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| 0    | ds_0        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 92                       |               |
| 1    | ds_1        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 92                       |               |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+

Execute Stop Write (optional).

This statement intercepts additions, deletions, and partial DistSQL and is optional. Some verification algorithms do not need to stop writing.

STOP RESHARDING SOURCE WRITING 'j51017f973ac82cb1edea4f5238a258c25e89';

Perform data consistency checks.

Example DistSQL

CHECK RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89' BY TYPE (NAME='CRC32_MATCH');

Data consistency checking algorithm type from：

SHOW RESHARDING CHECK ALGORITHMS;

For algorithm description, please refer to Consistency Check Algorithm List.

To query the progress of the data consistency check.

SHOW RESHARDING CHECK STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

Example results：

+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| tables  | result | finished_percentage | remaining_seconds | check_begin_time        | check_end_time          | duration_seconds | error_message |
+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| t_order | true   | 100                 | 0                 | 2022-10-25 10:13:33.220 | 2022-10-25 10:13:35.338 | 2                |               |
+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+

Switching metadata.

APPLY RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Preview the rules after they take effect:

PREVIEW SELECT * FROM t_order;

Example results:

+------------------+-----------------------------------------------------------------+
| data_source_name | actual_sql                                                      |
+------------------+-----------------------------------------------------------------+
| ds_0             | select * from v1_t_order_0 UNION ALL select * from v1_t_order_3 |
| ds_1             | select * from v1_t_order_1 UNION ALL select * from v1_t_order_4 |
| ds_2             | select * from v1_t_order_2 UNION ALL select * from v1_t_order_5 |
+------------------+-----------------------------------------------------------------+

Resume write-disable (optional).

If a write-stop was performed before, it needs to be restored.

RESTORE RESHARDING SOURCE WRITING 'j51017f973ac82cb1edea4f5238a258c25e89';

Complete the expansion and contraction operations.

Example:

COMMIT RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Scaling a Single Table to a Partitioned Table #

MySQL Preparation

DROP DATABASE IF EXISTS resharding_ds_0;
CREATE DATABASE resharding_ds_0 DEFAULT CHARSET utf8;
    
DROP DATABASE IF EXISTS resharding_ds_1;
CREATE DATABASE resharding_ds_1 DEFAULT CHARSET utf8;
    
DROP DATABASE IF EXISTS resharding_ds_2;
CREATE DATABASE resharding_ds_2 DEFAULT CHARSET utf8;

USE resharding_ds_0

CREATE TABLE t_order (
    order_id INT NOT NULL, 
    user_id INT NOT NULL, 
    status VARCHAR(45) NULL, 
    PRIMARY KEY (order_id));

INSERT INTO t_order (order_id, user_id, status) VALUES 
    (1,2,'ok'),
    (2,4,'ok'),
    (3,6,'ok'),
    (4,1,'ok'),
    (5,3,'ok'),
    (6,5,'ok');

Create logical libraries and add database sources

DROP DATABASE IF EXISTS sharding_db;

CREATE DATABASE sharding_db;

USE sharding_db;

REGISTER STORAGE UNIT ds_0 (
    URL="jdbc:mysql://192.168.xx.103:13306/resharding_ds_0?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="test",
    PASSWORD="Test@123"
), ds_1 (
    URL="jdbc:mysql://192.168.xx.104:13306/resharding_ds_1?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="test",
    PASSWORD="Test@123"
);

Create sharding rules


CREATE SHARDING TABLE RULE no_use(
STORAGE_UNITS(ds_0,ds_1),
SHARDING_COLUMN=order_id,
TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="4")),
KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

Add a new data source

REGISTER STORAGE UNIT ds_2 (
    URL="jdbc:mysql://192.168.xx.105:13306/resharding_ds_2?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="test",
    PASSWORD="Test@123"
);

Checklist t_order

mysql> show sharding table nodes t_order;
ERROR 20000 (44000): Can not find table rule with logic tables `[t_order]`.

mysql> preview select * from t_order;
+------------------+-----------------------+
| data_source_name | actual_sql            |
+------------------+-----------------------+
| ds_0             | select * from t_order |
+------------------+-----------------------+
1 row in set (0.21 sec)

Start the scaling task. Before starting the scaling task, you can adjust the threads based on the resource situation. Please refer to the Data Reshard job definition.

    STORAGE_UNITS(ds_0, ds_1, ds_2),
    SHARDING_COLUMN=order_id,
    TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="6")),
    KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

Check the status of the task and submit it when the task is completed

SHOW RESHARDING LIST;

SHOW RESHARDING STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

CHECK RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89' BY TYPE (NAME='CRC32_MATCH');

SHOW RESHARDING CHECK STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

APPLY RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

COMMIT RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Confirm the expansion results

mysql> SHOW SHARDING TABLE NODES t_order;
+---------+------------------------------------------------------------------------------------------------+
| name    | nodes                                                                                          |
+---------+------------------------------------------------------------------------------------------------+
| t_order | ds_0.t_order_0, ds_1.t_order_1, ds_2.t_order_2, ds_0.t_order_3, ds_1.t_order_4, ds_2.t_order_5 |
+---------+------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)

Automatic Scaling #

Prepare the corresponding data source in MySQL
Create logical libraries and rules in SphereEx-DBPLusEngine

CREATE DATABASE sharding_db;

USE sharding_db;

REGISTER STORAGE UNIT ds_0 (
    URL="jdbc:mysql://127.0.0.1:3306/resharding_ds_0?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="root",
    PASSWORD="root"
), ds_1 (
    URL="jdbc:mysql://127.0.0.1:3306/resharding_ds_1?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="root",
    PASSWORD="root"
);
    
CREATE SHARDING TABLE RULE t_order(
STORAGE_UNITS(ds_0,ds_1),
SHARDING_COLUMN=order_id,
TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="4")),
KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

CREATE TABLE t_order (order_id INT NOT NULL, user_id INT NOT NULL, status VARCHAR(45) NULL, PRIMARY KEY (order_id));

INSERT INTO t_order (order_id, user_id, status) VALUES (1,2,'ok'),(2,4,'ok'),(3,6,'ok'),(4,1,'ok'),(5,3,'ok'),(6,5,'ok');

Confirm the current shard

SHOW SHARDING TABLE NODES t_order;
+---------+----------------------------------------------------------------+
| name    | nodes                                                          |
+---------+----------------------------------------------------------------+
| t_order | ds_0.t_order_0, ds_1.t_order_1, ds_0.t_order_2, ds_1.t_order_3 |
+---------+----------------------------------------------------------------+

Add a new data source

REGISTER STORAGE UNIT ds_2 (
    URL="jdbc:mysql://127.0.0.1:3306/resharding_ds_2?useServerPrepStmts=true&serverTimezone=UTC&useSSL=false&characterEncoding=utf-8",
    USER="root",
    PASSWORD="root"
);

Start expansion

AUTO RESHARD TABLE t_order BY(
    STORAGE_UNITS(ds_0, ds_1, ds_2),
    SHARDING_COLUMN=order_id,
    TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="6")),
    KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

View task details

SHOW RESHARDING LIST;

SHOW RESHARDING STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| item | data_source | status                   | active | processed_records_count | inventory_finished_percentage | remaining_seconds | incremental_idle_seconds | error_message |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| 0    | ds_0        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 1                        |               |
| 1    | ds_1        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 1                        |               |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
2 rows in set (0.04 sec)

-- If the job cannot be queried, the job has been completed and you can go to the next step

Checking the effective status of the shard

SHOW SHARDING TABLE NODES t_order;
+---------+------------------------------------------------------------------------------------------------------------------+
| name    | nodes                                                                                                            |
+---------+------------------------------------------------------------------------------------------------------------------+
| t_order | ds_0.v1_t_order_0, ds_1.v1_t_order_1, ds_2.v1_t_order_2, ds_0.v1_t_order_3, ds_1.v1_t_order_4, ds_2.v1_t_order_5 |
+---------+------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)

At this point, the expansion is complete.

PostgreSQL User’s Manual #

Please refer to Scaling DistSQL for specific operation commands.

Environment Requirements #

Supported versions of PostgreSQL: 9.4 or above.

Permission requirements #

Enable test_decoding
Adjust the WAL configuration.

Example configuration of postgresql.conf :

wal_level = logical
max_wal_senders = 10
max_replication_slots = 10
max_connections = 600

Detail refers to Write Ahead Log and Replication

Configure PostgreSQL to allow the Proxy to have replication permissions.

Example configuration of pg_hba.conf.

host replication repl_acct 0.0.0.0/0 md5

For details, see The pg_hba.conf File

Complete Process Example #

Prerequisites #

The database is created in PostgreSQL.

Example:

DROP DATABASE IF EXISTS resharding_ds_0;
CREATE DATABASE resharding_ds_0;

DROP DATABASE IF EXISTS resharding_ds_1;
CREATE DATABASE resharding_ds_1;
    
DROP DATABASE IF EXISTS resharding_ds_2;
CREATE DATABASE resharding_ds_2;

Operation Steps #

Create a new logical database in proxy and configure resources and rules, create tables and initialize some data.

CREATE DATABASE sharding_db;

\c sharding_db;

REGISTER STORAGE UNIT ds_0 (
    URL="jdbc:postgresql://127.0.0.1:5432/resharding_ds_0",
    USER="root",
    PASSWORD="root"
), ds_1 (
    URL="jdbc:postgresql://127.0.0.1:5432/resharding_ds_1",
    USER="root",
    PASSWORD="root"
);
    
CREATE SHARDING TABLE RULE t_order(
STORAGE_UNITS(ds_0,ds_1),
SHARDING_COLUMN=order_id,
TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="4")),
KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

CREATE TABLE t_order (order_id INT NOT NULL, user_id INT NOT NULL, status VARCHAR(45) NULL, PRIMARY KEY (order_id));

INSERT INTO t_order (order_id, user_id, status) VALUES (1,2,'ok'),(2,4,'ok'),(3,6,'ok'),(4,1,'ok'),(5,3,'ok'),(6,5,'ok');

Add a new data source to the Proxy.

Example:

REGISTER STORAGE UNIT ds_2 (
    URL="jdbc:postgresql://127.0.0.1:5432/resharding_ds_2",
    USER="root",
    PASSWORD="root"
);

Start the scaling process. Before starting the scaling process, you can adjust the threads based on the resource situation. Please refer to the Data Reshard job definition.

A new data source, ds_2, has been added as an example:

RESHARD TABLE t_order BY(
    STORAGE_UNITS(ds_0, ds_1, ds_2),
    SHARDING_COLUMN=order_id,
    TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="6")),
    KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

View the list of expansion and contraction jobs.

SHOW RESHARDING LIST;

Example results.

+---------------------------------------+---------+----------------------+--------+---------------------+-----------+
| id                                    | tables  | sharding_total_count | active | create_time         | stop_time |
+---------------------------------------+---------+----------------------+--------+---------------------+-----------+
| j51017f973ac82cb1edea4f5238a258c25e89 | t_order | 2                    | true   | 2022-10-25 10:10:58 | NULL      |
+---------------------------------------+---------+----------------------+--------+---------------------+-----------+

View expansion and contraction details.

SHOW RESHARDING STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

Example results.

+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| item | data_source | status                   | active | processed_records_count | inventory_finished_percentage | remaining_seconds | incremental_idle_seconds | error_message |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| 0    | ds_0        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 92                       |               |
| 1    | ds_1        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 92                       |               |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+

Execute Stop Write (optional).

This statement intercepts additions, deletions, and partial DistSQL(optional).

STOP RESHARDING SOURCE WRITING 'j51017f973ac82cb1edea4f5238a258c25e89';

Perform data consistency checks.

CHECK RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Query data consistency verification progress.

SHOW RESHARDING CHECK STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

Example results:

+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| tables  | result | finished_percentage | remaining_seconds | check_begin_time        | check_end_time          | duration_seconds | error_message |
+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| t_order | true   | 100                 | 0                 | 2022-10-25 10:13:33.220 | 2022-10-25 10:13:35.338 | 2                |               |
+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+

Switching metadata.

APPLY RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Preview of the rules in effect:

PREVIEW SELECT * FROM t_order;

Example results:

+------------------+-----------------------------------------------------------------+
| data_source_name | actual_sql                                                      |
+------------------+-----------------------------------------------------------------+
| ds_0             | select * from v1_t_order_0 UNION ALL select * from v1_t_order_3 |
| ds_1             | select * from v1_t_order_1 UNION ALL select * from v1_t_order_4 |
| ds_2             | select * from v1_t_order_2 UNION ALL select * from v1_t_order_5 |
+------------------+-----------------------------------------------------------------+

Resume write-disable (optional)

If a write-stop was performed before, it needs to be restored.

RESTORE RESHARDING SOURCE WRITING 'j51017f973ac82cb1edea4f5238a258c25e89';

Complete the expansion and contraction operations.

Example:

COMMIT RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

openGauss User’s Manual #

Please refer to Scaling DistSQL for specific operation commands.

Environment Requirements #

Supported versions of openGauss: 2.0.1 ~ 3.0.0.

Permission requirements #

1. Adjust WAL configuration.

postgresql.conf Example configuration:

wal_level = logical
max_wal_senders = 10
max_replication_slots = 10
wal_sender_timeout = 0
max_connections = 600

See Write Ahead Log and Replication for details.

Configure PostgreSQL to allow the Proxy to have replication permissions.

Example configuration of pg_hba.conf：

host replication repl_acct 0.0.0.0/0 md5

For more information, please refer to Configuring Client Access Authentication and Example: Logic Replication Code .

Complete Process Example #

Prerequisites #

The database is created in openGauss.

Example:

DROP DATABASE IF EXISTS resharding_ds_0;
CREATE DATABASE resharding_ds_0;

DROP DATABASE IF EXISTS resharding_ds_1;
CREATE DATABASE resharding_ds_1;
    
DROP DATABASE IF EXISTS resharding_ds_2;
CREATE DATABASE resharding_ds_2;

Operation Steps #

Create a new logical database in proxy and configure resources and rules, create tables and initialize some data.

CREATE DATABASE sharding_db;

USE sharding_db;

REGISTER STORAGE UNIT ds_0 (
    URL="jdbc:opengauss://127.0.0.1:5432/resharding_ds_0",
    USER="root",
    PASSWORD="root"
), ds_1 (
    URL="jdbc:opengauss://127.0.0.1:5432/resharding_ds_1",
    USER="root",
    PASSWORD="root"
);
    
CREATE SHARDING TABLE RULE t_order(
STORAGE_UNITS(ds_0,ds_1),
SHARDING_COLUMN=order_id,
TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="4")),
KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

CREATE TABLE t_order (order_id INT NOT NULL, user_id INT NOT NULL, status VARCHAR(45) NULL, PRIMARY KEY (order_id));

INSERT INTO t_order (order_id, user_id, status) VALUES (1,2,'ok'),(2,4,'ok'),(3,6,'ok'),(4,1,'ok'),(5,3,'ok'),(6,5,'ok');

Add a new data source to the Proxy.

Example：

REGISTER STORAGE UNIT ds_2 (
    URL="jdbc:opengauss://127.0.0.1:5432/resharding_ds_2",
    USER="root",
    PASSWORD="root"
);

Start the expansion and contraction.

Added ds_2 data source, example:

RESHARD TABLE t_order BY(
    STORAGE_UNITS(ds_0, ds_1, ds_2),
    SHARDING_COLUMN=order_id,
    TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="6")),
    KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
);

View the list of expansion and contraction jobs.

SHOW RESHARDING LIST;

Example results:

+---------------------------------------+---------+----------------------+--------+---------------------+-----------+
| id                                    | tables  | sharding_total_count | active | create_time         | stop_time |
+---------------------------------------+---------+----------------------+--------+---------------------+-----------+
| j51017f973ac82cb1edea4f5238a258c25e89 | t_order | 2                    | true   | 2022-10-25 10:10:58 | NULL      |
+---------------------------------------+---------+----------------------+--------+---------------------+-----------+

View the expansion and contraction details.

SHOW RESHARDING STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

Example results:

+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| item | data_source | status                   | active | processed_records_count | inventory_finished_percentage | remaining_seconds | incremental_idle_seconds | error_message |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+
| 0    | ds_0        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 92                       |               |
| 1    | ds_1        | EXECUTE_INCREMENTAL_TASK | true   | 3                       | 100                           | 0                 | 92                       |               |
+------+-------------+--------------------------+--------+-------------------------+-------------------------------+-------------------+--------------------------+---------------+

Execute Stop Write (optional).

This statement intercepts additions, deletions, and partial DistSQL and is optional.

STOP RESHARDING SOURCE WRITING 'j51017f973ac82cb1edea4f5238a258c25e89';

Perform data consistency checks.

CHECK RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Query data consistency verification progress:

SHOW RESHARDING CHECK STATUS 'j51017f973ac82cb1edea4f5238a258c25e89';

Example results:

+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| tables  | result | finished_percentage | remaining_seconds | check_begin_time        | check_end_time          | duration_seconds | error_message |
+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| t_order | true   | 100                 | 0                 | 2022-10-25 10:13:33.220 | 2022-10-25 10:13:35.338 | 2                |               |
+---------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+

Switching metadata.

APPLY RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

Preview of the rules in effect.

PREVIEW SELECT * FROM t_order;

Example results:

+------------------+-----------------------------------------------------------------+
| data_source_name | actual_sql                                                      |
+------------------+-----------------------------------------------------------------+
| ds_0             | select * from v1_t_order_0 UNION ALL select * from v1_t_order_3 |
| ds_1             | select * from v1_t_order_1 UNION ALL select * from v1_t_order_4 |
| ds_2             | select * from v1_t_order_2 UNION ALL select * from v1_t_order_5 |
+------------------+-----------------------------------------------------------------+

Resume write-disable (optional).

If a write-stop was performed before, it needs to be restored.

RESTORE RESHARDING SOURCE WRITING 'j51017f973ac82cb1edea4f5238a258c25e89';

Complete the expansion and contraction operations.

Example:

COMMIT RESHARDING 'j51017f973ac82cb1edea4f5238a258c25e89';

FAQ #

In migration, can the shard key, shard algorithm and parameters, shard number, and unique value generation policy be modified?
Yes, you can modify them and move the corresponding data after modification.
If there are multiple tables that need to be expanded, can they be expanded in parallel?
Yes, they can be expanded in parallel.
Does the process of scaling up or down affect jobs when there is a master-slave switch in the database?

When data discovery capability is already configured, the scaling process can automatically adapt to changes in data sources, and a master-slave switch does not affect jobs.