KlustronDB customizable data sharding solution
KlustronDB customizable data sharding solution
Database and schema
In KlustronDB, data is organized according to three levels: database, schema, and table.
A schema is a logical grouping of tables and other database objects, and can also be understood as a namespace. A database can contain any number of schemas. After a database is created, the default schema is public, and users can add more schemas as needed at any time.
Any number of tables can be created within each schema. For all database objects such as tables, indexes, stored procedures, triggers, views, materialized views, etc., if no schema is specified, the public schema is always used; to explicitly specify (reference) a schema, prefix the name of the schema in front of the database object, for example, schemaName.tableName. When connecting to any compute node of KlustronDB, a database must be specified; this connection can only access the schemas, tables, and other database objects within that database, and cannot access schemas, tables, or other database objects in other databases. System metadata can be accessed when connected to any database, but only the metadata of the current database can be viewed.
Schema is also called a namespace in PostgreSQL and KlustronDB, because it is a logical grouping of database objects, that is, a namespace. The table that stores schema metadata is called pg_namespace. The metadata table that stores database metadata is called pg_database.
In KlustronDB, each schema in every database corresponds to a MySQL database in every storage shard in the backend to store all its table shards. The naming correspondence rule is that a schema named mySchema in a database named myDB in a compute node corresponds to a database named
myDB_$$_mySchema
Any regular (non-temporary) table or table partition in myDB.mySchema in the compute node corresponds to a data table in the storage shard and is stored in a certain storage shard
myDB_$$_mySchema
In this database, every time a new schema is created, it will automatically create the corresponding database for this database and schema in all current storage shards.
If a user creates a non-partitioned table myDB.mySchema.t1 through a compute node, the compute node will automatically allocate a storage shard to store t1, and KlustronDB will in that storage shard
myDB_$$_mySchema
Create a table with the same name in the database
myDB_$$_mySchema.t1
To store the data of myDB.mySchema.t1, and KlustronDB will record in the metadata table of the compute node that the data of t1 is stored in SS1.
In this way, when reading and writing data, the computing node can automatically find the storage node where t1 is located, and then interact with the target storage node to read and write data of the myDB.mySchema.t1 table.
If a user creates a partitioned table t2, the user will ultimately also need to create several table partitions of t2 to store data. Partitioned tables can have multiple levels of partitions, and the multi-level partitions of a table can be viewed as a tree, with the table partitions storing data being the leaf nodes of the partition tree. For example, the user needs to create table partitions t20, t21, t22, etc., and these table partitions will also be automatically assigned to a certain backend storage shard. The handling of each leaf node is similar to that of a single table, with the compute nodes automatically assigning a storage shard to it to create a table with the same name.
All single tables or table partitions in all schemas of KlustronDB's databases are automatically and evenly distributed across all backend storage shards. If it is necessary to add a storage shard, KlustronDB will automatically move some table shards to the new storage shard to achieve an even distribution.
So, if a user needs to group data according to business logic, and the data between the groups also needs to be joined, then these groups should be several schemas within the same database. At the same time, users need to decide whether a table needs partitioning based on their business and data characteristics (for example, if the data volume is not large, such as within 100,000 rows or 100 MB, partitioning may not be necessary). If it is decided to partition a table, users need to choose which columns to use as partition columns and how to partition (there are three options: hash, range, or list). Moreover, after deciding on the partitioning method, each specific partition also needs to be created by a DBA.
Since data from different databases in KlustronDB cannot be associated within the same database connection and must be used completely independently, we do not recommend creating many databases in a single KlustronDB cluster. If tables need to be separated by business, you can create several schemas for each business and then create tables within them.
This way, you can operate on all the tables in all the schemas within the same database in a single database connection. Different databases use different KlustronDB clusters. The advantage of doing this is that computing/storage resources can be allocated by database, and data backup/recovery/rollback management can also be performed by database.
Sharding-related metadata in kunlun-server
The metadata table of each compute node in a KlustronDB cluster contains the metadata information of all storage shards in that cluster.
The pg_shard table stores information about each storage shard, including ID, name, data volume, number of shards, etc.; while the pg_shard_node table stores metadata information for each storage node of each storage shard, including its IP, port, username, and password, etc.
In the pg_class table, we added the relshardid field, which stores the ID of the storage shard where the corresponding table/index/sequence data table is located. This ID is the ID column in the pg_shard table.
The compute nodes use this information to find the storage shard and node where each table and other database objects are located, and then connect to these nodes and interact with them to complete data reading, writing, and transaction processing.
Example Explanation
The KlustronDB cluster we use consists of two storage shards, with the details as follows:
postgres=# select t1.name, t2.shard_id, t2.hostaddr ip, t2.port, t2.user_name, t2.passwd from pg_shard t1, pg_shard_node t2 where t2.shard_id=t1.id;
name | shard_id | ip | port | user_name | passwd
--------+----------+-----------+------+-----------+---------
shard1 | 3 | 127.0.0.1 | 4001 | pgx | pgx_pwd
shard2 | 4 | 127.0.0.1 | 4002 | pgx | pgx_pwd
In the Postgres database, there are two schemas: public and benchmarksql, so on the storage shards of shard1 and shard2 there are
postgres_$$_public
and
postgres_$$_benchmarksql
Two databases.
Connecting their main nodes separately, you can see:
mysql> select @@port;
+--------+
| @@port |
+--------+
| 4001 |
+--------+
1 row in set (0.00 sec)
mysql> show databases;
+--------------------------+
| Database |
+--------------------------+
| information_schema |
| mysql |
| performance_schema |
| postgres_$$_benchmarksql |
| postgres_$$_public |
| regression_$$_public |
| sys |
+--------------------------+
7 rows in set (0.00 sec)
mysql> select @@port;
+--------+
| @@port |
+--------+
| 4002 |
+--------+
1 row in set (0.00 sec)
mysql> show databases;
+--------------------------+
| Database |
+--------------------------+
| information_schema |
| mysql |
| performance_schema |
| postgres_$$_benchmarksql |
| postgres_$$_public |
| regression_$$_public |
| sys |
+--------------------------+
7 rows in set (0.02 sec)
Use the following query to obtain the schema each table belongs to (that is, the nspname column), which storage shard the table files are in (that is, relshardid), and other information. Then, check the corresponding tables of these tables or table shards on the two storage shards respectively.
select t1.nspname, t2.relname, t2.relshardid, t2.relkind from pg_namespace t1 join pg_class t2 on t1.oid=t2.relnamespace where t2.relshardid != 0 order by t1.nspname;
postgres=# select t1.nspname, t2.relname, t2.relshardid, t2.relkind from pg_namespace t1 join pg_class t2 on t1.oid=t2.relnamespace where t2.relshardid != 0 order by t1.nspname;
nspname | relname | relshardid | relkind
--------------+--------------------+------------+---------
benchmarksql | customer | 4 | r
benchmarksql | hist_id_seq | 3 | S
benchmarksql | history | 4 | r
benchmarksql | oorder | 3 | r
benchmarksql | new_order | 4 | r
benchmarksql | order_line | 3 | r
benchmarksql | stock | 4 | r
benchmarksql | item | 3 | r
benchmarksql | warehouse | 4 | r
benchmarksql | district | 3 | r
public | t101_pkey | 4 | i
public | t102 | 3 | r
public | t102_pkey | 3 | i
public | tt14t | 4 | r
public | scores1 | 4 | r
public | scores1_pkey | 4 | i
public | t1 | 4 | r
public | uv_iocu_tab_a_seq | 4 | S
public | uv_iocu_tab | 4 | r
public | uv_iocu_tab_pkey | 4 | i
public | warehouse2 | 4 | r
public | warehouse2_pkey | 4 | i
public | district2 | 3 | r
public | warehouse1 | 4 | r
public | warehouse1_pkey | 4 | i
public | district1 | 3 | r
public | district1_pkey | 3 | i
public | customer1 | 4 | r
public | customer1_pkey | 4 | i
public | history1 | 3 | r
public | orders1 | 4 | r
public | orders1_pkey | 4 | i
public | new_orders1 | 3 | r
......Due to the large number of lines, several lines are omitted here......
Connect to the storage shard master node on port 4001, and you can see the databases corresponding to each database and schema combination of the compute nodes, as well as the data tables located in shard1.
mysql> use postgres_$$_benchmarksql
Database changed
mysql> show tables;
+------------------------------------+
| Tables_in_postgres_$$_benchmarksql |
+------------------------------------+
| district |
| item |
| oorder |
| order_line |
+------------------------------------+
4 rows in set (0.01 sec)
mysql> use postgres_$$_public
Database changed
mysql> show tables;
+------------------------------+
| Tables_in_postgres_$$_public |
+------------------------------+
| district1 |
| district2 |
| history1 |
| history2 |
| new_orders1 |
| new_orders2 |
| scores |
| stock1 |
| stock2 |
| t100 |
| t102 |
+------------------------------+
11 rows in set (0.01 sec)
Connect to the storage shard master node on port 4002, and you can see all the databases corresponding to combinations of database and schema for computing nodes that are not used internally by the system, as well as the data tables located in shard2 within them.
mysql> use postgres_$$_benchmarksql
Database changed
mysql> show tables;
+------------------------------------+
| Tables_in_postgres_$$_benchmarksql |
+------------------------------------+
| customer |
| history |
| new_order |
| stock |
| warehouse |
+------------------------------------+
5 rows in set (0.02 sec)
mysql> use postgres_$$_public
Database changed
mysql> show tables;
+------------------------------+
| Tables_in_postgres_$$_public |
+------------------------------+
| customer1 |
| customer2 |
| item1 |
| item2 |
| order_line1 |
| order_line2 |
| orders1 |
| orders2 |
| scores1 |
| students |
| t1 |
| t101 |
| tt14t |
| uv_iocu_tab |
| warehouse1 |
| warehouse2 |
+------------------------------+
16 rows in set (0.02 sec)
Table storage mechanism
Each partition of a partitioned table is stored in a certain backend storage shard; each non-partitioned table is also stored in a certain backend storage shard. In the compute nodes, all the metadata of these tables is available to complete query optimization and execution, but the data itself is stored in the corresponding tables in the storage shards. For example:
t10 is a partitioned table, with its three partitions t100, t101, and t102 located in partitions with ids 3, 4, and 3 respectively:
postgres=# \d+ t10;
Table "public.t10"
Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
--------+---------+-----------+----------+----------------------------------+---------+--------------+-------------
a | integer | | not null | "nextval"('t10_a_seq'::regclass) | plain | |
b | integer | | | | plain | |
Partition key: HASH (a)
Indexes:
"t10_pkey" PRIMARY KEY, btree (a)
Partitions: t100 FOR VALUES WITH (modulus 3, remainder 0),
t101 FOR VALUES WITH (modulus 3, remainder 1),
t102 FOR VALUES WITH (modulus 3, remainder 2)
postgres=# select relname, relshardid from pg_class where relname like 't10_' and relkind='r';
relname | relshardid
---------+------------
t100 | 3
t101 | 4
t102 | 3
Querying t10 from the compute node shows all its data, which is located in its three partitions:
postgres=# select*from t10;
a | b
----+----
2 | 2
4 | 4
6 | 6
8 | 8
15 | 13
3 | 3
7 | 7
10 | 10
13 | 11
14 |
1 | 1
5 | 5
9 | 9
11 |
12 |
(15 rows)
Check the definition of the table corresponding to partition t10 in each storage shard in sequence and examine the data in each table shard. It can be seen that the union of the data in these table shards is exactly all the data rows obtained by querying t10 on the computing node.
mysql> show create table t100;
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
| Table | Create Table |
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
| t100 | CREATE TABLE `t100` (
`a` int(11) NOT NULL,
`b` int(11) DEFAULT NULL,
PRIMARY KEY (`a`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 |
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)
mysql> show create table t102;
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
| Table | Create Table |
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
| t102 | CREATE TABLE `t102` (
`a` int(11) NOT NULL,
`b` int(11) DEFAULT NULL,
PRIMARY KEY (`a`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 |
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)
mysql> select*from t100;
+----+------+
| a | b |
+----+------+
| 2 | 2 |
| 4 | 4 |
| 6 | 6 |
| 8 | 8 |
| 15 | 13 |
+----+------+
5 rows in set (0.00 sec)
mysql> select*from t102;
+----+------+
| a | b |
+----+------+
| 1 | 1 |
| 5 | 5 |
| 9 | 9 |
| 11 | NULL |
| 12 | NULL |
+----+------+
5 rows in set (0.00 sec)
mysql> show create table t101;
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
| Table | Create Table |
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
| t101 | CREATE TABLE `t101` (
`a` int(11) NOT NULL,
`b` int(11) DEFAULT NULL,
PRIMARY KEY (`a`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 |
+-------+------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)
mysql> select*from t101;
+----+------+
| a | b |
+----+------+
| 3 | 3 |
| 7 | 7 |
| 10 | 10 |
| 13 | 11 |
| 14 | NULL |
+----+------+
5 rows in set (0.00 sec)