Distribution key

In Hologres, the distribution_key table property specifies the data distribution strategy for a table. The system guarantees that records with the same distribution key value are allocated to the same shard. You can set this property when you create a table with the following syntax:

-- Syntax supported by Hologres V2.1 and later
CREATE TABLE <table_name> (...) WITH (distribution_key = '[<columnName>[,...]]');
-- Syntax supported by all versions
BEGIN;
CREATE TABLE <table_name> (...);
call set_table_property('<table_name>', 'distribution_key', '[<columnName>[,...]]');
COMMIT;

The following table describes the parameters.

A correctly configured distribution key offers the following benefits:

• Improved computing performance

  Computations can run in parallel across different shards, improving overall computing performance.

• Improved queries per second (QPS)

  When you use the distribution key as a filter condition, Hologres can scan only the relevant shards. Otherwise, Hologres must scan all shards, which reduces the QPS.

• Significantly improved join performance

  When two tables are in the same table group and their join columns are also their distribution keys, Hologres ensures that data with matching join keys is located on the same shard. This allows a local join, where each node joins its own data without moving it between nodes, dramatically improving execution efficiency.

Follow these principles when you set a distribution key:

• Choose columns with high cardinality and an even data distribution for the distribution key. Uneven data distribution causes data skew and an unbalanced workload, reducing query efficiency. To troubleshoot data skew, see Detect and handle workload skew.

• Choose columns that are frequently used in GROUP BY clauses as the distribution key.

• For join operations, set the join columns as the distribution keys to enable a local join and avoid data shuffle. The tables being joined must be in the same table group.

• We recommend setting a distribution key with no more than two columns. If you use a multi-column key, queries might cause data shuffle if they do not include filters on all key columns. Avoid using a composite distribution key with redundant columns, such as two columns that always have the same value. This can cause data skew if the underlying data also has low cardinality.

• You can set a single column or multiple columns as the distribution key. If you use a single column, do not add extra spaces in the command. If you use multiple columns, separate the column names with commas (,) and do not add extra spaces. The order of columns in a multi-column distribution key does not affect the data layout or query performance.

• If a table has a primary key (PK), the distribution key must be the PK or a subset of the PK columns. A distribution key cannot be empty, which means you must specify at least one column. This requirement ensures that all data for a single record belongs to only one shard. If you do not explicitly specify a distribution key, Hologres uses the PK as the default distribution key.

Setting a distribution key

When you set a distribution key for a table, Hologres allocates data to different shards based on the key. Hologres uses the Hash(distribution_key) % shard_count algorithm to determine the target shard for each record. The system guarantees that records with the same distribution key value are placed on the same shard. The following examples show how to set a distribution key.

• Syntax supported by Hologres V2.1 and later:

  -- Set column 'a' as the distribution key. The system hashes the values in column 'a' and then applies a modulo operation: hash(a) % shard_count = shard_id. Records with the same result are distributed to the same shard.
  CREATE TABLE tbl (
      a int NOT NULL,
      b text NOT NULL
  )
  WITH (
      distribution_key = 'a'
  );
  -- Set columns 'a' and 'b' as the distribution key. The system hashes the values from both columns and then applies a modulo operation: hash(a,b) % shard_count = shard_id. Records with the same result are distributed to the same shard.
  CREATE TABLE tbl (
      a int NOT NULL,
      b text NOT NULL
  )
  WITH (
      distribution_key = 'a,b'
  );

• Syntax supported by all versions:

  -- Set column 'a' as the distribution key. The system hashes the values in column 'a' and then applies a modulo operation: hash(a) % shard_count = shard_id. Records with the same result are distributed to the same shard.
  begin;
  create table tbl (
  a int not null,
  b text not null
  );
  call set_table_property('tbl', 'distribution_key', 'a');
  commit;
  -- Set columns 'a' and 'b' as the distribution key. The system hashes the values from both columns and then applies a modulo operation: hash(a,b) % shard_count = shard_id. Records with the same result are distributed to the same shard.
  begin;
  create table tbl (
    a int not null,
    b text not null
  );
  call set_table_property('tbl', 'distribution_key', 'a,b');
  commit;
                              

The following figure illustrates this data distribution. When you set a distribution key, choose columns that distribute data evenly. In Hologres, the number of shards is related to the number of worker nodes. For more information, see Basic concepts. If you choose a key that results in uneven data distribution, data concentrates in a few shards. This forces a small number of worker nodes to handle most of the computational work, which can cause the long tail effect and reduce query efficiency. For information about how to identify and resolve data skew, see Detect and handle workload skew.

Omitting the distribution key

If you do not set a distribution key, data is distributed randomly across the shards. Records with the same value might be placed on the same shard or on different shards. The following example shows how to create a table without a distribution key.

-- Do not set a distribution key.
begin;
create table tbl (
a int not null,
b text not null
);
commit;

The following figure illustrates this random data distribution.

Distribution key for GROUP BY aggregation

If you set a distribution key, records with the same key value are located on the same shard. For GROUP BY aggregation queries, the system redistributes data based on the grouping keys. By setting the distribution key to the columns frequently used in GROUP BY clauses, you ensure that the data is already co-located on each shard. This reduces data redistribution between shards and improves query performance. The following example shows how to do this.

• Syntax supported by Hologres V2.1 and later:

  CREATE TABLE agg_tbl (
      a int NOT NULL,
      b int NOT NULL
  )
  WITH (
      distribution_key = 'a'
  );
  -- Example query: aggregate on column a.
  select a,sum(b) from agg_tbl group by a;

• Syntax supported by all versions:

  begin;
  create table agg_tbl (
  a int not null,
  b int not null
  );
  call set_table_property('agg_tbl', 'distribution_key', 'a');
  commit;
  -- Example query: aggregate on column a.
  select a,sum(b) from agg_tbl group by a;

Running the EXPLAIN command shows an execution plan with no redistribution operator. Because the table's distribution key (a) matches the GROUP BY column, Hologres can perform aggregation directly within each shard, which avoids data redistribution.

Distribution key for multi-table joins

Multi-table joins are more complex. Follow these general principles:

• If all tables are joined on the same columns, set those join columns as the distribution key for each table.

• If the tables are joined on different columns, prioritize the join between the largest tables. Set the join columns of the largest tables as their distribution keys.

The following examples use a three-table join to illustrate these cases. You can apply the same logic to joins involving more than three tables.

• All three tables have the same join columns

  If all three tables are joined on the same columns, the scenario is straightforward. You can set the common join columns as the distribution key for all three tables to enable a local join.

  • Syntax supported by Hologres V2.1 and later:

    BEGIN;
    CREATE TABLE join_tbl1 (
        a int NOT NULL,
        b text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    CREATE TABLE join_tbl2 (
        a int NOT NULL,
        d text NOT NULL,
        e text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    CREATE TABLE join_tbl3 (
        a int NOT NULL,
        e text NOT NULL,
        f text NOT NULL,
        g text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    COMMIT;
    -- 3-table join query
    SELECT * FROM join_tbl1
    INNER JOIN join_tbl2 ON join_tbl2.a = join_tbl1.a
    INNER JOIN join_tbl3 ON join_tbl2.a = join_tbl3.a;

  • Syntax supported by all versions:

    begin;
    create table join_tbl1(
    a int not null,
    b text not null
    );
    call set_table_property('join_tbl1', 'distribution_key', 'a');
    create table join_tbl2(
    a int not null,
    d text not null,
    e text not null
    );
    call set_table_property('join_tbl2', 'distribution_key', 'a');
    create table join_tbl3(
    a int not null,
    e text not null,
    f text not null,
    g text not null
    );
    call set_table_property('join_tbl3', 'distribution_key', 'a');
    commit;
    --3-table join query
    SELECT * FROM join_tbl1
    INNER JOIN join_tbl2 ON join_tbl2.a = join_tbl1.a
    INNER JOIN join_tbl3 ON join_tbl2.a = join_tbl3.a;

  The execution plan (EXPLAIN SQL) shows the following:

  • There is no redistribution operator. This indicates that data was not reshuffled and a local join was performed.

  • The exchange operator moves data between execution stages or nodes. This process improves query efficiency by ensuring that only data from the relevant shard is processed.

  QUERY PLAN
  Gather  (cost=0.00..16.44 rows=10000 width=36)
    -> Hash Join  (cost=0.00..15.57 rows=10000 width=36)
          Hash Cond: ((join_tbl2.a = join_tbl3.a) AND (join_tbl1.a = join_tbl3.a))
          -> Hash Join  (cost=0.00..10.31 rows=10000 width=20)
                Hash Cond: (join_tbl2.a = join_tbl1.a)
                -> Exchange (Gather Exchange)  (cost=0.00..5.11 rows=10000 width=12)
                      -> Decode  (cost=0.00..5.11 rows=10000 width=12)
                            -> Seq Scan on join_tbl2  (cost=0.00..5.00 rows=10000 width=12)
                -> Hash  (cost=5.11..5.11 rows=10000 width=8)
                      -> Exchange (Gather Exchange)  (cost=0.00..5.11 rows=10000 width=8)
                            -> Decode  (cost=0.00..5.11 rows=10000 width=8)
                                  -> Seq Scan on join_tbl1  (cost=0.00..5.00 rows=10000 width=8)
          -> Hash  (cost=5.12..5.12 rows=10000 width=16)
                -> Exchange (Gather Exchange)  (cost=0.00..5.12 rows=10000 width=16)
                      -> Decode  (cost=0.00..5.12 rows=10000 width=16)
                            -> Seq Scan on join_tbl3  (cost=0.00..5.00 rows=10000 width=16)
  Optimizer: HQO version 1.3.0

• The three tables have different join columns

  In practice, the join columns in a multi-table join might be different. In this case, use the following principles to set the distribution key:

  • The core optimization principle is to prioritize the join between the largest tables. Set the join columns of the large tables as the distribution key. Small tables contain less data, so their distribution strategy is a lower priority.

  • If the tables have roughly the same amount of data, set the distribution key to the join columns that are most frequently used in GROUP BY clauses.

  In the following example, three tables are joined, and their join columns are not identical. The best strategy is to choose the join columns of the largest table as the distribution key. The join_tbl_1 table has 10 million rows, while join_tbl_2 and join_tbl_3 each have 1 million rows. Therefore, join_tbl_1 is the primary target for optimization.

  • Syntax supported by Hologres V2.1 and later:

    BEGIN;
    -- join_tbl_1 contains 10 million rows.
    CREATE TABLE join_tbl_1 (
        a int NOT NULL,
        b text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    -- join_tbl_2 contains 1 million rows.
    CREATE TABLE join_tbl_2 (
        a int NOT NULL,
        d text NOT NULL,
        e text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    -- join_tbl_3 contains 1 million rows.
    CREATE TABLE join_tbl_3 (
        a int NOT NULL,
        e text NOT NULL,
        f text NOT NULL,
        g text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    COMMIT;
    -- When join keys are different, choose the join key of the largest table as the distribution key.
    SELECT * FROM join_tbl_1
    INNER JOIN join_tbl_2 ON join_tbl_2.a = join_tbl_1.a
    INNER JOIN join_tbl_3 ON join_tbl_2.d = join_tbl_3.f;

  • Syntax supported by all versions:

    begin;
    --join_tbl_1 contains 10 million rows.
    create table join_tbl_1(
    a int not null,
    b text not null
    );
    call set_table_property('join_tbl_1', 'distribution_key', 'a');
    --join_tbl_2 contains 1 million rows.
    create table join_tbl_2(
    a int not null,
    d text not null,
    e text not null
    );
    call set_table_property('join_tbl_2', 'distribution_key', 'a');
    --join_tbl_3 contains 1 million rows.
    create table join_tbl_3(
    a int not null,
    e text not null,
    f text not null,
    g text not null
    );
    call set_table_property('join_tbl_3', 'distribution_key', 'a');
    commit;
    -- When join keys are different, choose the join key of the largest table as the distribution key.
    SELECT * FROM join_tbl_1
    INNER JOIN join_tbl_2 ON join_tbl_2.a = join_tbl_1.a
    INNER JOIN join_tbl_3 ON join_tbl_2.d = join_tbl_3.f;

  The execution plan (EXPLAIN SQL) shows the following:

  • A redistribution operator appears for the join between join_tbl_2 and join_tbl_3. Because join_tbl_3 is a smaller table and its join column does not match its distribution key, Hologres redistributes its data.

  • There is no redistribution operator for the join between join_tbl_1 and join_tbl_2. Because both tables use their common join column as the distribution key, this part of the join requires no data redistribution.

  QUERY PLAN
  Gather  (cost=0.00..183.90 rows=1000000 width=49)
    -> Hash Join  (cost=0.00..64.87 rows=1000000 width=49)
        Hash Cond: (join_tbl_2.d = join_tbl_3.f)
        -> Redistribution  (cost=0.00..40.22 rows=1000000 width=27)
            Hash Key: join_tbl_2.d
            -> Hash Join  (cost=0.00..35.99 rows=1000000 width=27)
                Hash Cond: (join_tbl_1.a = join_tbl_2.a)
                -> Exchange (Gather Exchange)  (cost=0.00..18.45 rows=10000000 width=11)
                    -> Decode  (cost=0.00..18.08 rows=10000000 width=11)
                        -> Seq Scan on join_tbl_1  (cost=0.00..7.75 rows=10000000 width=11)
                -> Hash  (cost=6.93..6.93 rows=1000000 width=16)
                    -> Exchange (Gather Exchange)  (cost=0.00..6.93 rows=1000000 width=16)
                        -> Decode  (cost=0.00..6.88 rows=1000000 width=16)
                            -> Seq Scan on join_tbl_2  (cost=0.00..5.29 rows=1000000 width=16)
        -> Hash  (cost=10.97..10.97 rows=1000000 width=22)
            -> Redistribution  (cost=0.00..10.97 rows=1000000 width=22)
                Hash Key: join_tbl_3.f
                -> Exchange (Gather Exchange)  (cost=0.00..7.53 rows=1000000 width=22)
                    -> Decode  (cost=0.00..7.45 rows=1000000 width=22)
                        -> Seq Scan on join_tbl_3  (cost=0.00..5.31 rows=1000000 width=22)
  Optimizer: HQO version 1.3.0