NimでDuckDB用のインポート・エキスポートAPIを追加してみた

前回「NimでDuckDBのClient APIを勝手に作ってみた」の続きでDuckDBのAPIを追加しました。
追加したAPIは、DuckDB用のAppender関数で、全データを一括でデータベースに登録する関数と、カラム名を含めた全件データ抽出関数、DataFrameに変換する関数になります。
全データを一括で登録する関数は、元々は、nim-duckdbの作者が作成した関数で、前回の記事では、1日で全て取り入れれなかったので、今回追加しました。
ただ、ちょっとAPIを作成していて、ある違和感を覚えましたので、別途、Sqlite3との比較も行っていきます。

環境

私のPC環境はWindowsで動作させています。

• OS: Windows11
• Nim 2.0.8
• DuckDB 1.0.0

DuckDB用のNimAPI

下記に記載したduckdb.nimを、前回の記事を見て、プロジェクトのduckdb/duckdb.nimに差し替えて、ご利用ください。
前回からの変更点は、エラー用のプロシジャー名を変更しました。４つ同じ名前だと曖昧すぎるとコンパイル時に怒られたため、それぞれ名前を変えたぐらい。
また、Appender用のimportDb関数とexportDb関数、toDf関数を作成しました。
※Appender用の関数は、元々の作者が作成していたので、若干手直ししたぐらいです。
・importDb関数は、元々の作者がfastInsert関数として作成された物です。
・exportDb関数は、カラム名称を含めた、全件データ抽出
・toDf関数は、exportDb関数から出力されたデータをdatamancer用のDataFrameに変換

duckdb/duckdb.nim

const Lib = "duckdb/duckdb.dll" # windows特化

const DuckDBSuccess = 0
const DuckDBError = 1

type
  Db = distinct pointer  ## Generic database pointer.
  Con = distinct pointer  ## Generic connection pointer.
  DuckDBRow* = seq[string]
  DuckDBOperationError* = object of CatchableError

# DuckDB側のC言語構造体
type
  DuckDBConn* = ref object
    database: Db
    connection: Con
  DuckDBColum* = object
    deprecated_data: pointer
    deprecated_nullmask: pointer
    deprecated_type: cint   # duckdb_type
    deprecated_name: cstring
    internal_data: pointer
  DuckDBResult* = object
    deprecated_column_count: int
    deprecated_row_count: int
    deprecated_rows_changed: int
    deprecated_columns: ptr DuckDBColum
    deprecated_error_message: cstring
    internal_data: pointer
  DuckDBPreparedStatement* = pointer
  DuckDBAppender* = pointer

# DuckDBのC言語API　必要と思う物のみを抽出
{.push importc, cdecl, dynlib: Lib.}

proc duckdb_open(path: cstring, database: ptr Db): int
proc duckdb_close(database: ptr Db)
proc duckdb_connect(database: Db, connection: ptr Con): int
proc duckdb_disconnect(connection: ptr Con)
proc duckdb_query(connection: Con, query: cstring, result: ptr DuckDBResult): int
proc duckdb_value_varchar(result: ptr DuckDBResult, col: int, row: int): cstring
proc duckdb_free(v: pointer): void
proc duckdb_column_count(result: DuckDBResult): int
proc duckdb_row_count(result: DuckDBResult): int
proc duckdb_prepare(connection: Con, query: cstring, statement: ptr DuckDBPreparedStatement): int
proc duckdb_bind_varchar(statement: DuckDBPreparedStatement, param_idx: int, val: cstring): int
proc duckdb_execute_prepared(statement: DuckDBPreparedStatement, result: ptr DuckDBResult): int
proc duckdb_result_error(result: ptr DuckDBResult): cstring
proc duckdb_prepare_error(statement: DuckDBPreparedStatement): cstring
proc duckdb_destroy_result(result: ptr DuckDBResult)
proc duckdb_destroy_prepare(statement: ptr DuckDBPreparedStatement)
proc duckdb_appender_create(connection: Con, schema: cstring, table: cstring, appender: ptr DuckDBAppender): int
proc duckdb_append_null(appender: DuckDBAppender): int
proc duckdb_append_varchar(appender: DuckDBAppender, val: cstring): int
proc duckdb_appender_end_row(appender: DuckDBAppender): int
proc duckdb_appender_error(appender: DuckDBAppender): cstring
proc duckdb_appender_destroy(appender: ptr DuckDBAppender): int
proc duckdb_column_name(result: ptr DuckDBResult, col: int): cstring
proc duckdb_column_type(result: ptr DuckDBResult, col: int): cint

{.pop.}

# エラーチェック
proc checkStatus(status: int) =
  if status != DuckDBSuccess :
    raise newException(DuckDBOperationError, "DuckDB operation did not complete sucessfully.")

proc checkResult(status: int, resultParam: DuckDBResult) =
  if status != DuckDBSuccess :
    let msg = duckdb_result_error(resultParam.addr)
    raise newException(DuckDBOperationError, "DuckDB operation did not complete sucessfully. Reason:\n" & $msg)

proc checkStatement(status: int, statement: DuckDBPreparedStatement) =
  if status != DuckDBSuccess :
    let msg = duckdb_prepare_error(statement)
    raise newException(DuckDBOperationError, "DuckDB operation did not complete sucessfully. Reason:\n" & $msg)

proc checkAppender(status: int, appender: DuckDBAppender) =
  if status != DuckDBSuccess :
    let msg = duckdb_appender_error(appender)
    raise newException(DuckDBOperationError, "DuckDB operation did not complete sucessfully. Reason:\n" & $msg)

# 接続処理
proc connect*(path: string): DuckDBConn =
  result = new DuckDBConn
  duckdb_open(path.cstring, result.database.addr).checkStatus()
  duckdb_connect(result.database, result.connection.addr).checkStatus()

# 接続処理（in-memory）
proc connect*(): DuckDBConn =
  result = connect(":memory:")

# クローズ処理
proc close*(conn: DuckDBConn) =
  duckdb_close(conn.database.addr)

# 切断処理
proc disconnect*(conn: DuckDBConn) =
  duckdb_disconnect(conn.connection.addr)

iterator getRows(resultParam: DuckDBResult): DuckDBRow =
  var clmCnt = duckdb_column_count(resultParam)
  var rowCnt = duckdb_row_count(resultParam)
  var duckDBRow = newSeq[string](clmCnt)
  var v: cstring
  for row in 0 ..< rowCnt:
    for clm in 0 ..< clmCnt:
      v = duckdb_value_varchar(resultParam.addr, clm, row)
      duckDBRow[clm] = (
        if v.isNil():
          "NULL"
        else:
          $v
      )
      duckdb_free(v)
    yield duckDBRow

# 行毎にSelect処理
iterator rows*(conn: DuckDBConn, sql: string, args: varargs[string,`$`]): DuckDBRow =
  var resultParam: DuckDBResult
  var statement: DuckDBPreparedStatement
  defer:
    if args.len() != 0:
      duckdb_destroy_prepare(statement.addr)
    duckdb_destroy_result(resultParam.addr)

  if args.len() == 0:
    duckdb_query(conn.connection, sql.cstring, resultParam.addr).checkResult(resultParam)
  else:
    duckdb_prepare(conn.connection, sql.cstring, statement.addr).checkStatement(statement)

    for i, arg in args:
      duckdb_bind_varchar(statement, i+1, arg.cstring).checkStatus()
    duckdb_execute_prepared(statement, resultParam.addr).checkResult(resultParam)

  for duckDBRow in getRows(resultParam):
    yield duckDBRow

# 実行処理
proc exec*(conn: DuckDBConn, sql: string) =
  var resultParam: DuckDBResult
  duckdb_query(conn.connection, sql.cstring, resultParam.addr).checkResult(resultParam)
  duckdb_destroy_result(resultParam.addr)

# 最速の全件入力
proc importDb*(conn: DuckDBConn, table: string, data: seq[DuckDBRow]) =
  var appender: DuckDBAppender
  duckdb_appender_create(conn.connection, nil, table.cstring, appender.addr).checkAppender(appender)

  for row in data:
    for column in row:
      if column == "NULL": duckdb_append_null(appender).checkStatus()
      else: duckdb_append_varchar(appender, column.cstring).checkStatus()
    duckdb_appender_end_row(appender).checkStatus()

  duckdb_appender_destroy(appender.addr).checkStatus()

# ヘッダー付の全件出力
proc exportDb*(conn: DuckDBConn, sql: string, args: varargs[string,`$`]): seq[DuckDBRow] =
  var resultParam: DuckDBResult
  var statement: DuckDBPreparedStatement
  defer:
    if args.len() != 0:
      duckdb_destroy_prepare(statement.addr)
    duckdb_destroy_result(resultParam.addr)

  result = @[]
  if args.len() == 0:
    duckdb_query(conn.connection, sql.cstring, resultParam.addr).checkResult(resultParam)
  else:
    duckdb_prepare(conn.connection, sql.cstring, statement.addr).checkStatement(statement)

    for i, arg in args:
      duckdb_bind_varchar(statement, i+1, arg.cstring).checkStatus()
    duckdb_execute_prepared(statement, resultParam.addr).checkResult(resultParam)

  var clmCnt = duckdb_column_count(resultParam)
  var clmSeq = newSeq[string](clmCnt)
  for clm in  0 ..< clmCnt:
    var clmName = duckdb_column_name(resultParam.addr, clm)
    clmSeq[clm] = $clmName
  result.add(clmSeq)

  for duckDBRow in getRows(resultParam):
    result.add(duckDBRow)

# DataFrame変換
import datamancer
proc toDf*(data: seq[DuckDBRow]): DataFrame =

  if data.len == 0:
    return

  var csvdata: string
  for row in data:
    for clm in row:
      csvdata &= clm
      csvdata &=  ","
    csvdata[csvdata.len-1] = '\n'

  result = parseCsvString(csvdata, sep = ',')

動作検証

コメントを頂いて、トランザクション処理が入ってないから速度が遅くなってるとの、ご指摘を受けました。
DuckDBって、トランザクションが無いデータベースだと勘違いしておりました。ごめんなさい。
（DuckDBの公式サイトを数日しか覗いてなかったので、気づきませんでした！）
再度、計測し直した結果を下記に記載しておきます。

出来上がったDuckDBのAPIを使っての速度比較と、既存のSqlite3との速度比較を行ってみました。

1. DcuckDBによるinsertとimportの登録速度

下記のduckdb_insert.nimをプロジェクト配下に作成し、コンパイル実行を行う。
内容は、通常のinsert文とappender関数を使用したinsertで、各テーブルに1万レコードづつ作成して計測しています。
※コンパイルは実測値を計測するために、最適化を行って実行しています。

duckdb_insert.nim

import duckdb/duckdb
import std/[strformat, times, os]

var con: DuckDBConn
try:
  con = connect("duckdb.db")

  con.exec("DROP TABLE IF EXISTS my_table")
  con.exec("""CREATE TABLE my_table (
                  itema    INTEGER PRIMARY KEY,
                  itemb    INTEGER,
                  itemc    INTEGER,
                  itemd    INTEGER,
                  iteme    INTEGER
                )""")
  con.exec("DROP TABLE IF EXISTS appender_tbl")
  con.exec("""CREATE TABLE appender_tbl (
                  itema    INTEGER PRIMARY KEY,
                  itemb    INTEGER,
                  itemc    INTEGER,
                  itemd    INTEGER,
                  iteme    INTEGER
                )""")

  # 通常 insert 1万回
  var start_time = cpuTime()
  var sql: string
  con.exec("BEGIN TRANSACTION;")
  for i in 0 ..< 10_000:
    sql = fmt"""INSERT INTO my_table VALUES ({i+1}, {i+2}, {i+3}, {i+4}, {i+5});"""
    con.exec(sql)
  con.exec("COMMIT;")

  echo "exec insert time: ", cpuTime() - start_time, " sec"

  # 爆速 insert 1万回
  start_time = cpuTime()
  var data: seq[seq[string]]
  con.exec("BEGIN TRANSACTION;")
  for i in 0 ..< 10_000:
    var row = @[$(i+1), $(i+2), $(i+3), $(i+4), $(i+5)]
    data.add(row)
  con.importDb("appender_tbl", data)
  con.exec("COMMIT;")

  echo "appender insert time: ", cpuTime() - start_time, " sec"
except:
  echo getCurrentExceptionMsg()
finally:
  con.disconnect()
  con.close()

実行結果

$ nim c -d:release --opt:speed duckdb_insert.nim
$ .\duckdb_insert.exe
exec insert time: 4.045999999999999 sec
appender insert time: 0.09299999999999997 sec

結果は、1万レコードの登録では、appender関数を使用したinsert文に対して、通常のinsert文の43倍高速で登録出来ています。

2. DcuckDBによるselect読み込み速度

次に、下記のduckdb_select.nimをプロジェクト配下に作成し、コンパイル実行を行う。
内容は、先に作ったテーブルを全件読み込み、CSVファイルを作成して計測しています。

duckdb_select.nim

import duckdb/duckdb
import std/[strformat, times]

var con: DuckDBConn
var f : File = open("duckdb.csv" , FileMode.fmWrite)
try:
  con = connect("duckdb.db")

  var start_time = cpuTime()
  for item in con.rows("SELECT * FROM my_table ORDER BY itema;"):
    f.writeLine(fmt"""{item[0]:>5},{item[1]:>5},{item[2]:>5},{item[3]:>5},{item[4]:>5}""")

  echo "select time: ", cpuTime() - start_time, " sec"
except:
  echo getCurrentExceptionMsg()
finally:
  f.close()
  con.disconnect()
  con.close()

実行結果

$ nim c -d:release --opt:speed duckdb_select.nim
$ .\duckdb_select.exe
select time: 0.068 sec

結果は、1万レコードの読み込みでは、0.045秒とそれなりに速い結果が出ました。

3. DcuckDBによるselectとupdateの更新速度

次に、下記のduckdb_update.nimをプロジェクト配下に作成し、コンパイル実行を行う。
内容は、先に作ったテーブルを全件読み込み、itemb項目だけを値変更して計測を行いました。

duckdb_update.nim

import duckdb/duckdb
import std/[strformat, times, os]

var con: DuckDBConn
try:
  con = connect("duckdb.db")

  # 通常 select/update 1万回
  var start_time = cpuTime()
  var sql: string
  con.exec("BEGIN TRANSACTION;")
  for item in con.rows("SELECT * FROM my_table ORDER BY itema;"):
    sql = fmt"""UPDATE my_table SET itemb=1 where itema={item[0]};"""
    con.exec(sql)
  con.exec("COMMIT;")

  echo "exec update time: ", cpuTime() - start_time, " sec"

except:
  echo getCurrentExceptionMsg()
finally:
  con.disconnect()
  con.close()

実行結果

$ nim c -d:release --opt:speed duckdb_update.nim
$ .\duckdb_update.exe
exec update time: 4.205 sec

結果は、1万レコードの読み書きでは、4.205秒の結果になりました。

4. Sqlite3によるinsertの登録速度

次に、SQLiteへのアクセスを計測していきます。
プロジェクト配下に、sqlite_insert.nimを作成して計測します。
※nim2.0からDB関連は外だしになりましたので、予めnimble install db_connectorでライブラリを取り込んでください。
プログラム内容は、DuckDBと同じで1万回のレコード登録を行います。

sqlite_insert.nim

import db_connector/db_sqlite
import std/times

var db: DbConn
try:
  db = open("sqlite3.db", "", "", "")

  var start_time = cpuTime()
  db.exec(sql"DROP TABLE IF EXISTS my_table")
  db.exec(sql"""CREATE TABLE my_table (
                  itema    INTEGER PRIMARY KEY,
                  itemb    INTEGER,
                  itemc    INTEGER,
                  itemd    INTEGER,
                  iteme    INTEGER
                )""")

  db.exec(sql"BEGIN")
  for i in 0 ..< 10_000:
    db.exec(sql"INSERT INTO my_table (itema, itemb, itemc, itemd, iteme) VALUES (?, ?, ?, ?, ?);",
            i+1, i+2, i+3, i+4, i+5)
  db.exec(sql"COMMIT")

  echo "Sqlite insert time: ", cpuTime() - start_time, " sec"

except:
  echo getCurrentExceptionMsg()
finally:
  db.close()

実行結果

$ nim c -d:release --opt:speed sqlite_insert.nim
$ .\sqlite_insert.exe
Sqlite insert time: 0.377 sec

結果は、1万レコードの登録で、0.377秒。つまり、SqlLiteでのinsert登録の方が約10倍速い結果が出ました。
DuckDBのimportDb関数を使用した場合に較べれば、約4倍遅いですが、でもSQLiteの方が早いです。

5. Sqlite3によるselectの登録速度

次に、SQLiteのselectアクセスを計測していきます。
プロジェクト配下に、sqlite_select.nimを作成して計測します。
プログラム内容は、DuckDBと同じです。

sqlite_select.nim

import db_connector/db_sqlite
import std/[strformat, times, os]

var db: DbConn
var f : File = open("sqlite.csv" , FileMode.fmWrite)
try:
  db = open("sqlite3.db", "", "", "")

  var start_time = cpuTime()

  for item in db.fastRows(sql"SELECT * FROM my_table ORDER BY itema;"):
    f.writeLine(fmt"""{item[0]:>5},{item[1]:>5},{item[2]:>5},{item[3]:>5},{item[4]:>5}""")

  echo "Sqlite select time: ", cpuTime() - start_time, " sec"

except:
  echo getCurrentExceptionMsg()
finally:
  f.close()
  db.close()

実行結果

$ nim c -d:release --opt:speed sqlite_select.nim
$ .\sqlite_select.exe
Sqlite select time: 0.031 sec

結果は、1万レコードの読み込みで、0.031秒とかなり優秀です。

6. SQLite3によるselectとupdateの更新速度

次に、SQLiteのselect/updateアクセスを計測していきます。
プロジェクト配下に、sqlite_update.nimを作成して計測します。
プログラム内容は、DuckDBと同じです。

sqlite_update.nim

import db_connector/db_sqlite
import std/[strformat, times, os]

var db: DbConn
var f : File = open("sqlite.csv" , FileMode.fmWrite)
try:
  db = open("sqlite3.db", "", "", "")

  var start_time = cpuTime()

  db.exec(sql"BEGIN")
  for item in db.fastRows(sql"SELECT * FROM my_table ORDER BY itema;"):
    db.exec(sql"UPDATE my_table SET itemb=1 where itema=?;", item[0])
  db.exec(sql"COMMIT")

  echo "Sqlite update time: ", cpuTime() - start_time, " sec"

except:
  echo getCurrentExceptionMsg()
finally:
  f.close()
  db.close()

実行結果

$ nim c -d:release --opt:speed sqlite_update.nim
$ .\sqlite_update.exe
Sqlite update time: 0.167 sec

結果は、1万レコードの読み書きですが、DuckDBと較べて25倍速い結果で、こちらも優秀な結果になりました。

7. Python環境でのDuckDBによるinsertの登録速度

最後に、python環境でDuckDBのAPIを使ったInsertを書いて、どうなるか検証してみました。

duckdb_insert.py

import duckdb, time

conn = duckdb.connect('mydb.db')

conn.execute("drop TABLE test_table")
conn.execute("CREATE TABLE test_table (itema INTEGER PRIMARY KEY, itemb INTEGER, itemc INTEGER, itemd INTEGER, iteme INTEGER)")

start_time = time.perf_counter()
conn.begin()
for i in range(10000) :
  sql = f"insert into test_table values ({i+1}, {i+2}, {i+3}, {i+4}, {i+5})"
  conn.execute(sql)
conn.commit()

end_time = time.perf_counter()
execution_time = end_time - start_time
print(f"Program executed in: {execution_time: .5f} seconds")

conn.close()

実行結果

$ python -m venv venv
$ pip install duckdb
$ .\venv\Scripts\Activate.ps1
$ python duckdb_insert.py
Program executed in:  4.28799 seconds

プロジェクトにpythonの仮想環境を作り、pip install duckdbを実行しライブラリを取り入れます。
実行結果は、Nimとほぼ大して変わらないぐらいです。

8. 検証結果

1万レコードの読み書き計測結果

SQL実行コマンド	DuckDB速度	SQLite速度	評価結果
insert	4.0459 sec	0.377 sec	SQLiteの方が10倍速い
importDb(Appender)	0.0929 sec		SQLiteのinsertに較べて4倍速い
select	0.068 sec	0.031 sec	SQLiteの方が倍速い
update	4.205 sec	0.167 sec	SQLiteの方が25倍速い
python insert	4.2879 sec		やはりNimと余り変わらない

結論から言えば、DuckDBは、SQLiteよりは遅い。しかし、Appenderを使った一括インポートを使えば、SQLiteより4倍は速い。
（OLAPとしてインメモリを利用して、DataFrameに渡すから、登録更新とか問題ないんじゃない？と言う意見もあります）
また、NimとかPython関係なく、DuckDB側のinsert/update実行は遅いです。
※DuckDBのConfigurationをいじって、どれぐらいになるかは、後日調査します。

9. DataFrameに変換

最後に、今回作成したAPIで、DuckDBからエクスポートから、DataFrameに変換する方法も記載しておきます。

duckdb_dataframe.nim

import duckdb/duckdb
import std/[strformat, times]
import datamancer

var con: DuckDBConn
try:
  con = connect()               # インメモリ
  con.exec("INSTALL httpfs;")   # HTTPFSのExtensions機能をインストール・ロード設定
  con.exec("LOAD httpfs;")      # (設定しなくても読めたが、設定した方が良いと思います)
  var start_time = cpuTime()

  # DuckDBに登録せず、直接CSVファイルを読み込み、DataFrame型に変換し、DF側の条件で値を抽出
  var df = con.exportDb("""
          SELECT * 
          FROM 'https://raw.githubusercontent.com/duckdb/duckdb/main/data/csv/weather.csv'
          WHERE precipitation != ? AND snow_fall != ? AND snow_depth != ?;""",
          "T", "T", "T")
      .toDf()
      .filter(f{`maximum_temperature` < 80 and `minimum_temperature` > 60})
  echo df

  echo "export->dataframe time: ", cpuTime() - start_time, " sec"
except:
  echo getCurrentExceptionMsg()
finally:
  con.disconnect()
  con.close()

実行結果

$ nim c -d:release --opt:speed duckdb_dataframe.nim
$ .\duckdb_dataframe.exe
DataFrame with 7 columns and 23 rows:
     Idx                   date    maximum_temperature    minimum_temperature    average_temperature          precipitation              snow_fall             snow_depth
  dtype:                 object                    int                    int                  float                  float                  float                    int
       0           "2016-03-10"                     79                     63                     71                      0                      0                      0
       1           "2016-04-01"                     79                     61                     70                   0.02                      0                      0
       2           "2016-06-02"                     78                     62                     70                      0                      0                      0
       3           "2016-06-03"                     70                     63                   66.5                   0.04                      0                      0
       4           "2016-06-05"                     71                     65                     68                   0.91                      0                      0
       5           "2016-06-16"                     74                     65                   69.5                   0.22                      0                      0

export->dataframe time: 0.5 sec

サンプル用CSVファイルは、DuckDB公式のgithubからweather.csvファイルをダウンロードして、プロジェクトフォルダ直下に配置して利用してくださいをダウンロードせずに、拡張機能のHTTPFSで直接読み込む方法で記載変更しました。
上記プログラムでは、exportDbでSQL文の条件でデータを抽出します。
この時に、カラム名称も含めたシーケンスデータになりますので、それをtoDf関数に渡して、DataFrame型に変換しているだけのプログラムです。
後、datamancerの処理でfilterを利用して、条件抽出を行ってます。
datamancerの詳細については、1年半前に記事を書きましたので、そちらの「NimでDataFrameを操作」をご覧ください。

おわりに

色々なサイトで、DuckDBを取り上げていますが、はっきり言って、既存のデータベースとして使用するデータベースではないと思います。
但し、OLAPとして、データ間の変換としてであれば（CSVデータを何らかの条件で抽出した後、DataFrameに変換するとか）、かなり優秀なツールだとは思います。

後、何か記述に間違い事項がございましたら、コメントにて、どしどしご連絡下さい。（笑）

参考になったと思うなら、ハートマークを押せ。以上。