C#でNpgsqlを使ってPostgreSQLへ大量のバイナリデータを高速に処理する

Npgsqlの各機能を使用した時のベンチマークテストの記事に掲載する処理時間の検証を行っている時にふと気になったのが、PostgreSQLではバルクINSERTやCOPYコマンドがありますが、Npgsqlに同等の機能があるか否かです。

Npgsqlのドキュメンをそれっぽい文字列で検索してみたら発見したので検証してみたいと思います。

実際には頻繁に行うような機能ではありませんが、ある日突然必要になるかもしれないので今のうちに手法だけ覚えておきます。

• もくじ
  • バイナリデータの書き込み
  • バイナリデータの読み込み

Npgsqlの本家情報は

COPYに関する解説は

下記CPOYの文中にあるリンクはPostgreSQLのCOPYコマンドのリンクなのでおそらくPostgreSQLのCOPYコマンドを具現化していると思われます。

私の作成するサンプルソースファイルは

基本的なテーブルは下記構成となります。

テーブル名	型	概要
id	serial	自動的にセットされる通し番号
time	timestamp	トランザクション開始時刻または入力された日付
name	text	任意の文字列
numeric	integer	任意の数値

時間はCREATE TABLEで「time timestamp DEFAULT clock_timestamp()」としており、意図的に時間をセットできますが、何もINSERTしなければ自動的に現在の時間がセットされます。

バイナリデータの書き込み

カラム名「name」とカラム名「numeric」のデータを10万個用意してデータベースに書き込みます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using NpgsqlBinaryImporter writer = con.BeginBinaryImport("COPY data (name, numeric) FROM STDIN (FORMAT BINARY)");
for (int i = 0; i < 100000; i++)
{
    try
    {
        // 1つの行の書き込みを開始し、列を書き込む前に起動しなければなりません。
        writer.StartRow();
        // 書き込む値と型
        writer.Write($"name{i}", NpgsqlDbType.Text);
        writer.Write(i, NpgsqlDbType.Integer);
    }
    catch (InvalidOperationException)
    {
        // BeginBinaryImportで２つのカラムを指定しているのにStartRowメソッド以降Writeメソッドが不足して次のStartRowメソッドを実行すると例外「InvalidOperationException」が発生
    }
}
// コピー開始
try
{
    writer.Complete();
}
catch (PostgresException)
{
    // 違う型のデータを書き込もうとするとCompleteメソッドで例外「PostgresException」が発生
}

上記サンプルコードの3行目にあるNpgsqlConnectionのBeginBinaryImportメソッドの戻り値NpgsqlBinaryImporterを用いてデータをセットしてデータベースに書き込みを行います。

引数はまんま（近い）COPYクエリ文ですね。

9行から12行目までが書き込むデータの準備です。

順番としてはNpgsqlBinaryImporterのStartRowメソッドを実行した後、Writeメソッドを実行します。

注意しなければいけないのが3行目のクエリ文で指定したカラム名（今回はnameとnumeric）の数だけ必ずWriteメソッドを実行する必要があります。

Writeメソッドの実行数を省略すると次のStartRowメソッド実行時に例外「InvalidOperationException」が発生します。

Writeメソッドは2番めの引数を省略可能ですが、型を明確化したほうが無難かと思います。

また、デーテベーステーブルの型と書き込むデータの型が違うとキャスト可能な場合は運良く（運悪く？）書き込まれますが、キャストできない場合は22行目のNpgsqlBinaryImporterのCompletメソッドで例外「PostgresException」が発生します。

ちなみに11行目と12行目を入れ替えて違うカラムに書き込んだ場合も22行目のCompletメソッドで例外「PostgresException」が発生します。

上記手法で10万個のデータを書き込んで時間を計測してみました。

PostgreSQLは同じPCにインストールしネットワークのトラフィックに左右されない状態で実験していますので、ネットワーク上の別PCにあるPostgreSQLを対象とした場合私の結果とは変わってきます。

計測は10回行った平均で0.33秒と極めて高速にデータの書き込みができました。

別記事としますが、クエリ文を使用して同じデータを同じ数だけINSERTした場合、一番早い手法でも3.9秒なので約11倍早くなりました。

バイナリデータの読み込み

上記の書き込んたデータを読み込んでみます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using NpgsqlBinaryExporter reader = con.BeginBinaryExport("COPY data (name, numeric) TO STDOUT (FORMAT BINARY)");
while (reader.StartRow() != -1)
{
    try
    {
        string s = reader.Read<string>(NpgsqlDbType.Text);
        int i = reader.Read<int>(NpgsqlDbType.Integer);
    }
    catch (NullReferenceException)
    {
    // 型が違うと例外が発生
    }
    catch (InvalidCastException)
    {
        // Readメソッドで取得したデータがnullだとキャストの例外が発生
        // reader.Read<string>(NpgsqlDbType.Integer);などの型が不一致だと例外が発生
    }
}

読み込みはNpgsqlConnectionのBeginBinaryExportメソッドの戻り値NpgsqlBinaryExporterを用いて行います。

4行目のNpgsqlBinaryExporterのStartRowメソッドでデータが存在しているかの判断をします。

データが存在していればNpgsqlBinaryExporterのReadメソッドでデータを取得します。

この時取得するデータの型がRead<xxx>のxxxに該当します。

2番めの引数にNpgsqlDbTypeを引数としていますが、これは必ず必要ではありません。

しかし曖昧さを残すよりは明確にした方が後々の不具合にならないと思います。

取得するデータの型が違うと11行目の例外「NullReferenceException」が発生しますが、実験で8行目と9行目を入れ替えた時、9行目の数値を文字列で取得しようとした時は長い時間固まった状態となり、その後例外「NullReferenceException」が起きるのですが11行目の処理に行かず異常終了となってしまいます。

きちんとデバッグを行えば起こり得ない例外ですが、後でデータベースの構成を変えたりしたら注意が必要です。

例外「InvalidCastException」は取得したデータがnullだったりreader.Read<string>(NpgsqlDbType.Integer);などのキャストする型と指定したデータベースの型が不一致だと発生します。

これも同じく10万個のデータを取得する動作を10回行こない所要時間を計測してみました。

やっている事は取得するデータを選択し条件は付けないSELECTと同じです。

SELECTが最速で0.036秒、平均0.043秒に対してバイナリデータの読み込みは最速で0.029秒、平均で0.039秒となり、ほぼ誤差範囲と言ってもいいようなデータでした。

ドキュメントを見るとSELECTやINSERTより高速に処理が可能と説明されていますので、環境次第ではもっと差が出るかもしれません。