Colocate Join
shuffle join と broadcast join では、ジョイン条件が満たされると、2 つのジョインテーブルのデータ行が単一ノードにマージされてジョインが完了します。これら 2 つのジョイン方法のいずれも、ノード間のデータネットワーク伝送によって引き起こされる遅延やオーバーヘッドを回避することはできません。
コアアイデアは、同じ Colocation Group 内のテーブルに対して、バケッ ティングキー、コピーの数、およびコピーの配置を一貫して保つことです。ジョイン列がバケッティングキーである場合、計算ノードは他のノードからデータを取得せずにローカルジョインを行うだけで済みます。Colocate Join は等値ジョインをサポートしています。
このドキュメントでは、Colocate Join の原理、実装、使用法、および考慮事項を紹介します。
用語
- Colocation Group (CG): CG は 1 つ以上のテーブルを含みます。CG 内のテーブルは同じバケッティングとレプリカ配置を持ち、Colocation Group Schema を使用して記述されます。
- Colocation Group Schema (CGS): CGS には CG のバケッティングキー、バケット数、およびレプリカ数が含まれます。
原理
Colocate Join は、同じ CGS を持つ一連のテーブルで CG を形成し、これらのテーブルの対応するバケットコピーが同じ BE ノードセットに配置されることを保証します。CG 内のテーブルがバケット化された列でジョイン操作を行うとき、ローカルデータは直接ジョインされ、ノード間でデータを転送する時間を節約できます。
バケットシーケンスは hash(key) mod buckets によって取得されます。例えば、テーブルが 8 バケットを持つ場合、[0, 1, 2, 3, 4, 5, 6, 7] の 8 バケットがあり、各バケットには 1 つ以上のサ ブテーブルがあり、サブテーブルの数はパーティションの数に依存します。マルチパーティションテーブルの場合、複数のタブレットが存在します。
同じデータ分布を持つために、同じ CG 内のテーブルは次のことを遵守する必要があります。
- 同じ CG 内のテーブルは、同一のバケッティングキー（タイプ、数、順序）と同じ数のバケットを持つ必要があります。これにより、複数のテーブルのデータスライスを一対一で分配および制御できます。バケッティングキーは、テーブル作成ステートメント
DISTRIBUTED BY HASH(col1, col2, ...)で指定された列です。バケッティングキーは、データのどの列が異なるバケットシーケンスにハッシュされるかを決定します。同じ CG 内のテーブルでバケッティングキーの名前は異なる場合があります。作成ステートメントでバケッティング列は異なる場合がありますが、DISTRIBUTED BY HASH(col1, col2, ...)内の対応するデータ型の順序は完全に同じである必要があります。 - 同じ CG 内のテーブルは、異なる数のパーティションと異なるパーティションキーを持つことができます。
テーブルを作成する際、CG はテーブルの PROPERTIES で属性 "colocate_with" = "group_name" によって指定されます。CG が存在しない場合、テーブルは CG の最初のテーブルであり、親テーブルと呼ばれます。親テーブルのデータ分布（分割バケットキーのタイプ、数、順序、コピーの数、分割バケットの数）が CGS を決定します。CG が存在する場合、テーブルのデータ分布が CGS と一致しているかどうかを確認します。
同じ CG 内のテーブ ルのコピー配置は次の条件を満たします。
- すべてのテーブルのバケットシーケンスと BE ノードの間のマッピングは、親テーブルのマッピングと同じです。
- 親テーブル内のすべてのパーティションのバケットシーケンスと BE ノードの間のマッピングは、最初のパーティションのマッピングと同じです。
- 親テーブルの最初のパーティションのバケットシーケンスと BE ノードの間のマッピングは、ネイティブのラウンドロビンアルゴリズムを使用して決定されます。
一貫したデータ分布とマッピングにより、バケッティングキーで取得された同じ値を持つデータ行が同じ BE に配置されることが保証されます。したがって、バケッティングキーを使用して列をジョインする場合、ローカルジョインのみが必要です。
使用法
テーブル作成
テーブルを作成する際、PROPERTIES で属性 "colocate_with" = "group_name" を指定して、テーブルが Colocate Join テーブルであり、指定された Colocation Group に属することを示すことができます。
注意
バージョン 2.5.4 以降、異なるデータベースのテーブルに対して Colocate Join を実行できます。テーブルを作成する際に同じ
colocate_withプロパティを指定するだけです。
例:
CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
"colocate_with" = "group1"
);
指定されたグループが存在しない場合、StarRocks は現在のテーブルのみを含むグループを自動的に作成します。グループが存在する場合、StarRocks は現在のテーブルが Colocation Group Schema を満たしているかどうかを確認します。満たしている場合、テーブルを作成し、グループに追加します。同時に、テーブルは既存のグループのデータ分布ルールに基づいてパーティションとタブレットを作成します。
Colocation Group はデータベースに属します。Colocation Group の名前はデータベース内で一意です。内部ストレージでは、Colocation Group の完全な名前は dbId_groupName ですが、groupName だけを認識します。
注意
異なるデータベースからテーブルを関連付けるために同じ Colocation Group を指定する場合、Colocation Group はこれらのデータベースのそれぞれに存在します。異なるデータベースの Colocation Group を確認するには、
show proc "/colocation_group"を実行できます。
削除
完全な削除は、リサイクルビンからの削除です。通常、DROP TABLE コマンドでテーブルを削除した後、デフォルトでは 1 日間リサイクルビンに残り、その後削除されます。グループ内の最後のテーブルが完全に削除されると、グループも自動的に削除されます。
グループ情報の表示
次のコマンドを使用して、クラスター内に既に存在するグループ情報を表示できます。
SHOW PROC '/colocation_group';
+-------------+--------------+--------------+------------+----------------+----------+----------+
| GroupId | GroupName | TableIds | BucketsNum | ReplicationNum | DistCols | IsStable |
+-------------+--------------+--------------+------------+----------------+----------+----------+
| 10005.10008 | 10005_group1 | 10007, 10040 | 10 | 3 | int(11) | true |
+-------------+--------------+--------------+------------+----------------+----------+----------+
- GroupId: グループのクラスター全体で一意の識別子で、前半が db id で後半がグループ id です。
- GroupName: グループの完全な名前。
- TabletIds: グループ内のテーブルの id のリスト。
- BucketsNum: バケットの数。
- ReplicationNum: レプリカの数。
- DistCols: 分布列、つまりバケッティング列のタイプ。
- IsStable: グループが安定しているかどうか（安定性の定義については、Colocation Replica Balancing and Repair のセクションを参照してください）。
次のコマンドを使用して、グループのデータ分布をさらに表示できます。
SHOW PROC '/colocation_group/10005.10008';
+-------------+---------------------+
| BucketIndex | BackendIds |
+-------------+---------------------+
| 0 | 10004, 10002, 10001 |
| 1 | 10003, 10002, 10004 |
| 2 | 10002, 10004, 10001 |
| 3 | 10003, 10002, 10004 |
| 4 | 10002, 10004, 10003 |
| 5 | 10003, 10002, 10001 |
| 6 | 10003, 10004, 10001 |
| 7 | 10003, 10004, 10002 |
+-------------+---------------------+
- BucketIndex: バケットシーケンスの添字。
- BackendIds: バケッティングデータスライスが配置されている BE ノードの id。
注意: 上記のコマンドを使用するには、NODE 権限または
cluster_adminロールが必要です。通常のユーザーはアクセスできません。
テーブルグループ プロパティの変更
テーブルの Colocation Group プロパティを変更できます。例えば:
ALTER TABLE tbl SET ("colocate_with" = "group2");
テーブルが以前にグループに割り当てられていない場合、コマンドはスキーマをチェックし、テーブルをグループに追加します（グループが存在しない場合は最初に作成されます）。テーブルが以前に別のグループに割り当てられていた場合、コマンドはテーブルを元のグループから削除し、新しいグループに追加します（グループが存在しない場合は最初に作成されます）。
次のコマンドを使用して、テーブルの Colocation プロパティを削除することもできます。
ALTER TABLE tbl SET ("colocate_with" = "");
その他の関連操作
Colocation 属性を持つテーブルに ADD PARTITION を使用してパーティションを追加したり、コピーの数を変更したりする場合、StarRocks は操作が Colocation Group Schema に違反するかどうかを確認し、違反する場合は拒否します。
Colocation レプリカのバランスと修復
Colocation テーブルのレプリカ分布は、グループスキーマで指定された分布ルールに従う必要があるため、通常のシャーディングとは異なり、レプリカの修復とバランスが異なります。
グループ自体には stable プロパティがあります。stable が true の場合、グループ内のテーブルスライスに変更が加えられておらず、Colocation 機能が正常に動作していることを意味します。stable が false の場合、現在のグループ内の一部のテーブルスライスが修復または移行されており、影響を受けたテーブルの Colocate Join は通常のジョインに劣化します。
レプリカ修復
レプリカは指定された BE ノードにのみ保存できます。StarRocks は、利用できない BE（例: ダウン、廃止）を置き換えるために最も負荷の少ない BE を探します。置き換え後、古い BE 上のすべてのバケッティングデータスライスが修復されます。移行中、グループは Unstable とマークされます。
レプリカのバランス
StarRocks は、Colocation テーブルスライスをすべての BE ノードに均等に分配しようとします。通常のテーブルのバランスはレプリカレベルで行われ、各レプリカが個別に低負荷の BE ノードを見つけます。Colocation テーブルのバランスはバケットレベルで行われ、バケット内のすべてのレプリカが一緒に移動されます。BucketsSequnce をすべての BE ノードに均等に分配する単純なバランスアルゴリズムを使用しますが、レプリカの実際のサイズではなく、レプリカの数のみを考慮します。正確なアルゴリズムは ColocateTableBalancer.java のコードコメントにあります。
注意 1: 現在の Colocation レプリカのバランスと修復アルゴリズムは、異種展開を持つ StarRocks クラスターではうまく機能しない可能性があります。いわゆる異種展開とは、BE ノードのディスク容量、ディスク数、およびディスクタイプ（SSD と HDD）が一貫していないことを意味します。異種展開の場合、小容量の BE ノードが大容量の BE ノードと同じ数のレプリカを保存することがあるかもしれません。
注意 2: グループが不安定な状態にある場合、そのテーブルのジョインは通常のジョインに劣化し、クラスターのクエリパフォーマンスが大幅に低下する可能性があります。システムが自動的にバランスを取らないようにする場合は、FE 設定
disable_colocate_balanceを設定して自動バランスを無効にし、適切なタイミングで再度有効にします。（詳細は Advanced Operations (#Advanced Operations) セクションを参照してください）
クエリ
Colocation テーブルは通常のテーブルと同じ方法でクエリされます。Colocation テーブルが存在するグループが不安定な状態にある場合、それは自動的に通常のジョインに劣化します。以下の例で説明します。
テーブル 1:
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 6
PROPERTIES (
"colocate_with" = "group1"
);
INSERT INTO tbl1
VALUES
("2015-09-12",1000,1),
("2015-09-13",2000,2);
テーブル 2:
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 6
PROPERTIES (
"colocate_with" = "group1"
);
INSERT INTO tbl2
VALUES
("2015-09-12 00:00:00",3000,3),
("2015-09-12 00:00:00",4000,4);
クエリプランの表示:
EXPLAIN SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);
+-------------------------------------------------------------------------+
| Explain String |
+-------------------------------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:1: k1 | 2: k2 | 3: v1 | 4: k1 | 5: k2 | 6: v1 |
| PARTITION: UNPARTITIONED |
| |
| RESULT SINK |
| |
| 3:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 03 |
| UNPARTITIONED |
| |
| 2:HASH JOIN |
| | join op: INNER JOIN (COLOCATE) |
| | colocate: true |
| | equal join conjunct: 5: k2 = 2: k2 |
| | |
| |----1:OlapScanNode |
| | TABLE: tbl1 |
| | PREAGGREGATION: OFF. Reason: Has can not pre-aggregation Join |
| | partitions=1/2 |
| | rollup: tbl1 |
| | tabletRatio=6/6 |
| | tabletList=15344,15346,15348,15350,15352,15354 |
| | cardinality=1 |
| | avgRowSize=3.0 |
| | |
| 0:OlapScanNode |
| TABLE: tbl2 |
| PREAGGREGATION: OFF. Reason: None aggregate function |
| partitions=1/1 |
| rollup: tbl2 |
| tabletRatio=6/6 |
| tabletList=15373,15375,15377,15379,15381,15383 |
| cardinality=1 |
| avgRowSize=3.0 |
+-------------------------------------------------------------------------+
40 rows in set (0.03 sec)
Colocate Join が有効になると、Hash Join ノードに colocate: true が表示されます。
有効にならない場合、クエリプランは次のようになります。
+----------------------------------------------------+
| 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 |
+----------------------------------------------------+
HASH JOIN ノードは対応する理由を示します: colocate: false, reason: group is not stable。同時に EXCHANGE ノードが生成されます。
高度な操作
FE 設定項目
- disable_colocate_relocate
StarRocks の自動 Colocation レプリカ修復を無効にするかどうか。デフォルトは false で、オンになっています。このパラメータは、Colocation テーブルのレプリカ修復にのみ影響し、通常のテーブルには影響しません。
- disable_colocate_balance
StarRocks の自動 Colocation レプリカバランスを無効にするかどうか。デフォルトは false で、オンになっています。このパラメータは、Colocation テーブルのレプリカバランスにのみ影響し、通常のテーブルには影響しません。
-
disable_colocate_join
この変数を変更することで、セッション単位で Colocate join を無効にできます。
-
disable_colocate_join
この変数を変更することで、Colocate join 機能を無効にできます。
HTTP Restful API
StarRocks は、Colocate Join に関連する Colocation Group の表示および変更に関するいくつかの HTTP Restful API を提供しています。
この API は FE 上で実装されており、db_admin および user_admin 権限を使用して fe_host:fe_http_port でアクセスできます。
-
クラスターのすべての Colocation 情報を表示
curl --location-trusted -u<username>:<password> 'http://<fe_host>:<fe_http_port>/api/colocate'// Json 形式で内部の Colocation 情報を返します。
{
"colocate_meta": {
"groupName2Id": {
"g1": {
"dbId": 10005,
"grpId": 10008
}
},
"group2Tables": {},
"table2Group": {
"10007": {
"dbId": 10005,
"grpId": 10008
},
"10040": {
"dbId": 10005,
"grpId": 10008
}
},
"group2Schema": {
"10005.10008": {
"groupId": {
"dbId": 10005,
"grpId": 10008
},
"distributionColTypes": [{
"type": "INT",
"len": -1,
"isAssignedStrLenInColDefinition": false,
"precision": 0,
"scale": 0
}],
"bucketsNum": 10,
"replicationNum": 2
}
},
"group2BackendsPerBucketSeq": {
"10005.10008": [
[10004, 10002],
[10003, 10002],
[10002, 10004],
[10003, 10002],
[10002, 10004],
[10003, 10002],
[10003, 10004],
[10003, 10004],
[10003, 10004],
[10002, 10004]
]
},
"unstableGroups": []
},
"status": "OK"
} -
グループを安定または不安定としてマーク
# 安定としてマーク
curl -XPOST --location-trusted -u<username>:<password> 'http://<fe_host>:<fe_http_port>/api/colocate/group_stable?db_id=<dbId>&group_id=<grpId>'
# 不安定としてマーク
curl -XPOST --location-trusted -u<username>:<password> 'http://<fe_host>:<fe_http_port>/api/colocate/group_unstable?db_id=<dbId>&group_id=<grpId>'返された結果が
200の場合、グループは正常に安定または不安定としてマークされます。 -
グループのデータ分布を設定
このインターフェースを使用して、グループの数値分布を強制的に設定できます。
POST /api/colocate/bucketseq?db_id=10005&group_id= 10008Body:[[10004,10002],[10003,10002],[10002,10004],[10003,10002],[10002,10004],[10003,10002],[10003,10004],[10003,10004],[10003,10004],[10002,10004]]returns: 200Bodyは、ネストされた配列として表現されたBucketsSequenceであり、バケッティングスライスが配置されている BE の id です。このコマンドを使用するには、FE 設定
disable_colocate_relocateとdisable_colocate_balanceを true に設定し、システムが自動的に Colocation レプリカの修復とバランシングを 行わないようにする必要があるかもしれません。そうしないと、変更後にシステムによって自動的にリセットされる可能性があります。
