ApsaraDB for SelectDB：Filter runtime

Ikhtisar

Filter runtime secara dinamis dibuat untuk beberapa kueri join pada waktu proses guna mengurangi jumlah data yang dipindai atau dihitung serta mencegah I/O atau transmisi jaringan yang tidak perlu, sehingga mempercepat kueri. Untuk detail lebih lanjut mengenai desain, implementasi, dan kinerja filter runtime, lihat ISSUE 6116.

Istilah

• Tabel kiri: tabel di sebelah kiri kueri join. Tabel ini digunakan untuk operasi probe. Anda dapat menggunakan fitur pengurutan ulang join untuk menyesuaikan urutan.

• Tabel kanan: tabel di sebelah kanan kueri join. Tabel ini digunakan untuk operasi build. Anda dapat menggunakan fitur pengurutan ulang join untuk menyesuaikan urutan.

• Fragment: bagian dari pernyataan SQL. Node frontend (FE) membagi pernyataan SQL menjadi fragmen dan mengirimkan fragmen tersebut ke node backend (BE) dalam kluster terdistribusi untuk dieksekusi. Fragmen dieksekusi pada node BE, dan hasilnya dikumpulkan dan dikembalikan ke node FE.

Cara kerjanya

Filter runtime dibuat secara dinamis selama perencanaan kueri. HashJoinNode mengubah tabel kanan dari join menjadi kondisi filter dan mendorong kondisi filter ke bawah ke OlapScanNode. Selanjutnya, OlapScanNode memangkas tabel kiri berdasarkan kondisi filter selama pemindaian tabel, sehingga secara signifikan mengurangi data yang harus dibaca dan dihitung selama kueri serta meningkatkan kinerja kueri.

Sebagai contoh, kueri join dilakukan antara Tabel T1 dan Tabel T2 menggunakan mode HashJoin. T1 adalah tabel fakta dengan 1.000.000 baris data, sedangkan T2 adalah tabel dimensi dengan 200 baris data. Diagram berikut menunjukkan jumlah data yang perlu dipindai dalam hash join biasa.

|          >      HashJoinNode      <
|         |                          |
|         | 1000000                  | 200
|         |                          |
|   OlapScanNode              OlapScanNode
|         ^                          ^   
|         | 1000000                  | 200
|        T1                          T2
|

Seperti yang ditunjukkan pada diagram sebelumnya, sejumlah besar data di Tabel T1 perlu dipindai, dan sejumlah besar komputasi hash join harus dilakukan.

Jika sistem secara proaktif mengirimkan catatan data yang dipindai di Tabel T2 ke HashJoinNode, HashJoinNode dapat menghasilkan kondisi filter berdasarkan data di Tabel T2, seperti nilai maksimum atau minimum dari data di Tabel T2, atau membuat filter Bloom, dan kemudian mengirimkan kondisi filter ke OlapScanNode yang sedang menunggu untuk memindai Tabel T1. OlapScanNode menerapkan kondisi filter ini dan mengirimkan data yang telah difilter ke HashJoinNode. Dengan cara ini, jumlah operasi probe tabel hash dan overhead jaringan berkurang. Kondisi filter ini disebut filter runtime. Diagram berikut menunjukkan hasilnya.

|          >      HashJoinNode     <
|         |                         |
|         | 6000                    | 200
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 1000000                 | 200
|        T1                         T2
|

Dengan mendorong lebih banyak filter runtime ke mesin penyimpanan, Anda dapat menggunakan indeks untuk memangkas data dalam beberapa kasus, yang secara signifikan mengurangi jumlah data yang dibaca dan waktu pemindaian. Diagram berikut menunjukkan hasilnya.

|          >      HashJoinNode     <
|         |                         |
|         | 6000                    | 200
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 6000                    | 200
|        T1                        T2
|

Berdasarkan analisis sebelumnya, dapat disimpulkan bahwa filter runtime berbeda dari penurunan predikat dan pemangkasan partisi. Filter runtime adalah kondisi yang dihasilkan secara dinamis pada waktu proses. Klausul JOIN ON diurai untuk menentukan ekspresi filter selama kueri, dan ekspresi tersebut disiarkan ke OlapScanNode yang membaca tabel kiri. Hal ini mengurangi jumlah data yang dibaca dan diproses selama kueri, sehingga meningkatkan kinerja kueri secara signifikan.

Jenis filter runtime

ApsaraDB for SelectDB menyediakan jenis filter runtime berikut:

• Predikat IN: Struktur HashSet digunakan untuk mengimplementasikan predikat IN, yang kemudian didorong ke bawah ke OlapScanNode. Keuntungan dari predikat IN adalah bahwa penyaringannya efektif dan cepat. Dalam hal kekurangan, predikat IN hanya berlaku untuk join siaran dan menjadi tidak valid jika ukuran data di tabel kanan melebihi ambang batas. Ambang batasnya adalah 1.024 di ApsaraDB for SelectDB. Jika jumlah baris di tabel kanan melebihi 1.024, predikat IN menjadi tidak valid.

• Filter Bloom: Filter Bloom dibuat berdasarkan data di tabel hash, dan kemudian didorong ke bawah ke OlapScanNode. Filter Bloom dicirikan oleh kegunaannya yang serbaguna, berlaku untuk berbagai tipe data, dan memberikan kinerja yang baik. Kekurangannya adalah konfigurasinya kompleks dan overhead komputasi tinggi.

• Filter MinMax: Setelah rentang ditentukan berdasarkan data di tabel kanan, rentang tersebut dapat didorong ke bawah ke OlapScanNode sebagai filter MinMax. Keuntungan dari filter MinMax adalah overheadnya relatif rendah. Kerugiannya adalah filter MinMax tidak bekerja dengan baik pada kolom non-numerik.

Skenario

Filter runtime digunakan untuk mengoptimalkan join antara tabel besar dan tabel kecil. Jika jumlah data di tabel kiri terlalu kecil atau jumlah data di tabel kanan terlalu besar, filter runtime mungkin tidak berfungsi sesuai harapan. Filter runtime berlaku untuk skenario yang memenuhi persyaratan berikut:

• Tabel kiri besar, dan tabel kanan kecil. Hal ini karena pembuatan filter runtime membutuhkan biaya komputasi, termasuk overhead memori.

• Hasil join antara tabel kiri dan tabel kanan sedikit. Ini menunjukkan bahwa filter runtime dapat menyaring sebagian besar data di tabel kiri.

Penggunaan

Gunakan filter runtime dalam kueri

Secara default, fitur filter runtime diaktifkan di ApsaraDB for SelectDB. Ketika ApsaraDB for SelectDB memproses kueri pengguna, ApsaraDB for SelectDB secara otomatis menghasilkan predikat IN dan filter Bloom berdasarkan tabel dan pernyataan kueri untuk optimasi kueri.

Parameter filter runtime

runtime_filter_mode

Parameter ini menentukan rentang transmisi filter runtime di antara unit-unit terkecil eksekusi kueri.

Nilai valid: nilai numerik 0, 1, dan 2 atau string mnemonik yang sesuai yaitu OFF, LOCAL, dan GLOBAL. Nilai default: 2 (GLOBAL).

• LOCAL: Kebijakan ini bersifat relatif konservatif. Filter runtime yang dibuat hanya dapat digunakan dalam fragmen yang sama pada unit eksekusi kueri terkecil yang sama. Dalam hal ini, produsen filter runtime, yaitu HashJoinNode yang membuat filter, dan konsumen filter runtime, yaitu OlapScanNode yang menggunakan filter, berada dalam fragmen yang sama. Kebijakan ini biasanya digunakan untuk skenario seperti join siaran biasa.

• GLOBAL: Kebijakan ini bersifat relatif agresif dan memungkinkan Anda menggabungkan filter runtime serta mentransfernya ke fragmen berbeda pada unit eksekusi yang berbeda melalui jaringan. Sebagai contoh, produsen dan konsumen filter runtime dapat berada di fragmen yang berbeda selama join shuffle.

Selain skenario yang sesuai dengan kebijakan LOCAL, kebijakan GLOBAL dapat digunakan untuk mengoptimalkan kueri dalam berbagai skenario yang lebih luas. Namun, pada beberapa join shuffle, overhead pembuatan dan penggabungan filter runtime melebihi peningkatan kinerja kueri. Dalam situasi ini, Anda dapat mengubah kebijakan menjadi LOCAL. Jika kinerja kueri join dalam kluster tidak mendapatkan manfaat dari filter runtime, Anda dapat menonaktifkan fitur tersebut dengan menyetel parameter ini ke OFF.

Untuk informasi lebih lanjut mengenai alasan dan kebijakan penggabungan filter runtime saat membuat dan menerapkan filter runtime di fragmen yang berbeda, lihat ISSUE 6116.

runtime_filter_type

Jenis filter runtime.

Nilai valid: nilai numerik 1, 2, 4, 8, dan 16 atau string mnemonik yang sesuai IN, BLOOM_FILTER, MIN_MAX, IN_OR_BLOOM_FILTER, dan BITMAP_FILTER. Nilai default: 8 (IN_OR_BLOOM_FILTER). Gunakan koma (,) untuk memisahkan beberapa nilai dan tutup mereka dalam tanda kutip ("). Anda juga dapat menjumlahkan beberapa nilai numerik yang mewakili tipe. Contoh:

set runtime_filter_type="BLOOM_FILTER,IN,MIN_MAX";

Pengaturan di atas setara dengan pengaturan berikut:

set runtime_filter_type=7;

Tabel berikut menjelaskan jenis filter runtime.

runtime_filter_wait_time_ms

Durasi waktu menunggu filter runtime.

Nilai valid: bilangan bulat. Unit: ms. Nilai default: 1.000.

Setelah fitur filter runtime diaktifkan, OlapScanNode untuk tabel kiri akan menunggu selama periode waktu tertentu untuk setiap filter runtime yang ditugaskan sebelum pemindaian data dimulai. Sebagai contoh, jika OlapScanNode ditugaskan tiga filter runtime, maka OlapScanNode akan menunggu hingga 3.000 ms.

Pembuatan dan penggabungan filter runtime memerlukan waktu. OlapScanNode berupaya mendorong filter runtime yang diterima selama durasi menunggu ke mesin penyimpanan. Setelah durasi tersebut berakhir, OlapScanNode menggunakan filter runtime yang telah diterima untuk memindai data.

Jika filter runtime tiba setelah OlapScanNode mulai memindai, OlapScanNode tidak mendorong filter runtime ke bawah ke mesin penyimpanan. Sebagai gantinya, OlapScanNode melakukan penyaringan menggunakan ekspresi pada data yang telah dipindai dari mesin penyimpanan berdasarkan filter runtime. Filter runtime tidak diterapkan pada data yang telah dipindai, sehingga ukuran data perantara yang dihasilkan lebih besar daripada ukuran optimal, tetapi hal ini mencegah cracking parah.

Jika kluster sibuk dengan banyak kueri intensif sumber daya atau yang memakan waktu lama dieksekusi, Anda dapat memperpanjang durasi menunggu untuk mencegah kueri kompleks kehilangan peluang optimasi. Namun, jika beban kluster ringan dan sebagian besar kueri kecil hanya memerlukan beberapa detik untuk dieksekusi, Anda dapat mengurangi durasi menunggu guna mencegah peningkatan latensi sebesar 1 detik per kueri.

runtime_filters_max_num

Jumlah maksimum filter Bloom yang dapat diterapkan pada setiap kueri.

Nilai valid: bilangan bulat. Nilai default: 10.

Jumlah filter Bloom dibatasi karena filter Bloom lebih mahal untuk dibuat dan diterapkan dibandingkan dengan filter MinMax atau predikat IN.

Jika jumlah filter Bloom yang dihasilkan melebihi batas atas, filter Bloom dengan selektivitas lebih tinggi akan dipertahankan karena diharapkan dapat menyaring lebih banyak baris. Parameter ini mencegah filter Bloom mengonsumsi terlalu banyak overhead memori, yang berpotensi menyebabkan masalah.

Selektivitas = (Kardinalitas HashJoinNode/Kardinalitas anak kiri HashJoinNode)
-- Kardinalitas yang diperoleh oleh node FE tidak akurat. Selektivitas yang dihitung untuk filter Bloom menyimpang dari selektivitas aktual. Akibatnya, filter Bloom mungkin dipertahankan secara acak.

Parameter terkait panjang filter Bloom

Parameter runtime_bloom_filter_min_size, runtime_bloom_filter_max_size, dan runtime_bloom_filter_size digunakan untuk menentukan ukuran struktur data filter Bloom dalam byte.

Nilai valid: bilangan bulat.

Filter Bloom yang dibuat oleh HashJoinNode harus memiliki panjang yang sama agar dapat digabungkan. Oleh karena itu, node FE menghitung panjang filter Bloom selama perencanaan kueri.

Jika kardinalitas dalam statistik tabel kanan dapat diperoleh, sistem akan memperkirakan ukuran optimal filter Bloom berdasarkan kardinalitas tersebut dan membulatkannya ke pangkat dua terdekat, yang merupakan nilai log dengan basis 2. Namun, jika kardinalitas tabel kanan tidak dapat diperoleh, panjang filter Bloom default yang ditentukan oleh parameter runtime_bloom_filter_size akan digunakan. Parameter runtime_bloom_filter_min_size dan runtime_bloom_filter_max_size digunakan untuk membatasi panjang minimum dan maksimum filter Bloom yang digunakan.

Filter Bloom dengan ukuran lebih besar lebih efisien dalam memproses set input berkardinalitas tinggi, namun mengonsumsi lebih banyak memori. Sebagai contoh, jika Anda perlu menyaring kolom dengan kardinalitas tinggi, seperti kolom yang berisi jutaan nilai unik, Anda dapat menaikkan nilai parameter runtime_bloom_filter_size dan melakukan pengujian benchmark. Hal ini membantu meningkatkan akurasi penyaringan oleh filter Bloom serta mencapai peningkatan kinerja yang diharapkan.

Efektivitas filter Bloom bergantung pada distribusi data kueri. Oleh karena itu, Anda hanya perlu menyesuaikan panjang filter Bloom untuk kueri tertentu, bukan secara global. Penyesuaian parameter ini hanya diperlukan saat mengonfigurasi beberapa kueri join yang memakan waktu lama antara tabel besar.

Lihat filter runtime yang dihasilkan untuk kueri

Anda dapat mengeksekusi pernyataan EXPLAIN untuk melihat rencana kueri dari sebuah kueri. Rencana kueri mencakup informasi mengenai klausa JOIN ON yang digunakan oleh setiap fragmen, serta komentar terkait filter runtime yang dihasilkan dan digunakan dalam fragmen tersebut. Dengan demikian, Anda dapat menentukan apakah filter runtime telah diterapkan pada klausa JOIN ON yang diperlukan.

• Contoh komentar tentang filter runtime yang dihasilkan dalam fragmen: runtime filters: filter_id[type] <- table.column.

• Contoh komentar tentang filter runtime yang digunakan dalam fragmen: runtime filters: filter_id[type] -> table.column.

Dalam contoh berikut, filter runtime dengan ID RF000 digunakan untuk kueri.

CREATE TABLE test (t1 INT) DISTRIBUTED BY HASH (t1) BUCKETS 2;
INSERT INTO test VALUES (1), (2), (3), (4);

CREATE TABLE test2 (t2 INT) DISTRIBUTED BY HASH (t2) BUCKETS 2;
INSERT INTO test2 VALUES (3), (4), (5);

EXPLAIN SELECT t1 FROM test JOIN test2 where test.t1 = test2.t2;
+-------------------------------------------------------------------+
| Explain String                                                    |
+-------------------------------------------------------------------+
| PLAN FRAGMENT 0                                                   |
|  OUTPUT EXPRS:`t1`                                                |
|                                                                   |
|   4:EXCHANGE                                                      |
|                                                                   |
| PLAN FRAGMENT 1                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test`.`t1`  |
|                                                                   |
|   2:HASH JOIN                                                     |
|   |  join op: INNER JOIN (BUCKET_SHUFFLE)                         |
|   |  equal join conjunct: `test`.`t1` = `test2`.`t2`              |
|   |  runtime filters: RF000[in] <- `test2`.`t2`                   |
|   |                                                               |
|   |----3:EXCHANGE                                                 |
|   |                                                               |
|   0:OlapScanNode                                                  |
|      TABLE: test                                                  |
|      runtime filters: RF000[in] -> `test`.`t1`                    |
|                                                                   |
| PLAN FRAGMENT 2                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test2`.`t2` |
|                                                                   |
|   1:OlapScanNode                                                  |
|      TABLE: test2                                                 |
+-------------------------------------------------------------------+
-- Baris yang berisi "runtime filters" menunjukkan bahwa predikat IN dengan ID RF000 dihasilkan oleh 2:HASH JOIN dari PLAN FRAGMENT 1. 
-- Nilai-nilai `test2`.`t2` hanya diketahui pada waktu proses. 
-- Predikat IN digunakan oleh 0:OlapScanNode untuk menyaring data yang tidak perlu ketika node membaca `test`.`t1`. 

SELECT t1 FROM test JOIN test2 where test.t1 = test2.t2; 
-- Mengembalikan dua baris hasil: [3, 4].

-- Anda dapat melihat bagaimana kueri dilakukan secara internal dengan menggunakan profil kueri. Anda harus menjalankan perintah set enable_profile=true; untuk mengaktifkan profil. 
-- Profil mencakup apakah setiap filter runtime didorong ke bawah, waktu tunggu, dan durasi total dari waktu OLAP_SCAN_NODE disiapkan hingga runtime filter diterima. 
RuntimeFilter:in:
    -  HasPushDownToEngine:  true
    -  AWaitTimeCost:  0ns
    -  EffectTimeCost:  2.76ms

-- Selain itu, Anda dapat melihat efek penyaringan dan waktu yang dikonsumsi setelah runtime filter didorong ke bawah di bagian OLAP_SCAN_NODE profil. 
    -  RowsVectorPredFiltered:  9.320008M  (9320008)
    -  VectorPredEvalTime:  364.39ms

Aturan perencanaan untuk filter runtime

• Filter runtime hanya dapat dihasilkan untuk kondisi ekuivalen dalam klausa JOIN ON. Kondisi NULL-safe tidak termasuk karena nilai null di tabel kiri mungkin disaring.

• Filter runtime tidak dapat didorong ke bawah ke tabel kiri dari outer join kiri, full outer join, atau anti-join.

• Ekspresi sumber atau ekspresi target tidak boleh berupa konstanta.

• Ekspresi sumber tidak boleh setara dengan ekspresi target.

• Tipe ekspresi sumber tidak boleh setara dengan HLL atau BITMAP.

• Filter runtime hanya dapat didorong ke bawah ke OlapScanNode.

• Ekspresi target tidak boleh berisi ekspresi pemeriksaan NULL, seperti COALESCE, IFNULL, atau CASE. Jika klausa JOIN ON dari join lapisan atas lainnya berisi ekspresi pemeriksaan NULL dan filter runtime dihasilkan, hasilnya mungkin salah setelah filter runtime untuk outer join ini didorong ke bawah ke tabel kiri dari outer join.

• Kolom ekuivalen harus ada di tabel asli untuk kolom dalam ekspresi target.

• Konduksi kolom tidak didukung dalam skenario berikut:

  • Jika klausa JOIN ON berisi A.k = B.k and B.k = C.k, C.k hanya dapat didorong ke bawah ke B.k, bukan A.k.

  • Jika klausa JOIN ON berisi A.a + B.b = C.c dan A.a serta B.a adalah kolom ekuivalen sehingga A.a dapat dikonduksikan ke B.a, A.a dapat diganti dengan B.a, dan sistem mencoba mendorong filter runtime ke B. Jika A.a dan B.a bukan kolom ekuivalen, filter runtime tidak dapat didorong ke bawah ke B karena ekspresi target harus terikat hanya pada satu tabel kiri.

• Tipe ekspresi sumber dan ekspresi target harus sama karena filter Bloom adalah filter berbasis hash. Jika tipe berbeda, sistem mencoba mengonversi tipe ekspresi target menjadi tipe ekspresi sumber.

• Filter runtime yang dihasilkan oleh PlanNode.Conjuncts tidak dapat didorong ke bawah. Berbeda dengan eqJoinConjuncts dan otherJoinConjuncts dari HashJoinNode, PlanNode.Conjuncts menghasilkan filter runtime yang dapat menyebabkan hasil tidak benar berdasarkan pengujian. Sebagai contoh, jika subkueri IN diubah menjadi join, klausa JOIN ON yang dihasilkan secara otomatis disimpan dalam PlanNode.Conjuncts. Dalam hal ini, filter runtime dapat menyebabkan baris hilang dalam hasil.