WINDOW JOINを並列化・ベクトル化する方法

QuestDBは、取引フロアからミッションコントロールまで、要求の厳しいワークロードに対応するオープンソースの時系列データベースです。超低レイテンシー、高インジェストスループット、多層ストレージエンジンを提供します。ParquetとSQLのネイティブサポートにより、データはポータブルでAI対応となり、ベンダーロックインは発生しません。 トレーディングデスクで常に発生するワークロードを考えてみましょう。実行されたすべてのトレードについて、そのトレードの前後1秒間の平均買い気配値と売り気配値を付加します。専用の演算子がない場合、2つの結合（ウィンドウ開始時の繰越クォートに対するASOF JOINと、ウィンドウ内の行に対する範囲結合）が必要で、これらをUNION ALLで結合し、GROUP BYで集約します。 ```sql -- QuestDBのタイムスタンプはマイクロ秒なので、1_000_000は1秒です。 WITH prevailing AS ( -- ウィンドウ開始（トレードタイムスタンプ - 1秒）に対してASOFマッチングを行います。 -- トレードタイムスタンプ自体ではありません。 SELECT t.orig_ts ts, t.symbol, p.bid, p.ask FROM ( (SELECT (timestamp - 1000000) AS ts, symbol, timestamp AS orig_ts FROM trades) TIMESTAMP(ts) ) t ASOF JOIN prices p ON p.sym = t.symbol ), in_window AS ( SELECT t.timestamp ts, t.symbol, p.bid, p.ask FROM trades t JOIN prices p ON p.sym = t.symbol WHERE p.ts > t.timestamp - 1000000 AND p.ts <= t.timestamp + 1000000 ) SELECT ts, symbol, avg(bid) avg_bid, avg(ask) avg_ask FROM (SELECT * FROM prevailing UNION ALL SELECT * FROM in_window) GROUP BY ts, symbol; ``` これは機能しますが、単純な操作にしてはSQLが多すぎます。ASOF JOINと範囲JOINは、同じ質問の2つの半分に答えているにもかかわらず、pricesテーブルを独立して走査します。また、範囲JOINは、プランナーにsymでハッシュ化させ、その後一致するすべてのペアをBETWEEN述語に対して再フィルタリングさせることになります。tsに対する外側のGROUP BYはハッシュ集約であり、(ts, symbol)ペアごとに1行を具体化する必要があります。これは、私たちのテストデータでは5000万グループになります。ここには、オプティマイザが融合、きれいに並列化、またはベクトル化できるものは何もありません。 WINDOW JOINは、QuestDBが1つのテーブルを別のテーブルの各行の周りの時間ウィンドウで集約するための専用構文です。同じクエリを専用演算子で記述すると次のようになります。 ```sql SELECT t.*, avg(p.bid) avg_bid, avg(p.ask) avg_ask FROM trades t WINDOW JOIN prices p ON p.sym = t.symbol RANGE BETWEEN 1 second PRECEDING AND 1 second FOLLOWING; ``` これで演算子は何をしているのかを認識します。結合の左側（LHS - ここではtrades）の各行について、LHSのタイムスタンプの周りの[lo, hi]ウィンドウ内にタイムスタンプが収まる右側（RHS - prices）の行を見つけ、一致するシンボルキーに制限し、一連の集計関数でそれらを削減します。これを高速化することは、2つの部分に集約されます。LHSに対するデータレベルの並列処理と、我々がSAMPLE BYのために既に出荷しているSIMD集約カーネルがウィンドウのスライス上で変更なしで実行できるように、値を連続バッファにコピーする低カーディナリティの高速パスです。5000万行のtradesテーブルと1億5000万行のpricesテーブルを結合してTimescale、DuckDB、ClickHouseに対してベンチマークを行ったところ、並列+SIMDパスはQuestDB自身のシングルスレッドフォールバックよりも5.0倍速く、ClickHouseの最適な書き換えよりも25倍速く実行されました。 データレベルの並列処理 QuestDBはデータを時系列でパーティション分割された追記専用の列ファイルに保存します。クエリエンジンはそれらをページフレームのシーケンスとして読み取ります。ページフレームは、ファイルページに直接マッピングされる連続した列状のメモリのスラブです。フィルタリングと集約は両方ともこの粒度で機能します。ページフレームはワーカー スレッドへのディスパッチの単位です。WINDOW JOINも同じモデルに従います。LHSテーブルはページフレームにスライスされ、各ワーカーはフレームを受け取り、そのフレーム内のすべてのLHS行の集計結果を生成する責任を負います。そのためには、フレームがカバーするすべてのウィンドウのユニオン内に収まるRHS行が必要です。具体的には、LHSタイムスタンプがtLoからtHiまで実行され、[-w_lo, +w_hi]ウィンドウを持つフレームの場合、ワーカーは[tLo - w_lo, tHi + w_hi]のRHS行を必要とします。このスライスを安価に特定できることが、並列プランを可能にするものであり、それを可能にするのがQuestDBのストレージレイアウトです。両方のテーブルの行はディスク上で指定されたタイムスタンプ順に保持されているため、任意の時間範囲のRHSスライスは、LHS行ごとのスキャンではなく、ワーカーごとの単一の二分探索に集約されます。次に、LHSフレームに存在する結合キーに対して、ワーカーはRHSスライスから小さなインメモリインデックスを構築します。これは、キーごとのRHSタイムスタンプのリストと、集計する値のキーごとの配列です。このインデックスが構築されると、LHS行に対する内部ループは、ウィンドウの下限と上限に対する2つの二分探索にすぎず、その後、結果の連続範囲に対する集約が行われます。両方の二分探索は単調に前進するため、同じフレーム内の行全体で償却されます。 だいたい次のようになります。 ``` LHS page frames ┌─────┬─────┬─────┬─────┬─────┐ │ F0 │ F1 │ F2 │ F3 │ ... │ └──┬──┴──┬──┴──┬──┴──┬──┴─────┘ │ │ │ │ ┌──────┴┐ ┌──┴───┐ │worker0│ │worker1│ ... │ workers pulled from a shared pool └───┬───┘ └──┬────┘ │ │ one frame at a time │ │ ┌───────┴───────┐│ ┌───────┴───────┐ │ RHS slice + ││ │ RHS slice + │ │ per-key ││ │ per-key │ │ value arrays ││ │ value arrays │ └───────┬───────┘│ └───────┬───────┘ │ │ ▼ ▼ ▼ ┌───────────────────────────────────┐ │ for each LHS row in frame: │ │ bsearch lo, bsearch hi, │ │ aggregate(value_array[lo:hi]) │ └───────────────────────────────────┘ ``` 削減ステップはフレームごとであり、ロックフリーです。ワーカーはフレーム間で可変状態を共有しません。これが外部の並列処理をカバーします。内部ループ、つまり各LHS行がRHSスライスを集約する場所で、SIMDが登場します。 低カーディナリティキーの高速パス ベクトル化可能な集約と組み合わせた低カーディナリティ等価結合の場合、RHSの値をキーごとの連続バッファにコピーすることで、SAMPLE BY用に既に出荷しているSIMDカーネルをウィンドウのスライス上で変更なしで実行できます。SIMDカーネルは、合計、平均、最小、最大、およびその他いくつかの手作業でチューニングされた関数で、ループイテレーションごとに8つのdoubleを処理しますが、連続配列に対してのみです。ウィンドウ内のRHS行は連続しておらず、ページフレーム全体に散らばっており、混合された列を介してスレッド化されています。最初の実装では、一致する行を1つずつ走査し、各行に対してcomputeBatchを呼び出していました。これは正しかったのですが、これらのカーネルを未使用のままにしていました。この追加パスが有効になるクエリの形式は、実際には最も一般的なものです。低カーディナリティの等価結合（典型的なp.sym = t.symbolケース）と集約された数値列です。異なるキーが少ない場合、キーごとの値バッファはキャッシュに収まり、それらを具体化するコストは、各ウィンドウのスライスをベクトル化された集約に渡すことで十分に回収されます。 高速パスは2つの条件でゲートされます。クエリがベクトル化された実装を持つ集約（数値型に対する合計/平均/最小/最大/カウントのファミリー）を使用する必要があることと、結合条件がプランナーが認識する低カーディナリティの形式（現在、単一シンボル等価結合）に適合する必要があることです。両方が当てはまる場合、ワーカーはRHSスライスを1回反復して、キーごとのタイムスタンプインデックスを構築し、集約された列を同じパスでキーごとの値バッファにコピーします。バッファは型付けされ、パックされています。シンボルAAPLのすべてのbid値は、RHSタイムスタンプ順に1つの連続したdoubleのブロックに格納されます。条件を満たさないクエリの場合、バッファコピーを完全にスキップし、スカラーのcomputeBatchループにフォールバックします。内部ループがLHS行に到達すると、キーごとの値バッファはSIMDが望むように既に配置されています。2つの二分探索によってバッファ内の[rowLo, rowHi)が得られ、そのスライスを直接ベクトル化された集約に渡します。 ```java for (int i = 0; i < groupByFuncCount; i++) { groupByFunctions.getQuick(i).computeBatch( value, // running aggregate state bufferStart(i) + typeSize * rowLo, // pointer into the per-key buffer (int) (rowHi - rowLo), // slice length 0 ); } ``` RHSスライスに対する追加のパスは無料ではありません。各RHS列を2回読み取ります（1回は値をキーごとのバッファに入れるため、もう1回は集約がそれらを消費するため）。一般的なプランではこれは悪影響を及ぼすでしょう。しかし、QuestDBの集約ループは、入力が連続していればSIMDによって制限され、そうでなければ各行のcomputeBatchの呼び出しによって制限されます。doubleの場合、AVX2は我々にab