一文剖析PolarDB列存索引（IMCI）如何實作極緻TopK查詢性能？

作者簡介

餘南龍（花名：顧紳），來自阿裡雲瑤池資料庫-PolarDB新型存儲引擎組，從事PolarDB列存索引（In Memory Column Index，IMCI）的研發工作，有任何問題和咨詢請發郵件至：[email protected]。

導讀

在海量資料上求TopK是一個很經典的問題，特别是衍生出的深翻頁查詢，給分析型資料庫帶來了很大的挑戰。在本文中，我們将介紹雲原生關系型資料庫PolarDB的列存索引（In Memory Column Index，IMCI）特性如何應對這樣的挑戰，并最終在性能上顯著領先單表查詢性能優異的ClickHouse。

1.背景

業務系統中普遍存在這樣一種場景：根據給定條件篩選一批記錄，這些記錄按使用者指定的條件排序，以分頁的方式展示。例如，篩選出某個商家在售的商品，按商品銷量排序，以分頁的方式展示。

上述場景，反映在資料庫上，往往以ORDER BY column LIMIT n, m這樣的TopK查詢實作。例如，假設業務系統中每頁展示100條記錄，可以通過ORDER BY column LIMIT 0, 100來展示第1頁，通過ORDER BY column LIMIT 1000000, 100來展示第10001頁。

在沒有索引的情況下，這樣的查詢在資料庫中往往通過很經典的基于堆的TopK算法來實作：在記憶體中維護一個大小為K的堆，堆頂為目前排在第K位的記錄，算法執行過程中會實時維護這個堆，保證堆中的記錄始終是排在前K位的。當翻頁較淺時（如上文中展示第1頁），K較小，上述基于堆的TopK算法非常高效。

然而，業務場景中也存在翻頁較深的場景（下文中我們簡稱“深翻頁”），例如上文中展示第10001頁。該場景下的K非常大，記憶體中可能無法緩存大小為K的堆，也就無法使用上述方式獲得查詢結果。即便記憶體充裕，由于維護堆的操作訪存是亂序的，當堆非常大時，經典TopK算法的訪存效率較差，最終的性能表現也差強人意。

PolarDB IMCI最初也采用了上述方式來實作這樣的查詢，并在記憶體不足以緩存大小為K的堆時，退化為全表排序後取相應位置的記錄，是以在深翻頁時的性能表現也不是非常理想。為此，我們分析了深翻頁場景的特點和傳統方案存在的問題，并調研了相關研究和工業界實作，重新設計了PolarDB IMCI的Sort/TopK算子。在測試場景中，重新設計的Sort/TopK算子顯著提升了PolarDB IMCI在深翻頁場景的性能表現。

2.業界方案調研

TopK是一個非常經典的問題，存在多種方案來高效地實作TopK查詢，這些方案的核心都在于減少對非結果集資料的操作。已經在工業界中應用的方案主要有如下三種：

2.1 基于Priority Queue的TopK算法

在背景部分已簡單介紹，不再贅述。

2.2 歸并排序時基于offset和limit做truncate

當記憶體不足以緩存大小為K的Priority Queue時，一些資料庫會使用歸并排序來處理TopK查詢（如PolarDB IMCI，ClickHouse，SQL Server，DuckDB）。因為TopK查詢隻需要擷取排在第[offset, offset + limit)位的記錄，是以在每一次merge sorted run時，不需要對所有資料做排序，而是僅輸出長度為offset + limit的新的sorted run即可。上述merge時的truncation可以在保證結果正确性的同時減少對非結果集資料的操作。

2.3 Self-sharpening Input Filter

該方案最初是在Goetz Graefe的論文中提出的，ClickHouse目前采用了這種方案。該方案在執行過程中會維護一個cutoff value，并且保證大于cutoff value的記錄一定不會出現在TopK的結果集中。在生成new sorted run時，方案會使用目前的cutoff value對資料進行過濾。在生成new sorted run之後，如果K小于new sorted run的長度，則會使用new sorted run中第K條記錄替換目前cutoff value。由于new sorted run中的資料都是經過old cutoff value過濾的，是以必定有new cutoff value <= old cutoff value，即cutoff value是一個不斷sharpening的值。最後隻需要合并這些過濾後的sorted run即可得到結果集。

通過一個簡單的例子來說明上述算法：假設目前TopK查詢中K=3，讀取第一批資料後生成的sorted run為(1, 2, 10, 15, 21)，則cutoff value更新為10。接下來使用cutoff value=10過濾第二批資料，生成的第二個sorted run為(2, 3, 5, 6, 8)，則cutoff value更新為5。然後讀取并過濾第三批資料，生成的第三個sorted run為(1, 2, 3, 3, 3)，則cutoff value更新為3。依此類推，不斷sharpen cutoff value進而在接下來過濾更多的資料。

如果TopK查詢中K大于單個sorted run的長度，該方案會積累足夠的sorted run（包含的記錄數量大于K），然後對這些sorted run提前進行merge，進而獲得cutoff value。接下來會使用cutoff value進行過濾并繼續積累足夠的sorted run，進而獲得更小的cutoff value，依此類推。整個執行過程和K小于單個sorted run的情況是類似的，差別在于需要merge足夠的sorted run才能獲得cutoff value。

3.問題分析

深翻頁是TopK問題中一個較為特殊的場景，特殊之處在于所求的K特别大，但實際結果集很小。例如上文中展示第10001頁的例子，對于ORDER BY column LIMIT 1000000, 100，K=1,000,100，但最終結果集隻包含100條記錄。該特點會給上一節中所述方案帶來如下挑戰：

• 當記憶體充足時，如果采用基于Priority Queue的TopK算法，則需要維護一個非常大的Priority Queue，隊列操作對記憶體的通路操作是亂序的，訪存效率較差，影響算法實際運作的性能。
• 當記憶體不足時，如果使用歸并排序并基于offset和limit做truncate，則在歸并排序的前期階段，sorted run的長度可能小于offset + limit，無法進行truncate，所有資料都将參與排序，truncate的實際效果受到影響。

注：本文中的記憶體充足指的是，算法中用于管理至少K條記錄的資料結構可以在執行記憶體中緩存，而不是TopK查詢的輸入資料可以在執行記憶體中緩存。實際上本文讨論的場景，TopK查詢的輸入資料都是遠大于執行記憶體的。

另外，從系統設計的角度上看，設計深翻頁的解決方案時還應考慮如下兩點：

• 是否采用不同方案來實作深翻頁和淺翻頁？如果需要使用不同的方案來處理兩種場景，如果判斷深淺翻頁的界線？
• 如何根據可用執行記憶體的大小自适應地選擇記憶體算法或磁盤算法？

4.方案設計

4.1 總體設計

綜合上述調研和分析，我們基于現有的成熟方案，針對深翻頁帶來的挑戰，重新設計了PolarDB IMCI的Sort/TopK算子：

記憶體充足時：

• 采用Self-sharpening Input Filter的設計，避免訪存效率的問題。
• 并在此基礎上利用SIMD指令提升過濾效率。
• 深淺翻頁均采用該記憶體算法，不需要判斷深淺翻頁的界線。

記憶體不足時：

• 采用歸并排序時基于offset和limit做truncate的方案。
• 并在此基礎上利用ZoneMap在歸并排序的前期階段做pruning，盡可能地減少對非結果集資料的操作。

動态選擇記憶體磁盤算法：

不依賴固定的門檻值來選擇使用記憶體算法或磁盤算法，而是在執行過程中根據可用執行記憶體的大小，動态調整所用算法。

由于Self-sharpening Input Filter和歸并排序時基于offset和limit做truncate的方案在上一節中已經介紹，是以接下來僅介紹選擇這兩種方案的原因，并介紹利用SIMD指令提升過濾效率、利用ZoneMap做pruning以及動态選擇記憶體磁盤算法的部分。

4.2 SIMD Accelerated Self-sharpening Input Filter

在記憶體充足時，我們直接采用了Self-sharpening Input Filter的設計，主要基于兩個原因：

• Self-sharpening Input Filter不管是使用cutoff value進行過濾，還是pre-merge，通路記憶體的模式都是順序的，可以避免Priority Queue訪存效率的問題。
• 該設計無論翻頁深淺都具有優異的性能，在應用時不需要考慮深淺翻頁的界線。

實際上，Self-sharpening Input Filter在某種程度上和基于Priority Queue的算法是類似的，cutoff value類似堆頂，都用于過濾後續讀取的資料，兩者的不同之處在于，基于Priority Queue的算法會實時更新堆頂，而Self-sharpening Input Filter則将資料積累在sorted run中，以batch的方式更新cutoff value。

使用cutoff value進行過濾是Self-sharpening Input Filter中很重要的過程，涉及資料比較，操作簡單重複但非常頻繁，是以我們使用SIMD指令來加速這一過程。由于使用cutoff value過濾和TableScan中使用Predicate過濾是類似的，是以在具體實作中我們直接複用處理Predicate的表達式，提升過濾的效率，減少了計算TopK的時間。

4.3 Zonemap-based Pruning

在記憶體不足時，我們采用歸并排序，并基于offset和limit做truncate，主要原因如下：

• 如果在記憶體不足時繼續使用Self-sharpening Input Filter的設計，那麼就需要将積累的sorted run落盤，并且在pre-merge時同樣使用外排序算法，産生大量的讀寫磁盤的操作，相對于記憶體充足場景下的Self-sharpening Input Filter有額外的開銷。當K非常大時，pre-merge時的外排序可能還會涉及大量非結果集資料，因為我們最終隻需要擷取排在第[offset, offset + limit)位的記錄，而不關心排在第[0, offset)位的記錄。
• 在這種場景下，我們可以使用歸并排序，在生成sorted run的階段僅将sorted run落盤，然後使用統計資訊進行pruning，避免不必要的讀寫磁盤的操作，也可以盡可能地避免對非結果集資料的操作。

我們以下圖為例來說明使用統計資訊進行pruning的原理。下圖中，箭頭表示數軸，代表sorted run的矩形的左右兩端在數軸上對應的位置表示sorted run的min/max值，Barrier表示pruning所依賴的一個門檻值。

任意一個Barrier可以将所有sorted run分為三類：

• 類型A：min value of sorted run < Barrier && max value of sorted run < Barrier，如上圖中Run1，Run2。
• 類型B：min value of sorted run < Barrier && max value of sorted run > Barrier，如上圖中Run3。
• 類型C：min value of sorted run > Barrier && max value of sorted run > Barrier，如上圖中Run4，Run5。

對于任意一個Barrier，如果類型A和類型B中的資料量 < TopK查詢中的offset，那麼類型A中的資料必然排在第[0, offset)位，類型A中的sorted run可以不參與後續的merge。

對于任意一個Barrier，如果類型A中的資料量 > TopK查詢中的offset + limit，那麼類型C中的資料必然排在第[offset + limit, N)位，類型C中的sorted run可以不參與後續的merge。

根據上述原理，使用統計資訊進行pruning的具體流程如下：

• 建構包含sorted run的min/max資訊的Zonemap。
• 基于Zonemap尋找一個盡可能大的Barrier1滿足類型A和類型B中的資料量 < TopK查詢中的offset。
• 基于Zonemap尋找一個盡可能小的Barrier2滿足類型A中的資料量 > TopK查詢中的offset + limit。
• 使用Barrier1和Barrier2對相關sorted run進行pruning。

4.4 動态選擇記憶體磁盤算法

我們的方案中記憶體算法和磁盤算法不同，如果使用一個固定的門檻值來作為選擇記憶體算法或磁盤算法的依據（比如K小于門檻值時使用記憶體算法，否則使用磁盤算法），那麼針對不同的可用執行記憶體就需要設定不同的門檻值，帶來了人工幹預的開銷。

是以我們設計了一個簡單的回退機制，可以在執行過程中根據可用執行記憶體的大小，動态調整所用算法：

• 無論可用執行記憶體有多大，首先嘗試以記憶體算法計算TopK
• 在記憶體算法的執行過程中，如果記憶體始終都是充足的，那麼直接使用記憶體算法完成整個計算過程
• 在記憶體算法的執行過程中，如果出現記憶體不足的情況（例如，K比較大時，可用執行記憶體不足以緩存足夠的sorted run使其包含的記錄數量大于K，或者可用執行記憶體不足以完成pre-merge的過程），那麼執行回退機制
• 回退機制：采集記憶體中已積累的sorted run的min/max資訊，以便于後續使用Zonemap進行pruning，然後将sorted run落盤，這些sorted run将會參與磁盤算法的計算過程
• 執行完回退機制後，使用磁盤算法完成整個計算過程

由于記憶體算法和磁盤算法采用相同的資料組織格式，是以回退機制并不需要對資料進行重新組織，開銷較小。另外，記憶體算法隻會過濾非結果集的資料，是以直接使用記憶體算法已積累的sorted run參與磁盤算法的計算過程不會有正确性的問題。

4.5 番外1：延遲物化

延遲物化是一個工程實作方面的優化，指的是在生成sorted run時僅物化RowID和ORDER BY相關的表達式（列），在計算出TopK的結果集後，再根據結果集中的RowID從存儲上擷取查詢需要輸出的列。延遲物化相比于在生成sorted run時就物化查詢需要輸出的所有列有兩個優勢：

• 物化RowID的空間占用更小，在可用執行記憶體一定的情況下，可以使用記憶體算法處理更大的資料量
• 計算TopK的過程需要調整資料順序，涉及對資料的Copy/Swap。如果在生成sorted run時就物化查詢需要輸出的所有列，則計算過程中對一條記錄的Copy/Swap需要對每一列都進行相應操作，帶來很大的overhead。而如果僅物化RowID，則可以降低Copy/Swap的代價。

延遲物化的不足之處在于根據結果集中的RowID從存儲上擷取查詢需要輸出的列時，可能會産生一些随機IO。根據我們的分析，深翻頁場景雖然K特别大，但實際結果集很小，是以使用延遲物化時随機IO産生的overhead較小。

4.6 番外2：計算下推

應用Self-sharpening Input Filter時，我們會将不斷更新的cutoff value下推至table scan算子，作為SQL中一個新的predicate，在table scan算子擷取資料時根據這個新的predicate，複用pruner對pack（或稱為row group）進行過濾。

計算下推可以從兩個方面提升TopK查詢的性能：

1. 減少IO：table scan時避免讀取僅包含非結果集資料的pack/row group

2. 減少計算：被過濾的pack/row group中的資料将不再參與table scan上層算子的後續計算

5.實驗結果

我們在 TPC-H 100G 的資料集上對我們的方案進行簡單的驗證：

select
    l_orderkey,
    sum(l_quantity)
from
    lineitem
group by
    l_orderkey
order by
    sum(l_quantity) desc
limit
    1000000, 100;