Topik ini menjelaskan implementasi fitur colocation join di ApsaraDB for SelectDB serta cara menggunakannya untuk memilih metode join yang optimal dalam pengoptimalan kueri.
Ikhtisar
Fitur colocation join memberikan optimasi lokal untuk beberapa kueri join dengan mengurangi waktu transmisi data antar node dan mempercepat eksekusi kueri. Untuk informasi lebih lanjut tentang desain awal, implementasi, dan performa, lihat Issue 245. Fitur ini telah direvisi, sehingga desain dan penggunaannya sedikit berbeda dari versi awal.
Properti colocation tabel tidak dapat disinkronkan melalui replikasi antar kluster (CCR). Jika sebuah tabel memiliki properti is_being_synced = true, yang menunjukkan bahwa tabel tersebut disalin dari CCR, properti colocation akan dihapus.
Istilah
Colocation Group (CG): Berisi satu atau lebih tabel. Tabel dalam CG yang sama memiliki Colocation Group Schema (CGS) dan distribusi tablet yang identik.
CGS: Menggambarkan skema umum terkait colocation untuk tabel dalam CG, termasuk tipe kolom bucket dan jumlah bucket.
Cara kerjanya
Fitur colocation join membentuk CG yang terdiri dari tabel dengan CGS yang sama dan memastikan bahwa tablet dari tabel-tabel tersebut berada pada node backend (BE) yang sama. Dengan pendekatan ini, ketika tabel dalam CG digabungkan berdasarkan kolom bucket, join data lokal dapat dilakukan untuk mengurangi waktu transmisi data antar node.
Data dari sebuah tabel didistribusikan ke bucket setelah hashing dilakukan pada nilai kolom bucket dan operasi modulo diterapkan pada jumlah bucket. Sebagai contoh, jika jumlah bucket adalah delapan ([0, 1, 2, 3, 4, 5, 6, 7]), urutan tersebut disebut urutan bucket. Setiap bucket memiliki satu atau lebih tablet. Untuk tabel partisi tunggal, satu bucket hanya berisi satu tablet, sedangkan untuk tabel multi-partisi, satu bucket dapat berisi beberapa tablet.
Untuk memastikan distribusi data yang konsisten antar tabel dalam CG, kolom bucket dan jumlah bucket harus identik. Kolom bucket ditentukan dalam pernyataan pembuatan tabel menggunakan DISTRIBUTED BY HASH(col1, col2, ...). Kolom ini menentukan nilai yang digunakan untuk menghash data ke tablet yang berbeda. Pastikan tipe dan jumlah kolom bucket serta jumlah bucket sama untuk semua tabel dalam CG agar tablet dapat didistribusikan secara seragam.
Jumlah partisi, cakupan, dan tipe kolom partisi tidak harus sama untuk tabel dalam CG.
Setelah kolom bucket dan jumlah bucket ditentukan, tabel dalam CG memiliki urutan bucket yang sama. Misalnya, jika urutan bucket adalah [0, 1, 2, 3, 4, 5, 6, 7] dan node BE adalah [A, B, C, D], distribusi data mungkin terlihat seperti diagram berikut:
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
| 0 | | 1 | | 2 | | 3 | | 4 | | 5 | | 6 | | 7 |
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
| A | | B | | C | | D | | A | | B | | C | | D |
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+Data dari semua tabel dalam CG didistribusikan berdasarkan aturan ini, memastikan bahwa data dengan nilai kolom bucket yang sama berada pada node BE yang sama untuk join data lokal.
Penggunaan dasar
Buat tabel
Saat membuat tabel, Anda dapat menentukan "colocate_with" = "group_name" dalam PROPERTIES untuk menandai tabel sebagai tabel colocation dan memasukkannya ke dalam CG tertentu.
Contoh:
CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
"colocate_with" = "group1"
);Jika CG yang ditentukan tidak ada, ApsaraDB for SelectDB secara otomatis membuat CG baru yang hanya berisi tabel saat ini. Jika CG sudah ada, sistem memeriksa apakah tabel memenuhi persyaratan CGS. Jika ya, tabel dibuat dan ditambahkan ke CG. Tablet dan replika juga dibuat berdasarkan aturan distribusi data CG. CG milik database tertentu, dan nama CG unik dalam database. Secara internal, CG disimpan dengan format dbId_groupName, tetapi pengguna hanya perlu menyebutkan groupName.
ApsaraDB for SelectDB mendukung CG lintas database. Saat membuat tabel, Anda dapat menambahkan awalan __global__ ke nama CG untuk menentukan CG global. Contoh:
CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
"colocate_with" = "__global__group1"
);CG dengan nama yang diawali __global__ tidak termasuk dalam database dan bersifat unik secara global. Anda dapat menggunakan CG global untuk join colocation lintas database.
Hapus tabel
Penghapusan permanen berarti tabel dihapus dari Keranjang daur ulang. Umumnya, setelah pernyataan DROP TABLE dieksekusi, tabel dipindahkan ke Keranjang daur ulang dan tetap ada selama satu hari sebelum dihapus secara permanen. Setelah tabel terakhir dalam CG dihapus secara permanen, CG secara otomatis dihapus.
Kueri tabel
Anda dapat mengkueri tabel colocation seperti tabel biasa. Properti colocation tabel tidak terlihat oleh pengguna. Sistem secara otomatis menghasilkan rencana kueri yang menggunakan join colocation. Contoh berikut menunjukkan cara mengkueri tabel colocation:
Buat tabel.
Buat tabel tbl1.
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" );Buat tabel tbl2.
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" );Lihat rencana kueri untuk join tabel-tabel tersebut.
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 | +----------------------------------------------------+Jika join colocation berlaku, node HASH JOIN berisi
colocate: true.Contoh berikut menunjukkan rencana kueri di mana join colocation tidak berlaku:
+----------------------------------------------------+ | 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 | +----------------------------------------------------+Node HASH JOIN berisi alasan mengapa join colocation tidak berlaku:
colocate: false, reason: group is not stable. Node EXCHANGE juga dihasilkan.
Lihat informasi CG
Anda dapat melihat informasi tentang CG yang ada dalam kluster. Contoh:
SHOW PROC '/colocation_group';
+-------------+--------------+--------------+------------+----------------+----------+----------+
| GroupId | GroupName | TableIds | BucketsNum | ReplicationNum | DistCols | IsStable |
+-------------+--------------+--------------+------------+----------------+----------+----------+
| 10005.10008 | 10005_group1 | 10007, 10040 | 10 | 3 | int(11) | true |
+-------------+--------------+--------------+------------+----------------+----------+----------+Tabel berikut menjelaskan parameter.
Parameter | Deskripsi |
GroupId | Pengenal unik di seluruh kluster dari CG, di mana bagian pertama adalah ID database dan bagian kedua adalah ID CG. |
GroupName | Nama lengkap CG. |
TabletIds | ID tabel dalam CG. |
BucketsNum | Jumlah bucket. |
ReplicationNum | Jumlah replika. |
DistCols | Tipe kolom bucket. |
IsStable | Menunjukkan apakah CG stabil. Untuk informasi lebih lanjut tentang definisi stabilitas, lihat bagian " |
Lihat distribusi data CG. Contoh:
SHOW PROC '/colocation_group/10005.10008';
+-------------+---------------------+
| BucketIndex | BackendIds |
+-------------+---------------------+
| 0 | 10004 |
| 1 | 10003 |
| 2 | 10002 |
| 3 | 10003 |
| 4 | 10002 |
| 5 | 10003 |
| 6 | 10003 |
| 7 | 10003 |
+-------------+---------------------+Parameter | Deskripsi |
BucketIndex | Indeks bucket. |
BackendIds | ID node BE tempat tablet dalam bucket berada. |
Izin peran admin diperlukan untuk menjalankan perintah ini. Pengguna biasa tidak dapat menjalankannya.
Ubah properti colocation tabel
Anda dapat mengubah properti colocation tabel yang ada. Contoh:
ALTER TABLE tbl SET ("colocate_with" = "group2");Jika tabel belum ditambahkan ke CG, sistem memeriksa skema dan menambahkan tabel ke CG yang ditentukan. Jika CG tidak ada, sistem secara otomatis membuat CG.
Jika tabel sudah ditambahkan ke CG, sistem menghapus tabel dari CG asli dan menambahkannya ke CG yang ditentukan. Jika CG tidak ada, sistem secara otomatis membuat CG.
Anda juga dapat mengeksekusi pernyataan berikut untuk menghapus properti colocation tabel:
ALTER TABLE tbl SET ("colocate_with" = "");Operasi lainnya
Saat menambahkan partisi ke tabel colocation menggunakan pernyataan ADD PARTITION atau mengubah jumlah replika, ApsaraDB for SelectDB memeriksa apakah perubahan tersebut melanggar CGS. Jika ya, perubahan tersebut ditolak.
Penggunaan tingkat lanjut
Item konfigurasi FE
disable_colocate_relocateMenentukan apakah akan menonaktifkan perbaikan replika colocation otomatis di ApsaraDB for SelectDB. Nilai defaultnya adalah false, yang menunjukkan bahwa fitur ini diaktifkan. Parameter ini hanya memengaruhi tabel colocation, bukan tabel biasa.
disable_colocate_balanceMenentukan apakah akan menonaktifkan penyeimbangan replika colocation otomatis di ApsaraDB for SelectDB. Nilai defaultnya adalah false, yang menunjukkan bahwa fitur ini diaktifkan. Item konfigurasi ini hanya memengaruhi tabel colocation, bukan tabel biasa.
Anda dapat memodifikasi item konfigurasi ini secara dinamis. Untuk informasi lebih lanjut, eksekusi pernyataan HELP ADMIN SHOW CONFIG; dan HELP ADMIN SET CONFIG;.
disable_colocate_joinMenentukan apakah akan menonaktifkan fitur colocation join. Nilai defaultnya adalah false, yang menunjukkan bahwa fitur ini diaktifkan.
use_new_tablet_schedulerMenentukan apakah akan mengaktifkan logika penjadwalan replika baru. Nilai defaultnya adalah true, yang menunjukkan bahwa logika ini diaktifkan.
API HTTP RESTful
ApsaraDB for SelectDB menyediakan beberapa operasi API HTTP RESTful terkait colocation join yang dapat digunakan untuk melihat dan memodifikasi CG.
Operasi API ini diimplementasikan pada node FE dan dapat dipanggil dengan mengirim permintaan ke fe_host:fe_http_port. Izin peran admin diperlukan.
Lihat semua informasi colocation kluster. Contoh:
GET /api/colocate Mengembalikan informasi colocation internal dalam format JSON. { "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 }Tandai CG sebagai Stable atau Unstable. Contoh:
Tandai CG sebagai Stable.
DELETE /api/colocate/group_stable?db_id=10005&group_id=10008 Nilai balik: 200Tandai CG sebagai Unstable.
POST /api/colocate/group_stable?db_id=10005&group_id=10008 Nilai balik: 200
Konfigurasikan distribusi data CG.
Anda dapat memanggil operasi API ini untuk secara paksa mengonfigurasi distribusi data CG. Dalam hasil yang dikembalikan, bidang Body berisi urutan bucket yang diwakili oleh array bersarang dan ID node BE tempat tablet dalam bucket berada.
POST /api/colocate/bucketseq?db_id=10005&group_id=10008 Body: [[10004],[10003],[10002],[10003],[10002],[10003],[10003],[10003],[10003],[10002]] Nilai balik: 200CatatanUntuk menggunakan perintah ini, Anda harus menyetel item konfigurasi FE
disable_colocate_relocatedandisable_colocate_balanceke true. Dengan cara ini, perbaikan dan penyeimbangan replika colocation otomatis dinonaktifkan dalam sistem. Jika tidak, replika colocation akan diperbaiki dan diseimbangkan secara otomatis oleh sistem setelah modifikasi manual.
Penyeimbangan dan perbaikan replika colocation
Distribusi replika tabel colocation harus mengikuti aturan distribusi yang ditentukan dalam CGS. Oleh karena itu, penyeimbangan dan perbaikan untuk replika colocation berbeda dari penyeimbangan dan perbaikan untuk tablet biasa.
CG memiliki properti Stable. Jika nilai properti Stable adalah true, semua tablet dari tabel dalam CG tidak sedang berubah, dan fitur colocation join dapat digunakan. Jika nilai properti Stable adalah false, tablet dari beberapa tabel dalam CG sedang diperbaiki atau dimigrasi. Dalam hal ini, join colocation tabel yang terpengaruh akan menurun menjadi join biasa.
Perbaikan replika
Replika hanya dapat disimpan pada node BE tertentu. Jika node BE mati atau didekomisioning, node BE harus diganti dengan node BE lain. ApsaraDB for SelectDB mencari node BE dengan beban paling rendah untuk menggantikan node BE yang tidak tersedia. Setelah penggantian, semua tablet pada node BE asli harus diperbaiki. Selama migrasi, CG ditandai sebagai Unstable.
Penyeimbangan replika
ApsaraDB for SelectDB mendistribusikan tabel colocation secara merata di semua node BE. Replika tabel biasa diseimbangkan pada tingkat replika, sementara replika tabel colocation diseimbangkan pada tingkat bucket, di mana semua replika dalam bucket dipindahkan bersama-sama.
ApsaraDB for SelectDB menggunakan algoritma penyeimbangan sederhana yang mendistribusikan urutan bucket secara merata di semua node BE berdasarkan jumlah replika, bukan ukuran aktual replika.
Algoritma penyeimbangan dan perbaikan replika colocation saat ini mungkin tidak berfungsi dengan baik untuk instans ApsaraDB for SelectDB yang menggunakan penyebaran heterogen. Penyebaran heterogen menunjukkan bahwa node BE tidak memiliki kapasitas disk, jumlah disk, atau tipe disk yang sama. Dalam kasus ini, node BE berkapasitas kecil dan besar mungkin menyimpan jumlah replika yang sama.
Jika CG berada dalam keadaan Unstable, join tabel dalam CG akan menurun menjadi join biasa. Dalam hal ini, performa kueri kluster mungkin menurun secara signifikan. Jika Anda tidak ingin sistem secara otomatis menyeimbangkan replika, Anda dapat menyetel item konfigurasi FE
disable_colocate_balanceke true untuk menonaktifkan penyeimbangan otomatis. Jika penyeimbangan otomatis diperlukan nanti, Anda dapat secara manual menyetel item konfigurasi ke false.