Colocation Join

Overview

The colocation join feature optimizes certain join queries by ensuring data locality. This reduces data transfer time between nodes and accelerates queries. For details about the original design, implementation, and performance, see ISSUE 245. The colocation join feature has since been revised, and its design and usage differ slightly from the original version.

Key concepts

• colocation group (CG): A group that contains one or more tables. All tables within the same group share the same colocation group schema and have identical tablet distribution.

• colocation group schema (CGS): The set of schema properties that all tables in a colocation group must share. These properties include the data types of the bucket columns and the number of buckets.

How it works

The colocation join feature groups tables with the same CGS into a CG, ensuring that the corresponding tablets of these tables are placed on the same BE node. Consequently, when you join these tables on their bucket columns, the data is joined locally, reducing data transfer between nodes.

A table's data is distributed into buckets based on the hash value of its bucket columns, modulo the number of buckets. For example, if a table has 8 buckets, it has a bucket sequence of [0, 1, 2, 3, 4, 5, 6, 7]. Each bucket contains one or more tablets. For a single-partition table, a bucket contains only one tablet. For a multi-partition table, it contains multiple tablets.

To ensure a consistent data distribution, all tables within the same colocation group must have identical bucket columns and the same number of buckets. The bucket columns are specified in the DISTRIBUTED BY HASH(col1, col2, ...) clause of a CREATE TABLE statement and determine how data is hashed into different tablets. For tables to belong to the same CG, their bucket columns must match in type and number. This allows the system to manage the distribution of their tablets in a one-to-one correspondence.

Tables within the same colocation group do not need to have the same number of partitions, partition ranges, or partition column types.

After the bucket columns and the number of buckets are fixed, all tables in the same colocation group share the same bucket sequence. For example, assume a bucket sequence of [0, 1, 2, 3, 4, 5, 6, 7] and four BE nodes: [A, B, C, D]. A possible data distribution is as follows:

+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
| 0 | | 1 | | 2 | | 3 | | 4 | | 5 | | 6 | | 7 |
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
| A | | B | | C | | D | | A | | B | | C | | D |
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+

All tables in the colocation group distribute their data according to this rule. This ensures that data with the same bucket column values resides on the same BE node, enabling local joins.

Usage

Create a table

When you create a table, you can specify the "colocate_with" = "group_name" property in the PROPERTIES clause. This designates the table as a colocation table and assigns it to a specified colocation group.

Example:

CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
    "colocate_with" = "group1"
);

If the specified group does not exist, ApsaraDB for SelectDB automatically creates a new group that contains only this table. If the group already exists, ApsaraDB for SelectDB checks whether the table's schema is compatible with the colocation group schema. If they are compatible, the table is created and added to the group. The table's tablets and replicas are then created based on the data distribution rules of the existing group. A colocation group belongs to a database, and its name must be unique within that database. Internally, the full name of a group is stored as dbId_groupName, but you only need to specify the groupName.

ApsaraDB for SelectDB supports cross-database groups. When you create a table, you can prefix the group name with the __global__ keyword. Example:

CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
    "colocate_with" = "__global__group1"
);

A group with the __global__ prefix does not belong to a specific database, and its name is globally unique. By creating a global group, you can perform colocation joins across databases.

Delete a table

When the last table in a group is permanently deleted, the group is also automatically deleted. A permanent deletion means that the table is removed from the recycle bin. By default, after you delete a table by using the DROP TABLE command, it remains in the recycle bin for one day before it is permanently deleted.

Query a table

You can query a colocation table just like a regular table. You do not need to specify its colocation property, because the query planner automatically uses a colocation join when possible. The following example demonstrates how this works.

1. Create the tables.

   Create table tbl1. Example:

   CREATE TABLE `tbl1` (
       `k1` date NOT NULL COMMENT "",
       `k2` int(11) NOT NULL COMMENT "",
       `v1` int(11) SUM NOT NULL COMMENT ""
   ) ENGINE=OLAP
   AGGREGATE KEY(`k1`, `k2`)
   PARTITION BY RANGE(`k1`)
   (
       PARTITION p1 VALUES LESS THAN ('2019-05-31'),
       PARTITION p2 VALUES LESS THAN ('2019-06-30')
   )
   DISTRIBUTED BY HASH(`k2`) BUCKETS 8
   PROPERTIES (
       "colocate_with" = "group1"
   );

   Create table tbl2. Example:

   CREATE TABLE `tbl2` (
       `k1` datetime NOT NULL COMMENT "",
       `k2` int(11) NOT NULL COMMENT "",
       `v1` double SUM NOT NULL COMMENT ""
   ) ENGINE=OLAP
   AGGREGATE KEY(`k1`, `k2`)
   DISTRIBUTED BY HASH(`k2`) BUCKETS 8
   PROPERTIES (
       "colocate_with" = "group1"
   );

2. View the query plan. Example:

   DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);
   +----------------------------------------------------+
   | Explain String                                     |
   +----------------------------------------------------+
   | PLAN FRAGMENT 0                                    |
   |  OUTPUT EXPRS:`tbl1`.`k1` |                        |
   |   PARTITION: RANDOM                                |
   |                                                    |
   |   RESULT SINK                                      |
   |                                                    |
   |   2:HASH JOIN                                      |
   |   |  join op: INNER JOIN                           |
   |   |  hash predicates:                              |
   |   |  colocate: true                                |
   |   |    `tbl1`.`k2` = `tbl2`.`k2`                   |
   |   |  tuple ids: 0 1                                |
   |   |                                                |
   |   |----1:OlapScanNode                              |
   |   |       TABLE: tbl2                              |
   |   |       PREAGGREGATION: OFF. Reason: null        |
   |   |       partitions=0/1                           |
   |   |       rollup: null                             |
   |   |       buckets=0/0                              |
   |   |       cardinality=-1                           |
   |   |       avgRowSize=0.0                           |
   |   |       numNodes=0                               |
   |   |       tuple ids: 1                             |
   |   |                                                |
   |   0:OlapScanNode                                   |
   |      TABLE: tbl1                                   |
   |      PREAGGREGATION: OFF. Reason: No AggregateInfo |
   |      partitions=0/2                                |
   |      rollup: null                                  |
   |      buckets=0/0                                   |
   |      cardinality=-1                                |
   |      avgRowSize=0.0                                |
   |      numNodes=0                                    |
   |      tuple ids: 0                                  |
   +----------------------------------------------------+

   If a colocation join is used, the HASH JOIN node shows colocate: true.

   If a colocation join is not used, the query plan is as follows:

   +----------------------------------------------------+
   | Explain String                                     |
   +----------------------------------------------------+
   | PLAN FRAGMENT 0                                    |
   |  OUTPUT EXPRS:`tbl1`.`k1` |                        |
   |   PARTITION: RANDOM                                |
   |                                                    |
   |   RESULT SINK                                      |
   |                                                    |
   |   2:HASH JOIN                                      |
   |   |  join op: INNER JOIN (BROADCAST)               |
   |   |  hash predicates:                              |
   |   |  colocate: false, reason: group is not stable  |
   |   |    `tbl1`.`k2` = `tbl2`.`k2`                   |
   |   |  tuple ids: 0 1                                |
   |   |                                                |
   |   |----3:EXCHANGE                                  |
   |   |       tuple ids: 1                             |
   |   |                                                |
   |   0:OlapScanNode                                   |
   |      TABLE: tbl1                                   |
   |      PREAGGREGATION: OFF. Reason: No AggregateInfo |
   |      partitions=0/2                                |
   |      rollup: null                                  |
   |      buckets=0/0                                   |
   |      cardinality=-1                                |
   |      avgRowSize=0.0                                |
   |      numNodes=0                                    |
   |      tuple ids: 0                                  |
   |                                                    |
   | PLAN FRAGMENT 1                                    |
   |  OUTPUT EXPRS:                                     |
   |   PARTITION: RANDOM                                |
   |                                                    |
   |   STREAM DATA SINK                                 |
   |     EXCHANGE ID: 03                                |
   |     UNPARTITIONED                                  |
   |                                                    |
   |   1:OlapScanNode                                   |
   |      TABLE: tbl2                                   |
   |      PREAGGREGATION: OFF. Reason: null             |
   |      partitions=0/1                                |
   |      rollup: null                                  |
   |      buckets=0/0                                   |
   |      cardinality=-1                                |
   |      avgRowSize=0.0                                |
   |      numNodes=0                                    |
   |      tuple ids: 1                                  |
   +----------------------------------------------------+

   The HASH JOIN node shows the reason for the failure: colocate: false, reason: group is not stable. An EXCHANGE node also appears, indicating data movement across nodes.

View colocation groups

To view existing colocation groups in a cluster, run this command:

SHOW PROC '/colocation_group';

+-------------+--------------+--------------+------------+----------------+----------+----------+
| GroupId     | GroupName    | TableIds     | BucketsNum | ReplicationNum | DistCols | IsStable |
+-------------+--------------+--------------+------------+----------------+----------+----------+
| 10005.10008 | 10005_group1 | 10007, 10040 | 10         | 3              | int(11)  | true     |
+-------------+--------------+--------------+------------+----------------+----------+----------+

The following table describes the returned parameters.

To view the data distribution of a group, run the following command:

SHOW PROC '/colocation_group/10005.10008';

+-------------+---------------------+
| BucketIndex | BackendIds          |
+-------------+---------------------+
| 0           | 10004               |
| 1           | 10003               |
| 2           | 10002               |
| 3           | 10003               |
| 4           | 10002               |
| 5           | 10003               |
| 6           | 10003               |
| 7           | 10003               |
+-------------+---------------------+

Modify colocation property

To change a table's colocation group property, run this command:

ALTER TABLE tbl SET ("colocate_with" = "group2");

• If the table was not in a colocation group, this command adds it to the specified group after a schema compatibility check.

• If the table was previously in another group, this command moves it to the new group.

To remove the colocation property from a table, run the following command:

ALTER TABLE tbl SET ("colocate_with" = "");

Other related operations

When you perform operations such as ADD PARTITION or modify the replica count for a table with the colocation property, ApsaraDB for SelectDB validates the change against the colocation group schema. If the change violates the schema, the operation is rejected.

Advanced usage

FE parameters

• disable_colocate_relocate

  Specifies whether to disable automatic replica repair for colocation tables. The default value is false, which means automatic replica repair is enabled. This parameter only affects colocation tables.

• disable_colocate_balance

  Specifies whether to disable automatic replica rebalancing for colocation tables. The default value is false, which means automatic replica rebalancing is enabled. This parameter only affects colocation tables.

You can dynamically modify these parameters. For more information, run HELP ADMIN SHOW CONFIG; and HELP ADMIN SET CONFIG;.

• disable_colocate_join

  Specifies whether to disable the colocation join feature. The default value is false, which means the feature is enabled.

• use_new_tablet_scheduler

  Specifies whether to enable the new replica scheduling logic. The default value is true, which means it is enabled.

HTTP RESTful API

ApsaraDB for SelectDB provides several HTTP RESTful APIs that you can use to view and modify colocation groups.

These APIs run on the FE and can be accessed at fe_host:fe_http_port. Calling these APIs requires admin privileges.

• View all colocation information in the cluster. Example:

  GET /api/colocate
  
  The response provides internal colocation information in JSON format.
  
  {
      "msg": "success",
      "code": 0,
      "data": {
          "infos": [
              ["10003.12002", "10003_group1", "10037, 10043", "1", "1", "int(11)", "true"]
          ],
          "unstableGroupIds": [],
          "allGroupIds": [{
              "dbId": 10003,
              "grpId": 12002
          }]
      },
      "count": 0
  }

• Mark a group as stable or unstable. Example:

  • Mark as stable.

    DELETE /api/colocate/group_stable?db_id=10005&group_id=10008
    
    Response: 200

  • Mark as unstable.

    POST /api/colocate/group_stable?db_id=10005&group_id=10008
    
    Response: 200

• Set the data distribution for a group. Example:

  You can use this API to force a specific data distribution for a group. The body of the POST request is a nested array that represents the bucket sequence and the corresponding BE node ID for each bucket's tablets.

  POST /api/colocate/bucketseq?db_id=10005&group_id=10008
  
  Body:
  [[10004],[10003],[10002],[10003],[10002],[10003],[10003],[10003],[10003],[10002]]
  
  Response: 200

  Note

  Before you use this command, you must set the disable_colocate_relocate and disable_colocate_balance FE configuration items to true. This action disables the system's automatic replica repair and rebalancing. If you do not disable these features, the system might override your manual changes.