列ストアインデックスにおける TopK 演算子の実装 - PolarDB

ディープページングクエリ（ページ番号が非常に大きく、一方で結果セットは小さい）は、分析用データベースにおいて著しいパフォーマンス劣化を引き起こします。PolarDB for MySQL では、インメモリー列指向インデックス（IMCI）機能に内蔵される Sort/TopK 演算子を再設計し、ディープページングを効率的に処理できるようにしました。具体的には、メモリ内実行には SIMD 加速型の自己鋭敏化入力フィルターを、ディスクベースのフォールバックには ZoneMap を活用したプルーニングを採用しています。100 GB の TPC-H データセットを用いた評価では、再設計された演算子によるディープページングクエリの実行時間は 7.72 秒であり、ClickHouse の 23.07 秒、MySQL の 353.15 秒と比較して大幅な高速化を実現しています。

ディープページングの課題

業務システムにおける一般的なクエリパターンとして、レコードの絞り込み、特定カラムによる並べ替え、および結果のページングがあります。SQL では次のように表現されます。

ORDER BY 列 LIMIT オフセット, カウント

1 ページあたり 100 件のレコードを取得する場合：

ページ	クエリ
ページ 1	`ORDER BY カラム LIMIT 0, 100`
ページ 10,001	`ORDER BY カラム LIMIT 1000000, 100`

2 番目のクエリでは、返却されるレコード数がわずか 100 件であるにもかかわらず、K（オフセット＋カウント）は 1,000,100 となります。この K と結果セットサイズの非対称性こそが ディープページング を定義するものであり、標準的な TopK アルゴリズムの弱点を露呈させます。

業界のアプローチとそのトレードオフ

実際の運用では、主に以下の 3 つのアプローチが採用されています。いずれも結果セット外のデータに対する操作を削減することを目指しますが、ディープページング下ではそれぞれ異なる形で限界に直面します。

優先度付きキュー（ヒープベース TopK）

システムはサイズ K の最大ヒープを維持します。スキャンされる各レコードについて、そのレコードが上位 K 件に含まれるかどうかを判定し、該当する場合はヒープの先頭要素を置き換え、ヒープを再構成します。全スキャン終了後、ヒープには上位 K 件のレコードが格納されます。

これは K が小さい場合には良好に動作しますが、K が 1,000,100 のような大規模な値になると、以下の 2 つの問題が顕在化します。

メモリ圧迫：ヒープがメモリに収まらない可能性があります。
キャッシュ効率の低下：ヒープ操作はランダムなメモリアクセスを必要とします。K が大きいと、頻繁なキャッシュミスが発生し、スループットが低下します。

マージソート中の切り捨て

PolarDB、ClickHouse、SQL Server、DuckDB などで採用されています。システムはソート中にソート済みの「ラン」を生成し、各マージ済みランを オフセット + リミット 個に切り捨てます。[オフセット, オフセット + リミット) の範囲に含まれるレコードのみが有効であるため、全データをソートする必要はありません。

Truncation based on an offset and a limit during merge sort

この手法は、メモリが不足した場合にディスクへスケールできます。ただし、切り捨ての恩恵は、ソート済みランの長さがある程度十分に長い場合にのみ得られます。K が非常に大きいディープページングでは、マージ初期段階のソート済みランはしばしば オフセット + リミット より短く、切り捨てが有効になる前に全データセットのソートが必要となる場合があります。

自己シャープ入力フィルター

Goetz Graefe によって初めて提案され、ApsaraDB for ClickHouse でも採用されています。システムは カットオフ値 — 上位 K 件の結果に出現しうるレコードの上限値 — を維持します。カットオフ値を超えるレコードは、ソート済みランへの入力前に除外されます。各ソート済みランの構築後、そのランの長さが K より大きい場合、K 番目のレコードの値が新しいカットオフ値となります。新しいカットオフ値は常に旧カットオフ値以下であるため、フィルターは継続的に鋭敏化（シャープ化）されます。

例（K = 3）：

バッチ	ソート済みラン	カットオフ値
1	`(1, 2, 10, 15, 21)`	10
2	`(2, 3, 5, 6, 8)`（カットオフ値 10 で事前フィルタリング）	5
3	`(1, 2, 3, 3, 3)`（カットオフ値 5 で事前フィルタリング）	3

ヒープ操作とは異なり、自己鋭敏化フィルターはカットオフフィルタリングおよびソート済みランの蓄積の両方において、メモリを逐次アクセスします。これにより、ヒープメンテナンスに伴うランダムアクセスのペナルティを回避できます。

なぜディープページングが両アプローチを破綻させるのか

LIMIT 1000000, 100 の場合、K は 1,000,100 ですが、返却されるレコードはわずか 100 件です。これは各アプローチの限界を明確に浮き彫りにします。

ヒープベース：1,000,100 個のヒープエントリを維持すると、たとえメモリが十分に確保されていても、極めて大きなランダムメモリアクセスオーバーヘッドが発生します。
マージソートの切り捨て：ソート初期段階で、1,000,100 件を超える長さのソート済みランが生成されることは稀であり、切り捨てが有効になる前に全データセットのソートが必要になります。

説明

本トピックにおける「十分なメモリ」とは、K 個のレコードを管理するデータ構造がメモリに収まることを意味します — 入力データ全体がメモリに収まることを意味するものではありません。本トピックで説明するシナリオでは、入力データは利用可能なメモリをはるかに上回ります。

さらに、以下の 2 つの設計要件が追加されます。

単一の統合アルゴリズムが、浅いページングから深いページングまで、明確なしきい値を設けずに両方を処理できる必要があります。
システムは、静的な構成ではなく、利用可能なメモリ量に応じて動的にメモリ実行またはディスク実行を選択する必要があります。

ソリューション設計

PolarDB IMCI の再設計された Sort/TopK 演算子は、既存のアプローチの長所を統合するとともに、ディープページング下での失敗点を解消するよう設計されています。

メモリーアルゴリズム：SIMD 加速の自己シャープ入力フィルター

メモリが十分に確保されている場合、IMCI は優先度付きキューの代わりに自己鋭敏化入力フィルターを採用します。大規模な K において優先度付きキューを用いない理由は以下のとおりです。

ヒープメンテナンス中のランダムメモリアクセスにより、K が大きい場合のパフォーマンスが劣化します。
ヒープサイズは K に等しくなるため、メモリ圧迫は K に対して線形に増加します。

セルフシャープニングフィルターは、両方の問題を回避します：

カットオフフィルタリングおよびソート済みランの蓄積の両方において、メモリを逐次アクセスします。
フィルターは任意の K で正しく動作し、境界条件を設けることなく、浅いページングから深いページングまでをカバーします。

SIMD 加速：カットオフフィルタリングは単純・反復的・高頻度であり、各レコードを現在のカットオフ値と比較する処理です。IMCI では、単一命令・複数データ（SIMD）命令を活用してこれを加速します。SIMD は、同一の比較処理を複数のレコードに対して並列に適用します。このフィルターはテーブルスキャンの述語フィルターと同一の式評価基盤を再利用するため、追加のコードパスは不要です。

ディスクアルゴリズム：ZoneMap ベースのプルーニングを伴うマージソート

メモリが不足している場合、IMCI は切り捨てを伴うマージソートを採用します。ディスク上で自己鋭敏化フィルターを用いない理由は以下のとおりです。

蓄積されたソート済みランをディスクに保存し、事前マージ中に外部ソートを実行すると、大量のディスク I/O が発生します。
K が大きい場合、有用なカットオフ値が確立される前に、[オフセット, オフセット + リミット) の範囲外の大量のデータが事前マージで処理される可能性があります。

切り捨てを伴うマージソートはこれらの問題を回避し、さらに ZoneMap ベースのプルーニングにより、ソート済みランの最小値／最大値統計情報を活用して、結果に寄与しないランをスキップすることで、I/O をさらに削減します。

ZoneMap プルーニングの仕組み：

各ソート済みランは、その最小値および最大値を保持します。バリア 値を用いて、すべてのソート済みランを以下の 3 種類に分類します。

種別	条件	例
タイプ A	最小値と最大値の両方がバリア未満	Run1、Run2
タイプ B	最小値がバリア未満、かつ最大値がバリア超	Run3
タイプ C	最小値と最大値の両方が > barrier	Run4、Run5

以下の 2 つのプルーニング規則により、結果に影響を与えないソート済みランを除外します。

タイプ A とタイプ B のレコード総数が オフセット より小さい場合、タイプ A のすべてのレコードは [0, オフセット) の範囲に収まります。以降のマージからタイプ A を除外します。
タイプ A のレコード総数が オフセット + リミット を超える場合、タイプ C のすべてのレコードは [オフセット + リミット, N) の範囲に収まります。以降のマージからタイプ C を除外します。

プルーニングの手順：

各ソート済みランの最小値および最大値を含む ZoneMap を構築します。
タイプ A ＋タイプ B のレコード数が オフセット より小さいという条件を満たす、可能な限り大きな バリア 1 を求めます。
タイプ A のレコード数が オフセット + リミット を超えるという条件を満たす、可能な限り小さな バリア 2 を求めます。
バリア 1 およびバリア 2 を用いて、対応するソート済みランを以降のマージから除外します。

動的アルゴリズム選択

固定の K しきい値を用いるのではなく、IMCI はフォールバック機構を用いてアルゴリズムを動的に選択します。

常にメモリアルゴリズムから開始してください。
実行中にメモリが十分に確保され続ける場合、計算をメモリ内で完了します。
メモリが不足した場合（例：K 件を超えるレコードを含むソート済みランをキャッシュするのに十分なスペースがない、または事前マージを完了するのに十分なスペースがない）にフォールバックをトリガーします。
- メモリ内のソート済みランの最小値／最大値を収集し、ZoneMap を構築します。
- それらのソート済みランをディスクに保存します。
- 残りの計算をディスクアルゴリズムに切り替えます。
ディスクアルゴリズムを用いて計算を完了します。

両アルゴリズムは同一のデータ構造を用いるため、フォールバック時にデータの再構成は不要です。メモリフェーズで蓄積されたソート済みランは、精度の損失なしにディスクアルゴリズムで直接活用されます。

エンジニアリング最適化

遅延マテリアライズ

ソート済みランを構築する際、IMCI は ORDER BY で参照される行 ID およびカラム／式のみをマテリアライズします。出力カラムは、TopK 結果セットが確定した後にストレージから取得します。

これにより、以下の 2 つの観点でオーバーヘッドが削減されます。

行 ID はコンパクトであるため、同一のメモリ予算でより多くのレコードを保持でき、メモリアルゴリズムが適用可能な範囲が拡大します。
TopK 計算中には Copy および Swap 操作を通じてレコードの並び替えが頻繁に行われます。行 ID のみをマテリアライズすることで、これらの操作におけるレコード単位のコストが最小限に抑えられます。

トレードオフとして、ソート後の行 ID を用いた出力カラムの取得はランダム I/O を伴う可能性があります。しかし、ディープページングでは実際の結果セットは小さく（例：100 件）、このランダム I/O は無視できるレベルです。

計算のプッシュダウン

自己鋭敏化フィルター実行中に、現在のカットオフ値はテーブルスキャン演算子へ新たな述語としてプッシュダウンされます。テーブルスキャンは既存のプルーナー基盤を活用してこの述語を適用し、データが TopK 演算子に到達する前に、パックまたは行グループ単位でフィルタリングを行います。

これにより、以下の 2 つの観点でオーバーヘッドが削減されます。

I/O 削減：カットオフ値を超えるレコードのみを含むパックまたは行グループは、完全にスキップされます。
計算負荷削減：フィルタリングされたパックまたは行グループは、上位レイヤーの演算子によって処理されません。

テスト結果

以下のクエリを 100 GB の TPC-H データセットに対して実行しました。

SELECT
    l_orderkey,
    sum(l_quantity)
FROM
    lineitem
GROUP BY
    l_orderkey
ORDER BY
    sum(l_quantity) DESC
LIMIT
    1000000, 100;

システム	実行時間
PolarDB IMCI	7.72 秒
ClickHouse	23.07 秒
MySQL	353.15 秒