RDBMSを自作して学ぶ：第13章 作ったデータベースを動かす
前回はWALとリカバリを実装し、クラッシュしてもコミット済みのデータが失われず、未コミットの変更が取り消されるようになりました。これで第1章で「実装する範囲」として挙げたコンポーネントがすべてそろいました。今回は、それらを1つにまとめて、CLIから対話的にSQLを実行できるデータベースを完成させます。実際にSQLを打ち込んで動かし、プロセスを強制終了してリカバリを確かめ、最後にベンチマークでボトルネックを探します。
全体の構成
これまでに作ったコンポーネントは、次のように積み重なっています。
┌──────────────────────────────────────────────────┐
│ CLI（main.go） ← 本章 │
├──────────────────────────────────────────────────┤
│ DB・Session（db/） ← 本章 │
├──────────────────────────────────────────────────┤
│ Lexer・Parser（第7章） │
│ Planner・Executor（第8章） Catalog（第9章） │
├──────────────────────────────────────────────────┤
│ TransactionManager（第10章） │
│ LockManager（第11章） LogManager・Recovery（第12章）│
├──────────────────────────────────────────────────┤
│ HeapFile（第5章） B+木（第6章） │
│ BufferPoolManager（第4章） │
│ DiskManager（第3章） │
└──────────────────────────────────────────────────┘本章では、全体を束ねる DB と、1つのクライアント（接続）に相当する Session を db パッケージに作り、その上にCLIを載せます。
DB：全体を束ねる
DB は、データファイル・カタログ・WALを1つのディレクトリにまとめて管理します。
// db/db.go
package db
import (
"fmt"
"os"
"path/filepath"
"github.com/yourname/go-rdbms/sql/catalog"
"github.com/yourname/go-rdbms/sql/executor"
"github.com/yourname/go-rdbms/storage/buffer"
"github.com/yourname/go-rdbms/storage/disk"
"github.com/yourname/go-rdbms/transaction"
"github.com/yourname/go-rdbms/transaction/lock"
"github.com/yourname/go-rdbms/transaction/log"
)
// Config はデータベースの設定
type Config struct {
Dir string // データファイル・カタログ・WALを置くディレクトリ
BufferSize int // バッファプールのフレーム数（0なら1024 = 4MB）
}
// DB はすべてのコンポーネントを束ねたデータベース本体
type DB struct {
dm *disk.FileDiskManager
bpm *buffer.BufferPoolManager
logMgr *log.LogManager
catalog *catalog.Catalog
txMgr *transaction.TransactionManager
exec *executor.Executor
recovered bool
}
// Open はデータベースを開く（なければ新しく作る）
// 前回が異常終了だった場合は、ここでリカバリが行われる
func Open(cfg Config) (*DB, error) {
if cfg.BufferSize == 0 {
cfg.BufferSize = 1024
}
if err := os.MkdirAll(cfg.Dir, 0755); err != nil {
return nil, err
}
// ストレージ層とWAL
dm, err := disk.NewFileDiskManager(filepath.Join(cfg.Dir, "data.db"))
if err != nil {
return nil, err
}
bpm := buffer.NewBufferPoolManager(dm, cfg.BufferSize)
logMgr, err := log.NewLogManager(filepath.Join(cfg.Dir, "wal.log"))
if err != nil {
dm.Close()
return nil, err
}
bpm.SetLogFlusher(func(lsn uint64) error { return logMgr.FlushTo(log.LSN(lsn)) })
// リカバリ → カタログ → インデックス再構築 → チェックポイント の順に行う
recovered, err := log.NewRecoveryManager(bpm, logMgr).Recover()
if err != nil {
dm.Close()
return nil, fmt.Errorf("db: リカバリに失敗しました: %w", err)
}
cat, err := catalog.NewCatalog(bpm, filepath.Join(cfg.Dir, "catalog.json"))
if err != nil {
dm.Close()
return nil, err
}
if recovered {
if err := executor.RebuildIndexes(cat); err != nil {
dm.Close()
return nil, fmt.Errorf("db: インデックスの再構築に失敗しました: %w", err)
}
}
if err := log.Checkpoint(bpm, logMgr); err != nil {
dm.Close()
return nil, err
}
// ロックとトランザクション、SQLの実行
lockMgr := lock.NewLockManager()
return &DB{
dm: dm,
bpm: bpm,
logMgr: logMgr,
catalog: cat,
txMgr: transaction.NewTransactionManager(lockMgr, logMgr),
exec: executor.New(cat, lockMgr),
recovered: recovered,
}, nil
}
// Recovered は、Open時にクラッシュからのリカバリを行ったかどうかを返す
func (db *DB) Recovered() bool { return db.recovered }
// ListTables はテーブル名の一覧を返す
func (db *DB) ListTables() []string { return db.catalog.ListTables() }
// NewSession はSQLを実行するためのセッションを作る
func (db *DB) NewSession() *Session { return &Session{db: db} }
// Close はチェックポイントを取ってからデータベースを閉じる
// 実行中のトランザクションがあれば、それが終わるまで待つ
func (db *DB) Close() error {
err := db.txMgr.Checkpoint(func() error {
return log.Checkpoint(db.bpm, db.logMgr)
})
if err != nil {
return err
}
if err := db.logMgr.Close(); err != nil {
return err
}
return db.dm.Close()
}Open の処理の順番は、第12章の「起動処理の順番」のとおりです。リカバリを行ってからカタログを読み込み、リカバリを行った場合はインデックスを再構築し、最後にチェックポイントを取ります。
Close では、TransactionManager.Checkpoint を通してチェックポイントを取ります。実行中のトランザクションがあれば、それが終わるのを待ってから、ダーティなページをすべて書き出してログを切り詰めます。正常に Close すればログにはCHECKPOINTレコードしか残らないので、次に開くときはリカバリが不要です。
Session：トランザクションの状態を持つ
Session は、1つのクライアントに相当し、そのクライアントが BEGIN で始めたトランザクションを覚えておきます。
// db/session.go
package db
import (
"errors"
"fmt"
"github.com/yourname/go-rdbms/sql/executor"
"github.com/yourname/go-rdbms/sql/parser"
"github.com/yourname/go-rdbms/transaction"
"github.com/yourname/go-rdbms/transaction/lock"
)
// Session は1つのクライアント（接続）に相当し、実行中のトランザクションを覚えておく
// 1つのSessionを複数のゴルーチンから同時に使ってはいけない
type Session struct {
db *DB
tx *transaction.Transaction // BEGIN〜COMMIT/ROLLBACKの間だけnilでない
}
// InTransaction はBEGINで始めたトランザクションの途中かどうかを返す
func (s *Session) InTransaction() bool { return s.tx != nil }
// Exec はSQL文を1つ実行する
// BEGINしていなければ、1文ごとに自動でコミット（失敗したらロールバック）する
func (s *Session) Exec(sql string) (*executor.Result, error) {
stmt, err := parser.Parse(sql)
if err != nil {
return nil, err
}
switch stmt.(type) {
case *parser.BeginStatement:
if s.tx != nil {
return nil, fmt.Errorf("db: 既にトランザクションの途中です")
}
s.tx = s.db.txMgr.Begin()
return &executor.Result{Message: "BEGIN"}, nil
case *parser.CommitStatement:
if s.tx == nil {
return nil, fmt.Errorf("db: トランザクションが開始されていません")
}
tx := s.tx
s.tx = nil
if err := s.db.txMgr.Commit(tx); err != nil {
s.db.txMgr.Rollback(tx)
return nil, err
}
return &executor.Result{Message: "COMMIT"}, nil
case *parser.RollbackStatement:
if s.tx == nil {
return nil, fmt.Errorf("db: トランザクションが開始されていません")
}
tx := s.tx
s.tx = nil
if err := s.db.txMgr.Rollback(tx); err != nil {
return nil, err
}
return &executor.Result{Message: "ROLLBACK"}, nil
}
// BEGINしたトランザクションの中で実行する
if s.tx != nil {
res, err := s.db.exec.Execute(s.tx, stmt)
if errors.Is(err, lock.ErrDeadlock) {
// デッドロックの犠牲になったトランザクションは続けられないのでロールバックする
s.db.txMgr.Rollback(s.tx)
s.tx = nil
return nil, fmt.Errorf("%w（トランザクションをロールバックしました）", err)
}
return res, err
}
// 自動コミット
tx := s.db.txMgr.Begin()
res, err := s.db.exec.Execute(tx, stmt)
if err != nil {
s.db.txMgr.Rollback(tx)
return nil, err
}
if err := s.db.txMgr.Commit(tx); err != nil {
s.db.txMgr.Rollback(tx)
return nil, err
}
return res, nil
}
// Close は途中のトランザクションがあればロールバックする
func (s *Session) Close() error {
if s.tx == nil {
return nil
}
tx := s.tx
s.tx = nil
return s.db.txMgr.Rollback(tx)
}Exec の動きをまとめると、次のようになります。
| 状況 | 動作 |
|---|---|
BEGIN |
トランザクションを開始し、セッションに覚えておく |
COMMIT / ROLLBACK |
覚えておいたトランザクションをコミット／ロールバックする |
| トランザクションの途中のSQL | そのトランザクションの中で実行する。デッドロックの犠牲になったら、ロールバックしてエラーを返す |
BEGIN していないときのSQL |
その1文だけのトランザクションで実行し、成功したらコミット、失敗したらロールバック（自動コミット） |
第11章で残しておいた「ErrDeadlock を受け取ったらロールバックする」処理は、ここで実装しています。
CLI：対話的にSQLを実行する
最後に、SQLを1行ずつ読み込んで実行し、結果を表形式で表示するCLI（REPL: Read-Eval-Print Loop）を作ります。
// main.go
package main
import (
"bufio"
"flag"
"fmt"
"os"
"strings"
"unicode/utf8"
"github.com/yourname/go-rdbms/db"
"github.com/yourname/go-rdbms/sql/executor"
)
func main() {
dir := flag.String("dir", "go-rdbms-data", "データを置くディレクトリ")
flag.Parse()
database, err := db.Open(db.Config{Dir: *dir})
if err != nil {
fmt.Fprintf(os.Stderr, "データベースを開けませんでした: %v\n", err)
os.Exit(1)
}
defer database.Close()
session := database.NewSession()
defer session.Close()
if database.Recovered() {
fmt.Println("前回は正常に終了しなかったため、リカバリを行いました。")
}
fmt.Println("go-rdbms へようこそ！ SQL文は ; で終えてください（\\help でヘルプ、\\q で終了）")
scanner := bufio.NewScanner(os.Stdin)
var buf strings.Builder
for {
switch {
case buf.Len() > 0:
fmt.Print(" -> ")
case session.InTransaction():
fmt.Print("go-rdbms*> ") // * はトランザクションの途中であることを表す
default:
fmt.Print("go-rdbms> ")
}
if !scanner.Scan() {
fmt.Println()
return
}
line := strings.TrimSpace(scanner.Text())
// \ で始まる行はメタコマンド
if buf.Len() == 0 && strings.HasPrefix(line, `\`) {
if quit := handleMetaCommand(database, line); quit {
return
}
continue
}
if line == "" {
continue
}
// ; が来るまで複数行を1つの文としてためる
buf.WriteString(line)
buf.WriteString(" ")
if !strings.HasSuffix(line, ";") {
continue
}
sql := buf.String()
buf.Reset()
result, err := session.Exec(sql)
if err != nil {
fmt.Printf("エラー: %v\n", err)
continue
}
printResult(result)
}
}
// handleMetaCommand はメタコマンドを処理する。終了する場合は true を返す
func handleMetaCommand(database *db.DB, cmd string) bool {
switch cmd {
case `\q`:
return true
case `\tables`:
tables := database.ListTables()
if len(tables) == 0 {
fmt.Println("テーブルはありません")
}
for _, t := range tables {
fmt.Println(t)
}
case `\help`:
fmt.Println(`\tables テーブルの一覧を表示する`)
fmt.Println(`\help このヘルプを表示する`)
fmt.Println(`\q 終了する`)
default:
fmt.Printf("不明なコマンドです: %s\n", cmd)
}
return false
}
// printResult は実行結果を表形式で表示する
func printResult(r *executor.Result) {
switch {
case r.Message != "":
fmt.Println(r.Message)
return
case r.Columns == nil:
fmt.Printf("%d行を挿入しました\n", r.RowsAffected)
return
}
// 各列の幅を、見出しと値のうち最も長いものに合わせる
widths := make([]int, len(r.Columns))
for i, c := range r.Columns {
widths[i] = utf8.RuneCountInString(c)
}
cells := make([][]string, len(r.Rows))
for i, row := range r.Rows {
cells[i] = make([]string, len(row))
for j, v := range row {
cells[i][j] = v.String()
widths[j] = max(widths[j], utf8.RuneCountInString(cells[i][j]))
}
}
printRow(r.Columns, widths)
seps := make([]string, len(widths))
for i, w := range widths {
seps[i] = strings.Repeat("-", w)
}
fmt.Println(strings.Join(seps, "-+-"))
for _, row := range cells {
printRow(row, widths)
}
fmt.Printf("(%d行)\n", len(r.Rows))
}
func printRow(values []string, widths []int) {
padded := make([]string, len(values))
for i, v := range values {
padded[i] = v + strings.Repeat(" ", widths[i]-utf8.RuneCountInString(v))
}
fmt.Println(strings.TrimRight(strings.Join(padded, " | "), " "))
}- SQLは
;で終わるまで複数行にわたって入力できます - トランザクションの途中は、プロンプトが
go-rdbms*>になります \tables（テーブル一覧）、\help、\q（終了）のメタコマンドを用意しましたmainのdeferは後に書いたものから実行されるので、終了時にはまずsession.Close()で途中のトランザクションをロールバックし、その後database.Close()でチェックポイントを取ります
動かしてみる
ビルドして起動する
go build -o go-rdbms .
./go-rdbms -dir data実際に動かした結果です。
go-rdbms へようこそ！ SQL文は ; で終えてください（\help でヘルプ、\q で終了）
go-rdbms> CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(20), age INT);
テーブル USERS を作成しました
go-rdbms> INSERT INTO users VALUES (1, 'Alice', 30);
1行を挿入しました
go-rdbms> INSERT INTO users VALUES (2, 'Bob', 25);
1行を挿入しました
go-rdbms> INSERT INTO users (id, name, age)
-> VALUES (3, 'Charlie', 35);
1行を挿入しました
go-rdbms> SELECT * FROM users;
ID | NAME | AGE
---+---------+----
1 | Alice | 30
2 | Bob | 25
3 | Charlie | 35
(3行)
go-rdbms> SELECT name, age FROM users WHERE age >= 30 AND id <> 1;
NAME | AGE
--------+----
Charlie | 35
(1行)
go-rdbms> CREATE TABLE orders (id INT PRIMARY KEY, user_id INT, item VARCHAR(20));
テーブル ORDERS を作成しました
go-rdbms> INSERT INTO orders VALUES (10, 1, 'book');
1行を挿入しました
go-rdbms> INSERT INTO orders VALUES (11, 3, 'pen');
1行を挿入しました
go-rdbms> SELECT users.name, orders.item FROM users JOIN orders ON users.id = orders.user_id;
USERS.NAME | ORDERS.ITEM
-----------+------------
Alice | book
Charlie | pen
(2行)
go-rdbms> \tables
ORDERS
USERS複数行の入力、WHERE句、JOINが動いています。
トランザクションとロールバック
go-rdbms> BEGIN;
BEGIN
go-rdbms*> INSERT INTO users VALUES (4, 'Dave', 22);
1行を挿入しました
go-rdbms*> SELECT * FROM users WHERE id = 4;
ID | NAME | AGE
---+------+----
4 | Dave | 22
(1行)
go-rdbms*> ROLLBACK;
ROLLBACK
go-rdbms> SELECT * FROM users WHERE id = 4;
ID | NAME | AGE
---+------+----
(0行)
go-rdbms> INSERT INTO users VALUES (1, 'Duplicate', 99);
エラー: executor: 主キー 1 は既に存在します
go-rdbms> \qトランザクションの中では自分が挿入した行が見え、ROLLBACKすると消えます。主キーの重複もエラーになります。
プロセスを強制終了してリカバリを確かめる
次は、トランザクションの途中でプロセスを kill -9 で強制終了してみます。kill -9（SIGKILL）はプロセスに後片付けの機会を与えないので、Close もチェックポイントも実行されず、バッファプールの中身は失われます。
$ ./go-rdbms -dir data2
go-rdbms へようこそ！ SQL文は ; で終えてください（\help でヘルプ、\q で終了）
go-rdbms> CREATE TABLE accounts (id INT PRIMARY KEY, owner VARCHAR(20), balance INT);
テーブル ACCOUNTS を作成しました
go-rdbms> INSERT INTO accounts VALUES (1, 'Alice', 10000);
1行を挿入しました
go-rdbms> INSERT INTO accounts VALUES (2, 'Bob', 5000);
1行を挿入しました
go-rdbms> BEGIN;
BEGIN
go-rdbms*> INSERT INTO accounts VALUES (3, 'Carol', 7000);
1行を挿入しました
go-rdbms*>
（別のターミナルで pkill -9 go-rdbms を実行して強制終了）もう一度起動します。
$ ./go-rdbms -dir data2
前回は正常に終了しなかったため、リカバリを行いました。
go-rdbms へようこそ！ SQL文は ; で終えてください（\help でヘルプ、\q で終了）
go-rdbms> SELECT * FROM accounts;
ID | OWNER | BALANCE
---+-------+--------
1 | Alice | 10000
2 | Bob | 5000
(2行)
go-rdbms> SELECT owner FROM accounts WHERE id = 2;
OWNER
-----
Bob
(1行)自動コミットで挿入したAliceとBobの行は、ページがディスクに書かれていなくてもWALからREDOで復元されています。一方、コミットしていなかったCarolの行はありません。インデックスも再構築されているので、WHERE id = 2 の検索（IndexScan）も正しく動いています。
テストを書く
db パッケージのテストでは、セッションのトランザクション、正常終了と再オープン、異常終了からのリカバリ、2つのセッションの間のデッドロックを確かめます。
// db/db_test.go
package db_test
import (
"errors"
"fmt"
"testing"
"time"
"github.com/yourname/go-rdbms/db"
"github.com/yourname/go-rdbms/transaction/lock"
)
func mustOpen(t *testing.T, dir string) *db.DB {
t.Helper()
database, err := db.Open(db.Config{Dir: dir, BufferSize: 64})
if err != nil {
t.Fatal(err)
}
return database
}
func mustExec(t *testing.T, s *db.Session, sql string) {
t.Helper()
if _, err := s.Exec(sql); err != nil {
t.Fatalf("%s: %v", sql, err)
}
}
func count(t *testing.T, s *db.Session, table string) int {
t.Helper()
res, err := s.Exec("SELECT * FROM " + table)
if err != nil {
t.Fatal(err)
}
return len(res.Rows)
}
func TestDB_SessionTransaction(t *testing.T) {
database := mustOpen(t, t.TempDir())
defer database.Close()
s := database.NewSession()
mustExec(t, s, "CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(20))")
mustExec(t, s, "INSERT INTO users VALUES (1, 'Alice')") // 自動コミット
mustExec(t, s, "BEGIN")
mustExec(t, s, "INSERT INTO users VALUES (2, 'Bob')")
if got := count(t, s, "users"); got != 2 {
t.Errorf("トランザクション内では自分の挿入が見える: want 2, got %d", got)
}
mustExec(t, s, "ROLLBACK")
if got := count(t, s, "users"); got != 1 {
t.Errorf("ROLLBACK後: want 1, got %d", got)
}
if _, err := s.Exec("COMMIT"); err == nil {
t.Error("BEGINしていないCOMMITがエラーにならなかった")
}
}
func TestDB_CloseAndReopen(t *testing.T) {
dir := t.TempDir()
database := mustOpen(t, dir)
s := database.NewSession()
mustExec(t, s, "CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(20))")
mustExec(t, s, "INSERT INTO users VALUES (1, 'Alice')")
if err := database.Close(); err != nil {
t.Fatal(err)
}
database2 := mustOpen(t, dir)
defer database2.Close()
if database2.Recovered() {
t.Error("正常終了したのにリカバリが行われた")
}
if got := count(t, database2.NewSession(), "users"); got != 1 {
t.Errorf("再オープン後: want 1, got %d", got)
}
}
func TestDB_CrashRecovery(t *testing.T) {
dir := t.TempDir()
database := mustOpen(t, dir)
s := database.NewSession()
mustExec(t, s, "CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(20))")
for i := 1; i <= 100; i++ {
mustExec(t, s, fmt.Sprintf("INSERT INTO users VALUES (%d, 'user%d')", i, i))
}
mustExec(t, s, "BEGIN")
mustExec(t, s, "INSERT INTO users VALUES (1000, 'uncommitted')")
// Closeせずに放置して、プロセスが落ちた状況を模す
database2 := mustOpen(t, dir)
defer database2.Close()
if !database2.Recovered() {
t.Error("異常終了後なのにリカバリが行われていない")
}
s2 := database2.NewSession()
if got := count(t, s2, "users"); got != 100 {
t.Errorf("リカバリ後の件数: want 100, got %d", got)
}
res, _ := s2.Exec("SELECT name FROM users WHERE id = 42")
if len(res.Rows) != 1 || res.Rows[0][0].Str != "user42" {
t.Errorf("インデックス経由の検索結果が不正: %v", res.Rows)
}
}
func TestDB_Deadlock(t *testing.T) {
database := mustOpen(t, t.TempDir())
defer database.Close()
s1, s2 := database.NewSession(), database.NewSession()
mustExec(t, s1, "CREATE TABLE a (id INT PRIMARY KEY)")
mustExec(t, s1, "CREATE TABLE b (id INT PRIMARY KEY)")
mustExec(t, s1, "BEGIN")
mustExec(t, s2, "BEGIN")
mustExec(t, s1, "SELECT * FROM a")
mustExec(t, s2, "SELECT * FROM b")
done := make(chan error, 1)
go func() {
_, err := s1.Exec("INSERT INTO b VALUES (1)") // s1 は s2 を待つ
done <- err
}()
time.Sleep(100 * time.Millisecond)
_, err := s2.Exec("INSERT INTO a VALUES (1)") // 閉路ができる
if !errors.Is(err, lock.ErrDeadlock) {
t.Fatalf("デッドロックになるはず: %v", err)
}
t.Logf("s2: %v", err)
if s2.InTransaction() {
t.Error("デッドロックの犠牲になったトランザクションが残っている")
}
if err := <-done; err != nil {
t.Fatal(err)
}
mustExec(t, s1, "COMMIT")
if got := count(t, s1, "b"); got != 1 {
t.Errorf("s1の挿入がコミットされていない: %d件", got)
}
}go test ./db/... -v期待される出力（=== RUN の行は省略）：
db_test.go:132: s2: lock: デッドロックを検出しました（トランザクションをロールバックしました）
--- PASS: TestDB_SessionTransaction (0.04s)
--- PASS: TestDB_CloseAndReopen (0.06s)
--- PASS: TestDB_CrashRecovery (0.45s)
--- PASS: TestDB_Deadlock (0.16s)
ok github.com/yourname/go-rdbms/db 0.848sベンチマーク ── どこがボトルネックか
ベンチマークを書く
Goの testing パッケージには、ベンチマークの仕組みも入っています。Benchmark で始まる関数を書くと、b.N 回の繰り返しにかかった時間から、1回あたりの時間を測ってくれます。次の5つを測ります。
- 1件ずつ自動コミットでINSERTする
- 1つのトランザクションにまとめてINSERTする
- 1万件のテーブルから、主キーで1件を引く（IndexScan）
- 1万件のテーブルから、インデックスのないカラムで1件を引く（SeqScan + Filter）
- 1万件を全件読む
// bench/bench_test.go
package bench_test
import (
"fmt"
"testing"
"github.com/yourname/go-rdbms/db"
)
const numRows = 10000
// setup はテーブルを作ってnumRows件のデータを入れたデータベースを返す
func setup(b *testing.B, rows int) (*db.DB, *db.Session) {
b.Helper()
database, err := db.Open(db.Config{Dir: b.TempDir(), BufferSize: 1024})
if err != nil {
b.Fatal(err)
}
s := database.NewSession()
exec(b, s, "CREATE TABLE bench (id INT PRIMARY KEY, val INT, memo VARCHAR(30))")
exec(b, s, "BEGIN")
for i := 0; i < rows; i++ {
exec(b, s, fmt.Sprintf("INSERT INTO bench VALUES (%d, %d, 'memo-%d')", i, i, i))
}
exec(b, s, "COMMIT")
return database, s
}
func exec(b *testing.B, s *db.Session, sql string) {
if _, err := s.Exec(sql); err != nil {
b.Fatalf("%s: %v", sql, err)
}
}
// 1件ごとにコミットする（毎回WALのfsyncが走る）
func BenchmarkInsertAutoCommit(b *testing.B) {
database, s := setup(b, 0)
defer database.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec(b, s, fmt.Sprintf("INSERT INTO bench VALUES (%d, %d, 'memo')", i, i))
}
}
// 1つのトランザクションにまとめてコミットする（fsyncは最後の1回だけ）
func BenchmarkInsertInTransaction(b *testing.B) {
database, s := setup(b, 0)
defer database.Close()
b.ResetTimer()
exec(b, s, "BEGIN")
for i := 0; i < b.N; i++ {
exec(b, s, fmt.Sprintf("INSERT INTO bench VALUES (%d, %d, 'memo')", i, i))
}
exec(b, s, "COMMIT")
}
// 主キーで1件を引く（IndexScan）
func BenchmarkPointQueryIndexScan(b *testing.B) {
database, s := setup(b, numRows)
defer database.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec(b, s, fmt.Sprintf("SELECT * FROM bench WHERE id = %d", i%numRows))
}
}
// インデックスのないカラムで1件を引く（SeqScan + Filter）
func BenchmarkPointQuerySeqScan(b *testing.B) {
database, s := setup(b, numRows)
defer database.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec(b, s, fmt.Sprintf("SELECT * FROM bench WHERE val = %d", i%numRows))
}
}
// 全件を読む
func BenchmarkFullScan(b *testing.B) {
database, s := setup(b, numRows)
defer database.Close()
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec(b, s, "SELECT * FROM bench")
}
}go test ./bench/... -bench=. -benchmem最初の結果
Apple M4 のMacで実行した結果は次のとおりです（数値は環境によって大きく変わります）。
BenchmarkInsertAutoCommit-10 574 4141773 ns/op 2073 B/op 43 allocs/op
BenchmarkInsertInTransaction-10 1429554 1691 ns/op 2137 B/op 40 allocs/op
BenchmarkPointQueryIndexScan-10 573 4090728 ns/op 2425 B/op 47 allocs/op
BenchmarkPointQuerySeqScan-10 306 8173355 ns/op 4726977 B/op 100168 allocs/op
BenchmarkFullScan-10 307 8104216 ns/op 4741725 B/op 90173 allocs/op分析1：INSERTはfsyncが支配的
自動コミットのINSERTは1件あたり約4.1ms、トランザクションにまとめたINSERTは約1.7µsで、2000倍以上の差があります。違いは、コミットのたびにCOMMITレコードをfsyncするかどうかだけです。つまり自動コミットのINSERTの時間のほとんどは、fsyncを待つ時間です。
特にmacOSでは、Goの File.Sync は通常の fsync ではなく fcntl(F_FULLFSYNC) を呼びます。macOSの fsync はディスク装置のキャッシュまでは書き出さないので、Goは確実に永続化するためにこちらを使っています。そのぶん1回あたり数msかかります。
改善の方向は2つあります。
- トランザクションにまとめる: アプリケーション側で複数のINSERTを1つのトランザクションにすれば、fsyncは最後の1回で済みます
- グループコミット: 同時にコミットしようとしている複数のトランザクションのCOMMITレコードを、1回のfsyncでまとめて永続化します。本シリーズの
FlushToも、バッファにある他のトランザクションのログをまとめて書き出すので、その素地はあります。PostgreSQL（commit_delayの設定）やMySQLも、同じ考え方の仕組みを持っています
分析2：読み取りなのにfsyncしている
おかしな結果が1つあります。主キーで1件を引くだけのIndexScanが約4.1msで、自動コミットのINSERTとほぼ同じ時間になっています。B+木を3ページたどるだけの処理に、数msもかかるはずがありません。
原因は、読み取りしかしていないトランザクションでも、コミットのたびにCOMMITレコードをfsyncしていることです。SELECTも自動コミットの1文トランザクションとして実行されるので、そのたびにfsyncを待っていました。
何も書き込んでいないトランザクションは、COMMITレコードがディスクに届く前にクラッシュしても、リカバリで取り消すべき変更がないので、結果は変わりません。つまりfsyncを待つ必要はありません。PostgreSQLでも、データを変更しなかったトランザクションはCOMMITのWALレコードを書きません。そこで Commit を次のように変えます。
@@ -58,6 +58,10 @@ func (tm *TransactionManager) Commit(tx *Transaction) error {
// COMMITログをfsyncし終えた時点で、コミットが確定する（Durability）
lsn := tm.logMgr.Append(&log.LogRecord{TxID: tx.id, Type: log.RecordCommit})
- if err := tm.logMgr.FlushTo(lsn); err != nil {
- return fmt.Errorf("transaction: COMMITログを永続化できません: %w", err)
+ // 何も書き込んでいないトランザクションは、fsyncを待たなくてよい。
+ // COMMITログが失われても、取り消すべき変更がないので結果は変わらない
+ if len(tx.writes) > 0 {
+ if err := tm.logMgr.FlushTo(lsn); err != nil {
+ return fmt.Errorf("transaction: COMMITログを永続化できません: %w", err)
+ }
}
if err := tm.finish(tx, StatusCommitted); err != nil {改善後の結果
BenchmarkInsertAutoCommit-10 577 4113918 ns/op 2073 B/op 43 allocs/op
BenchmarkInsertInTransaction-10 1315028 1716 ns/op 2114 B/op 40 allocs/op
BenchmarkPointQueryIndexScan-10 1390017 1471 ns/op 2783 B/op 47 allocs/op
BenchmarkPointQuerySeqScan-10 776 2653807 ns/op 4726930 B/op 100169 allocs/op
BenchmarkFullScan-10 1076 2144394 ns/op 4741711 B/op 90173 allocs/op| ベンチマーク | 改善前 | 改善後 |
|---|---|---|
| 主キーで1件（IndexScan） | 4.09 ms | 1.47 µs |
| 非キーで1件（SeqScan） | 8.17 ms | 2.65 ms |
| 全件読み込み | 8.10 ms | 2.14 ms |
IndexScanは約2800倍速くなりました。読み取りのトランザクションが書き込みと同じ重さのfsyncを払っていた、という問題は、ベンチマークを取ってみて初めて気づけたものです。
分析3：インデックスの効果と、スキャンのコスト
改善後の数字を比べると、主キーで1件を引くIndexScan（約1.5µs）は、インデックスのないカラムで1件を探すSeqScan（約2.7ms）の約1800倍速くなっています。SeqScanは1万件すべてを読んでFilterで捨てているのに対し、IndexScanはB+木で目的の1件のRIDを直接求めているからです。第6章で見たO(log N)とO(N)の差が、はっきり数字に表れています。
では、1万件の全件読み込みにかかる約2.1ms（1件あたり約210ns）は、何に使われているのでしょうか。Goのプロファイラ（pprof）で調べてみます。
go test ./bench/ -run xxx -bench=FullScan -cpuprofile cpu.out
go tool pprof -top bench.test cpu.out結果を見ると、CPU時間の多くがGC（ガベージコレクション）とメモリ確保に関係するランタイムの関数で占められていました。-benchmem の結果でも、1回の全件読み込みで約9万回（1行あたり約9回）、合計約4.7MBのメモリを確保しています。ページからレコードをコピーする、フィールドごとにスライスを作る、Value やタプルを作る、といった処理を1行ごとに繰り返しているためです。また、スキャナが1行ごとにページを FetchPage / UnpinPage していること（第5章）も、ミューテックスとLRUリストの操作として現れていました。
改善するとしたら、次のような方向が考えられます。
- スキャン中は同じページをピンしたまま、そのページのスロットをまとめて読む
- レコードのコピーや、行ごとのスライスの確保を減らす（バッファを使い回す）
- 1行ずつではなく複数行をまとめて処理する（ベクトル化実行）
どれも本シリーズの範囲を超えるので、ここでは演習として残しておきます。
ベンチマークからわかったこと
- 書き込みの性能は、fsyncの回数でほぼ決まる。自動コミットは、1文ごとにfsyncのコストを払う
- 何も書き込まないトランザクションは、コミット時にfsyncを待たなくてよい
- インデックスは、1件を探す処理を桁違いに速くする
- 全件読み込みのコストは、行ごとのメモリ確保とページのピン操作に多く使われている
完成したコード全体像
go-rdbms/
├── main.go ← CLI（本章）
├── db/
│ ├── db.go ← DB：全体の組み立て・Open・Close（本章）
│ └── session.go ← Session：トランザクションの状態管理（本章）
├── bench/
│ └── bench_test.go ← ベンチマーク（本章）
├── storage/ ← ストレージエンジン
│ ├── page/ ← ページ・スロット付きページ（第3・5章）
│ ├── disk/ ← DiskManager（第3章）
│ ├── buffer/ ← BufferPoolManager（第4章）
│ ├── heap/ ← ヒープファイル（第5章）
│ └── btree/ ← B+木（第6章）
├── sql/ ← SQLエンジン
│ ├── lexer/ ← 字句解析（第7章）
│ ├── parser/ ← 構文解析（第7章）
│ ├── planner/ ← 実行計画（第8章）
│ ├── executor/ ← 実行エンジン（第8章）
│ └── catalog/ ← カタログ（第8・9章）
└── transaction/ ← トランザクション
├── transaction.go ← Transaction（第10章）
├── transaction_manager.go ← TransactionManager（第10〜12章）
├── txid/ ← トランザクションID（第10章）
├── lock/ ← LockManager（第11章）
└── log/ ← WAL・リカバリ（第12章）第2章で最初に描いたディレクトリ構成に、db（全体の組み立て）、bench（ベンチマーク）、transaction/txid（循環importを避けるための型）が加わった形になりました。
まとめ
DBで全コンポーネントを組み立て、Sessionで1クライアントぶんのトランザクションの状態を管理したBEGINしていないSQLは、1文ごとの自動コミットで実行する。デッドロックの犠牲になったトランザクションはセッションがロールバックする- CLIからSQLを実行でき、
kill -9で強制終了してもリカバリでデータが正しく戻ることを確かめた - ベンチマークで、書き込みの性能がfsyncの回数で決まること、読み取りのトランザクションまでfsyncしていたことがわかり、後者を改善した
- インデックスの効果と、全件読み込みのボトルネック（メモリ確保とページのピン操作）も数字で確かめた
次章では、本シリーズで作ったデータベースを、PostgreSQLやSQLiteといった実際のデータベースと比べます。どのコンポーネントが対応していて、何を省略したのかを整理します。