Hologres:Distribution key

In Hologres, the distribution_key property is used to specify the distribution policy of data in tables. After this property is configured, the entries with the same distribution key value are distributed to the same shard. When you create a table, you can use the following syntax to configure the distribution_key property:

-- Syntax supported in Hologres V2.1 and later
CREATE TABLE <table_name> (...) WITH (distribution_key = '[<columnName>[,...]]');

-- Syntax supported in all Hologres versions
BEGIN;
CREATE TABLE <table_name> (...);
call set_table_property('<table_name>', 'distribution_key', '[<columnName>[,...]]');
COMMIT;

The following table describes the parameters in the preceding syntax.

A distribution key is important for distributed computing. Appropriate settings of the distribution key can achieve the following effects:

• Improved computing performance

  Parallel computing can be performed on different shards. This improves computing performance.

• Improved queries per second (QPS)

  You can filter data based on the distribution key. In this case, Hologres scans data only in the shards that meet filter conditions. Otherwise, Hologres performs computing on all shards. This decreases the QPS.

• Improved join efficiency

  If Table A and Table B belong to the same table group and the fields used to join the tables are configured as the distribution keys, a local join operation can be performed because the entries with the same distribution key value in the two tables are distributed to the same shard. This greatly improves execution efficiency.

Take note of the following rules when you specify a distribution key:

• Specify the columns in which data is evenly distributed to constitute a distribution key. Otherwise, data skew issues may occur due to uneven allocation of computing resources. This reduces query efficiency. For more information about how to troubleshoot data skew issues, see Query the shard allocation among workers.

• Specify the columns on which the GROUP BY operation is frequently performed to constitute a distribution key.

• In scenarios where you want to create and join tables, specify the join columns as the distribution keys of the tables. This way, the system can implement a local join, which helps prevent data shuffling. The tables that you want to join must belong to the same table group.

• We recommend that you specify no more than two columns to constitute a distribution key for a table. If you specify more than two columns to constitute a distribution key and any of the distribution key columns is not hit during queries, data shuffling may occur.

After you specify a distribution key for a table, data in the table is distributed to shards based on the distribution key by using the Hash(distribution_key)%shard_count algorithm. The returned result is the shard ID. Entries with the same distribution key value are distributed to the same shard. Examples:

• Syntax supported in Hologres V2.1 and later:

  -- Specify Column a as a distribution key. Hologres performs a hash operation on the values in Column a and performs a modulo operation on the results. The returned result is the shard ID. Entries with the same distribution key value are distributed to the same shard.
  CREATE TABLE tbl (
      a int NOT NULL,
      b text NOT NULL
  )
  WITH (
      distribution_key = 'a'
  );
  
  -- Specify Column a and Column b to constitute a distribution key. Hologres performs a hash operation on the values in Column a and Column b and performs a modulo operation on the results. The returned result is the shard ID. Entries with the same distribution key value are distributed to the same shard.
  CREATE TABLE tbl (
      a int NOT NULL,
      b text NOT NULL
  )
  WITH (
      distribution_key = 'a,b'
  );

• Syntax supported in all Hologres versions:

  -- Specify Column a as a distribution key. Hologres performs a hash operation on the values in Column a and performs a modulo operation on the hash results. The returned result is the shard ID. Entries with the same distribution key value 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;
  
  -- Specify Columns a and b to constitute a distribution key. Hologres performs a hash operation on the values in Columns a and b and performs a modulo operation on the hash results. The returned result is the shard ID. Entries with the same distribution key value 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 shows the data distribution.If you specify a distribution key, make sure that the data of the distribution key columns is evenly distributed. The shard count in Hologres is proportional to the number of worker nodes. For more information, see Basic concepts. If you specify columns in which data is unevenly distributed to constitute a distribution key, data is distributed to some shards and most computing resources are consumed by some worker nodes. As a result, long tails occur. This reduces query efficiency. For more information about how to troubleshoot data skew, see Query the shard allocation among workers.

If you do not specify a distribution key, data is randomly distributed to shards. The entries with the same value may be distributed to different shards. Example:

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

The following figure shows the data distribution if no distribution key is specified.

After you specify a distribution key for a table, the entries with the same distribution key value are distributed to the same shard. If the GROUP BY operation is performed, the data is redistributed based on the specified distribution key during computing. You can specify the columns on which the GROUP BY operation is frequently performed to constitute a distribution key. In this case, the same data in the same shard is aggregated. This reduces data redistribution among the shards and improves query performance. Example:

• Syntax supported in Hologres V2.1 and later:

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

• Syntax supported in all Hologres versions:

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

You can query the execution plan by executing the Explain statement. The following figure shows that the execution plan does not contain the redistribution operator. This indicates that no data is redistributed.

Take note of the following rules when you join multiple tables:

• If the join columns of the tables are the same, you can specify these columns as distribution keys.

• If the join columns of the tables are different, you can specify the join columns of the large tables as distribution keys.

In the following examples, three tables are joined. If you want to join more than three tables, use one of the following examples as a reference:

• The join columns of the three tables are the same.

  If the join columns of the three tables are the same, you can specify these columns as distribution keys to implement a local join. This scenario is the simplest scenario.

  • Syntax supported in 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;
    
    -- Execute the following statement to join the three tables:
    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 in all Hologres 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;
    
    -- Execute the following statement to join the three tables:
    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;

  You can query the execution plan by executing the Explain statement.

  • The redistribution operator is not included in the execution plan. This indicates that no data is redistributed and that a local join operation is performed.

  • The exchange operator indicates a shift from file-level aggregation to shard-level aggregation. In this case, only data in the corresponding shard is aggregated. This improves data query efficiency.

  

• The join columns of the three tables are different.

  In actual business scenarios, if you want to join multiple tables, the join columns may be different. In this case, take note of the following rules when you specify a distribution key:

  • Compared with small tables, large tables have a higher priority because small tables contain only a small amount of data. You can preferentially specify the join columns of the large tables as distribution keys.

  • If the amount of data in the tables is nearly the same, specify columns on which the GROUP BY operation is frequently performed as distribution keys.

  In the following example, three tables named join_tbl_1, join_tbl_2, and join_tbl_3 are to be joined, and the join columns are not exactly the same. In this case, specify the join columns of the large tables as distribution keys. The join_tbl_1 table contains 10 million data entries. Both the join_tbl_2 table and the join_tbl_3 table contain one million data entries. Use the join_tbl_1 table as the main optimization object.

  • Syntax supported in Hologres V2.1 and later:

    BEGIN;
    -- The join_tbl_1 table contains 10 million data entries.
    CREATE TABLE join_tbl_1 (
        a int NOT NULL,
        b text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    -- The join_tbl_2 table contains one million data entries.
    CREATE TABLE join_tbl_2 (
        a int NOT NULL,
        d text NOT NULL,
        e text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    -- The join_tbl_3 table contains one million data entries.
    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;
    
    -- If the join columns of the tables that you want to join are different, specify the join columns of the large tables as distribution keys. 
    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 in all Hologres versions:

    begin;
    -- The join_tbl_1 table contains 10 million data entries.
    create table join_tbl_1(
    a int not null,
    b text not null
    );
    call set_table_property('join_tbl_1', 'distribution_key', 'a');
    
    -- The join_tbl_2 table contains one million data entries.
    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');
    
    --The join_tbl_3 table contains one million data entries.
    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;
    
    -- If the join columns of the tables that you want to join are different, specify the join columns of the large tables as distribution keys. 
    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;

  You can query the execution plan by executing the Explain statement.

  • The redistribution operator is used for the computing of the join_tbl_2 table and the join_tbl_3 table. The join_tbl_3 table is a small table, and the join column of the small table is different from the distribution key. Therefore, the data is redistributed.

  • No redistribution operator is used for the computing of the join_tbl_1 table and the join_tbl_2 table. The join columns of the two tables are specified as the distribution keys. Therefore, the data is not redistributed.