Fluentd向けApache Arrowプラグインについて

構想は半年ほど前？ここ一ヶ月ほど集中して開発に取り組んでいた、Fluentd向けApache Arrowプラグインがようやく動くようになったので、今回はこちらのモジュールについてご紹介します。

そもそもPG-Stromは、IoT/M2M領域で大量に発生するデータを高速に処理できますというのがセールスポイントで、GPU-Direct SQLはじめ、各種の機能によってそれを実現しているワケですが、実際に運用する際には、発生したデータを『どうやってSQLで処理できるようDBにインポートするか？』という問題があります。
例えば、PostgreSQLに一行ずつINSERTするというのも一つの解です。ただし、単純なI/Oに比べると、DBへの書き込みはどうしても処理ボトルネックになりがちです。

そこで、大量に収集するログデータを、少ない時間ロスで（つまり一時ファイルに保存したデータを再度DBにインポートするなどの手間をかける事なく）検索や集計できる状態に持って行くために、以下のように Fluentd から Apache Arrow 形式ファイルを出力し、それを直接 PG-Strom から読み出すというスキームを作りました。

Fluentdとは Treasure Data の古橋貞之氏によって開発されたログ収集ツールで、SyslogのようなサーバログからIoT/M2M機器のデバイスログに至るまで、多種多様なログデータを集積・保存するために事実上のスタンダードとして利用されているソフトウェアです。
Ruby で記述されたプラグインを追加する事で、ログデータの入出力や加工を自在にカスタマイズすることができます。

arrow-file プラグイン

Fluentdのプラグインにはいくつかカテゴリがあり、外部からログを受け取るInputプラグイン、ログを成形するParserプラグイン、受信したログを一時的に蓄積するBufferプラグイン、ログを出力するOutputプラグイン、などの種類があります。

Fluentdがログを受け取ると、Input/Parserプラグインによってログは共通の内部形式へと変換されます。
これは、ログの振り分けに利用できる識別子のtag、ログのタイムスタンプtimeおよび、生ログを整形した連想配列であるrecordです。
Bufferプラグインは、ログを Output プラグインに渡して書き出すまでの間、一時的にこれを保持します。これにより、渡すまでの間、一時的にこれを保持します。これにより、複数レコードをまとめて書き込む事で出力のパフォーマンスが向上したり、障害時のリトライを単純化する事ができます。
最後に、OutputプラグインがBufferプラグインから渡されたログをそれぞれのプラグインに応じた出力先に書き出します。

今回、作成したfluent-plugin-arrow-fileモジュールは、この Output プラグインに相当するもので、出力先として指定されたファイルに Apache Arrow ファイル形式で書き込みます。

インストール

ここでは、Treasure Data社の提供する Fluentd の安定板 td-agent を利用します。
また、arrow-fileプラグインのインストールにはrake-compilerモジュールも必要ですので、予めインストールしておきます。

Fluentdのインストール詳細については、こちらを参照してください。

$ curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent4.sh | sh
$ sudo /opt/td-agent/bin/fluent-gem install rake-compiler

次に、PG-Stromのソースコードをダウンロードし、fluentd ディレクトリ以下の物件をビルドします。

$ git clone https://github.com/heterodb/pg-strom.git
$ cd pg-strom/fluentd
$ make TD_AGENT=1 gem
$ sudo make TD_AGENT=1 install

Fluentdのプラグインがインストールされている事を確認するため、以下のコマンドを実行します。
fluent-plugin-arrow-fileが表示されていれば、インストールは成功です。

動かしてみる

では実際に動かしてみる事にします。

簡単な例として、ローカルのApache Httpdサーバのログを監視し、それをフィールド毎にパースしてApache Arrow形式ファイルに書き込みます。
<source>で/var/log/httpd/access_logをデータソースとして指定しているほか、apache2のParseプラグインを用いて、host, user, time, method, path, code, size, referer, agentの各フィールドを切り出しています。
（これは公式サイトのExampleからのコピペです）

後半の<match>以下がarrow-fileプラグインの設定です。
pathで出力先を指定しています。ここでは/tmp/mytest%Y%m%d.%p.arrowと記述していますが、書き込み時に、%Y、%m、%dはそれぞれ年、月、日に、%pはプロセスのPIDに置き換えられます。
schema_defsでは、出力先 Apache Arrow ファイルのスキーマ構造を定義します。
tsがタイムスタンプ、host、method、path、referer、agentがそれぞれ文字列（Utf8）で、codeとsizeはInt32で設定しています。

また、バッファに関してはもう少し大きなサイズを指定すべきですが、ここでは動作確認のため比較的小さなサイズ（4MB、200行）で、かつ書き出しのインターバルを10sに指定しています。実際にはPG-StromがGPU-Direct SQLを発動するのに向いたサイズのバッファサイズを指定する事をお勧めします。（例えばデフォルト値の 256MB など）

<source>
  @typetail
  path /var/log/httpd/access_log
  pos_file /var/log/td-agent/httpd_access.pos
  tag httpd
  format apache2
  <parse>
    @typeapache2
    expression /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>(?:[^\"]|\\.)*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>(?:[^\"]|\\.)*)" "(?<agent>(?:[^\"]|\\.)*)")?$/
    time_format %d/%b/%Y:%H:%M:%S %z
  </parse>
</source>

<match httpd>
  @typearrow_file
  path /tmp/mytest%Y%m%d.%p.arrow
  schema_defs "ts=Timestamp[sec],host=Utf8,method=Utf8,path=Utf8,code=Int32,size=Int32,referer=Utf8,agent=Utf8"
  ts_column "ts"
  <buffer>
    flush_interval 10s
    chunk_limit_size 4MB
    chunk_limit_records 200
  </buffer>
</match>

さて、td-agentを起動します。

sudo systemctl start td-agent

以下のように、Apache Httpdのログが path で設定した /tmp/mytest%Y%m%d.%p.arrow が展開された先である /tmp/mytest20220124.3206341.arrow に書き出されています。

中身を見てみると、それっぽい感じになっているのが分かります。

$ arrow2csv /tmp/mytest20220124.3206341.arrow --head --offset 300 --limit 10
"ts","host","method","path","code","size","referer","agent"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/js/theme_extra.js",200,195,"http://buri/docs/ja/fluentd/","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/js/theme.js",200,4401,"http://buri/docs/ja/fluentd/","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/img/fluentd_overview.png",200,121459,"http://buri/docs/ja/fluentd/","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/search/main.js",200,3027,"http://buri/docs/ja/fluentd/","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/fonts/Lato/lato-regular.woff2",200,182708,"http://buri/docs/ja/css/theme.css","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/fonts/fontawesome-webfont.woff2?v=4.7.0",200,77160,"http://buri/docs/ja/css/theme.css","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/fonts/RobotoSlab/roboto-slab-v7-bold.woff2",200,67312,"http://buri/docs/ja/css/theme.css","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:42","192.168.77.95","GET","/docs/ja/fonts/Lato/lato-bold.woff2",200,184912,"http://buri/docs/ja/css/theme.css","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:43","192.168.77.95","GET","/docs/ja/search/worker.js",200,3724,"http://buri/docs/ja/fluentd/","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
"2022-01-24 06:13:43","192.168.77.95","GET","/docs/ja/img/favicon.ico",200,1150,"http://buri/docs/ja/fluentd/","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"

これを PG-Strom のArrow_Fdwを用いてPostgreSQLにマッピングしてみます。

postgres=# IMPORT FOREIGN SCHEMA mytest
           FROM SERVER arrow_fdw INTO public
           OPTIONS (file '/tmp/mytest20220124.3206341.arrow');
IMPORT FOREIGN SCHEMA

postgres=# SELECT ts, host, path FROM mytest WHERE code = 404;
         ts          |     host      |         path
---------------------+---------------+----------------------
 2022-01-24 12:02:06 | 192.168.77.73 | /~kaigai/ja/fluentd/
(1 row)

postgres=# EXPLAIN SELECT ts, host, path FROM mytest WHERE code = 404;
                                  QUERY PLAN
------------------------------------------------------------------------------
 Custom Scan (GpuScan) on mytest  (cost=4026.12..4026.12 rows=3 width=72)
   GPU Filter: (code = 404)
   referenced: ts, host, path, code
   files0: /tmp/mytest20220124.3206341.arrow (read: 128.00KB, size: 133.94KB)
(4 rows)

生成された Apache Arrow ファイルを外部テーブルとしてマッピングし、これをSQLから参照しています。

Fluentd側で成形されたログの各フィールドを参照する検索条件を与える事ができます。 上記の例では、HTTPステータスコード404のログを検索し、1件がヒットしています。

まとめ

以上のように、Fluentdで受け取ったログを Apache Arrow 形式ファイルとして書き出し、それをそのまま、つまり改めてデータをインポートする事なく PostgreSQL から参照する事ができる事が分かりました。
これは、ログ集積系のシステムから、検索・分析系のシステムへデータを移送するという手間なしにSQL処理を発行できる事を意味するほか、例えば、もう使わなくなった古いログデータをOS上でコピーして退避すれば、それだけでアーカイブ作業が終了します。（Apache Arrow形式の場合、ファイルにスキーマ構造も内包しているため、後になって『あれ？このテーブルのDDLは？』なんて事もありません）

加えて、Fluentdのarrow-fileプラグインはタイムスタンプに統計情報を付加する事もできるため、検索条件に日付時刻範囲の絞り込みを含むケースでは大幅な検索時間の高速化を見込むことができます。
kaigai.hatenablog.com

課題としては、現状、まだ「動くようになった」というレベルですので、実際に Fluentd のインスタンスを何台も立てて検証済みである、という訳ではありません。ですので、この辺はぜひ『一緒に検証しませんか？』という方がいらっしゃいましたら、お声がけいただければと思います。

半精度浮動小数点型（Float2）について

このエントリはPostgreSQL Advent Calendar 2021に参加しています。

実は、現在開発中の別の機能について書きたかったのですが、間に合いませんでした。反省。
そこで、急遽ネタを用意したのが、反省、はんせい、はんせいど…ふどうしょうすうてん（ﾋﾟｺｰﾝ!!

という事で、PostgreSQLで利用できる半精度浮動小数点型（float2）の事について書こうと思います。

半精度浮動小数点とは

Cで言えば32bitのfloatに対して64bitのdoubleを倍精度と呼ぶように、floatの半分である16bitの浮動小数点フォーマットが半精度浮動小数点形式です。
もちろん、データ量が少ない分、表現できる範囲や精度に制限はあるのですが、一方で必要なストレージ領域は小さく、またSIMDやGPUといったベクトル演算を行う場合にはメモリバスを有効活用できることから、機械学習の分野などで活用が進んでいます。

型名	ビット幅	指数部	仮数部
倍精度	64bit	11bit	52bit
単精度	32bit	8bit	23bit
半精度	16bit	5bit	10bit

PostgreSQL で float2 型を定義する

さて、この半精度浮動小数点型ですが、PostgreSQL本体ではまだ対応していません。
そもそもがApache Arrowとのデータ交換に必要であったので PG-Strom 拡張モジュールの一部として作成したモノ・・・ではあるのですが、別にこれ自体はGPUやNVMEを必要とするものではありませんので、これ単体を切り出して利用する事も可能です。

github.com

x86_64のCPUでは今のところ半精度浮動小数点をそのまま計算する事はできませんので、内部的にはこれをfloat4やfloat8に変換した上で演算を行っています。GPU側であればfloat2のまま計算する事もできるのですが。

float2 -> float4/float8 への変換はそれほど難しいことではありません。指数部も仮数部もより幅が広くなる方に動くので、float2で表現できる値は確実にfloat4/float8へと変換する事ができます。

例えば、float2からfloat4への変換は、このような単純なビット操作だけで可能です。

static inline float
fp16_to_fp32(half_t fp16val)
{
    uint32_t    sign = ((uint32_t)(fp16val & 0x8000) << 16);
    int32_t     expo = ((fp16val & 0x7c00) >> 10);
    int32_t     frac = ((fp16val & 0x03ff));
    uint32_t    result;

    if (expo == 0x1f)
    {
        if (frac == 0)
            result = (sign | 0x7f800000);   /* +/-Infinity */
        else
            result = 0xffffffff;            /* NaN */
    }
    else if (expo == 0 && frac == 0)
        result = sign;                      /* +/-0.0 */
    else
    {
        if (expo == 0)
        {
            expo = FP16_EXPO_MIN;
            while ((frac & 0x400) == 0)
            {
                frac <<= 1;
                expo--;
            }
            frac &= 0x3ff;
        }
        else
            expo -= FP16_EXPO_BIAS;

        expo += FP32_EXPO_BIAS;

        result = (sign | (expo << FP32_FRAC_BITS) | (frac << 13));
    }
    return int_as_float(result);
}

一方で、float4/float8 -> float2 への変換は、表現可能な範囲を超えると+/-Infに発散して島唄め、注意が必要です。

postgres=# select 65000::float2;
 float2
--------
 64992
(1 row)

postgres=# select 66000::float2;
  float2
----------
 Infinity
(1 row)

テーブル定義で float2 を用いる

半精度浮動小数点データ型は PG-Strom に含まれているため、以下のようにCREATE EXTENSIONコマンドでインストールする事ができます。

postgres=# CREATE EXTENSION pg_strom ;
CREATE EXTENSION
postgres=# \dT float2
        List of data types
   Schema   |  Name  | Description
------------+--------+-------------
 pg_catalog | float2 |
(1 row)

早速、テーブルを定義して、データを流し込んでみます。

postgres=# CREATE TABLE fp16_test (
              id int,
              a  float2,
              b  float2,
              c  float2,
              d  float2,
              e  float2,
              f  float2,
              g  float2,
              h  float2
            );
postgres=# insert into fp16_test (select x, 1000*random(),
                                            1000*random(), 
                                            1000*random(),
                                            1000*random(),
                                            1000*random(),
                                            1000*random(),
                                            1000*random(),
                                            1000*random() from generate_series(1, 4000000) x);
INSERT 0 4000000

一方、比較のために倍精度浮動小数点で同じようにテーブルを定義してみます。

postgres=# CREATE TABLE fp64_test (
              id int,
              a  float8,
              b  float8,
              c  float8,
              d  float8,
              e  float8,
              f  float8,
              g  float8,
              h  float8
            );
CREATE TABLE
postgres=# insert into fp64_test (select x, 1000*random(),
                                            1000*random(), 
                                            1000*random(),
                                            1000*random(),
                                            1000*random(),
                                            1000*random(),
                                            1000*random(),
                                            1000*random() from generate_series(1, 4000000) x);
INSERT 0 4000000

あたり前の話ではありますが、大きくサイズが変わってきます。
（ただし、タプルのヘッダ 24バイト分は必ずくっつくので、単純に4倍違う、とはなりませんが）

postgres=# \d+
                                        List of relations
 Schema |       Name        |       Type        | Owner  | Persistence |    Size    | Description
--------+-------------------+-------------------+--------+-------------+------------+-------------
 public | fp16_test         | table             | kaigai | permanent   | 199 MB     |
 public | fp64_test         | table             | kaigai | permanent   | 386 MB     |

インデックスを張ることもできます。

postgres=# create index on fp16_test(b);
CREATE INDEX
postgres=# explain select * from fp16_test where b between 100 and 150;
                                       QUERY PLAN
-----------------------------------------------------------------------------------------
 Bitmap Heap Scan on fp16_test  (cost=21238.43..61716.43 rows=1000000 width=20)
   Recheck Cond: ((b >= '100'::double precision) AND (b <= '150'::double precision))
   ->  Bitmap Index Scan on fp16_test_b_idx  (cost=0.00..20988.43 rows=1000000 width=0)
         Index Cond: ((b >= '100'::double precision) AND (b <= '150'::double precision))
(4 rows)

なぜ半精度浮動小数点がGPUで好まれるのか？

最後に、なぜGPUアクセラレーションの文脈で半精度浮動小数点形式が使われるようになってきたのかを説明します。

GPUのように多数のコアが並列に動作するとき、とりわけNVIDIAのGPUではWarpと呼ばれる32スレッド単位でのスケジューリングが行われますが*1、隣接したコアが隣接したメモリからデータをロードする際、coalescingといって、一回のメモリトランザクションで複数スレッド分のデータをロードする事があります。
例えば、一回のメモリトランザクションで32byte（= 256bit）分のデータをL2キャッシュからロード*2できる場合、これが32bitの単精度浮動小数点なら、最大で8スレッドにデータを供給できる。一方、これが16bitの半精度浮動小数点なら、最大で16スレッドにデータを供給できるという計算になる。
通常、この手のワークロードであれば、メモリアクセスが最大の律速要因となってしまうので、そうすると、大量の計算をこなさねばならない機械学習のようなワークロードで、計算精度にある程度目をつぶれる（-1.0～1.0を十分に表現できればよい、など）場合には、単位時間あたりの計算量を増やすためにデータ量を削るという判断もアリとなる。

この辺については、CUDA C++ Programming GuideのMaximize Memory Throughputの章が詳しい。

言うまでもないが、これはデータが単純配列の形で並んでいるような場合の話で、例えばPostgreSQLの行データ（Heap形式）ではそれ以前の段階である。ただし、大抵のSQL条件式の検索というのは、特定の列を一回だけ参照してWHERE X BETWEEN 100 AND 200の条件を評価するものであるので、このためだけに行⇒列変換というのはリーズナブルではない。
（かつて一度実装したことがあり、やめたｗ）

PG-Stromでも対応している Apache Arrow 形式や、あるいはGPU Cache機能のように、データが単純配列のように並ぶことになっているデータ形式であれば、こういったcoalesced accessによるメモリ読み出しの高速化効果というものも期待できるかもしれない。
（ただし、RAM => GPUへの転送というのがそれよりずっと遅いので、あまり差分は見えてこないかも…。）

*1:しかしこれもVolta以降では条件分岐に絡めてズレる事もあるのでもはや正確な表現とも言い難いが…。

*2:Global Device Memoryからのロードは全てL2を介するとマニュアルには書いてある

PCI-E 4.0がやってきた！

突然ですが、サーバを新調しました。

昨年、先行して NVIDIA A100 を調達していたのですが、手持ちのサーバ自体はSkylake-SpでPCI-E 3.0世代なので、まだGPU自体の持つポテンシャルを評価できず・・・といったところでした。

調達したサーバは、Supermicro社のAS-2014CS-TRというモデルで、構成自体は以下の通りとなります。

• 筐体: AS-2014CS-TR
• CPU: AMD EPYC 7402P (24C; 2.85GHz) x1
• RAM: 128GB [16GB DDR4-3200 (ECC) x8]
• GPU: NVIDIA A100 (PCI-E; 40GB) x1
• SSD: Intel D7-P5510 (U.2; 3.84TB) x4
• HDD: Toshiba 3.5 1.0TB (7.2krpm; 6.0Gb/s) x2
• N/W: AIOM 2-port 10Gbase-T x1

今回、はじめてCPUとSSDもPCI-E 4.0に対応した世代のものを調達して、これでCPU/GPU/SSDをPCI-E 4.0に揃えてのベンチマークが可能となります。
CPUにMillan世代ではなくRoma世代を選んだのは、この世代ではI/O周りの変化がない（らしい）とされている事と、昨今の半導体不足の影響でタダでさえ長い納期がさらに延びる懸念が・・・というワケです。（そもそもこのサーバを発注したのは６月だｗ）

さて、早速、いつものように SSBM のデータベースをSF=999で構築し、13本のクエリの応答速度を計測してみます。
最もサイズの大きなlineorderテーブルのサイズは 875GB となるので、ストレージ中心のベンチマークには十分なサイズです。

まず結果はこの通り。
分かりやすさのためスループット表記でグラフにしていますが、これは単純に（875GB ÷ クエリ応答時間）ですので、縦軸の『Query Execution Throughput [MB/s]』の値が大きいほど処理性能が高い事を示しています。

Filesystem I/Oの場合、元々PCI-E 3.0世代でもSSD性能を遥かに下回る処理性能しか出せていませんでしたが、H/Wが新しくなっても同じような性能値であるという事は、ストレージ読み出しではなく、バッファコピーの繰り返しなど別のところにボトルネックがある事を示唆しています。

一方、GPU-Direct SQLの場合、クエリによっては19GB/sに迫る値を出しています。
PCI-E3.0のSkylake-Sp世代では8.5GB/s～9.0GB/s程度で頭打ちになっていた事を考えると、中々のスコアといえるでしょうか。

続いて、iostatで測定したnvme0～nvme3の各デバイスのクエリ実行中の読み出しスループットを計測すると、これはこれで中々ひどい。
同じサイズのデータを読み出すにも、短時間でガツん！と読み出すか、時間をかけてチンタラ読み出すかというのが可視化されていると思います。

ここまでは同じデータ形式の場合。

では、この875GB/60億行のlineorderテーブルを Pg2Arrow でApache Arrow形式に変換し、列データとして読み出す場合であればどうか？
行データと列データの処理性能をスループットで測るのは適切とは言えませんので、今度は（60億 ÷ クエリ応答時間）を『１秒あたり処理した行数』としてプロットしてみます。すると、I/Oの効率が良い分、さらに差が広がる結果となりました。

細かいところで言うと、『秒速で10億行』を達成しているQ1_1と、Q1_2およびQ1_3のデータ読み出しサイズは等しいので、処理時間もそれと同じ程度になっていてほしいのですが、オプティマイザがいま一つ、おバカな実行計画を作ってしまったようです。この辺は要改善といったところでしょうか。

Q1_1のEXPLAIN ANALYZE結果

postgres=# explain analyze
select sum(lo_extendedprice*lo_discount) as revenue
from flineorder,date1
where lo_orderdate = d_datekey
and d_year = 1993
and lo_discount between 1 and 3
and lo_quantity < 25;
                                                                      QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=9717634.19..9717634.20 rows=1 width=8) (actual time=4670.374..4781.510 rows=1 loops=1)
   ->  Gather  (cost=9717633.96..9717634.17 rows=2 width=8) (actual time=4387.503..4781.496 rows=3 loops=1)
         Workers Planned: 2
         Workers Launched: 2
         ->  Parallel Custom Scan (GpuPreAgg)  (cost=9716633.96..9716633.97 rows=1 width=8) (actual time=4359.970..4359.976 rows=1 loops=3)
               Reduction: NoGroup
               Combined GpuJoin: enabled
               GPU Preference: GPU0 (NVIDIA A100-PCIE-40GB)
               ->  Parallel Custom Scan (GpuJoin) on flineorder  (cost=17101.66..9716355.33 rows=594409 width=8) (never executed)
                     Outer Scan: flineorder  (cost=17060.26..9711145.81 rows=4162493 width=12) (actual time=156.771..564.225 rows=5993990673 loops=1)
                     Outer Scan Filter: ((lo_discount >= 1) AND (lo_discount <= 3) AND (lo_quantity < 25))
                     Rows Removed by Outer Scan Filter: 5209361385
                     Depth 1: GpuHashJoin(plan nrows: 4162493...1426582, actual nrows: 784629288...119025391)
                              HashSize: 20.50KB (estimated: 63.28KB)
                              HashKeys: flineorder.lo_orderdate
                              JoinQuals: (flineorder.lo_orderdate = date1.d_datekey)
                     GPU Preference: GPU0 (NVIDIA A100-PCIE-40GB) with GPUDirect SQL
                     referenced: lo_orderdate, lo_quantity, lo_extendedprice, lo_discount
                     files0: /opt/pgdata13/flineorder.arrow (read: 89.37GB, size: 681.05GB)
                     ->  Seq Scan on date1  (cost=0.00..78.95 rows=365 width=4) (actual time=0.060..0.317 rows=365 loops=1)
                           Filter: (d_year = 1993)
                           Rows Removed by Filter: 2191
 Planning Time: 2.334 ms
 Execution Time: 4910.442 ms
(24 rows)

上記のようにGpuPreAggの直下にGpuJoinが入っており、加えて「Combined GpuJoin: enabled」と出力されています。
このとき、GpuJoinの処理結果は一度CPUに戻される事なくGPU上で集約されるため、最も効率のよいパターンとなります。
また、Apache Arrowファイルは全体で681GBあり、そのうち89GBを読み出した事が分かります。つまり、20GB/s程度の読み出し速度があれば、5秒程度で処理を終えるのは不思議な事ではないという事になります。

Q1_2のEXPLAIN ANALYZE結果

postgres=# explain analyze
select sum(lo_extendedprice*lo_discount) as revenue
from flineorder, date1
where lo_orderdate = d_datekey
  and d_yearmonthnum = 199401
  and lo_discount between 4 and 6
  and lo_quantity between 26 and 35;
                                                                                  QUERY PLAN
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Finalize Aggregate  (cost=10085531.30..10085531.31 rows=1 width=8) (actual time=13043.719..13155.506 rows=1 loops=1)
   ->  Gather  (cost=10085531.08..10085531.29 rows=2 width=8) (actual time=12720.924..13155.496 rows=3 loops=1)
         Workers Planned: 2
         Workers Launched: 2
         ->  Partial Aggregate  (cost=10084531.08..10084531.09 rows=1 width=8) (actual time=12709.551..12709.556 rows=1 loops=3)
               ->  Hash Join  (cost=17686.41..10084527.30 rows=757 width=8) (actual time=4078.065..12645.401 rows=1404451 loops=3)
                     Hash Cond: (flineorder.lo_orderdate = date1.d_datekey)
                     ->  Parallel Custom Scan (GpuScan) on flineorder  (cost=17607.07..10084283.79 rows=62438 width=12) (actual time=313.555..6825.444 rows=108984171 loops=3)
                           GPU Filter: ((lo_discount >= 4) AND (lo_discount <= 6) AND (lo_quantity >= 26) AND (lo_quantity <= 35))
                           Rows Removed by GPU Filter: 5667038161
                           GPU Preference: GPU0 (NVIDIA A100-PCIE-40GB) with GPUDirect SQL
                           referenced: lo_orderdate, lo_quantity, lo_extendedprice, lo_discount
                           files0: /opt/pgdata13/flineorder.arrow (read: 89.37GB, size: 681.05GB)
                     ->  Hash  (cost=78.95..78.95 rows=31 width=4) (actual time=0.521..0.522 rows=31 loops=3)
                           Buckets: 1024  Batches: 1  Memory Usage: 10kB
                           ->  Seq Scan on date1  (cost=0.00..78.95 rows=31 width=4) (actual time=0.171..0.513 rows=31 loops=3)
                                 Filter: (d_yearmonthnum = '199401'::numeric)
                                 Rows Removed by Filter: 2525
 Planning Time: 2.308 ms
 Execution Time: 13268.534 ms
(20 rows)

一方、Q1_2の結果を見てみると、GpuPreAggが選択されておらず、GpuScanによるフィルタを実行した後はCPUでHashJoinとAggregateを実行しています。GpuScanの返す行数の推定値が62438行足らずなので『そんな程度の処理にGPUを使うまでもない』というのは分かるのですが、実際には3億行ちょい（1億行×3ワーカー）を読み出している事になるので、読みが大ハズレだったというわけです。

ハードウェアの増強だけでは高速化を達成する事はできないというよい例ですが、この辺は、追って調べてみたいと思います。

なお、今回のこの測定結果を含む、大量データを処理するための PG-Strom の諸機能については、11月12日（金）のPostgreSQL Conference Japan 2021にて発表を行います。

このご時世、なかなか顔を突き合わせたディスカッションの機会が少ないのですが、こちらはオフラインでの開催予定となっております。
ぜひのご参加をお待ちしております。

HyperLogLogを使ったカーディナリティの推測（補足）

少し、こちらのフォローアップ記事となります。

kaigai.hatenablog.com

ブログ公開後、何件かコメントをいただきました。

なるほど確かに、GUCパラメータの値に応じてCOUNT(distinct KEY)を置き換える構造だと、そのつもりがないのに、HyperLogLogを使ったカーディナリティの推計を行ってしまう・・・という事故が発生してしまうかもしれぬ。

という事で、前回の記事で説明したpg_strom.enable_hll_countは廃止し、代わりに、ユーザが明示的にHyperLogLogを使用する事を教えるために、hll_count(KEY)という集約関数を追加しています。

使用法としてはこんな感じ。

=# select hll_count(lo_custkey) from lineorder ;
 hll_count
-----------
   2005437
(1 row)

さらにもう一点、hll_count(KEY)はHyperLogLogを使って作成したHLL Sketch（前回記事でHLL Registersと呼んでいたもの。用語を統一。）を元に推計値を出す関数ですが、推計値を出すのではなく、そのままHLL Sketchをbytea型で保存できるようにしました。
こちらは、hll_sketch(KEY)という集約関数になり、あとで保存しておいたHLL Sketchをhll_merge(SKETCH)に食わせて、改めて推計値を出力できるようになります。

使い方としては、例えば、予め週次や月次のデータで HLL Sketch を作成しておけば、あとで必要な範囲だけの HLL Sketch をマージしてカーディナリティの推計値を出力するといった使い方が考えられます。

使用法としてはこんな感じ。

--- 年単位で HLL Sketch を出力する
=# select lo_orderdate / 10000 as year, hll_sketch(lo_custkey) as sketch
     into pg_temp.annual
     from lineorder group by 1;
SELECT 7

--- HLL Sketchをヒストグラムにして出力する
=# select year, hll_sketch_histogram(sketch) from pg_temp.annual order by year;
 year |                 hll_sketch_histogram
------+-------------------------------------------------------
 1992 | {0,0,0,0,0,0,0,0,0,22,73,132,118,82,39,26,12,2,4,2}
 1993 | {0,0,0,0,0,0,0,0,0,9,59,118,125,96,50,30,15,2,6,2}
 1994 | {0,0,0,0,0,0,0,0,0,4,33,111,133,113,53,36,17,4,6,2}
 1995 | {0,0,0,0,0,0,0,0,0,2,21,99,131,121,62,42,18,5,7,3,1}
 1996 | {0,0,0,0,0,0,0,0,0,1,17,84,119,131,73,50,20,5,7,4,1}
 1997 | {0,0,0,0,0,0,0,0,0,0,14,71,118,128,82,53,23,10,7,4,2}
 1998 | {0,0,0,0,0,0,0,0,0,0,13,64,114,126,86,61,23,11,8,4,2}
(7 rows)

--- 累積値で lo_custkey のカーディナリティを推測
=# select max_y, (select hll_merge(sketch) from pg_temp.annual where year < max_y)
     from generate_series(1993,1999) max_y;
 max_y | hll_merge
-------+-----------
  1993 |    854093
  1994 |   1052429
  1995 |   1299916
  1996 |   1514915
  1997 |   1700274
  1998 |   1889527
  1999 |   2005437
(7 rows)

例えば、ユニークユーザ数の集計を日次・週次で集計する時など、毎回 COUNT(distinct KEY)でやっていては遅くてたまらない、みたいな状況であれば、利用価値のある手法かもしれません。

本日、PostgreSQL Unconference (online) にてこの辺のトピックについて話しますので、お時間ある方はぜひご覧ください。
pgunconf.connpass.com

HyperLogLogを使ったカーディナリティの推測

高校生の頃までは滋賀県に住んでいた事もあり、夜、勉強の合間に、KBS京都で放送されていた『日髙のり子のはいぱぁナイト』を聞いており、日々ネタを考えては、番組へハガキを投稿する常連だった*1のですが（←勉強はどうした）、今回は、PG-Stromに実装した『はいぱぁ』な機能を紹介したいと思います。

ja.wikipedia.org

SELECT COUNT(distinct KEY) は結構難しい

SELECT COUNT(KEY) FROM my_table;

が

SELECT COUNT(distinct KEY) FROM my_table;

になった瞬間、特にサイズの大きなテーブルをスキャンする場合には、非常に難しい問題になってしまいます。

最初の例は、KEYが非NULLである行数を全部カウントして返せば良いのですが、後者の場合はKEYが重複する場合にはカウントしないため、重複排除を行うための工夫が必要になります。これをカーディナリティを計算すると言います。

これをDBで実装するには２通りの方法が考えられます。

方法①
入力ストリームを予めKEY値でソートしておき、KEY値が変わるたびにカウンタをインクリメントする。
KEY値にインデックスが張られている場合などには有効な方法だが、そうでなければ、実行時にテーブル全体のソートが必要になる。しかも、領域分割による並列処理が不可能であるので、仮に入力レコードが数億行あったとすると、律儀にCOUNT(distinct KEY)関数を数億回実行せねばならない。

方法②
集約関数を実行する Agg ノードでハッシュ表を持っておき、KEY値がそれまでにスキャンしたレコードに含まれているかどうかを判定する。最終的なCOUNT(distinct KEY)の結果は、このハッシュ表のエントリ数となる。
ソートは必要ないが、メモリ消費量が事前に予測不可能で、ハッシュ表のサイズによっては並列処理も難しい。（通常、メモリ消費が問題になるような状況ではマージ処理も大変な負荷になる）

なので、大量のデータセットの中から正確なカーディナリティを出力しようとすると、そこそこ大変（= 処理時間がかかってしまう）という事になります。

「ざっくり」でもよくないですか？

ただこれは、厳密な重複排除を行った集計を行う上での制限事項で、世の中には「ざっくりとした数が知りたい」で十分なケースが存在します。例えば、アクセスログからアクティブなユーザ数を集計してグラフに出したい、といった場合など、多少の誤差は許容できるユースケースです。

これを比較的精度よく推定できる方法として、HyperLogLogという手法が知られており、いくつかのビッグデータ処理向けデータベースに実装されているものもあります。

en.wikipedia.org

• Amazon RedShift ... HLL_CARDINALITY 関数 - Amazon Redshift
• Google BigQuery ... 標準 SQL の HyperLogLog++ 関数
• Microsoft CitusDB ... Distributed Distinct Count with HyperLogLog on Postgres — Citus 10.2 documentation

今回は、GPU上でのGROUP BY処理を行うGpuPreAgg機能の拡張として、PG-StromにHyperLogLogを実装してみました。

HyperLogLogアルゴリズムの考え方

HyperLogLogアルゴリズムの考え方をざっくり説明します。

• 前提①：COUNT(distinct KEY)のKEY値をハッシュ関数にかけると、ランダムなビット列が生成されるハズである。
• 前提②：KEY値のカーディナリティが高ければ、...10100000のように0が連続するパターンも含まれるハズである。

したがって、テーブルをスキャンしてKEY値のハッシュを計算し、その中で下位ビットから連続する0の個数の最大値を記録しておけば、その集合のカーディナリティは2n$2^n$程度であると推定する事ができます。
もちろん、このようにnの値に応じて2n$2^n$で増えていく推定値というのはあまりにも誤差が大きいですので、もう少し工夫を加えます。
ハッシュ値の下位bビット分をｍ互いに独立したカウンタであるHLLレジスタのインデックスと見なし、残りのビット列から連続する0の個数をカウントして、インデックスされたHLLレジスタにその最大値を記録します。
最後に、これの平均値を計算する事で、しばしば混じってしまう例外的なハッシュ値の影響を排除し、もう少し真の値に近いKEY値のカーディナリティを推定する．．．という流れになります。

この手法の良いところは、入力値をソートする必要がなく、また、テーブルを分割統治して互いに独立なHLLレジスタを作ったとしても、それほど大量のメモリを消費しないため、並列処理に向いているところです。
例えば、64bitのハッシュ値でレジスタのセレクタに10bitを使った場合、各レジスタは8bitあればカウンタとして十分に機能するため、HLLレジスタとして必要なのは僅か1.0kBだけという事になります。

PG-StromにおけるHyperLogLog

ここでは例として、Star Schema Benchmarkデータセットの lineorder テーブルから、lo_custkey*2のカーディナリティを調べてみる事にします。
scale factorは100なので、テーブルのサイズは概ね87GBとなります。

nvme=# \d+
                            List of relations
 Schema |   Name    | Type  | Owner  | Persistence |  Size  | Description
--------+-----------+-------+--------+-------------+--------+-------------
 public | customer  | table | kaigai | permanent   | 406 MB |
 public | date1     | table | kaigai | permanent   | 416 kB |
 public | lineorder | table | kaigai | permanent   | 87 GB  |
 public | part      | table | kaigai | permanent   | 160 MB |
 public | supplier  | table | kaigai | permanent   | 132 MB |
(5 rows)

nvme=# explain select count(distinct lo_custkey) from lineorder;
                                  QUERY PLAN
------------------------------------------------------------------------------
 Aggregate  (cost=18896094.80..18896094.81 rows=1 width=8)
   Output: count(DISTINCT lo_custkey)
   ->  Seq Scan on public.lineorder  (cost=0.00..17396057.84 rows=600014784 width=6)
         Output: lo_orderkey, ...(snip)..., lo_shipmode
(4 rows)

デフォルト設定では、このように count(distinct ...) を含むクエリをGPUで実行できません。
これは、HyperLogLogによる推定値で count(distinct ...) を代替する事で結果が変わってしまうため、デフォルトでは無効化されているためです。

nvme=# select count(distinct lo_custkey) from lineorder;
  count
---------
 2000000
(1 row)

Time: 409851.751 ms (06:49.852)

実行すると、厳密なcount(distinct lo_custkey)は 2,000,000 である一方、その実行には 409 秒を要している事が分かります。
（Sortの高速化を目的としたCPU並列クエリすら有効になっていないので、当然と言えば当然と言えます。）

次に、PG-StromのHyperLogLog機能による count(distinct ...) の置き換えを有効にします。

nvme=# set pg_strom.enable_hll_count = on;
SET

実行計画を見てみましょう。

nvme=# explain verbose select count(distinct lo_custkey) from lineorder;
                                                   QUERY PLAN
-----------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=7444397.37..7444397.38 rows=1 width=8)
   Output: pgstrom.hll_count((pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey))))
   ->  Gather  (cost=7444397.14..7444397.35 rows=2 width=32)
         Output: (pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))
         Workers Planned: 2
         ->  Parallel Custom Scan (GpuPreAgg) on public.lineorder  (cost=7443397.14..7443397.15 rows=1 width=32)
               Output: (pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))
               GPU Output: (pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))
               GPU Setup: pgstrom.hll_hash(lo_custkey)
               Reduction: NoGroup
               Outer Scan: public.lineorder  (cost=2833.33..7365270.22 rows=250006160 width=6)
               GPU Preference: GPU0 (Tesla V100-PCIE-16GB)
               Kernel Source: /var/lib/pgdata/pgsql_tmp/pgsql_tmp_strom_35128.1.gpu
               Kernel Binary: /var/lib/pgdata/pgsql_tmp/pgsql_tmp_strom_35128.2.ptx
(14 rows)

GPUを用いた集約関数であるGpuPreAggが選択されているほか、元々count(DISTINCT lo_custkey)を出力していた Aggregate ノードが、代わりにpgstrom.hll_count((pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))) の実行結果を出力するように書き換えられています。

内側から順に説明すると、pgstrom.hll_hash(lo_custkey)関数は、HyperLogLogに使用するハッシュ値を計算するための関数で、ここでは軽量かつ比較的ランダムな64bitのハッシュ値が得られるという事でSipHashアルゴリズムを使用しています。
次に、pgstrom.hll_pcount(HASH)関数は、HLLレジスタ配列をセットアップし、引数として与えられた64bitのハッシュ値を元にこれを次々と更新していきます。重要なのは、pgstrom.hll_pcount(HASH)関数はHLLレジスタ配列だけを出力するため、どれだけ巨大なテーブルをスキャンする事になったとしても、pgstrom.hll_pcount(HASH)関数より後の工程ではたった1行しか（GROUP BY句が指定されている場合はグループの数だけしか）返さないという事です。

したがって、各ワーカープロセスから返却されるものも含め、CPUでHLLレジスタ配列をマージする事になるpgstrom.hll_count()関数は、僅か1行 x 3プロセス分の結果を処理するだけで、HyperLogLogによるlo_custkey値のカーディナリティの推定が可能になるという事です。

このような性質により、GPU/CPUの並列処理の恩恵を最大限に受ける事ができるため、パフォーマンスも良好です。
厳密な集計値を導出するために409秒を要していた一方、実際の値と 0.3% 程度しかズレのない 2,005,437 という推定値を9.2秒で導出しています。

nvme=# select count(distinct lo_custkey) from lineorder;
  count
---------
 2005437
(1 row)

Time: 9212.712 ms (00:09.213)

実行計画の詳細を見てみましょう。

nvme=# explain (verbose, analyze) select count(distinct lo_custkey) from lineorder;
                                                                           QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=4992387.95..4992387.96 rows=1 width=8) (actual time=9045.729..9081.690 rows=1 loops=1)
   Output: pgstrom.hll_count((pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey))))
   ->  Gather  (cost=4992387.72..4992387.93 rows=2 width=32) (actual time=8892.195..9081.633 rows=3 loops=1)
         Output: (pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))
         Workers Planned: 2
         Workers Launched: 2
         ->  Parallel Custom Scan (GpuPreAgg) on public.lineorder  (cost=4991387.72..4991387.73 rows=1 width=32) (actual time=8760.881..8760.885 rows=1 loops=3)
               Output: (pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))
               GPU Output: (pgstrom.hll_pcount(pgstrom.hll_hash(lo_custkey)))
               GPU Setup: pgstrom.hll_hash(lo_custkey)
               Reduction: NoGroup
               Outer Scan: public.lineorder  (cost=2833.33..4913260.79 rows=250006160 width=6) (actual time=159.316..2800.578 rows=600037902 loops=1)
               GPU Preference: GPU0 (Tesla V100-PCIE-16GB)
               GPUDirect SQL: load=11395910
               Kernel Source: /var/lib/pgdata/pgsql_tmp/pgsql_tmp_strom_39266.2.gpu
               Kernel Binary: /var/lib/pgdata/pgsql_tmp/pgsql_tmp_strom_39266.3.ptx
               Worker 0:  actual time=8694.640..8694.644 rows=1 loops=1
               Worker 1:  actual time=8699.829..8699.833 rows=1 loops=1
 Planning Time: 0.129 ms
 Execution Time: 9194.200 ms
(20 rows)

このクエリには全体で9.2秒を要していますが、そのうち8.760秒が GpuPreAgg での実行に要しています。
ここでは GPU-Direct SQL を用いて、4台のNVME-SSDから 10GB/s 程度のスループットで合計6億行を読み出していますが、GpuPreAggが出力しているのはHLLレジスタ配列の1行だけであるので、非常に効率的なデータ転送が行われていると言えます。

下の図で言えば、テーブル（ディスク）からデータを読み出し、GPU上で実行される hll_pcount() 関数にロードするところまでが、スループット番長である PG-Strom の真骨頂で、クエリの書き換えとアルゴリズムの工夫により、厄介なCOUNT(distinct KEY)をこのような形態の処理に書き換えるところが HyperLogLog の恩恵と言えるでしょう。

結論

• COUNT(distinct KEY)関数で「大まかな推定値」を得れば十分である場合、HyperLogLogを使って相応に精度の良い推定値を得る事ができる。
• COUNT(distinct KEY)関数を、distinct句の付かない集約関数に書き換える事で、領域分割と並列処理が可能な形式に変換できる。このパターンに落とす事ができれば、GPU-Direct SQLでほぼほぼハードウェアの限界に近い速度で集計処理を回すことができる。

ひとまず、現状では論文に書かれている内容をそのまま何も考えた形なので、例えばカーディナリティが小さい時の推定値のズレや、より正確な推定値を得るための補正（関連研究でそういうのがあるらしい）については、全く何も入っていません。誰かそういうのに強い人がパッチを書いてくれたりすると助かります（ボソッ

*1:ちなみに、『リスナーと電話をつないでクイズに答える』というコーナーで、日髙さんと一度だけ15秒くらい喋った事がある。

*2:customer表に対するキー

Apache Arrowの統計情報を使ったログ検索の爆速化

PostgreSQLにはBRINインデックス（Block Range Index）という機能があり、ログデータに付属するタイムスタンプ値など、近しい値を持ったデータが物理的に近接するという特徴を持っているとき、検索範囲を効率的に絞り込むために使用する事ができる。

この機能はPG-Stromでも対応しており、その詳細は以前のエントリでも解説している。

kaigai.hatenablog.com

かいつまんで説明すると、時系列のログデータのように大半が追記（Insert-Only）であり、かつタイムスタンプ値のように近しい値同士が近接している場合、1MBのブロック((pages_per_rangeがデフォルトの128の場合、8kB * 128 = 1MB))ごとにその最小値／最大値を記録しておくことで『明らかに検索条件にマッチしない範囲』を読み飛ばす事ができる。

例えば以下の例であれば、WHERE ymd BETWEEN '2016-01-01' AND '2016-12-31'という検索条件は、ブロック内に含まれる個々の行をチェックしなくとも、最小値・最大値を参照すれば１番目と５番目のブロックに該当する行が無いのは自明である。そのため、検索条件にマッチする可能性のある２，３，４番目のブロックのみを読み出せばよい、というのがBRINインデックスの考え方である。

なお、B-treeインデックスとは異なり、BRINインデックスは個々の行を指し示すデータを持っていないため、1MBのブロック単位で読む/読まないを決めた後は、全件スキャンと同じ処理となる。

Apache Arrowに統計情報を埋め込む

話を Apache Arrow 形式に戻す。

Apache Arrow は列フォーマットの構造化データ形式であるが、例えば『このArrowファイルは1億件のデータを保持している』と言っても、単純に要素数1億の配列が並んでいるわけではない。
内部的にはRecord-Batchと呼ばれるブロックが複数並んだ構造となっており、このブロックの内側に、例えば1億件のうち100万件といったより小規模なデータの配列を保持している。（こうしたデータ構造を取る事により、より追記を行いやすくするという狙いがあるものと思われる。）

ファイルの末尾にはフッタ（Footer）があり、ここを読めば、ファイルのどこにRecord-Batchが配置されているかが分かる。
それと同時に、フッタにはデータ構造の定義を記述する Schema Definition の部分がある。これは基本的にはファイル先頭の Schema Definition のコピーであるものの、custom_metadataフィールドに独自のKey-Value値を埋め込んで書き込む程度の事は認められているようである。

このcustom_metadataフィールドは、テーブルだけでなく、カラムにも付与する事ができる。
そうすると、どういった使い方ができるか。フィールド毎のKey-Value値として、Record-Batchごとの最小値／最大値の配列（以下、統計情報）をApache Arrowファイルに埋め込む事ができる。

しかもこれはフッタ領域なので、Apache Arrowファイルを追記して Record-Batch を更新するたびに、統計情報をアップデートして書き直す事までできる。Yeah!!

pg2arrow の --stat オプション

早速、統計情報付きのApache Arrowファイルを生成してみる事にする。
これにはpg2arrowコマンドの--statオプションを使用する事ができ、統計情報を埋め込む対象列を指定する。現在のところ、Int、FloatingPoint、Decimal、Date、Time、Timestampの各データ型に対応している。

$ pg2arrow -d postgres -o /dev/shm/flineorder_sort.arrow -t lineorder_sort --stat=lo_orderdate --progress
RecordBatch[0]: offset=1640 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[1]: offset=268438720 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[2]: offset=536875800 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[3]: offset=805312880 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[4]: offset=1073749960 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[5]: offset=1342187040 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[6]: offset=1610624120 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[7]: offset=1879061200 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[8]: offset=2147498280 length=268437080 (meta=920, body=268436160) nitems=1303085
RecordBatch[9]: offset=2415935360 length=55668888 (meta=920, body=55667968) nitems=270231
$ ls -lh /dev/shm/flineorder_sort.arrow
-rw-r--r--. 1 kaigai users 2.4G  7月 19 10:45 /dev/shm/flineorder_sort.arrow

上記の例は、タイムスタンプに相当するlo_orderdate列でソート済みの*1テーブルをダンプし、Apache Arrowファイルとして保存したもの。

生のメタデータを可読な形式でダンプしてみると、確かにlo_orderdate列のフィールド定義にmin_valuesおよびmax_valuesのメタデータが埋め込まれ、それぞれ19920101とか19920919とか「それっぽい」値が見えている。

なお、custom-metadataを用いたデータの埋め込みはApache Arrow形式で認められているものなので、例えば、PG-Stromのコードベースと全く関係のない PyArrow を使って上記のファイルを読み出す場合でも、きちんとメタデータとして認識される。
（特にスキャンの最適化などにも使われないが）

$ python3
Python 3.6.8 (default, Aug 24 2020, 17:57:11)
[GCC 8.3.1 20191121 (Red Hat 8.3.1-5)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import pyarrow as pa
>>> X = pa.RecordBatchFileReader('/dev/shm/flineorder_sort.arrow')
>>> X.schema
lo_orderkey: decimal(30, 8)
lo_linenumber: int32
lo_custkey: decimal(30, 8)
lo_partkey: int32
lo_suppkey: decimal(30, 8)
lo_orderdate: int32
  -- field metadata --
  min_values: '19920101,19920919,19930608,19940223,19941111,19950730,1996' + 31
  max_values: '19920919,19930608,19940223,19941111,19950730,19960417,1997' + 31
lo_orderpriority: fixed_size_binary[15]
lo_shippriority: fixed_size_binary[1]
lo_quantity: decimal(30, 8)
lo_extendedprice: decimal(30, 8)
lo_ordertotalprice: decimal(30, 8)
lo_discount: decimal(30, 8)
lo_revenue: decimal(30, 8)
lo_supplycost: decimal(30, 8)
lo_tax: decimal(30, 8)
lo_commit_date: fixed_size_binary[8]
lo_shipmode: fixed_size_binary[10]
-- schema metadata --
sql_command: 'SELECT * FROM lineorder_sort'

統計情報を使って Apache Arrow のスキャンを最適化する

さて、埋め込んだ統計情報も、集計処理の段でこれを参照しなければカピバラに小判、カピバラに真珠である。

という事で、Apache ArrowファイルをPostgreSQLからスキャンするためのArrow_Fdwに統計情報を参照するための機能を付加してみた。
検索条件が統計情報を含む列を参照し、その大小関係に基づく絞り込みを行う場合、統計情報に基づいて『明らかに無駄なRecord-Batch』の読み出しをスキップする。

postgres=# IMPORT FOREIGN SCHEMA flineorder_sort
           FROM SERVER arrow_fdw INTO public
           OPTIONS (file '/dev/shm/flineorder_sort.arrow');
IMPORT FOREIGN SCHEMA

postgres=# EXPLAIN ANALYZE
           SELECT count(*) FROM flineorder_sort
            WHERE lo_orderpriority = '2-HIGH'
              AND lo_orderdate BETWEEN 19940101 AND 19940630;

                                                                 QUERY PLAN
------------------------------------------------------------------------------------------
 Aggregate  (cost=33143.09..33143.10 rows=1 width=8) (actual time=115.591..115.593 rows=1loops=1)
   ->  Custom Scan (GpuPreAgg) on flineorder_sort  (cost=33139.52..33142.07 rows=204 width=8) (actual time=115.580..115.585 rows=1 loops=1)
         Reduction: NoGroup
         Outer Scan: flineorder_sort  (cost=4000.00..33139.42 rows=300 width=0) (actual time=10.682..21.555 rows=2606170 loops=1)
         Outer Scan Filter: ((lo_orderdate >= 19940101) AND (lo_orderdate <= 19940630) AND (lo_orderpriority = '2-HIGH'::bpchar))
         Rows Removed by Outer Scan Filter: 2425885
         referenced: lo_orderdate, lo_orderpriority
         Stats-Hint: (lo_orderdate >= 19940101), (lo_orderdate <= 19940630)  [loaded: 2, skipped: 8]
         files0: /dev/shm/flineorder_sort.arrow (read: 217.52MB, size: 2357.11MB)
 Planning Time: 0.210 ms
 Execution Time: 153.508 ms
(11 rows)

postgres=# SELECT count(*) FROM flineorder_sort
            WHERE lo_orderpriority = '2-HIGH'
              AND lo_orderdate BETWEEN 19940101 AND 19940630;
 count
--------
 180285
(1 row)

上の実行計画は、先ほどのApache ArrowファイルをPostgreSQLから外部テーブルとして参照できるようにしたflineorder_sortに、lo_orderdate列とそれ以外の検索条件を付加して検索したものである。

実行計画のStats-Hintを見ると、lo_orderdate BETWEEN 19940101 AND 19940630を展開した(lo_orderdate >= 19940101), (lo_orderdate <= 19940630)という条件を用いて不要Record-Batchの読み飛ばしを行い、全体で10個のRecord-Batchがある中で8個をスキップ、2個だけを読み出したと出力されている。

その結果、読み出した2個のRecord-BatchをGPUで評価し、2425885行を除去して残りが180285行であると集計している。

統計情報を使わない場合はどうなるのか？
以下の実行結果をご覧頂きたい。

postgres=# SET arrow_fdw.stats_hint_enabled = off;
SET
postgres=# EXPLAIN ANALYZE
           SELECT count(*) FROM flineorder_sort
            WHERE lo_orderpriority = '2-HIGH'
              AND lo_orderdate BETWEEN 19940101 AND 19940630;

                                                                 QUERY PLAN
------------------------------------------------------------------------------------------
 Aggregate  (cost=33143.09..33143.10 rows=1 width=8) (actual time=185.985..185.986 rows=1 loops=1)
   ->  Custom Scan (GpuPreAgg) on flineorder_sort  (cost=33139.52..33142.07 rows=204 width=8) (actual time=185.974..185.979 rows=1 loops=1)
         Reduction: NoGroup
         Outer Scan: flineorder_sort  (cost=4000.00..33139.42 rows=300 width=0) (actual time=10.626..100.734 rows=11997996 loops=1)
         Outer Scan Filter: ((lo_orderdate >= 19940101) AND (lo_orderdate <= 19940630) AND (lo_orderpriority = '2-HIGH'::bpchar))
         Rows Removed by Outer Scan Filter: 11817711
         referenced: lo_orderdate, lo_orderpriority
         files0: /dev/shm/flineorder_sort.arrow (read: 217.52MB, size: 2357.11MB)
 Planning Time: 0.186 ms
 Execution Time: 231.942 ms
(10 rows)

postgres=# SELECT count(*) FROM flineorder_sort
            WHERE lo_orderpriority = '2-HIGH'
              AND lo_orderdate BETWEEN 19940101 AND 19940630;
 count
--------
 180285
(1 row)

arrow_fdw.stats_hint_enabledにoffを設定すると、統計情報を使わなくなる。
その場合、Rows Removed by Outer Scan Filterが2,425,885から11,817,711へと増加しているが、条件に該当しない行を除いた最終結果は180285で同一である。
つまり、”明らかにマッチする行が一つもない” Record-Batch を読み出した挙句、検索条件をGPUで評価してフィルタしているワケである。大変にご苦労様なことである。

サイズの大きなApache Arrowファイルによるベンチマーク

数GB程度のスケールだと中々差が見えないので、60億件規模のlineorderテーブル((ログデータに似た構造とするため、予めlo_orderdateでソートしておいた))を検索するワークロードで、クエリの応答時間を比較してみた。
元のPostgreSQLテーブルでは875GBあり、これをApache Arrowにダンプすると682GBあった*2。

単純にストレージからの読み出しの部分だけに絞って考えても、PostgreSQLテーブル（Heap）のような行データであれば、集計処理を行うために全てのデータを読み出さねばならない。
一方、Apache Arrowのような列形式データの場合は被参照列のみ読み出せば集計処理を行う事ができるため、相対的にI/O量が小さくなり、それが高速な処理速度の要因のひとつとなっている。

これに加えて、予め収集した統計情報を元にRecord-Batchを読み飛ばすという事は、列方向に加えて行方向に関してもI/O量を小さくするという事と同義である。どの程度、集計クエリの応答速度が短くなるのか実測してみた。

測定パターンの全てで概ね妥当なクエリ実行計画を生成するサンプルとして、Star Schema BenchmarkのQ1_2をピックアップしてみた。

select sum(lo_extendedprice*lo_discount) as revenue
  from flineorder_sort, date1
 where lo_orderdate = d_datekey
   and d_yearmonthnum = 199401
   and lo_discount between 4 and 6
   and lo_quantity between 26 and 35;

この人は、lo_orderdateが『1994年1月』のデータを抽出するように作られている。
ただ、これはdate1テーブルとのJOINによって記述されており、そのままでは統計情報を用いて不要Record-Batchをスキップする形には落ちてくれない。そのため、以下のように一部クエリを修正して条件を付加する事にした。

select sum(lo_extendedprice*lo_discount) as revenue
  from flineorder_sort, date1
 where lo_orderdate = d_datekey
   and d_yearmonthnum = 199401
   and lo_discount between 4 and 6
   and lo_quantity between 26 and 35;
   and lo_orderdate between 19940101 and 19940131;

その結果が、以下のグラフである。縦軸は Q1_2 クエリの応答時間であるので、結果が小さいほど性能が良いという事ができる。
また、行データをスキャンするパターン（青、橙）に関しては、Apache Arrowファイルをマップした flineorder_sort テーブルではなく、その元になった lineorder_sort テーブルをスキャンした。

ご覧の通り、I/O量の低減が功を奏してかなり極端な結果が出ている。
スケールが違いすぎて分かりにくいが、PostgreSQL v13.3で254秒を要した60億件のレコードのスキャンが、GPUDirect SQL、Apache Arrow、および統計情報による読み飛ばしの効果により1.2秒弱で終わっている。

以下のEXPLAIN ANALYZE結果を見ると、全体で2725個のRecord-Batchを含み、その大半（2688個）は検索条件に該当する行ナシとして捨ててしまっている。GPUが高速とはいえ、そもそも「何もしない」に勝る結果は無いのであるから、妥当な結果と言えるだろう。

postgres=# explain analyze
select sum(lo_extendedprice*lo_discount) as revenue
  from flineorder_sort, date1
 where lo_orderdate = d_datekey
    and d_yearmonthnum = 199401
    and lo_discount between 4 and 6
    and lo_quantity between 26 and 35
    and lo_orderdate between 19940101 and 19940131;
                                                                                           QUERY PLAN

------------------------------------------------------------------------------------------------------------------------------------------------------- Finalize Aggregate  (cost=15117511.53..15117511.54 rows=1 width=8) (actual time=1042.000..1092.485 rows=1 loops=1)
   ->  Gather  (cost=15117511.31..15117511.52 rows=2 width=8) (actual time=846.545..1092.475 rows=3 loops=1)
         Workers Planned: 2
         Workers Launched: 2
         ->  Partial Aggregate  (cost=15116511.31..15116511.32 rows=1 width=8) (actual time=781.303..781.308 rows=1 loops=3)
               ->  Hash Join  (cost=25060.71..15116511.29 rows=4 width=8) (actual time=435.810..719.086 rows=1404451 loops=3)
                     Hash Cond: (flineorder_sort.lo_orderdate = date1.d_datekey)
                     ->  Parallel Custom Scan (GpuScan) on flineorder_sort  (cost=24981.37..15116431.13 rows=312 width=12) (actual time=435.301..553.115 rows=1404451 loops=3)
                           GPU Filter: ((lo_discount >= 4) AND (lo_discount <= 6) AND (lo_quantity >= 26) AND (lo_quantity <= 35) AND (lo_orderdate >= 19940101) AND (lo_orderdate <= 1
9940131))
                           Rows Removed by GPU Filter: 77197339
                           GPU Preference: GPU0 (NVIDIA A100-PCIE-40GB) with GPUDirect SQL
                           referenced: lo_orderdate, lo_quantity, lo_extendedprice, lo_discount
                           Stats-Hint: (lo_orderdate >= 19940101), (lo_orderdate <= 19940131)  [loaded: 37, skipped: 2688]
                           files0: /opt/nvme0/flineorder_sort.arrow (read: 89.37GB, size: 681.05GB)
                     ->  Hash  (cost=78.95..78.95 rows=31 width=4) (actual time=0.472..0.473 rows=31 loops=3)
                           Buckets: 1024  Batches: 1  Memory Usage: 10kB
                           ->  Seq Scan on date1  (cost=0.00..78.95 rows=31 width=4) (actual time=0.140..0.464 rows=31 loops=3)
                                 Filter: (d_yearmonthnum = '199401'::numeric)
                                 Rows Removed by Filter: 2525
 Planning Time: 4.661 ms
 Execution Time: 1190.959 ms
(21 rows)

PostgreSQLパーティションとの比較

最後に、タイムスタンプ等を用いたデータ分散と集計処理の際の読み飛ばしという点で、類似する機能を有する PostgreSQL パーティションについても考察する。

PostgreSQLパーティションの場合、親テーブルと全く同一のスキーマ定義を持つ子テーブルに、他の子テーブルと重複しないようパーティションキー値の取り得る範囲を設定する。
通常のPostgreSQLテーブルの場合、これはINSERT/UPDATE/DELETE処理の際に自動的に子テーブル側の更新処理に振り分けられ、テーブルの内容に矛盾が生じる事はない。
一方、Apache Arrowファイルをマッピングする場合には、パーティション子テーブルとして振舞うArrow_Fdw外部テーブルに紐づいたApache Arrowファイルの内容に責任を持つのは、ユーザやDB管理者の役割である。

ログを保存するApache Arrowファイルを特定の時刻にバチッと切り替えるなら対応可能かもしれないが、例えば、一定のサイズに達したApache Arrowファイルは一旦クローズして新しいファイルを作ったり、非同期でマルチスレッドが次々とデータを書き出すために、明確なタイムスタンプの区切りを設けるのが難しい場合、パーティション設定と辻褄を合わせるのは中々難しい作業となる。

最小／最大値の統計情報であれば、機械的に Apache Arrow ファイルに埋め込まれるだけなので、明確な境界値を設ける必要はなく、また、各Apache Arrowファイルに含まれるタイムスタンプ値に関しても、これを管理者が意識する必要はない。
つまり、どこかのディレクトリを『時系列ログをApache Arrow形式で放り込む場所』と決めてしまいさえすれば、あとはポンポンとファイルをコピーするだけで、それを分析する事ができるようになるのである。

現在のところ、Apache Arrowファイルに最小／最大値の統計情報を埋め込む事ができるのはpg2arrowだけであるが、Apache Arrowファイルの構造上、末尾のフッタ部分だけを書き換えれば統計情報を付加する事ができるようになるの。
したがって、他のツールによって生成されたファイルであっても、簡単なバッチ処理で統計情報を付加できるようにする事ができるハズであるし、そういったツールは追って提供する事としたい。

*1:ジェネレータで生成したデータはランダム分布なので、ログデータのようなタイムスタンプの局所性を満たさない

*2:行ごとのヘッダ

GPUDirect SQL on NFS-over-RDMAを試す

タイトルでほぼほぼ出オチですが、先日、NVIDIAからCUDA Toolkit 11.4と共にリリースされた新機能GPUDirect Storage 1.0のドキュメントを読んでいると、面白い記述を見つけた。

曰く、MOFEDドライバ5.3以降と、Mellanox Connect-X4/5の組み合わせで、NFS-over-RDMAとGPUDirect Storageを組み合わせ、リモートのNFS区画からローカルのGPUへと直接のデータ転送を行う事ができるようになる、と。

14.10. NFS Support with GPUDirect Storage
This section provides information about NFS support with GDS.

14.10.2. Install GPUDirect Storage Support for the NFS Client
Here is some information about installing GDS support for the NFS client.
To install a NFS client with GDS support complete the following steps:
Note: The client must have a Mellanox connect-X4/5 NIC with MLNX_OFED 5.3 or later installed.
:

結構な事である。
PG-Strom v3.0以前では、ローカルのNVME-SSDまたはリモートのNVME-oF区画（実験的）を Ext4 ファイルシステムで初期化したパターンに限って GPUDirect SQL が対応していたため、

• 段階的にストレージを拡張するのに困難を伴った。
• 共有ファイルシステムではないので、複数台のノードから書き込みができなかった。

という課題があった。NFS自体はものすごく高速なファイルシステム、というワケではないが、DB/GPUサーバからストレージを分離し、かつ複数のノードから書き込みができるのであれば、例えば、IoT/M2M系のワークロードでログデータを収集し、これをNFSサーバ上に置いておきさえすれば、DB/GPUサーバからこれを参照してGPUDirect SQLの処理スピードでもってコレを分析する事ができる。

結論：結構イケてる

セットアップ手順などは長くなるので後回しにするとして、ひとまずSSBM (Star Schema Benchmark) の結果を一言でまとめると「結構イケてる」という印象。

測定環境は以下の図の通りで、今回は1UサーバのSYS-1019GP-TTにNFSサーバになってもらった。この人には、エンクロージャ経由でNVME-SSD（Intel DC P4510[1.0TB; U.2]）を4台接続し、また Mellanox Connect-X5 という100Gb-NIC を接続している。
GPU/DBサーバには4UのSYS-4029GP-TRTを使い、この人には、同じPCI-Eスイッチの配下にGPUとConnect-X5を接続したペアと、もう一つGPUとNVME-SSD（同 DC P4510）を4台接続したペアを作った。これはローカルNVME-SSDとの性能比較用である。

NFSサーバは、SSD x4台をmd-raid0でストライピングした区画をNFSクライアントにエクスポートし、NFSクライアントは直結の100Gbネットワーク*1を介して、これをNFS-over-RDMAモードでマウント。

GPU/DBサーバ側では以下のようなストレージ構成となっている。
/opt/nvme0には、ローカルのNVME-SSD x4台をmd-raid0でストライピングした区画をマウント、/opt/nvme1には、1Uサーバ（192.168.80.106）のNFS区画が見えている。

[kaigai@kujira ~]$ df -h
Filesystem                   Size  Used Avail Use% Mounted on
devtmpfs                      94G     0   94G   0% /dev
tmpfs                         94G  257M   94G   1% /dev/shm
tmpfs                         94G   19M   94G   1% /run
tmpfs                         94G     0   94G   0% /sys/fs/cgroup
/dev/mapper/vg_disk-root     246G   15G  218G   7% /
/dev/nvme0n1p1               1.8T   35G  1.7T   2% /opt
/dev/md0p1                   3.6T  1.4T  2.1T  41% /opt/nvme0
/dev/sda2                    976M  189M  721M  21% /boot
/dev/mapper/vg_disk-home     393G   24G  349G   7% /home
/dev/sda1                    599M  6.9M  592M   2% /boot/efi
tmpfs                         19G     0   19G   0% /run/user/1000
192.168.80.106:/mnt/nfsroot  2.0T  1.2T  697G  64% /opt/nvme1

で、それぞれの区画に保持されているlineorderテーブルへの参照を含むSSBMクエリの実行速度は以下の通り。
分かりやすいように、（総DBサイズ）÷（クエリ応答時間）で導出した『クエリ処理スループット』で表記している。

見ての通り、ローカルのNVME-SSDに比べるとNFS-over-RDMAは１割程度遅いと*2言えるが、これは、１割程度遅いだけでストレージの拡張性やリモートアクセスといった特性を得られるという事を意味する。

クエリ実行中のストレージからの読み出し速度を見てみても、クエリ実行中の100Gbのネットワークで8.0GB/s強を出せているので、まずまずのパフォーマンスと言える。
なお、ローカルのNVME-SSDの場合、後半で突然読み出し速度が増しで10.0GB/s程度まで増速しているが、これについては現時点で謎である…。

結論

• PG-StromのGPUDirect SQLとNFS-over-RDMAの併用、低コストのログ集積＆分析基盤としては結構アリかもよ。
• ログデータを Apache Arrow 形式で書き込んでおけば、データをインポートする必要すらなくなります。

NFS-over-RDMAのセットアップ手順

NFS-over-RDMAのセットアップ手順は、以下のブログを参考にした…というか、ほとんどそのまま。
https://community.mellanox.com/s/article/howto-configure-nfs-over-rdma--roce-x

ソフトウェアの構成はざっくり以下の通り

• CentOS 8.3 (kernel-4.18.0-240.22.1.el8_3.x86_64)
• CUDA Toolkit 11.4 (NVIDIA Driver R470.42.01)
• MOFED 5.3-1.0.0.1 (RHEL8.3; x86_64)
• PostgreSQL v13.3 (PG-Strom v3.0-3)

MOFEDOドライバのインストール

まず、MellanoxのサイトからMOFEDドライバの最新版をダウンロードする。

[Version]->[OS Distribution]->[OS Distribution Version]->[Architecture]と選択していくと、バイナリパッケージを含む tgz のパッケージと、ソースコードの tgz パッケージの両方が表示されるので、両方ともダウンロード。実はソースコードも後で使います。

tgzファイルをダウンロードすると、まず GPUDirect Storage のドキュメント通りにドライバのインストールを行う。
途中、不足するパッケージがある場合には、インストールスクリプトがサジェスト通りに`dnf install ...`すればよいので、その通りに進めればMOFEDドライバのインストールは行えるはず。

$ sudo ./mlnxofedinstall --with-nvmf --with-nfsrdma --enable-gds --add-kernel-support
Note: This program will create MLNX_OFED_LINUX TGZ for rhel8.3 under /tmp/MLNX_OFED_LINUX-5.3-1.0.0.1-4.18.0-240.22.1.el8_3.x86_64 directory.
See log file /tmp/MLNX_OFED_LINUX-5.3-1.0.0.1-4.18.0-240.22.1.el8_3.x86_64/mlnx_iso.225746_logs/mlnx_ofed_iso.225746.log

Checking if all needed packages are installed...
Building MLNX_OFED_LINUX RPMS . Please wait...
    :
  <snip>
    :
$ sudo dracut -f
$ sudo shutdown -r now

これを、NFSサーバ側、NFSクライアント側の両方で行い、システムを再起動。

NFSサーバの設定

1UサーバのSYS-1019GP-TT側では、ローカルのNVME-SSDを4本束ねたmd-raid0区画を`/mnt/nfsroot`にマウントしている。
これを以下の手順でNFS-over-RDMA区画としてエクスポートする。

1. IPアドレス他ネットワーク設定

今回は安直に192.168.80.0/24を直結用のネットワークとして使用。
静的に192.168.80.106/24をConnect-X5デバイスに設定し、MTU=9000でNICを有効化しました。

2. /etc/exportsを記述。特にセキュリティとか何も考えてない設定です。

# cat /etc/exports
/mnt/nfsroot *(rw,async,insecure,no_root_squash)

3. RDMA Transport Kernel Moduleをロード。これはMOFEDドライバによって提供されるモジュール。

# modprobe svcrdma
# modinfo svcrdma
filename:       /lib/modules/4.18.0-240.22.1.el8_3.x86_64/extra/mlnx-nfsrdma/svcrdma.ko
version:        2.0.1
license:        Dual BSD/GPL
description:    svcrdma dummy kernel module
author:         Alaa Hleihel
rhelversion:    8.3
srcversion:     F7C50654667EBC6F832D608
depends:        mlx_compat
name:           svcrdma
vermagic:       4.18.0-240.22.1.el8_3.x86_64 SMP mod_unload modversions

4. NFSサーバを起動

# systemctl start nfs-server

5. RDMA転送用のポート番号を設定。一応、任意のポート番号を使用できるが、20049というのがwell-known defaultとのこと。

# echo rdma 20049 > /proc/fs/nfsd/portlist
# cat /proc/fs/nfsd/portlist
rdma 20049
rdma 20049
tcp 2049
tcp 2049

NFSクライアントの設定

1. IPアドレス他ネットワーク設定

サーバー側と同様、静的に192.168.80.108/24をConnect-X5デバイスに設定し、MTU=9000でNICを有効化しました。
ネットワークの有効化が終わったら、pingなどで導通確認。

$ ping 192.168.80.106
PING 192.168.80.106 (192.168.80.106) 56(84) bytes of data.
64 bytes from 192.168.80.106: icmp_seq=1 ttl=64 time=0.178 ms
64 bytes from 192.168.80.106: icmp_seq=2 ttl=64 time=0.197 ms
^C

2. クライアント側のRDMA Transport Kernel Moduleをロード。これもMOFEDドライバに含まれるモジュール。

# modprobe rpcrdma
# modinfo rpcrdma
filename:       /lib/modules/4.18.0-240.22.1.el8_3.x86_64/extra/mlnx-nfsrdma/rpcrdma.ko
alias:          xprtrdma
alias:          svcrdma
license:        Dual BSD/GPL
description:    RPC/RDMA Transport
author:         Open Grid Computing and Network Appliance, Inc.
rhelversion:    8.3
srcversion:     EFB4ED2B09C65AA7DA8D887
depends:        ib_core,sunrpc,mlx_compat,rdma_cm
name:           rpcrdma
vermagic:       4.18.0-240.22.1.el8_3.x86_64 SMP mod_unload modversions

3. 前節でエクスポートしたNFS区画をマウント

# mount -o rdma,port=20049 192.168.80.106:/mnt/nfsroot /opt/nvme1
# df -h
Filesystem                   Size  Used Avail Use% Mounted on
devtmpfs                      94G     0   94G   0% /dev
tmpfs                         94G  257M   94G   1% /dev/shm
tmpfs                         94G   19M   94G   1% /run
tmpfs                         94G     0   94G   0% /sys/fs/cgroup
/dev/mapper/vg_disk-root     246G   15G  218G   7% /
/dev/nvme0n1p1               1.8T   35G  1.7T   2% /opt
/dev/md0p1                   3.6T  1.4T  2.1T  41% /opt/nvme0
/dev/sda2                    976M  189M  721M  21% /boot
/dev/mapper/vg_disk-home     393G   24G  349G   7% /home
/dev/sda1                    599M  6.9M  592M   2% /boot/efi
tmpfs                         19G     0   19G   0% /run/user/1000
192.168.80.106:/mnt/nfsroot  2.0T  1.2T  697G  64% /opt/nvme1

これで準備完了。
導通確認を兼ねて、巨大なファイルの転送を行ってみる。

# dd if=/opt/nvme1/100GB of=/dev/null iflag=direct bs=32M
3106+1 records in
3106+1 records out
104230305696 bytes (104 GB, 97 GiB) copied, 11.8926 s, 8.8 GB/s

これは速い！ 8.8GB/s も出ている。

一方、NFS-over-RDMAを使わないパターンだと。

# mount 192.168.80.106:/mnt/nfsroot /mnt/
# dd if=/mnt/100GB of=/dev/null iflag=direct bs=32M
3106+1 records in
3106+1 records out
104230305696 bytes (104 GB, 97 GiB) copied, 32.6171 s, 3.2 GB/s

御意。

GPUDirect StorageでNFS区画⇒GPUへの直接Readを行う

続いて本番。GPUDirect Storageを使って、リモートのNFS区画からGPUへの直接Readを行う。

今現在、NFS区画からGPUDirect Storageによる直接読み出しが可能な状態になっているかどうか、CUDA 11.4に添付のgdscheckというコマンドで確認する事ができる。。。。が、あらら。Unsupportedと表示されている。

# /usr/local/cuda/gds/tools/gdscheck -p
 GDS release version: 1.0.0.82
 nvidia_fs version:  2.7 libcufile version: 2.4
 ============
 ENVIRONMENT:
 ============
 =====================
 DRIVER CONFIGURATION:
 =====================
 NVMe               : Supported
 NVMeOF             : Supported
 SCSI               : Unsupported
 ScaleFlux CSD      : Unsupported
 NVMesh             : Unsupported
 DDN EXAScaler      : Unsupported
 IBM Spectrum Scale : Unsupported
 NFS                : Unsupported
 WekaFS             : Unsupported
 Userspace RDMA     : Unsupported
 --Mellanox PeerDirect : Enabled
 --rdma library        : Not Loaded (libcufile_rdma.so)
 --rdma devices        : Not configured
 --rdma_device_status  : Up: 0 Down: 0
        :

これは2時間くらいかけて調べたところ、どうやら、MOFEDドライバでバイナリ配布されているrpcrdmaモジュールでGPUDirect Storage対応のコードが有効化されないままビルド、配布されてしまっているという事のようである。

MOFEDドライバのソースコードを見てみると、もしCONFIG_GPU_DIRECT_STORAGE=yつきでビルドされているのであれば、/proc/kallsymsにnvfs_opsという関数ポインタ表が出現してしかるべきであるのだが、それが出現していない。

# grep nvfs_ops /proc/kallsyms
ffffffffc0c256c0 b nvfs_ops     [nvme_rdma]
ffffffffc00dc718 b nvfs_ops     [nvme]

という事で、当該モジュールを野良ビルドしてみる事にする。
（なお、NVIDIAの開発チームにはエスカレーション済み。Mellanoxへも展開してくれるでしょう。）

ソースコードの tgz には SRPM が含まれているので、rpcrdmaモジュールを含むmlnx-nfsrdmaのSRPMを展開し、これにCONFIG_GPU_DIRECT_STORAGE=yを付加してビルドする。

これをinsmodしてみると、rpcrdmaモジュールにもnvfs_opsシンボルがエクスポートされているのがわかる。

$ wget http://www.mellanox.com/downloads/ofed/MLNX_OFED-5.3-1.0.0.1/MLNX_OFED_SRC-5.3-1.0.0.1.tgz
$ tar zxvf MLNX_OFED_SRC-5.3-1.0.0.1.tgz
$ cd MLNX_OFED_SRC-5.3-1.0.0.1
$ rpm2cpio SRPMS/mlnx-nfsrdma-5.3-OFED.5.3.0.3.8.1.src.rpm | cpio -idu
$ tar zxvf mlnx-nfsrdma-5.3.tgz
$ cd mlnx-nfsrdma-5.3
$ make CONFIG_GPU_DIRECT_STORAGE=y
$ sudo insmod rpcrdma.ko
$ sudo grep nvfs_ops /proc/kallsyms
ffffffffc319ddc8 b nvfs_ops     [rpcrdma]
ffffffffc0c256c0 b nvfs_ops     [nvme_rdma]
ffffffffc00dc718 b nvfs_ops     [nvme]

この状態で、再度gdscheckコマンドを実行してみると。

$ /usr/local/cuda/gds/tools/gdscheck -p
 GDS release version: 1.0.0.82
 nvidia_fs version:  2.7 libcufile version: 2.4
 ============
 ENVIRONMENT:
 ============
 =====================
 DRIVER CONFIGURATION:
 =====================
 NVMe               : Supported
 NVMeOF             : Supported
 SCSI               : Unsupported
 ScaleFlux CSD      : Unsupported
 NVMesh             : Unsupported
 DDN EXAScaler      : Unsupported
 IBM Spectrum Scale : Unsupported
 NFS                : Supported
 WekaFS             : Unsupported
 Userspace RDMA     : Unsupported
 --Mellanox PeerDirect : Enabled
 --rdma library        : Not Loaded (libcufile_rdma.so)
 --rdma devices        : Not configured
 --rdma_device_status  : Up: 0 Down: 0
        :

イヤッホゥゥゥゥ！！！

早速、GPUDirect StorageのRaw-I/O性能を測定してみる事にする。

$ /usr/local/cuda/gds/tools/gdsio -x 0 -f /mnt/100GB -d 1 -s 96G -i 16M -w 6
IoType: READ XferType: GPUD Threads: 6 DataSetSize: 63143936/100663296(KiB) IOSize: 16384(KiB) Throughput: 7.642794 GiB/sec, Avg_Latency: 12874.833807 usecs ops: 3854 total_time 7.879154 secs

イヤッホゥゥゥゥ！！！

サーバ機材は有り合わせなので、もしかするとSkylake-SP内蔵のPCI-Eコントローラで詰まっているかも（帯域的にはそんな感じがしないでもない）しれないが、NFSという言葉から受ける印象とはずいぶん違ったレベルのパフォーマンスを出しているように見える。

さて、それでは、最も重要な PG-Strom でGPUDirect SQLを用いた場合のパフォーマンスを計測してみる事にする。
（⇒先頭に戻る）

8/21追記：5.4-1.0.3.0 ドライバでは直ってた

上記、rpcrdmaモジュールがGPUDirect Storage対応でビルドされていなかった問題ですが、本エントリを書いた時点のMOFEDドライバ（5.3-1.0.0.1）ではなく、最新の 5.4-1.0.3.0 を使用すれば GPUDirect Storage 関連の機能を有効にしてビルドされるようです。

ドライバ標準のインストールスクリプトを実行しただけの状態で

[root@magro ~]# modinfo rpcrdma
filename:       /lib/modules/4.18.0-305.12.1.el8_4.x86_64/extra/mlnx-nfsrdma/rpcrdma.ko
alias:          xprtrdma
alias:          svcrdma
license:        Dual BSD/GPL
description:    RPC/RDMA Transport
author:         Open Grid Computing and Network Appliance, Inc.
rhelversion:    8.4
srcversion:     6144CA5B71903B01293DD5F
depends:        ib_core,sunrpc,mlx_compat,rdma_cm
name:           rpcrdma
vermagic:       4.18.0-305.12.1.el8_4.x86_64 SMP mod_unload modversions
[root@magro ~]# modprobe rpcrdma
[root@magro ~]# grep nvfs_ops /proc/kallsyms
ffffffffc0f20dc8 b nvfs_ops     [rpcrdma]
ffffffffc0970700 b nvfs_ops     [nvme_rdma]
ffffffffc02ce718 b nvfs_ops     [nvme]

[root@magro ~]# /usr/local/cuda/gds/tools/gdscheck -p
 GDS release version: 1.0.1.3
 nvidia_fs version:  2.7 libcufile version: 2.4
 ============
 ENVIRONMENT:
 ============
 =====================
 DRIVER CONFIGURATION:
 =====================
 NVMe               : Supported
 NVMeOF             : Supported
 SCSI               : Unsupported
 ScaleFlux CSD      : Unsupported
 NVMesh             : Unsupported
 DDN EXAScaler      : Unsupported
 IBM Spectrum Scale : Unsupported
 NFS                : Supported
 WekaFS             : Unsupported
 Userspace RDMA     : Unsupported
 --Mellanox PeerDirect : Enabled
 --rdma library        : Not Loaded (libcufile_rdma.so)
 --rdma devices        : Not configured
 --rdma_device_status  : Up: 0 Down: 0
 =====================
 CUFILE CONFIGURATION:
 =====================
 properties.use_compat_mode : true
 properties.gds_rdma_write_support : true
 properties.use_poll_mode : false
 properties.poll_mode_max_size_kb : 4
 properties.max_batch_io_timeout_msecs : 5
 properties.max_direct_io_size_kb : 16384
 properties.max_device_cache_size_kb : 131072
 properties.max_device_pinned_mem_size_kb : 33554432
 properties.posix_pool_slab_size_kb : 4 1024 16384
 properties.posix_pool_slab_count : 128 64 32
 properties.rdma_peer_affinity_policy : RoundRobin
 properties.rdma_dynamic_routing : 0
 fs.generic.posix_unaligned_writes : false
 fs.lustre.posix_gds_min_kb: 0
 fs.weka.rdma_write_support: false
 profile.nvtx : false
 profile.cufile_stats : 0
 miscellaneous.api_check_aggressive : false
 =========
 GPU INFO:
 =========
 GPU index 0 Tesla V100-PCIE-16GB bar:1 bar size (MiB):16384 supports GDS
 ==============
 PLATFORM INFO:
 ==============
 IOMMU: disabled
 Platform verification succeeded

*1:100GbのN/Wスイッチて結構高いんです。涙。

*2:Q3_1で逆転している原因については調査中