For transactions that affect only a single worker node, atomicity, durability, and the required isolation levels are ensured by Postgres Pro as usual. However, to handle distributed transactions, pg_shardman relies on two-phase commit (2PC) protocol for atomicity and Clock-SI algorithm for transaction isolation.

To turn on 2PC support, use the postgres_fdw.use_twophase variable. With 2PC enabled, each transaction is prepared on all nodes before the commit. A successful PREPARE on the node means that this node is ready to commit the transaction, but it will only be committed if all the other nodes can commit this transaction as well. Otherwise, the transaction will be aborted. This approach allows to ensure transaction atomicity across multiple nodes.

This parameter can be changed at any time. Since the transaction behavior is determined by the postgres_fdw.use_twophase setting at commit time, some distributed transactions may use 2PC while others may not. Regardless of this setting, 2PC is never used for write transactions only affect a single node.

A well-known shortcoming of 2PC is that it is a blocking protocol: if the transaction coordinator has failed, some transactions might hang in the PREPARE state. pg_shardman provides shardman.recover_xacts() function to resolve them.

To ensure cluster-wide transaction isolation, use the following variables:

To ensure fault tolerance, pg_shardman allows creating shard replicas to store data with redundancy. For each sharded table, you can define the redundancy level — the number of replicas to store for each shard. Replicas are automatically distributed between the cluster nodes. If the node holding the primary shard fails, you can promote its most advanced replica to become primary.

To avoid performance degradation when using a large number replicas, all nodes are arranged into replication groups — subsets of nodes that can create replicas on each other. Primary shards and their replicas can only be located on different nodes that belong to the same replication group, as explained in Section F.45.3.3.2.

pg_shardman uses logical replication to synchronize data between the primary shards and their replicas. You can use synchronous or asynchronous replication types. The trade-off is well-known: synchronous replication is slower, but committed transactions are unlikely to be dropped. Asynchronous replication is faster, but allows replicas to lag behind the primary for arbitrary time, which might lead to a loss of some recently committed transactions, or WAL bloating if a replica fails.

Replication type is defined by the shardman.sync_replication variable. If set to off (default), this variable enables asynchronous replication. Otherwise, synchronous replication is used: although transactions on the primary will be committed locally right after the COMMIT request, the client will not get a transaction confirmation until it is committed on all replicas.

By default, cluster redundancy level is 0, so replicas are not created. You can change the redundancy level as follows:

To configure the nodes, you must modify the postgresql.conf file on all nodes as described below.

When all settings are configured, restart the nodes to preload the newly added shared libraries, and execute the following statement on each node:

CREATE EXTENSION pg_shardman CASCADE;

Postgres Pro Enterprise installs pg_shardman, together with pg_pathman and postgres_fdw extensions it depends on. The pg_shardman extension is installed into the shardman schema, which is currently fixed and cannot be changed. You now need to connect all worker nodes to the shardlord as explained in Section F.45.3.2.

Before you can start sharding tables, all worker nodes must be explicitly united into a single cluster with the shardlord node. For general recommendations on the optimal number nodes, see Section F.45.3.3.1.

To add a worker node to the cluster, run the following function on the shardlord:

shardman.add_node(super_conn_string text, conn_string text DEFAULT NULL, repl_group text DEFAULT 'default')

You can specify two types of connection strings. The super_conn_string is mandatory, as superuser privileges are required to configure logical replication between the nodes. The optional conn_string parameter is used for configuring FDWs; this allows to access the data without superuser rights. If you omit this parameter, pg_shardman uses super_conn_string for all purposes. The repl_group parameter defines the replication group to add the node to. If you omit this, the node is added to an automatically created new group. For details on replication groups, see Section F.45.3.3.2.

Each cluster node is assigned a unique ID and appears in Section F.45.5.3.1. Node IDs start with 1 and are incremented by one for each added node.

If the node previously belonged to another cluster managed by a different shardlord, its state will be reset. Once all nodes are added to the cluster, you can start sharding tables, as explained in Section F.45.4.1.

SELECT shardman.get_my_id();

To remove the node from the cluster, run the following function on the shardlord:

shardman.rm_node(rm_node_id int, force bool DEFAULT false)

where node_id is the ID of the node to be removed. By default, the node is not excluded from the cluster if it holds at least one primary shard. To overwrite this behavior, set the force parameter to true. In this case, the node is excluded from the cluster, and the most advanced replicas of the removed shards are promoted on other nodes to become primary.

The shardman.rm_node() function does not delete tables with data and foreign tables on the removed node. To delete a table with all the data, including shard replicas, use shardman.rm_table(rel_name regclass) . Caution: this operation cannot be undone and does not require any confirmation.

pg_shardman only allows sharding empty tables created on the shardlord, so a typical workflow for sharding a table is as follows:

Once a table is sharded and filled with data, you can execute DML statements on this table from any pg_shardman worker node. DML statements can access more than one shard, remote or local. It is implemented using the standard Postgres Pro inheritance mechanism: all partitions are derived from the parent table. If the required partition is located on another node, it will be accessed using postgres_fdw.

DDL support is very limited. Function shardman.alter_table(relation regclass, alter_clause text) alters the parent table on all nodes and updates its definition in cluster metadata; it can be used to add, remove, or rename columns, as well as create NOT NULL constraints. However, changing the column used as the sharding key is not supported. Foreign keys pointing to sharded tables are not supported either. Although you can create a UNIQUE constraint, it will only be enforced on per-partition basis.

If you create an index on the parent table before sharding, it will be included into the table definition and created on all shards and replicas automatically. There is no easy way to define an index on the already sharded table, since the index must be created separately for each partition, as described in pg_pathman wiki. To execute the same statement on all nodes, you can use the shardman.forall(sql text, use_2pc bool DEFAULT false, including_shardlord bool DEFAULT false) function.

To delete a table with all the data, including shard replicas, use shardman.rm_table(rel_name regclass) . Caution: this operation cannot be undone and does not require any confirmation.

Apart from sharded tables, an application may need to have a local table on one of the nodes and/or shared tables that store the same data on all cluster nodes.

Local tables store data that is unique for a particular node. For example, it can be some data in temporary tables. Local tables do not need to be handled by pg_shardman. You can create and use them locally at each node, just like in regular Postgres Pro clusters.

Shared tables store the same data on all cluster nodes. Such tables can be used for dictionaries that provide rarely updated data required for all queries. For each shared table, pg_shardman chooses the master node where this table will be stored and broadcasts the data to all other nodes using logical replication. If run on a different node, all updates of shared tables are automatically redirected to the corresponding master node.

To create a shared table, run the following function:

shardman.create_shared_table( rel regclass, master_node_id int DEFAULT 1)

The master_node_id parameter is a unique identifier assigned to the node when it is added to the cluster.

SELECT shardman.get_my_id();

When new nodes are added to the cluster, they are initially empty. To start using these nodes, you need to rebalance the data. Data rebalancing is performed in the background. To minimize the impact on the normal work of the cluster, shards and replicas are moved sequentially, one at time.

pg_shardman offers the following functions for rebalancing the specified primary shards and replicas, respectively:

shardman.rebalance(table_pattern text DEFAULT '%')

shardman.rebalance_replicas(table_pattern text DEFAULT '%')

All shards/replicas with the names conforming to the given pattern are evenly distributed between all nodes within each replication group. The naming pattern for primary shards and replicas is the same:

${sharded_table_name}_${shard_number}

For example, if you have sharded the films table, the following command will rebalance all its primary shards:

SELECT shardman.rebalance('films%')

Similarly, to rebalance all replicas, run:

SELECT shardman.rebalance_replicas('films%')

If you omit this argument, shards or replicas will be rebalanced for all sharded tables in the cluster.

To achieve a more fine-grained control over data distribution, you can move the specified primary shard or its replica to the exact node within the same replication group using the following functions:

shardman.mv_partition( part_name text, dst_node_id int)

shardman.mv_replica( part_name text, src_node_id int, dst_node_id int)

To move a primary shard, it is enough to specify the shard name and destination node, as pg_shardman knows the location of each primary shard. For replicas, you must also specify the source node as different nodes can hold replicas with the same name.

pg_shardman does not support automatic failure detection and recovery. If a node fails, it should be manually excluded from the cluster as described in the sections below.

To check for and resolve distributed deadlocks, you can use the shardman.monitor(). For details, see Section F.45.4.5.3.

To manage sharded cluster, pg_shardman provides regular Postgres Pro functions in the shardman schema. All the functions except for shardman.get_my_id() must be executed on the shardlord. pg_shardman functions return immediately, without waiting for the operation to complete, except for shardman.ensure_redundancy() , which is specifically tailored to wait for data synchronization.

pg_shardman provides several tables to store the sharded cluster metadata and monitor the state of the nodes.