星環科技分佈式搜索引擎 Transwarp Scope 查詢優化技術解讀
概述
分佈式數據庫系統是物理上分佈而邏輯上集中的數據庫系統，為了提高性能並最大限度地減少資源爭用，其被廣泛用于海量數據處理的場景中。在這種情況下，**數據庫查詢速度是系統性能表現的決定性指標。**而由於數據分佈在不同節點上並通過網絡通信在不同節點間傳輸，分佈式查詢的處理流程比單機集中式查詢更加複雜。與傳統的集中式數據庫系統相比，對分佈式查詢的構建和優化需要同時考慮CPU、I/O成本以及網絡通信成本。
本文旨在從分佈式集群視角，對Transwarp Scope查詢相關原理和優化技術進行較為全面的解讀。
整體流程
對於分佈式搜索引擎來說，一般情況下， 一次查詢涉及到多台機器的多個分片，正確的結果需要匯總多個分片的各自結果之後才能獲得。因此，無論是Transwarp Scope還是es，其查詢過程都包括一個Merger的角色存在，這個Merger在es中是Coordinating node, 而在NS中是Client。而整個流程以Phase劃分，可以分為DFS, QUERY, FETCH三類Phase。
專用詞與明確
分片一般也被成為shard/tablet
Phase簡介
**DFS Phase:**統計數據收集階段，對於文本信息來說，其在單個text中的freq等信息是準確的。但是類似與idf這樣的全局統計信息而言，每個分片只能明確該文本在分片內部的idf，也就是一個局部的idf。如果不進行全局idf綜合統計，僅以local idf計算score，得出來的分數是不準確的。所以，在很多對打分結果準確性要求較高的場景下， 都會有dfs這個階段進行全局統計信息匯總。當然，也因為多了這個階段，相應地響應速度也會受到影響。
**Query Phase:**查詢階段， 根據client輸入的信息在各個分片上找到匹配的文檔集合。這一階段基本上會做3件事情：**match（匹配），score(打分），local_sort(本地排序）。**各個分片會將匹配的doc_id集合，返回給Merger節點。Merger節點會對各個分片彙報上來的doc_set進行merge + global_sort。然後根據client設定的from,size, 從global_result_set中cut出[from, from + size]，再進行下一階段。
**Fetch階段：**獲取doc原始內容的phase。該Phase會根據Query Phase結束後的global_result_set向各個分片索要目標的doc_set, 包括文檔的原始內容以及可能的某些再加工內容，比如Highlight。由於要真正地加載文檔內容[source]，所以 Fetch階段會產生比較大的io負載（page cache缺失的情況下）。因此，如果是一些大寬表（500列+）的場景，其行數據size比較大的情況下，更可行的方式其實是把ES/NS作為一張純粹的Index Table，即只對目標列設置索引 + 對外表主鍵列存儲source。如此，當query階段階段執行完之後，進行fetch phase的時候只需要加載rowkey這一列的值，再global_result_set中的外表rowkey值去外部行數據庫中拿到原始內容，這樣做能明顯減輕es/ns集群的存儲和讀寫壓力。
從整體上來看，查詢部分基本的架構原則就是**用各種不同的Phase拼接執行不同的查詢動作，即Compose Phases into Action.**如上圖示意。
查詢操作類型簡介
查詢操作本身可以按照如上圖這樣進行細分， 各自含義如下表：
點查詢圖解
點查，或者說排序查詢是核心功能，舉例如下。
對於一張成績表schema=(姓名、數學成績、語文成績、 英語成績），整張表格有3個tablet, 現在要獲取全部成績的前3名，則整體流程如下圖所示。
如上圖所示，即為單次點查詢的原理示意圖。在Query階段，所有Tablet都將自己的數學成績的前3名匯總給Merger, Merger進行全局排序之後，發現真正的前三名是tablet1的11,4號, tablet3的4號。然後在Fetch階段，將這些對應doc標識發送給tablet1, tablet3, 再拿到對應的文檔原始內容，這裡有2處細節值得提及。
二維全局rowKey。在上圖所示數據分佈體系中， 用以表示全局唯一row或者doc的標識是一個（tablet, docId)的二元組，及tablet1和tablet3都有doc4, 但2者沒有關係。
上圖所示是在全局數據本身無序分佈的情況下進行排序查詢的流程，如果對數據本身就是有序分佈的， 那麼流程會大大簡化，這一點會在後續內容中討論。
分頁查詢
所謂分頁查詢，或者掃描，就是當結果集比較大的時候，分成多次rpc返回結果。
1.並發分頁查詢
所謂並發分頁，如下圖所示，就是client同時向所有的tablet發送request。這種情況下，每一頁的具體流程以排序/不排序分可以對應上文點查/輕量點查。
2.順序分頁查
所謂順序分頁查，如上右圖所示，指的是**每一頁並不是將rpc同時發送給所有tablet, 而是對所有tablet進行逐個掃描，**tablet1，tablet2，tablet3。這種掃描方式的明顯好處就是大幅度減少了rpc的數量，降低了集群整體負載。又因為每個rpc只有1個tablet的結果，所以也不需要進行多個tablet結果的合併，降低了client的處理負載。
3.動態超分頁查詢
對於查詢操作來說，緩存是很有效果的優化措施。尤其是對一些單線程掃描全表的應用，其客戶端內存可能大量閑置。這種場景下，合理地使用客戶端內存作為緩存來優化查詢速度，就是動態超分頁查詢的思想，其基本原理仍以是否排序分2種情況討論。
對於不排序場景，緩存的策略很簡單，如上圖所示，就是一次rpc取n個整頁，放在客戶端內存中備用，從第二次之後，直接從本地內存中取用。而為了在保證穩定性的基礎上儘可能地加快scan，對於N這個值採用二進制試探+回退的方式進行控制。即最開始只取一頁，然後是2,4,8,16。在這個過程中，保存Page的平均大小和已經使用的內存量，綜合jvm內存大小，從而計算出**下一次scan最大能拿多少頁。**從而讓N回退，降低client內存壓力，保證客戶端程序的穩定。在實際使用中，一般會限定客戶端jvm_heap的8%作為scan_cache的上限。
此外，為了避免N過大導致延遲過長問題，當單次時間超過一定閾值的時候，N也會相應回退，避免讓客戶端感覺到太明顯的卡頓。
對於排序場景，緩存不能像no sort場景下這麼魯莽。因為排序本身存在一個回收率（1/s)的問題，即前文所提及的，3個shard， 取前3名，則實際上需要拿到3×3=9行數據，最終有效返回卻是只有3行，所以回收率=1/3。在超大集群場景下，一張大表可能有500+個shard，此時如果貿然地擴大N倍，一次性從server端取回4000-5000個page，很有可能造成client劇烈的gc, 影響程序穩定。因此，排序場景下客戶端緩存，Transwarp Scope採用了客戶端復用的方式來進行。
如上圖所示，續前文所述排序場景下QueryThenFetch的流程，當第一輪Fetch結束之後，真正的全局前三被fetch之後，剩餘的(圖中標紅的）T1-doc5, T2-doc11-doc22-doc15和T3-doc32-doc5，**一定是下一輪全局排序的備選項，**所以下一輪query階段並不需要再從每個tablet拿3個了，對於tablet1，只需要再拿2個，tablet3再拿3個， 而tablet2則不需要在round2進行query階段。在超大表的場景下，以500shard, page_size=1000為例， 那麼98%的row都可以在客戶端進行復用，從而大大減少了rpc次數和server端查詢排序的開銷。當然，實際生產環境中也要考慮到rpc_size的問題，配合整頁緩存一起使用。
查詢優化的基礎：分區
分區是最直接有效的查詢加速手段，尤其是對於超大規模的集群的大表（1000+ shard， 單表50T)這樣的場景，如果能在查詢真正開始之前將搜索範圍縮小到全量數據集合的1-2%，即10-20shard，500G-1000G這個規模。那麼實際表現出來的性能就是百毫秒到秒級級別。
最常見的兩種分區機制，是Range分區和Hash分區。
1.Range分區
如上圖所示，即為Range分區的基本原理示意圖，所有的row, 按照age這一列進行劃分partition。當select (14,19)之間的row時，就可以通過partition prune將查詢限制在一個tablet1上，從而避免了全表搜索，大幅度減少了集群負載。
另外，在排序場景下，如果要獲取全局age最大的5個row， 那麼在已有範圍分區的情況下，只需要對tablet1和tablet2的數據進行排序， 填滿結果集即可，避免了對Tablet1的無效查詢和排序。
又如上圖所示，在Range分區的基礎上，配合分片內部的預排序，就可以保證整張表格數據的全局有序。此時的升序掃表動作，就轉換成了順序依次掃描每個shard，從而完全避免了分片級別/表級別的排序動作，極大提升速度。
2.Hash分區
hash分區的即是根據指定列的hash值進行分區，如上圖所示，當搜索age=13的所有row時， 由於13的hash值是1，所以搜索可以被剪枝到tablet1上，從而避免了tablet0, tablet2的無效搜索。
3.混合分區
分區的實際意義在於，通過對數據進行**物理分佈上的隔離，**從而查詢時進行大片的剪枝。在實際使用中，真實數據可能有很多的細化查詢需求，需要對數據進行不止一層或一種分區，這就對應了混合分區的概念。
如上圖所示，數據全集採用2層分區進行物理隔離，在shard級別，按照age進行範圍分區。在每個shard內部，再按照rid進行hash分區。那麼對於如上圖sql, 查詢操作能立刻通過partiton prune將範圍縮小到shard1的P0 Parition上，查詢範圍大大縮小。
注意，在同一個物理隔離級別上，只能有一個Range分區標準，否則會有歧義導致無法排序。而Hash分區可以組合多個。
總結
本文分別從客戶端和集群的視角，介紹了Transwarp Scope的查詢的基本流程、基本原理、實現方式以及不同類型分區對查詢速度帶來的優化。
