Create and manage distributed tables

1. Choose a distribution column

The distribution column determines how data is distributed across different data nodes (DNs). The system routes data rows based on the hash value of the specified column.

• Selection principle: Typically, choose the primary key or a unique identifier of the table as the distribution column to ensure even data distribution.

• Important constraint: If the table has a primary key or unique constraint, the distribution column must be one of the columns that form these constraints.

Example: Convert a standard table t into a distributed table, using the id column as the distribution column.

1. Create a standard table t.

   CREATE TABLE t (id int primary key, data text);

2. Convert it to a distributed table, using the id column as the distribution column.

   SELECT create_distributed_table('t', 'id');

   The following result is returned:

    create_distributed_table 
   --------------------------
    
   (1 row)

2. (Optional) Specify the shard count

A shard is a physical storage unit of a distributed table. By default, each distributed table is created with 32 shards. You can also explicitly specify the shard count at creation time, or configure it globally using the polar_cluster.shard_count parameter.

Example: Create a distributed table with 4 shards.

1. Create a standard table t1.

   CREATE TABLE t1 (id int primary key, data text);

2. Explicitly specify the shard count.

   Using the shard_count parameter

   1. Specify the shard count using the shard_count parameter during conversion.

      SELECT create_distributed_table('t1', 'id', shard_count := 4);

      The following result is returned:

       create_distributed_table 
      --------------------------
       
      (1 row)

   2. Query the shard count.

      SELECT * FROM pg_dist_shard WHERE logicalrelid = 't1'::regclass;

      The following result is returned:

      logicalrelid | shardid | shardstorage | shardminvalue | shardmaxvalue 
      --------------+---------+--------------+---------------+---------------
       t1           |  102072 | t            | -2147483648   | -1073741825
       t1           |  102073 | t            | -1073741824   | -1
       t1           |  102074 | t            | 0             | 1073741823
       t1           |  102075 | t            | 1073741824    | 2147483647
      (4 rows)

   Using the polar_cluster.shard_count parameter

   1. Set the shard count globally using the polar_cluster.shard_count parameter.

       SET polar_cluster.shard_count TO 4;

   2. Convert to a distributed table.

      SELECT create_distributed_table('t1', 'id');

      The following result is returned:

       create_distributed_table 
      --------------------------
       
      (1 row)

   3. Query the shard count.

      SELECT * FROM pg_dist_shard WHERE logicalrelid = 't1'::regclass;

      The following result is returned:

      logicalrelid | shardid | shardstorage | shardminvalue | shardmaxvalue 
      --------------+---------+--------------+---------------+---------------
       t1           |  102072 | t            | -2147483648   | -1073741825
       t1           |  102073 | t            | -1073741824   | -1
       t1           |  102074 | t            | 0             | 1073741823
       t1           |  102075 | t            | 1073741824    | 2147483647
      (4 rows)

3. (Optional) Use co-location groups to optimize JOIN performance

In a business model, related information about an entity may be spread across multiple tables, typically requiring a JOIN query to retrieve the complete information. For example, the user_info table stores all user data, and user_order stores all user orders. The two tables need to be joined on user_id.

In a distributed database, if the data in these two tables is randomly distributed across different nodes, join queries will trigger cross-node data transfers, which incurs significant overhead. To address this issue,PolarDB for PostgreSQL Distributed Editionintroduces the concept of co-location groups.

• Purpose: Ensures that rows with the same distribution key value across multiple tables (for example, all related records where user_id is 1001) are always physically stored on the same data node. This way, JOIN operations based on the distribution key can be completed efficiently within a single node, achieving performance equivalent to a local query.

• Usage:PolarDB for PostgreSQL Distributed Editionprovides two approaches for managing co-location groups: implicit (default behavior) and explicit (recommended).

  • Default co-location (implicit behavior): When you create a distributed table without specifying the colocate_with parameter, the system automatically places the table into a default co-location group based on two characteristics: the distribution column type and the shard count. This means that all tables with the same distribution column type and shard count are considered co-located by default.

    Note

    Default co-location (implicit behavior): Even if two distributed tables have the same distribution column data type and shard count, the data in the two tables is not necessarily related.

  • Explicit control (recommended approach): The default behavior may incorrectly group unrelated tables together. To precisely control table co-location relationships, we recommend using explicit declarations:

    1. When creating the first table in a co-location group, set colocate_with:= 'none' in the create_distributed_table function. This creates a new, independent co-location group for the table.

    2. When creating subsequent tables that need to be co-located, set colocate_with:= 'first_table_name' to accurately add them to the existing co-location group.

Example: Place the user tables and order tables in the same co-location group, and the animal-related tables in a separate group.

1. Create user and order related tables and place them in the same co-location group.

   -- Create a new co-location group
   CREATE TABLE user_info (user_id int, user_data text);
   SELECT create_distributed_table('user_info', 'user_id', colocate_with := 'none');
   -- Join an existing co-location group
   CREATE TABLE user_order (user_id int, order_id int, order_data text);
   SELECT create_distributed_table('user_order', 'user_id', colocate_with := 'user_info');

2. Create animal-related tables and place them in another co-location group.

   -- Create a new co-location group
   CREATE TABLE animal (animal_id int, animal_data text);
   SELECT create_distributed_table('animal', 'animal_id', colocate_with := 'none');
   -- Join an existing co-location group
   CREATE TABLE animal_class (animal_id int, class_id int, class_data text);
   SELECT create_distributed_table('animal_class', 'animal_id', colocate_with := 'animal');

3. Verify that the two groups of distributed tables have different co-location group IDs, which means each group has been added to a separate co-location group.

   SELECT table_name, colocation_id, polar_cluster_table_type, distribution_column, shard_count
   FROM polar_cluster_tables
   WHERE table_name IN (
     'user_info'::regclass, 'user_order'::regclass,
     'animal'::regclass, 'animal_class'::regclass)
   ORDER BY colocation_id;

   The following result is returned:

     table_name  | colocation_id | polar_cluster_table_type | distribution_column | shard_count 
   --------------+---------------+--------------------------+---------------------+-------------
    user_info    |             3 | distributed              | user_id             |           4
    user_order   |             3 | distributed              | user_id             |           4
    animal       |             4 | distributed              | animal_id           |           4
    animal_class |             4 | distributed              | animal_id           |           4
   (4 rows)

Convert a distributed table back to a standard table

If distributed features are no longer needed, you can use the undistribute_table function to convert a distributed table or replicated table back to a single-node table. Data is automatically collected from all shards and moved to the primary CN node.

Example: Convert the distributed table t to a standard table.

SELECT undistribute_table('t');

The following result is returned:

NOTICE:  creating a new table for public.t
NOTICE:  moving the data of public.t
NOTICE:  dropping the old public.t
NOTICE:  renaming the new table to public.t
 undistribute_table
--------------------

(1 row)

Other DDL operations

For distributed tables that have already been created, you can perform other DDL operations just as you would with standard PostgreSQL tables. These operations are automatically propagated to all physical shards to maintain consistency between the logical and physical table structures.

• Drop a table: DROP TABLE table_name;

• Create an index: CREATE INDEX index_name ON table_name (column_name);