RDBMSを自作して学ぶ：第12章 WAL（Write-Ahead Logging）とリカバリ
前回はロックマネージャを実装し、同時に動くトランザクションが互いに干渉しないようにしました。しかし、まだクラッシュには無防備です。コミットした変更がバッファプールにしか載っていない状態でプロセスが落ちれば変更は失われますし、ディスクに書き出された未コミットの変更は、そのまま残ってしまいます。今回は、データベースの変更をすべて記録するWAL（Write-Ahead Logging）と、再起動時にWALを使ってデータを正しい状態に戻すリカバリを実装します。
クラッシュが起きるとどうなるか
バッファプールとディスクのずれ
第4章のバッファプールは、ダーティなページをすぐにはディスクに書きません。ページが追い出されるときか、明示的にフラッシュしたときに初めて書き込みます。そのため、メモリ上のデータとディスク上のデータには、常にずれがあります。
このずれが、クラッシュ時に2つの問題を引き起こします。
問題1: コミットした変更が失われる（永続性の違反）
TX1: users に行を挿入（バッファプール上のページだけが変わる）
TX1: COMMIT → 成功
← ここでクラッシュ！（ページはまだディスクに書かれていない）
再起動後: 挿入したはずの行がない問題2: コミットしていない変更が残る（原子性の違反）
TX2: users に大量の行を挿入（まだコミットしていない）
バッファプールが満杯になり、TX2が変更したページがディスクに書き出される
← ここでクラッシュ！
再起動後: コミットしていない行がディスクに残っている単純な解決策とその問題
これらを避ける最も単純な方法は、次の2つを守ることです。
- FORCE: コミットのときに、そのトランザクションが変更したページをすべてディスクに書く（問題1を防ぐ）
- NO-STEAL: コミットしていないトランザクションが変更したページは、ディスクに書き出さない（問題2を防ぐ）
しかしFORCEでは、コミットのたびにディスク上のあちこちのページへランダムに書き込むことになり、非常に遅くなります。NO-STEALでは、大きなトランザクションが変更したページをすべてバッファプールに載せ続けなければなりません。
そこで実際のデータベースは、逆のSTEAL（未コミットのページも書き出してよい）とNO-FORCE（コミット時にページを書かなくてよい）を選び、その代わりにログを使って問題1・問題2の両方を解決します。本シリーズも同じ方針を取ります。
WALの考え方
すべての変更をログに書く
WALは、データベースへの変更を1件ずつログレコードとして、追記専用のログファイルに記録する仕組みです。ログは常にファイルの末尾に追記するだけなので、ディスクへの書き込みがシーケンシャルになり、あちこちのページを書き換えるよりずっと速く書けます。
ログがあれば、クラッシュしたときに次のことができます。
- ディスクに反映されていなかったコミット済みの変更を、ログから**再実行（REDO）**する
- ディスクに書き出されてしまった未コミットの変更を、ログを見て取り消す（UNDO）
ディスクに書く順序の2つのルール
ログで復旧できるためには、ログがディスクにあることが前提です。WAL（Write-Ahead Logging）という名前は、「データより先にログを書く」ことに由来します。具体的には次の2つのルールを守ります。
ルール1: データページをディスクに書く前に、そのページへの変更を記録したログをディスクに書く
未コミットの変更がディスクに書き出されたとき（STEAL）、その変更のログがディスクになければ、クラッシュ後に「何を取り消せばよいか」がわかりません。
ルール2: コミットを完了する前に、そのトランザクションのログ（COMMITレコードを含む）をディスクに書く
コミットした変更のページがディスクになくても（NO-FORCE）、ログさえあれば再実行できます。言い換えると、COMMITレコードをfsyncし終えた瞬間が、コミットが確定する瞬間です。
LSNとページLSN
2つのルールを実装するために、ログレコードには1から始まる通し番号 LSN（Log Sequence Number） を振ります。そして各ページに、「このページに最後に反映したログレコードのLSN」をページLSNとして記録します。第5章と第6章で、ページの [16:24] を予約しておいたのはこのためです。
ページLSNは2か所で使います。
- ルール1の実装: バッファプールがページをディスクに書く前に、「ページLSNまでのログがディスクにあるか」を確かめ、なければ先にログをフラッシュする
- リカバリ: ログレコードを再実行するとき、ページLSN ≥ ログのLSN なら、その変更はすでにページに反映済みなので何もしない。これにより、同じログを何度再実行しても結果が変わらなくなる（冪等になる）
ログレコードの設計
何を記録するか
本シリーズでログに記録するのは、次のレコードです。
| 種類 | 内容 | 用途 |
|---|---|---|
| BEGIN | トランザクションの開始 | 分析 |
| COMMIT | コミット | 分析（これがあればコミット済み） |
| ABORT | ロールバックの完了 | 分析（取り消しが済んでいる） |
| INSERT | ページID・スロット番号・レコードの中身 | REDO・UNDO |
| DELETE | ページID・スロット番号 | REDO（ロールバックの取り消し操作で出る） |
| NEWPAGE | 新しいページのIDと、つないだ元のページのID | REDO |
| CHECKPOINT | チェックポイント | ログの先頭の目印 |
INSERTのレコードは、「どのページ（物理的な位置）の、どのスロットに（ページ内の論理的な操作）、何を入れたか」を記録します。ページは物理的に特定し、ページ内の変更は操作として記録するこの方式を、**物理論理ロギング（physiological logging）**と呼びます。
DELETEのレコードは、SQLのDELETE文ではなく（本シリーズのSQLにはDELETEがありません）、ロールバックでINSERTを取り消したときに出るレコードです。取り消し操作もログに記録しておくことで、「ロールバックの途中でクラッシュした」場合でも、どこまで取り消したかを再現できます。取り消し操作を記録したこのようなログは、ARIES（後述）では補償ログレコード（CLR: Compensation Log Record）と呼ばれます。
NEWPAGEは、ヒープファイルに新しいページをつないだことを記録します。この操作は、新しいページの初期化と元の末尾ページのリンクの書き換えという、2つのページにまたがる変更です。
記録しないもの
次のものはログに記録しません。
- B+木の変更: ノードの分割は複数のページにまたがる複雑な操作で、ログに記録して再実行するのは大変です。本シリーズでは、リカバリの後でヒープファイルの中身からインデックスを作り直すことにします
- CREATE TABLE: 第9章で実装したとおり、テーブルを作った時点でページとカタログをディスクに書くので、ログは不要です
ファイル形式
ログファイルには、レコードを次の形式で順に書きます。
[本体の長さ(4B)][本体のCRC32(4B)][本体]
[本体の長さ(4B)][本体のCRC32(4B)][本体]
...ログの末尾に書いている途中でクラッシュすると、最後のレコードが中途半端に残ります。長さが足りない、またはCRC（チェックサム）が合わないレコードを見つけたら、そこから先は書きかけだったと判断して捨てます。
実装：ページLSNとバッファプール
ページLSN
予約しておいた位置に、ページLSNを読み書きするメソッドを追加します。
@@ -66,4 +66,17 @@ func (p *Page) SetNumSlots(n uint32) {
}
+// ページLSNの位置（第5章のヒープページ・第6章のB+木ノードで予約しておいた8バイト）
+const offsetLSN = 16
+
+// GetLSN はこのページに最後に反映したWALレコードのLSNを返す
+func (p *Page) GetLSN() uint64 {
+ return binary.BigEndian.Uint64(p[offsetLSN : offsetLSN+8])
+}
+
+// SetLSN はページLSNを設定する
+func (p *Page) SetLSN(lsn uint64) {
+ binary.BigEndian.PutUint64(p[offsetLSN:offsetLSN+8], lsn)
+}
+
// Data はヘッダを除いたデータ領域のスライスを返す
func (p *Page) Data() []byte {バッファプールでルール1を守る
バッファプールには、「ページを書き出す前に呼ぶ関数」を登録できるようにします。ログマネージャはバッファプールより上の層にあるので、バッファプールがログマネージャのパッケージを直接importすると、依存関係が逆向きになってしまいます。そこで関数を外から渡してもらう形にしています。
@@ -26,4 +26,5 @@ type BufferPoolManager struct {
freeList []FrameID
replacer Replacer
+ flushLog func(lsn uint64) error // WAL: ページを書き出す前に呼ぶ（第12章）
}
@@ -133,4 +134,23 @@ func (bpm *BufferPoolManager) UnpinPage(pageID page.PageID, isDirty bool) error
}
+// SetLogFlusher はページをディスクに書き出す前に呼ぶ関数を登録する
+// 「ページLSNまでのログがディスクにある」ことを保証するために使う（WALのルール）
+func (bpm *BufferPoolManager) SetLogFlusher(f func(lsn uint64) error) {
+ bpm.mu.Lock()
+ defer bpm.mu.Unlock()
+ bpm.flushLog = f
+}
+
+// writeBack はフレームの内容をディスクに書き出す
+// 先にページLSNまでのログをフラッシュしてから、ページを書く
+func (bpm *BufferPoolManager) writeBack(f *frame) error {
+ if bpm.flushLog != nil {
+ if err := bpm.flushLog(f.page.GetLSN()); err != nil {
+ return fmt.Errorf("buffer: ページ %d のログのフラッシュに失敗しました: %w", f.pageID, err)
+ }
+ }
+ return bpm.diskManager.WritePage(f.pageID, &f.page)
+}
+
// FlushPage は指定ページをディスクに書き戻す
func (bpm *BufferPoolManager) FlushPage(pageID page.PageID) error {
@@ -150,5 +170,5 @@ func (bpm *BufferPoolManager) flushFrame(pageID page.PageID) error {
return nil
}
- if err := bpm.diskManager.WritePage(pageID, &f.page); err != nil {
+ if err := bpm.writeBack(f); err != nil {
return fmt.Errorf("buffer: ページ %d の書き戻しに失敗しました: %w", pageID, err)
}
@@ -190,5 +210,5 @@ func (bpm *BufferPoolManager) allocateFrame() (FrameID, error) {
victim := &bpm.frames[victimID]
if victim.isDirty {
- if err := bpm.diskManager.WritePage(victim.pageID, &victim.page); err != nil {
+ if err := bpm.writeBack(victim); err != nil {
// 書き戻せなかったページは追い出さず、追い出し候補に戻しておく
bpm.replacer.Unpin(victimID)ページを書き出す2か所（追い出しと明示的なフラッシュ）を writeBack にまとめ、書き出しの前に必ずページLSNまでのログをフラッシュするようにしました。
実装：ログレコードとLogManager
ログレコード
// transaction/log/log_record.go
package log
import (
"encoding/binary"
"fmt"
"github.com/yourname/go-rdbms/storage/page"
"github.com/yourname/go-rdbms/transaction/txid"
)
// LSN (Log Sequence Number) はログレコードに振る、1から単調増加する番号
type LSN uint64
// RecordType はログレコードの種類
type RecordType uint8
const (
RecordBegin RecordType = iota + 1 // トランザクション開始
RecordCommit // コミット
RecordAbort // ロールバック完了
RecordInsert // レコードの挿入（REDO・UNDO用）
RecordDelete // レコードの削除（REDO用。ロールバックの取り消し操作で出る）
RecordNewPage // ヒープへのページ追加（REDO用）
RecordCheckpoint // チェックポイント
)
func (t RecordType) String() string {
names := map[RecordType]string{
RecordBegin: "BEGIN", RecordCommit: "COMMIT", RecordAbort: "ABORT",
RecordInsert: "INSERT", RecordDelete: "DELETE", RecordNewPage: "NEWPAGE",
RecordCheckpoint: "CHECKPOINT",
}
return names[t]
}
// LogRecord は1件のログレコード
type LogRecord struct {
LSN LSN
TxID txid.TxID
Type RecordType
PageID page.PageID // INSERT/DELETE: 対象のページ, NEWPAGE: 追加したページ
SlotID uint16 // INSERT/DELETE: 対象のスロット
PrevPageID page.PageID // NEWPAGE: 新しいページをつないだ元のページ
Data []byte // INSERT: 挿入したレコード
}
// 固定長部分: LSN(8) + TxID(4) + Type(1) + PageID(4) + SlotID(2) + PrevPageID(4) + Dataの長さ(4)
const recordHeaderSize = 27
// encode はログレコードをバイト列にする
func (r *LogRecord) encode() []byte {
buf := make([]byte, recordHeaderSize, recordHeaderSize+len(r.Data))
binary.BigEndian.PutUint64(buf[0:], uint64(r.LSN))
binary.BigEndian.PutUint32(buf[8:], uint32(r.TxID))
buf[12] = byte(r.Type)
binary.BigEndian.PutUint32(buf[13:], uint32(r.PageID))
binary.BigEndian.PutUint16(buf[17:], r.SlotID)
binary.BigEndian.PutUint32(buf[19:], uint32(r.PrevPageID))
binary.BigEndian.PutUint32(buf[23:], uint32(len(r.Data)))
return append(buf, r.Data...)
}
// decodeRecord はバイト列をログレコードに戻す
func decodeRecord(buf []byte) (*LogRecord, error) {
if len(buf) < recordHeaderSize {
return nil, fmt.Errorf("log: レコードが短すぎます")
}
dataLen := int(binary.BigEndian.Uint32(buf[23:]))
if len(buf) != recordHeaderSize+dataLen {
return nil, fmt.Errorf("log: レコードの長さが不正です")
}
return &LogRecord{
LSN: LSN(binary.BigEndian.Uint64(buf[0:])),
TxID: txid.TxID(binary.BigEndian.Uint32(buf[8:])),
Type: RecordType(buf[12]),
PageID: page.PageID(binary.BigEndian.Uint32(buf[13:])),
SlotID: binary.BigEndian.Uint16(buf[17:]),
PrevPageID: page.PageID(binary.BigEndian.Uint32(buf[19:])),
Data: append([]byte(nil), buf[recordHeaderSize:]...),
}, nil
}LogManager
// transaction/log/log_manager.go
package log
import (
"bufio"
"encoding/binary"
"errors"
"fmt"
"hash/crc32"
"io"
"os"
"sync"
)
// ログファイル上では、1件のレコードを次の形式で書く
//
// [本体の長さ(4B)][本体のCRC32(4B)][本体]
//
// 書き込みの途中でクラッシュすると末尾のレコードが中途半端に残るが、
// 長さが足りない・CRCが合わないことで検出して捨てられる
const frameHeaderSize = 8
// LogManager はWALファイルへの追記・フラッシュ・読み出しを管理する
type LogManager struct {
mu sync.Mutex
path string
file *os.File
buf []byte // まだファイルに書いていないログ
nextLSN LSN
flushedLSN LSN // このLSNまでのログはfsync済み
}
// NewLogManager はWALファイルを開く。既存のログがあれば最後のLSNの続きから番号を振る
func NewLogManager(path string) (*LogManager, error) {
f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0644)
if err != nil {
return nil, fmt.Errorf("log: %s を開けません: %w", path, err)
}
records, validEnd, err := readRecords(f)
if err != nil {
f.Close()
return nil, err
}
// 末尾の壊れたレコードを切り捨て、続きはその位置から書く
if err := f.Truncate(validEnd); err != nil {
f.Close()
return nil, err
}
if _, err := f.Seek(validEnd, io.SeekStart); err != nil {
f.Close()
return nil, err
}
lm := &LogManager{path: path, file: f, nextLSN: 1}
if len(records) > 0 {
lm.nextLSN = records[len(records)-1].LSN + 1
}
lm.flushedLSN = lm.nextLSN - 1
return lm, nil
}
// Append はログレコードにLSNを振ってメモリ上のバッファに追加し、そのLSNを返す
// この時点ではまだディスクに書かれていない
func (lm *LogManager) Append(rec *LogRecord) LSN {
lm.mu.Lock()
defer lm.mu.Unlock()
rec.LSN = lm.nextLSN
lm.nextLSN++
lm.buf = appendFrame(lm.buf, rec)
return rec.LSN
}
// appendFrame はレコードを [長さ][CRC32][本体] の形式でbufの後ろに付け足す
func appendFrame(buf []byte, rec *LogRecord) []byte {
body := rec.encode()
buf = binary.BigEndian.AppendUint32(buf, uint32(len(body)))
buf = binary.BigEndian.AppendUint32(buf, crc32.ChecksumIEEE(body))
return append(buf, body...)
}
// FlushTo はlsnまでのログがディスクに永続化されていることを保証する
func (lm *LogManager) FlushTo(lsn LSN) error {
lm.mu.Lock()
defer lm.mu.Unlock()
if lsn <= lm.flushedLSN {
return nil // もうfsync済み
}
return lm.flushLocked()
}
// Flush はバッファにあるログをすべてディスクに永続化する
func (lm *LogManager) Flush() error {
lm.mu.Lock()
defer lm.mu.Unlock()
return lm.flushLocked()
}
// flushLocked はバッファの内容をすべてファイルに書いてfsyncする（lm.muを持った状態で呼ぶ）
func (lm *LogManager) flushLocked() error {
if len(lm.buf) > 0 {
if _, err := lm.file.Write(lm.buf); err != nil {
return fmt.Errorf("log: 書き込みに失敗しました: %w", err)
}
lm.buf = lm.buf[:0]
}
if err := lm.file.Sync(); err != nil {
return fmt.Errorf("log: fsyncに失敗しました: %w", err)
}
lm.flushedLSN = lm.nextLSN - 1
return nil
}
// ReadAll はログファイルの全レコードを先頭から読む（リカバリ用）
func (lm *LogManager) ReadAll() ([]*LogRecord, error) {
lm.mu.Lock()
defer lm.mu.Unlock()
if err := lm.flushLocked(); err != nil {
return nil, err
}
f, err := os.Open(lm.path)
if err != nil {
return nil, err
}
defer f.Close()
records, _, err := readRecords(f)
return records, err
}
// Truncate はログを捨て、CHECKPOINTレコード1件だけのログに置き換える
// チェックポイント（すべての変更がデータファイルに反映済み）のときにだけ呼ぶこと。
// LSNは続きの番号を使うので、データファイル上のページLSNより必ず大きくなる
func (lm *LogManager) Truncate() error {
lm.mu.Lock()
defer lm.mu.Unlock()
rec := &LogRecord{LSN: lm.nextLSN, Type: RecordCheckpoint}
lm.nextLSN++
// 新しいログを一時ファイルに書いてfsyncし、リネームで差し替える。
// 途中で落ちても、古いログか新しいログのどちらかが必ず残る
tmp := lm.path + ".tmp"
f, err := os.Create(tmp)
if err != nil {
return err
}
if _, err := f.Write(appendFrame(nil, rec)); err != nil {
f.Close()
return err
}
if err := f.Sync(); err != nil {
f.Close()
return err
}
if err := os.Rename(tmp, lm.path); err != nil {
f.Close()
return err
}
lm.file.Close()
lm.file = f // 以降はこの新しいファイルの末尾に追記する
lm.buf = lm.buf[:0]
lm.flushedLSN = rec.LSN
return nil
}
// Close はバッファを書き出してファイルを閉じる
func (lm *LogManager) Close() error {
lm.mu.Lock()
defer lm.mu.Unlock()
if err := lm.flushLocked(); err != nil {
return err
}
return lm.file.Close()
}
// readRecords はファイルの先頭からレコードを読み、正しく読めた末尾の位置も返す
func readRecords(f *os.File) ([]*LogRecord, int64, error) {
if _, err := f.Seek(0, io.SeekStart); err != nil {
return nil, 0, err
}
r := bufio.NewReader(f)
var records []*LogRecord
var offset int64
header := make([]byte, frameHeaderSize)
for {
// io.ReadFull は指定した長さを読み切るまで読む（足りなければエラー）
if _, err := io.ReadFull(r, header); err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
return records, offset, nil
}
return nil, 0, err
}
body := make([]byte, binary.BigEndian.Uint32(header[0:4]))
if _, err := io.ReadFull(r, body); err != nil {
return records, offset, nil // 書きかけのレコード
}
if crc32.ChecksumIEEE(body) != binary.BigEndian.Uint32(header[4:8]) {
return records, offset, nil // 壊れたレコード
}
rec, err := decodeRecord(body)
if err != nil {
return records, offset, nil
}
records = append(records, rec)
offset += int64(frameHeaderSize + len(body))
}
}ポイントを見ていきます。
Appendはメモリに書くだけ
Append はLSNを振ってメモリ上のバッファ（buf）に追加するだけで、ディスクには書きません。レコードを1件書くたびにfsyncしていたら非常に遅くなるからです。
FlushToは「そこまで」を保証する
FlushTo(lsn) は、lsn までのログがディスクにあることを保証します。すでにfsync済みなら何もしません。まだなら、バッファにあるログをすべて書いてfsyncします。バッファプールがページを書き出すとき（ルール1）と、コミットするとき（ルール2）に呼ばれます。
起動時は続きのLSNから振る
NewLogManager は既存のログファイルを読み、最後のレコードの次のLSNから番号を振ります。ページLSNとの比較で使うので、LSNが巻き戻ったり重複したりしてはいけません。末尾に書きかけのレコードがあれば、ファイルを切り詰めて取り除きます。
読み込みには io.ReadFull を使う
bufio.Reader の Read は、要求した長さより短いデータを返すことがあります。io.ReadFull なら、指定した長さを読み切るか、エラーを返すかのどちらかです。
Truncateはチェックポイント用
Truncate は、ログをCHECKPOINTレコード1件だけのファイルに置き換えます。詳しくは後のチェックポイントの節で説明します。
実装：ヒープファイルにWALを組み込む
Loggerインターフェース
ヒープファイルがページを変更するたびに、ログを書いてページLSNを更新するようにします。ただし、ストレージ層の heap パッケージが上位のログマネージャ（transaction/log）をimportすると、依存が逆向きになります。しかも log パッケージはRIDを使うために heap をimportしているので、循環importにもなってしまいます。そこで、ログを書くためのインターフェース Logger を heap パッケージの側に定義し、実装は上位から渡してもらいます。
@@ -11,4 +11,12 @@ import (
)
+// Logger はヒープへの変更をWALに記録する（第12章）
+// 各メソッドは書き込んだログレコードのLSNを返す
+type Logger interface {
+ LogInsert(rid RID, data []byte) uint64
+ LogDelete(rid RID) uint64
+ LogNewPage(prevPageID, newPageID page.PageID) uint64
+}
+
// HeapFile は1つのテーブルのレコードを、連結されたスロット付きページの列として管理する
type HeapFile struct {
@@ -59,5 +67,6 @@ func (hf *HeapFile) FirstPageID() page.PageID {
// InsertRecord はレコードを格納してRIDを返す
-func (hf *HeapFile) InsertRecord(data []byte) (RID, error) {
+// lgがnilでなければ、変更をWALに記録してページLSNを更新する
+func (hf *HeapFile) InsertRecord(lg Logger, data []byte) (RID, error) {
if len(data) > page.MaxRecordSize {
return RID{}, fmt.Errorf("heap: レコードが大きすぎます（%dバイト、上限%dバイト）",
@@ -77,5 +86,9 @@ func (hf *HeapFile) InsertRecord(data []byte) (RID, error) {
slotID, err := sp.InsertRecord(data)
if err == nil {
- return RID{PageID: lastID, SlotID: slotID}, hf.bpm.UnpinPage(lastID, true)
+ rid := RID{PageID: lastID, SlotID: slotID}
+ if lg != nil {
+ p.SetLSN(lg.LogInsert(rid, data))
+ }
+ return rid, hf.bpm.UnpinPage(lastID, true)
}
if !errors.Is(err, page.ErrNoSpace) {
@@ -93,4 +106,10 @@ func (hf *HeapFile) InsertRecord(data []byte) (RID, error) {
newSP.Init()
sp.SetNextPageID(newID)
+ if lg != nil {
+ // ページの追加は、2つのページ（元の末尾と新しいページ）にまたがる変更
+ lsn := lg.LogNewPage(lastID, newID)
+ p.SetLSN(lsn)
+ newP.SetLSN(lsn)
+ }
if err := hf.bpm.UnpinPage(lastID, true); err != nil {
return RID{}, err
@@ -103,5 +122,9 @@ func (hf *HeapFile) InsertRecord(data []byte) (RID, error) {
return RID{}, err
}
- return RID{PageID: newID, SlotID: slotID}, hf.bpm.UnpinPage(newID, true)
+ rid := RID{PageID: newID, SlotID: slotID}
+ if lg != nil {
+ newP.SetLSN(lg.LogInsert(rid, data))
+ }
+ return rid, hf.bpm.UnpinPage(newID, true)
}
@@ -120,17 +143,24 @@ func (hf *HeapFile) GetRecord(rid RID) ([]byte, error) {
// DeleteRecord はRIDのレコードを削除する
-func (hf *HeapFile) DeleteRecord(rid RID) error {
+func (hf *HeapFile) DeleteRecord(lg Logger, rid RID) error {
hf.mu.Lock()
defer hf.mu.Unlock()
+ return DeleteAt(hf.bpm, lg, rid)
+}
- p, err := hf.bpm.FetchPage(rid.PageID)
+// DeleteAt はRIDのレコードを削除する（HeapFileを経由せずに使えるようにしたもの）
+func DeleteAt(bpm *buffer.BufferPoolManager, lg Logger, rid RID) error {
+ p, err := bpm.FetchPage(rid.PageID)
if err != nil {
return err
}
if err := page.NewSlottedPage(p).DeleteRecord(rid.SlotID); err != nil {
- hf.bpm.UnpinPage(rid.PageID, false)
+ bpm.UnpinPage(rid.PageID, false)
return err
}
- return hf.bpm.UnpinPage(rid.PageID, true)
+ if lg != nil {
+ p.SetLSN(lg.LogDelete(rid))
+ }
+ return bpm.UnpinPage(rid.PageID, true)
}
変更のパターンはどれも同じで、ページをピンしている間に「ページを変更 → ログを書く → ページLSNを設定」の順で行い、それからアンピンします。ピン中のページは追い出されず、第9章の FlushAllPages もピン中のページを書き出さないので、「ページは変更済みなのに、ログもページLSNもまだ」という状態のページがディスクに書かれることはありません。
DeleteAt は、HeapFile を経由せずにバッファプールとRIDだけでレコードを削除する関数です。リカバリでは、カタログを読み込む前にページ単位で処理するので、この形が必要になります。lg に nil を渡すとログを書かないので、第5章のテストなどはそのまま使えます（引数に nil を足すだけです）。
TxLogger
トランザクションごとに、自分のTxIDでログを書く Logger の実装を用意します。
// transaction/log/tx_logger.go
package log
import (
"github.com/yourname/go-rdbms/storage/heap"
"github.com/yourname/go-rdbms/storage/page"
"github.com/yourname/go-rdbms/transaction/txid"
)
// TxLogger は1つのトランザクションのヒープ操作をWALに記録する（heap.Loggerの実装）
type TxLogger struct {
lm *LogManager
txID txid.TxID
}
var _ heap.Logger = (*TxLogger)(nil)
func NewTxLogger(lm *LogManager, txID txid.TxID) *TxLogger {
return &TxLogger{lm: lm, txID: txID}
}
func (l *TxLogger) LogInsert(rid heap.RID, data []byte) uint64 {
return uint64(l.lm.Append(&LogRecord{
TxID: l.txID, Type: RecordInsert, PageID: rid.PageID, SlotID: rid.SlotID, Data: data,
}))
}
func (l *TxLogger) LogDelete(rid heap.RID) uint64 {
return uint64(l.lm.Append(&LogRecord{
TxID: l.txID, Type: RecordDelete, PageID: rid.PageID, SlotID: rid.SlotID,
}))
}
func (l *TxLogger) LogNewPage(prevPageID, newPageID page.PageID) uint64 {
return uint64(l.lm.Append(&LogRecord{
TxID: l.txID, Type: RecordNewPage, PageID: newPageID, PrevPageID: prevPageID,
}))
}実装：TransactionManagerにWALを組み込む
トランザクションごとに TxLogger を持たせます。
@@ -5,4 +5,5 @@ import (
"github.com/yourname/go-rdbms/sql/catalog"
"github.com/yourname/go-rdbms/storage/heap"
+ "github.com/yourname/go-rdbms/transaction/log"
"github.com/yourname/go-rdbms/transaction/txid"
)
@@ -31,4 +32,5 @@ type Transaction struct {
status Status
writes []WriteRecord
+ logger *log.TxLogger
}
@@ -36,4 +38,7 @@ func (tx *Transaction) ID() txid.TxID { return tx.id }
func (tx *Transaction) Status() Status { return tx.status }
+// Logger はこのトランザクションのヒープ操作をWALに記録するLoggerを返す
+func (tx *Transaction) Logger() heap.Logger { return tx.logger }
+
// AddWrite は書き込み操作を記録する（ロールバック時に逆順で取り消す）
func (tx *Transaction) AddWrite(w WriteRecord) {TransactionManagerは、開始時にBEGIN、コミット時にCOMMIT、ロールバックの完了時にABORTを記録します。
@@ -9,4 +9,5 @@ import (
"github.com/yourname/go-rdbms/storage/btree"
"github.com/yourname/go-rdbms/transaction/lock"
+ "github.com/yourname/go-rdbms/transaction/log"
"github.com/yourname/go-rdbms/transaction/txid"
)
@@ -14,16 +15,22 @@ import (
// TransactionManager はトランザクションの開始・コミット・ロールバックを管理する
type TransactionManager struct {
- mu sync.Mutex
- nextID txid.TxID
- active map[txid.TxID]*Transaction
- lockMgr *lock.LockManager
+ mu sync.Mutex
+ cond *sync.Cond // トランザクションの終了・チェックポイントの完了を知らせる
+ nextID txid.TxID
+ active map[txid.TxID]*Transaction
+ lockMgr *lock.LockManager
+ logMgr *log.LogManager
+ checkpointing bool
}
-func NewTransactionManager(lockMgr *lock.LockManager) *TransactionManager {
- return &TransactionManager{
+func NewTransactionManager(lockMgr *lock.LockManager, logMgr *log.LogManager) *TransactionManager {
+ tm := &TransactionManager{
nextID: 1,
active: make(map[txid.TxID]*Transaction),
lockMgr: lockMgr,
+ logMgr: logMgr,
}
+ tm.cond = sync.NewCond(&tm.mu)
+ return tm
}
@@ -33,7 +40,12 @@ func (tm *TransactionManager) Begin() *Transaction {
defer tm.mu.Unlock()
+ for tm.checkpointing {
+ tm.cond.Wait() // チェックポイント中は新しいトランザクションを始めない
+ }
tx := &Transaction{id: tm.nextID, status: StatusActive}
+ tx.logger = log.NewTxLogger(tm.logMgr, tx.id)
tm.nextID++
tm.active[tx.id] = tx
+ tm.logMgr.Append(&log.LogRecord{TxID: tx.id, Type: log.RecordBegin})
return tx
}
@@ -41,9 +53,15 @@ func (tm *TransactionManager) Begin() *Transaction {
// Commit はトランザクションをコミットする
func (tm *TransactionManager) Commit(tx *Transaction) error {
+ if tx.status != StatusActive {
+ return fmt.Errorf("transaction: TX %d は既に終了しています", tx.id)
+ }
+ // 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)
+ }
if err := tm.finish(tx, StatusCommitted); err != nil {
return err
}
- // 第12章でここにCOMMITログの書き込みとfsyncが入る
-
// 厳格2PL: ロックはトランザクションの終了時にまとめて解放する
tm.lockMgr.ReleaseAll(tx.id)
@@ -66,8 +84,10 @@ func (tm *TransactionManager) Rollback(tx *Transaction) error {
}
}
- if err := w.Table.Heap.DeleteRecord(w.RID); err != nil {
+ // 取り消し操作もWALに記録する（クラッシュ後のREDOで再現できるように）
+ if err := w.Table.Heap.DeleteRecord(tx.logger, w.RID); err != nil {
return err
}
}
+ tm.logMgr.Append(&log.LogRecord{TxID: tx.id, Type: log.RecordAbort})
if err := tm.finish(tx, StatusAborted); err != nil {
return err
@@ -88,4 +108,21 @@ func (tm *TransactionManager) finish(tx *Transaction, status Status) error {
tx.status = status
delete(tm.active, tx.id)
+ tm.cond.Broadcast()
return nil
}
+
+// Checkpoint は新しいトランザクションの開始を止め、実行中のトランザクションが
+// すべて終わるのを待ってから fn（ページの書き出しとログの切り詰め）を実行する
+func (tm *TransactionManager) Checkpoint(fn func() error) error {
+ tm.mu.Lock()
+ defer tm.mu.Unlock()
+
+ tm.checkpointing = true
+ for len(tm.active) > 0 {
+ tm.cond.Wait()
+ }
+ err := fn()
+ tm.checkpointing = false
+ tm.cond.Broadcast()
+ return err
+}Commit は、COMMITレコードを追加した後、FlushTo でそこまでのログをfsyncしてから、トランザクションを完了扱いにしてロックを放します（ルール2）。一方、ロールバックではfsyncを待ちません。ABORTレコードがディスクに届く前にクラッシュしても、リカバリがそのトランザクションを未完了とみなして取り消し直すだけで、結果は同じだからです。
Checkpoint は、後述のチェックポイントのために、新しいトランザクションの開始を止めて、実行中のトランザクションがすべて終わるのを待つ関数です。
Executorは、ヒープへの挿入にトランザクションの Logger を渡すようにします。
// sql/executor/executor.go（変更部分）
rid, err := table.Heap.InsertRecord(tx.Logger(), data)リカバリ
ARIESの考え方
クラッシュ後のリカバリ手法として最も有名なのが、IBMのC. Mohanらが1992年に発表したARIES（Algorithm for Recovery and Isolation Exploiting Semantics）です。ARIESのリカバリは次の3つのフェーズからなります。
1. 分析（Analysis）: ログを読み、クラッシュ時点で終わっていなかったトランザクション（loser）を洗い出す
2. REDO: ログを先頭から順に再実行し、クラッシュ直前の状態を再現する
3. UNDO: loserの変更を、ログの後ろから順に取り消すARIESの重要な考え方は、REDOフェーズでコミットしたかどうかに関係なく、すべての変更を再実行することです（repeating history）。いったんクラッシュ直前とまったく同じ状態を作り、そのうえで未完了のトランザクションを通常のロールバックと同じ要領で取り消します。こうすると、REDOとUNDOの役割がはっきり分かれ、ロールバックの途中でクラッシュした場合のような込み入った状況も素直に扱えます。
本シリーズのリカバリも、この3フェーズの構成に従います。ARIESとの主な違いは次のとおりです。
- ARIESは、チェックポイントに記録した情報（ダーティなページやアクティブなトランザクションの一覧）を使って、ログのどこからREDOを始めればよいかを絞り込みます。本シリーズは後述の方法でチェックポイントのたびにログを空にするので、常にログ全体を読みます
- ARIESのCLRは「次に取り消すべきレコード」へのポインタを持ち、取り消し済みの操作を再び取り消さないようにしています。本シリーズでは、取り消す操作がINSERTだけで、取り消し（スロットの削除）が済んでいるかはページを見ればわかるので、削除済みなら飛ばすだけで済ませています
REDOを行う関数
ページ単位でログを再実行する関数を heap パッケージに用意します。どれも、ページLSNがログのLSN以上なら何もしません。
// storage/heap/redo.go
package heap
import (
"fmt"
"github.com/yourname/go-rdbms/storage/buffer"
"github.com/yourname/go-rdbms/storage/page"
)
// ここにあるのは、リカバリ（第12章）でWALを再適用するための関数。
// どれも「ページLSN >= ログのLSN なら、そのページには反映済み」と判断して何もしない。
// そのため同じログを何度適用しても結果は変わらない（冪等）。
// RedoInsert はINSERTのログを再適用する
func RedoInsert(bpm *buffer.BufferPoolManager, rid RID, data []byte, lsn uint64) error {
p, err := bpm.FetchPage(rid.PageID)
if err != nil {
return err
}
if p.GetLSN() >= lsn {
return bpm.UnpinPage(rid.PageID, false)
}
sp := page.NewSlottedPage(p)
// ページはこのログの直前の状態のはずなので、次に割り当てられるスロットが一致する
if sp.NumSlots() != rid.SlotID {
bpm.UnpinPage(rid.PageID, false)
return fmt.Errorf("heap: REDO先のスロットが一致しません（ページ %d: 期待 %d, 実際 %d）",
rid.PageID, rid.SlotID, sp.NumSlots())
}
if _, err := sp.InsertRecord(data); err != nil {
bpm.UnpinPage(rid.PageID, false)
return err
}
p.SetLSN(lsn)
return bpm.UnpinPage(rid.PageID, true)
}
// RedoDelete はDELETEのログを再適用する
func RedoDelete(bpm *buffer.BufferPoolManager, rid RID, lsn uint64) error {
p, err := bpm.FetchPage(rid.PageID)
if err != nil {
return err
}
if p.GetLSN() >= lsn {
return bpm.UnpinPage(rid.PageID, false)
}
if err := page.NewSlottedPage(p).DeleteRecord(rid.SlotID); err != nil {
bpm.UnpinPage(rid.PageID, false)
return err
}
p.SetLSN(lsn)
return bpm.UnpinPage(rid.PageID, true)
}
// RedoNewPage はページ追加のログを再適用する（新しいページの初期化と、元の末尾ページからのリンク）
func RedoNewPage(bpm *buffer.BufferPoolManager, prevPageID, newPageID page.PageID, lsn uint64) error {
newP, err := bpm.FetchPage(newPageID)
if err != nil {
return err
}
dirty := false
if newP.GetLSN() < lsn {
page.NewSlottedPage(newP).Init()
newP.SetLSN(lsn)
dirty = true
}
if err := bpm.UnpinPage(newPageID, dirty); err != nil {
return err
}
prevP, err := bpm.FetchPage(prevPageID)
if err != nil {
return err
}
dirty = false
if prevP.GetLSN() < lsn {
page.NewSlottedPage(prevP).SetNextPageID(newPageID)
prevP.SetLSN(lsn)
dirty = true
}
return bpm.UnpinPage(prevPageID, dirty)
}RedoInsert では、再実行する直前のページは「このログを書いたときの直前の状態」になっているはずなので、次に割り当てられるスロット番号がログのスロット番号と一致することを確かめています。第5章でスロットを常に末尾に追加し、再利用しないと決めたので、この確認ができます。
RecoveryManager
// transaction/log/recovery.go
package log
import (
"errors"
"fmt"
"github.com/yourname/go-rdbms/storage/buffer"
"github.com/yourname/go-rdbms/storage/heap"
"github.com/yourname/go-rdbms/storage/page"
"github.com/yourname/go-rdbms/transaction/txid"
)
// RecoveryManager は起動時にWALを読み、ヒープを一貫した状態に戻す
type RecoveryManager struct {
bpm *buffer.BufferPoolManager
lm *LogManager
}
func NewRecoveryManager(bpm *buffer.BufferPoolManager, lm *LogManager) *RecoveryManager {
return &RecoveryManager{bpm: bpm, lm: lm}
}
// Recover はリカバリを行う。前回が正常終了でなかった
// （ログにCHECKPOINT以外のレコードが残っていた）場合は true を返す
func (rm *RecoveryManager) Recover() (bool, error) {
records, err := rm.lm.ReadAll()
if err != nil {
return false, err
}
// 1. 分析: 最後まで終わらなかったトランザクション（loser）を洗い出す
losers := map[txid.TxID]bool{}
unclean := false
for _, rec := range records {
switch rec.Type {
case RecordCheckpoint:
continue
case RecordCommit, RecordAbort:
delete(losers, rec.TxID)
default:
losers[rec.TxID] = true
}
unclean = true
}
if !unclean {
return false, nil
}
// 2. REDO: コミットしたかどうかに関係なく、すべての変更を先頭から再適用し、
// クラッシュ直前のバッファプールの状態を再現する
for _, rec := range records {
if err := rm.redo(rec); err != nil {
return true, fmt.Errorf("recovery: LSN %d のREDOに失敗しました: %w", rec.LSN, err)
}
}
// 3. UNDO: loserが挿入したレコードを、後ろから順に削除する。
// 削除もログに記録するので、リカバリ中に再びクラッシュしてもやり直せる
for i := len(records) - 1; i >= 0; i-- {
rec := records[i]
if rec.Type != RecordInsert || !losers[rec.TxID] {
continue
}
rid := heap.RID{PageID: rec.PageID, SlotID: rec.SlotID}
err := heap.DeleteAt(rm.bpm, NewTxLogger(rm.lm, rec.TxID), rid)
if errors.Is(err, page.ErrRecordDeleted) {
continue // クラッシュ前のロールバックで既に取り消されていた
}
if err != nil {
return true, fmt.Errorf("recovery: LSN %d のUNDOに失敗しました: %w", rec.LSN, err)
}
}
for tx := range losers {
rm.lm.Append(&LogRecord{TxID: tx, Type: RecordAbort})
}
return true, rm.lm.Flush()
}
func (rm *RecoveryManager) redo(rec *LogRecord) error {
lsn := uint64(rec.LSN)
switch rec.Type {
case RecordInsert:
return heap.RedoInsert(rm.bpm, heap.RID{PageID: rec.PageID, SlotID: rec.SlotID}, rec.Data, lsn)
case RecordDelete:
return heap.RedoDelete(rm.bpm, heap.RID{PageID: rec.PageID, SlotID: rec.SlotID}, lsn)
case RecordNewPage:
return heap.RedoNewPage(rm.bpm, rec.PrevPageID, rec.PageID, lsn)
}
return nil // BEGIN / COMMIT / ABORT / CHECKPOINT はページを変更しない
}
// Checkpoint はダーティページをすべてデータファイルに書き出してから、ログを空にする
// 実行中のトランザクションがない状態で呼ぶこと（TransactionManager.Checkpoint 経由で呼ぶ）
func Checkpoint(bpm *buffer.BufferPoolManager, lm *LogManager) error {
if err := bpm.FlushAllPages(); err != nil {
return err
}
return lm.Truncate()
}UNDOフェーズでは、loserのINSERTを後ろから順に DeleteAt で削除します。削除もログに記録するので、リカバリの途中でもう一度クラッシュしても、次のリカバリで同じ手順をやり直せます。すでに削除済み（ロールバックの途中でクラッシュした場合など）なら ErrRecordDeleted が返るので、飛ばします。最後にloserごとにABORTを記録して、ログをフラッシュします。
インデックスの再構築
B+木の変更はログに記録していないので、クラッシュ後のB+木のページは、どこまで書かれているかわからない状態です。そこでリカバリの後、ヒープファイルの中身からインデックスを作り直します。レコードから主キーの値を取り出すにはスキーマに従ってデコードする必要があるので、この処理は executor パッケージに置きます。
// RebuildIndexes はヒープの内容から、すべてのテーブルの主キーインデックスを作り直す
// インデックスの変更はWALに記録していないので、クラッシュからの復旧後に呼ぶ（第12章）
func RebuildIndexes(cat *catalog.Catalog) error {
for _, name := range cat.ListTables() {
table, err := cat.GetTable(name)
if err != nil {
return err
}
pk := table.Schema.PrimaryKeyIndex()
if pk < 0 {
continue
}
if err := table.Index.Reset(); err != nil {
return err
}
scanner := table.Heap.Scan()
for {
rid, data, ok, err := scanner.Next()
if err != nil {
return err
}
if !ok {
break
}
values, err := DecodeTuple(data, table.Schema)
if err != nil {
return err
}
if err := table.Index.Insert(values[pk].Int, rid); err != nil {
return err
}
}
}
return nil
}B+木には、ルートを空の葉に戻す Reset を追加します。第6章でルートのPageIDを固定にしたので、カタログを書き換えずに作り直せます。古いノードのページは解放されずに残りますが、本シリーズでは空きページの管理を実装していないので、そのままにしています。
// storage/btree/btree.go（追加）
// Reset はルートを空の葉ノードに戻して、木を空にする（第12章のインデックス再構築で使う）
// 古いノードのページは解放せずにそのまま残る（空きページの管理は実装していない）
func (t *BPlusTree) Reset() error {
t.mu.Lock()
defer t.mu.Unlock()
p, err := t.bpm.FetchPage(t.rootPageID)
if err != nil {
return err
}
node{p}.initLeaf()
return t.bpm.UnpinPage(t.rootPageID, true)
}起動処理の順番
データベースを開くときは、次の順番で処理します（第13章の db.Open でもこの順番になります）。
1. データファイル・バッファプール・ログマネージャを開き、バッファプールにログのフラッシュ関数を登録
2. リカバリ（分析 → REDO → UNDO）
3. カタログを読み込む（ヒープファイルとB+木を開く）
4. リカバリを行った場合は、インデックスを再構築
5. チェックポイントカタログの読み込みをリカバリの後にしている点が重要です。heap.OpenHeapFile は、ページのリンクをたどって末尾のページを見つけます。REDOでNEWPAGEが再実行されるとリンクの先にページが増えるので、リカバリの前に開くと、古い末尾のページを覚えてしまいます。その状態で新しい行を挿入すると、REDOで追加されたページへのリンクを上書きしてしまいます。
チェックポイント
ログを無限に伸ばさないために
ログは追記される一方なので、放っておくとどこまでも大きくなり、リカバリのたびに読む量も増えていきます。そこでチェックポイントを取って、「ここより前のログはもう要らない」という地点を作ります。
本シリーズの方法：静止チェックポイント
本シリーズでは、最も単純な**静止チェックポイント（quiescent checkpoint）**を実装します。
1. 新しいトランザクションの開始を止める
2. 実行中のトランザクションがすべて終わるのを待つ
3. ダーティなページをすべてディスクに書き出す（FlushAllPages）
4. ログをCHECKPOINTレコード1件だけに置き換える
5. トランザクションの開始を再開する手順2・3の後では、コミットしたトランザクションの変更はすべてデータファイルに反映済みで、未完了のトランザクションは存在しません。ログはもう1件も要らないので、捨ててかまいません。手順1・2は TransactionManager.Checkpoint が、手順3・4は log.Checkpoint が担当します。
手順4でログを空のファイルにしてしまわず、CHECKPOINTレコードを1件残しているのには理由があります。データファイル上のページには、これまでのLSNがページLSNとして残っています。ログを空にしてLSNを1から振り直すと、新しいログのLSNがページLSNより小さくなり、リカバリで「反映済み」と誤って判断されてしまいます。CHECKPOINTレコードに続きのLSNを振っておけば、次の起動時にも NewLogManager がその続きから番号を振れます。ログの置き換えは、第9章のカタログと同じく一時ファイルとリネームで行うので、途中でクラッシュしても古いログか新しいログのどちらかが残ります。
チェックポイントは、起動時（リカバリの直後）と、第13章で実装するデータベースのクローズ時に取ります。正常に終了した場合、ログにはCHECKPOINTレコードしか残っていないので、次の起動時にはリカバリを行う必要がありません。
実際のデータベースのチェックポイント
静止チェックポイントは、実行中のトランザクションが終わるまで新しいトランザクションを止めてしまうので、稼働中のシステムでは使えません。PostgreSQLやARIESは、トランザクションを止めずに取れるファジーチェックポイントを使います。ダーティなページを少しずつ書き出しながら、「リカバリはこの位置のログから始めればよい」という情報をチェックポイントレコードに記録しておく方式です。
テストを書く
LogManagerのテスト
LSNが1から振られること、Append しただけではディスクに書かれないこと、開き直すとLSNが続きから振られること、末尾の書きかけのレコードが捨てられることを確かめます（Truncate のテストは省略しています）。
// transaction/log/log_manager_test.go（抜粋）
func TestLogManager_AppendFlushRead(t *testing.T) {
path := filepath.Join(t.TempDir(), "wal.log")
lm, err := log.NewLogManager(path)
if err != nil {
t.Fatal(err)
}
lsn1 := lm.Append(&log.LogRecord{TxID: 1, Type: log.RecordBegin})
lsn2 := lm.Append(&log.LogRecord{TxID: 1, Type: log.RecordInsert, PageID: 3, SlotID: 7, Data: []byte("hello")})
lsn3 := lm.Append(&log.LogRecord{TxID: 1, Type: log.RecordCommit})
if lsn1 != 1 || lsn2 != 2 || lsn3 != 3 {
t.Fatalf("LSNは1から順に振られるはず: %d, %d, %d", lsn1, lsn2, lsn3)
}
// Append しただけではファイルは空のまま
if info, _ := os.Stat(path); info.Size() != 0 {
t.Errorf("Append直後にファイルへ書かれている: %dバイト", info.Size())
}
if err := lm.FlushTo(lsn3); err != nil {
t.Fatal(err)
}
lm.Close()
// 開き直すと、レコードが読めて、LSNは続きから振られる
lm2, err := log.NewLogManager(path)
if err != nil {
t.Fatal(err)
}
defer lm2.Close()
records, err := lm2.ReadAll()
if err != nil {
t.Fatal(err)
}
if len(records) != 3 {
t.Fatalf("レコード数: want 3, got %d", len(records))
}
if r := records[1]; r.Type != log.RecordInsert || r.PageID != 3 || r.SlotID != 7 || string(r.Data) != "hello" {
t.Errorf("INSERTレコードの中身が不正: %+v", r)
}
if lsn := lm2.Append(&log.LogRecord{TxID: 2, Type: log.RecordBegin}); lsn != 4 {
t.Errorf("再オープン後のLSNは4から: got %d", lsn)
}
}
func TestLogManager_TornTail(t *testing.T) {
path := filepath.Join(t.TempDir(), "wal.log")
lm, _ := log.NewLogManager(path)
lm.Append(&log.LogRecord{TxID: 1, Type: log.RecordBegin})
lm.Append(&log.LogRecord{TxID: 1, Type: log.RecordCommit})
lm.Close()
// 書き込みの途中でクラッシュしたように、末尾に中途半端なバイト列を足す
f, _ := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0644)
f.Write([]byte{0, 0, 0, 40, 1, 2, 3})
f.Close()
lm2, err := log.NewLogManager(path)
if err != nil {
t.Fatal(err)
}
lm2.Append(&log.LogRecord{TxID: 2, Type: log.RecordBegin})
records, err := lm2.ReadAll()
if err != nil {
t.Fatal(err)
}
// 壊れた末尾は捨てられ、その位置から新しいレコードが続く
if len(records) != 3 || records[2].LSN != 3 {
t.Errorf("壊れた末尾が正しく切り捨てられていない: %d件", len(records))
}
lm2.Close()
}リカバリのテスト
リカバリのテストでは、crash でバッファプールの中身をディスクに書かずにデータファイルを閉じ、プロセスが異常終了した状況を作ります。その後、open で起動処理（リカバリ → カタログ → インデックス再構築 → チェックポイント）を行い、データが正しい状態になっているかを確かめます。
// transaction/log/recovery_test.go（抜粋）
// testDB はデータファイル・カタログ・WALを束ねたテスト用のデータベース
type testDB struct {
t *testing.T
dm *disk.FileDiskManager
bpm *buffer.BufferPoolManager
lm *log.LogManager
tm *transaction.TransactionManager
executor *executor.Executor
}
// open は起動処理を行う: リカバリ → カタログの読み込み → インデックス再構築 → チェックポイント
func open(t *testing.T, dir string) *testDB {
t.Helper()
dm, err := disk.NewFileDiskManager(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatal(err)
}
bpm := buffer.NewBufferPoolManager(dm, 16) // 小さめにして追い出しを起こりやすくする
lm, err := log.NewLogManager(filepath.Join(dir, "wal.log"))
if err != nil {
t.Fatal(err)
}
bpm.SetLogFlusher(func(lsn uint64) error { return lm.FlushTo(log.LSN(lsn)) })
recovered, err := log.NewRecoveryManager(bpm, lm).Recover()
if err != nil {
t.Fatal(err)
}
// カタログはリカバリの後に開く（REDOでヒープにページが増えている場合があるため）
cat, err := catalog.NewCatalog(bpm, filepath.Join(dir, "catalog.json"))
if err != nil {
t.Fatal(err)
}
if recovered {
if err := executor.RebuildIndexes(cat); err != nil {
t.Fatal(err)
}
}
if err := log.Checkpoint(bpm, lm); err != nil {
t.Fatal(err)
}
lockMgr := lock.NewLockManager()
return &testDB{
t: t, dm: dm, bpm: bpm, lm: lm,
tm: transaction.NewTransactionManager(lockMgr, lm),
executor: executor.New(cat, lockMgr),
}
}
// crash はバッファプールの中身を書き出さずにデータファイルを閉じる（プロセスの異常終了を模す）
func (db *testDB) crash() {
db.dm.Close()
}
func TestRecovery_CommittedDataSurvivesCrash(t *testing.T) {
dir := t.TempDir()
db := open(t, dir)
db.createUsers()
tx := db.tm.Begin()
for i := 1; i <= 300; i++ {
db.exec(tx, fmt.Sprintf("INSERT INTO users VALUES (%d, 'user%03d-padding-padding-padding')", i, i))
}
if err := db.tm.Commit(tx); err != nil {
t.Fatal(err)
}
db.crash() // コミット済みの変更の多くはまだバッファプールの中にしかない
db2 := open(t, dir)
if got := len(db2.query("SELECT * FROM users").Rows); got != 300 {
t.Errorf("コミット済みの行が失われた: want 300, got %d", got)
}
// インデックスも再構築されているので、主キーで引ける
res := db2.query("SELECT name FROM users WHERE id = 123")
if len(res.Rows) != 1 || res.Rows[0][0].Str != "user123-padding-padding-padding" {
t.Errorf("インデックス経由の検索結果が不正: %v", res.Rows)
}
}
func TestRecovery_UncommittedChangesAreUndone(t *testing.T) {
dir := t.TempDir()
db := open(t, dir)
db.createUsers()
tx1 := db.tm.Begin()
db.exec(tx1, "INSERT INTO users VALUES (1, 'committed')")
db.tm.Commit(tx1)
// TX2 は大量に挿入したがコミットしていない
tx2 := db.tm.Begin()
for i := 100; i < 400; i++ {
db.exec(tx2, fmt.Sprintf("INSERT INTO users VALUES (%d, 'uncommitted-padding-padding-padding')", i))
}
// 未コミットの変更がデータファイルに書き出された状態を作る
if err := db.bpm.FlushAllPages(); err != nil {
t.Fatal(err)
}
db.crash()
db2 := open(t, dir)
res := db2.query("SELECT * FROM users")
if len(res.Rows) != 1 || res.Rows[0][1].Str != "committed" {
t.Errorf("未コミットの行が残っている: %d件", len(res.Rows))
}
// 取り消したキーは再び使える
tx := db2.tm.Begin()
db2.exec(tx, "INSERT INTO users VALUES (100, 'reused')")
db2.tm.Commit(tx)
}
func TestWAL_LogIsFlushedBeforePage(t *testing.T) {
dir := t.TempDir()
db := open(t, dir)
db.createUsers()
tx := db.tm.Begin()
db.exec(tx, "INSERT INTO users VALUES (1, 'not committed yet')")
// コミット前なので、まだINSERTのログはディスクにない
if countOnDisk(t, dir, log.RecordInsert) != 0 {
t.Fatal("コミット前なのにINSERTのログがディスクにある")
}
// ページを書き出すと、そのページLSNまでのログが先にディスクへ書かれる
if err := db.bpm.FlushAllPages(); err != nil {
t.Fatal(err)
}
if countOnDisk(t, dir, log.RecordInsert) != 1 {
t.Error("ページを書き出したのにINSERTのログがディスクにない（WALのルール違反）")
}
db.tm.Commit(tx)
}それぞれのテストが確かめていることは次のとおりです。
| テスト | 確かめていること |
|---|---|
| CommittedDataSurvivesCrash | コミット済みの変更は、ページがディスクになくてもREDOで復元される（インデックスも再構築される） |
| UncommittedChangesAreUndone | 未コミットの変更は、ページがディスクに書き出されていてもUNDOで取り消される |
| RollbackBeforeCrash | ロールバック（取り消しのDELETEとABORT）もREDOで再現される |
| Idempotent | チェックポイントを挟まずにリカバリを2回実行しても、結果は変わらない |
| LogIsFlushedBeforePage | ページを書き出すと、その前にログがディスクに書かれる（ルール1） |
go test ./transaction/log/... -v期待される出力（=== RUN の行は省略）：
--- PASS: TestLogManager_AppendFlushRead (0.03s)
--- PASS: TestLogManager_TornTail (0.02s)
--- PASS: TestLogManager_Truncate (0.03s)
--- PASS: TestRecovery_CommittedDataSurvivesCrash (0.06s)
--- PASS: TestRecovery_UncommittedChangesAreUndone (0.08s)
--- PASS: TestRecovery_RollbackBeforeCrash (0.05s)
--- PASS: TestRecovery_Idempotent (0.07s)
--- PASS: TestWAL_LogIsFlushedBeforePage (0.05s)
PASS
ok github.com/yourname/go-rdbms/transaction/log 0.625s試しに RecoveryManager.redo とUNDOのループを何もしないように書き換えてテストを実行すると、リカバリの4つのテストが失敗します。テストが実際にリカバリの効果を確かめていることを、こうして確認しておくと安心です。
この実装の制限
DiskManager.AllocatePageはファイルを拡張するときにfsyncしないので、OSごと落ちた場合に、ログにはあるページがファイルから消えている可能性があります- B+木はログに記録しないので、クラッシュ後の起動では、テーブルの大きさに比例した時間をかけてインデックスを作り直します
- テーブルの作成（DDL）はトランザクションの対象外です
- チェックポイントの間は、新しいトランザクションを開始できません
ここまでのコード全体像
go-rdbms/
├── storage/
│ ├── page/page.go ← ページLSN（本章）
│ ├── buffer/ ← 書き出し前にログをフラッシュ（本章）
│ ├── heap/
│ │ ├── heap_file.go ← 変更をログに記録（本章）
│ │ └── redo.go ← REDO用の関数（本章）
│ └── btree/btree.go ← Reset（本章）
├── sql/
│ └── executor/executor.go ← Loggerの受け渡し・インデックス再構築（本章）
└── transaction/
├── txid/ lock/
├── log/
│ ├── log_record.go ← ログレコード（本章）
│ ├── log_manager.go ← 追記・フラッシュ・読み出し・切り詰め（本章）
│ ├── tx_logger.go ← トランザクションごとのLogger（本章）
│ └── recovery.go ← リカバリ・チェックポイント（本章）
├── transaction.go
└── transaction_manager.go ← BEGIN・COMMIT・ABORTの記録（本章）WALとリカバリの設計で意識したポイント
① 書く順序で正しさを保証する
「ページより先にログ」「コミット完了より先にログ」という2つの順序さえ守れば、どのタイミングでクラッシュしても、ログからデータを正しい状態に戻せます。
② ページLSNでREDOを冪等にする
ページLSNとログのLSNを比べるだけで、変更が反映済みかどうかを判断できます。同じログを何度再実行しても結果は変わらないので、リカバリ中のクラッシュにも耐えられます。
③ まず歴史を再現してから、取り消す
コミット済みかどうかに関係なく、すべての変更をいったん再実行してから、未完了のトランザクションを取り消します。取り消し操作もログに記録するので、何度中断しても同じ手順でやり直せます。
④ 記録しにくいものは作り直す
B+木の分割のように記録と再実行が難しい変更は、ログに残さず、リカバリ後にヒープから作り直すことにしました。正しさを保ちながら、実装の量を抑えています。
まとめ
- バッファプールがあると、メモリとディスクの間にずれが生じ、クラッシュでコミット済みの変更が失われたり、未コミットの変更が残ったりする
- 実際のデータベースは STEAL / NO-FORCE を選び、WALで両方の問題を解決する
- WALのルールは「ページを書く前にそのページのログを書く」「コミットを完了する前にログを書く」の2つ
- ページLSNを使うと、バッファプールでルール1を守れるうえ、リカバリのREDOを冪等にできる
- リカバリはARIESにならって、分析 → REDO（歴史の再現）→ UNDO の3フェーズで行う
- 静止チェックポイントで、ダーティなページを書き出してからログを切り詰める
次章では、ここまで作ってきたすべてのコンポーネントを1つにまとめ、CLIから対話的にSQLを実行できるデータベースを完成させます。実際にプロセスを強制終了してリカバリを確かめたり、ベンチマークでボトルネックを探したりもします。