RDBMSを自作して学ぶ：第10章 トランザクションとACID特性

前回はカタログを永続化し、データベースを開き直してもテーブルを使い続けられるようになりました。今回からは第4部「トランザクション」に入ります。まずは、業務でもよく耳にするトランザクションとACID特性が、実装上どういう意味を持つのかを整理します。そのうえで、トランザクションの開始・コミット・ロールバックを管理するTransactionManagerの骨格を実装します。

トランザクションとは

トランザクションは、「複数のデータ操作を、1つの不可分な処理単位としてまとめたもの」です。

典型的な例は銀行の振込です。「Aさんの残高を1万円減らす」と「Bさんの残高を1万円増やす」は、両方が実行されるか、どちらも実行されないかのどちらかでなければなりません。片方だけが実行されると、お金が消えたり増えたりしてしまいます。

BEGIN;
  UPDATE accounts SET balance = balance - 10000 WHERE owner = 'A';
  UPDATE accounts SET balance = balance + 10000 WHERE owner = 'B';
COMMIT;

（本シリーズのSQLはUPDATEをサポートしていませんが、ここでは考え方の説明のために使っています。）

ACID特性

トランザクションが満たすべき性質は、頭文字をとって ACID と呼ばれます。それぞれの性質が「何を保証するのか」と「どう実装するのか」を分けて見ていきます。

A: Atomicity（原子性）

トランザクション内の操作は、すべて反映されるか、まったく反映されないかのどちらかです。途中でエラーが起きたり、ユーザーがROLLBACKしたりした場合は、それまでの変更をすべて取り消します。

実装上は、「変更を取り消すための情報（UNDO情報）」を記録しておき、ロールバック時にそれを使って元に戻します。本章では、この仕組みをメモリ上で実装します。サーバーがクラッシュした場合でも取り消せるようにするには、UNDO情報をディスクに残す必要があり、それは第12章のWALで扱います。

C: Consistency（一貫性）

トランザクションの前後で、データベースが満たすべき制約が守られていることです。本シリーズのデータベースが守る制約は、主キーの一意性、カラムの型、VARCHARの最大文字数です。これらに違反するINSERTはエラーになり、データは変更されません。

ただし「振込の前後で全員の残高の合計が変わらない」のような業務上のルールは、データベースは知りません。一貫性は、データベースが守る制約と、アプリケーションが正しいトランザクションを書くことの両方で成り立つ性質です。

I: Isolation（分離性）

同時に実行されている複数のトランザクションが、互いに干渉しないことです。理想的には、トランザクションを1つずつ順番に実行した場合と同じ結果になるべきです。

実装方法には、データにロックをかけて他のトランザクションを待たせる方法（第11章で実装します）と、データの複数の版を持って読み手と書き手が互いを待たなくて済むようにするMVCC（多版型同時実行制御）があります。

D: Durability（永続性）

コミットが完了したトランザクションの変更は、その後にクラッシュや電源断が起きても失われないことです。

実装上は、コミットを完了とする前に、変更内容をログとしてディスクに書き、fsyncで確実に永続化します。これが第12章で実装するWAL（Write-Ahead Logging）です。

ACIDと実装の対応

分離レベル

分離性を完全に保証するとトランザクションどうしの待ち合わせが増え、性能が落ちます。そこでSQL標準では、どの程度の干渉まで許すかを分離レベルとして4段階に定めています。

それぞれの現象の具体例は次章で詳しく見ます。本シリーズでは、次章で実装するロックによってSERIALIZABLEを実現します。

トランザクションのライフサイクル

トランザクションは BEGIN で始まり、COMMIT か ROLLBACK で終わります。

            BEGIN
              │
              ▼
         ┌─────────┐   COMMIT    ┌───────────┐
         │ ACTIVE  │───────────▶│ COMMITTED │
         │(実行中) │             └───────────┘
         └─────────┘
              │ ROLLBACK
              ▼
         ┌─────────┐
         │ ABORTED │
         └─────────┘

一度COMMITTEDかABORTEDになったトランザクションは、もう操作できません。

BEGIN を書かずに実行したSQL文も、実際には「その1文だけのトランザクション」として実行され、成功すれば自動的にコミットされます（自動コミット）。このふるまいは第13章のセッションで実装します。

何を記録して、どう取り消すか

本シリーズのSQLでデータを書き込むのはINSERTだけです。INSERTは次の2つを変更します。

1. ヒープファイルにレコードを追加する（RIDが決まる）
2. 主キーがあれば、B+木にキーとRIDの組を追加する

そこで、INSERTのたびに「どのテーブルの、どのRIDに、どの主キーで挿入したか」を書き込みの記録（WriteRecord）としてトランザクションに追加しておきます。ロールバックするときは、記録を後ろから順にたどり、B+木からキーを削除し、ヒープファイルからレコードを削除します。

後ろから取り消すのは、後の操作が前の操作の結果に依存している場合があるからです。「靴下を履いてから靴を履いた」なら、脱ぐときは靴から脱ぐ、というのと同じです。

第5章で「削除済みスロットを再利用しない」と決めたのは、ここで効いてきます。スロットを再利用すると、ロールバックで削除しようとしたRIDに、別のトランザクションが挿入した無関係なレコードが入っているかもしれません。スロット番号を使い回さなければ、記録したRIDは必ず自分が挿入したレコードを指します。

実装

トランザクションIDの型

トランザクションを識別する**トランザクションID（TxID）**の型を定義します。この型は、本章の transaction パッケージだけでなく、第11章のロックマネージャ（transaction/lock）と第12章のログマネージャ（transaction/log）からも使います。一方で transaction パッケージは、ロックマネージャとログマネージャを使う側です。TxIDを transaction パッケージに置くと、lock → transaction → lock という循環importになってしまうので、TxIDだけを小さな独立したパッケージに切り出しておきます。

// transaction/txid/txid.go

// Package txid はトランザクションIDの型だけを持つ小さなパッケージ。
// transaction・lock・log の各パッケージから参照するため、循環importを避けて独立させている
package txid

// TxID はトランザクションを一意に識別するID（1から単調増加）
type TxID uint32

// Invalid は無効なトランザクションIDを表す
const Invalid TxID = 0

Transaction

// transaction/transaction.go
package transaction

import (
    "github.com/yourname/go-rdbms/sql/catalog"
    "github.com/yourname/go-rdbms/storage/heap"
    "github.com/yourname/go-rdbms/transaction/txid"
)

// Status はトランザクションの状態
type Status int

const (
    StatusActive    Status = iota // 実行中
    StatusCommitted               // コミット済み
    StatusAborted                 // ロールバック済み
)

// WriteRecord はロールバックのために覚えておく書き込み操作（本シリーズではINSERTだけ）
type WriteRecord struct {
    Table  *catalog.TableInfo
    RID    heap.RID // 挿入したレコードの位置
    Key    int64    // 挿入した主キー（HasKeyがtrueのときだけ有効）
    HasKey bool
}

// Transaction は1つのトランザクション
// 1つのトランザクションを複数のゴルーチンから同時に使うことはない前提
type Transaction struct {
    id     txid.TxID
    status Status
    writes []WriteRecord
}

func (tx *Transaction) ID() txid.TxID  { return tx.id }
func (tx *Transaction) Status() Status { return tx.status }

// AddWrite は書き込み操作を記録する（ロールバック時に逆順で取り消す）
func (tx *Transaction) AddWrite(w WriteRecord) {
    tx.writes = append(tx.writes, w)
}

WriteRecord は *catalog.TableInfo を直接持つので、ロールバック時にカタログを引き直す必要はありません。

TransactionManager

// transaction/transaction_manager.go
package transaction

import (
    "errors"
    "fmt"
    "sync"

    "github.com/yourname/go-rdbms/storage/btree"
    "github.com/yourname/go-rdbms/transaction/txid"
)

// TransactionManager はトランザクションの開始・コミット・ロールバックを管理する
type TransactionManager struct {
    mu     sync.Mutex
    nextID txid.TxID
    active map[txid.TxID]*Transaction
}

func NewTransactionManager() *TransactionManager {
    return &TransactionManager{
        nextID: 1,
        active: make(map[txid.TxID]*Transaction),
    }
}

// Begin は新しいトランザクションを開始する
func (tm *TransactionManager) Begin() *Transaction {
    tm.mu.Lock()
    defer tm.mu.Unlock()

    tx := &Transaction{id: tm.nextID, status: StatusActive}
    tm.nextID++
    tm.active[tx.id] = tx
    return tx
}

// Commit はトランザクションをコミットする
func (tm *TransactionManager) Commit(tx *Transaction) error {
    if err := tm.finish(tx, StatusCommitted); err != nil {
        return err
    }
    // 第12章でここにCOMMITログの書き込みとfsyncが入る
    return nil
}

// Rollback はトランザクションの書き込みをすべて取り消す
func (tm *TransactionManager) Rollback(tx *Transaction) error {
    if tx.status != StatusActive {
        return fmt.Errorf("transaction: TX %d は既に終了しています", tx.id)
    }

    // 後に行った操作から順に取り消す
    for i := len(tx.writes) - 1; i >= 0; i-- {
        w := tx.writes[i]
        if w.HasKey {
            // インデックスへの登録前に失敗した場合はキーがないので、見つからなくてもよい
            if err := w.Table.Index.Delete(w.Key); err != nil && !errors.Is(err, btree.ErrKeyNotFound) {
                return err
            }
        }
        if err := w.Table.Heap.DeleteRecord(w.RID); err != nil {
            return err
        }
    }
    return tm.finish(tx, StatusAborted)
}

// finish はトランザクションを終了状態にして、実行中の一覧から外す
func (tm *TransactionManager) finish(tx *Transaction, status Status) error {
    tm.mu.Lock()
    defer tm.mu.Unlock()

    if tx.status != StatusActive {
        return fmt.Errorf("transaction: TX %d は既に終了しています", tx.id)
    }
    tx.status = status
    delete(tm.active, tx.id)
    return nil
}

TxIDは1から始めて、トランザクションを開始するたびに1ずつ増やします。

Rollback でB+木からキーを削除するとき、ErrKeyNotFound は無視しています。ExecutorのINSERTは「ヒープへの挿入 → 書き込みの記録 → B+木への挿入」の順で進むので、B+木への挿入の前に失敗した場合は、記録はあってもキーがないからです。

状態のチェックと変更は finish にまとめ、ミューテックスで守っています。同じトランザクションを2回コミットしようとしたり、コミット済みのトランザクションをロールバックしようとしたりすると、エラーになります。

Executorをトランザクションに対応させる

Execute がトランザクションを受け取るようにし、INSERTのたびに書き込みを記録します。

@@ -8,4 +8,5 @@ import (
     "github.com/yourname/go-rdbms/sql/parser"
     "github.com/yourname/go-rdbms/sql/planner"
+    "github.com/yourname/go-rdbms/transaction"
 )
 
@@ -28,11 +29,11 @@ func New(cat *catalog.Catalog) *Executor {
 }
 
-// Execute はSQL文を1つ実行する
-func (e *Executor) Execute(stmt parser.Statement) (*Result, error) {
+// Execute はトランザクションtxの中でSQL文を1つ実行する
+func (e *Executor) Execute(tx *transaction.Transaction, stmt parser.Statement) (*Result, error) {
     switch s := stmt.(type) {
     case *parser.SelectStatement:
         return e.executeSelect(s)
     case *parser.InsertStatement:
-        return e.executeInsert(s)
+        return e.executeInsert(tx, s)
     case *parser.CreateTableStatement:
         return e.executeCreateTable(s)
@@ -115,5 +116,5 @@ func (e *Executor) build(plan planner.Plan) (Operator, error) {
 // ---- INSERT ----
 
-func (e *Executor) executeInsert(stmt *parser.InsertStatement) (*Result, error) {
+func (e *Executor) executeInsert(tx *transaction.Transaction, stmt *parser.InsertStatement) (*Result, error) {
     table, err := e.catalog.GetTable(stmt.Table)
     if err != nil {
@@ -143,4 +144,11 @@ func (e *Executor) executeInsert(stmt *parser.InsertStatement) (*Result, error)
         return nil, err
     }
+    // ロールバックに備えて、インデックスを更新する前に書き込みを記録しておく
+    w := transaction.WriteRecord{Table: table, RID: rid}
+    if pk >= 0 {
+        w.Key, w.HasKey = values[pk].Int, true
+    }
+    tx.AddWrite(w)
+
     if pk >= 0 {
         if err := table.Index.Insert(values[pk].Int, rid); err != nil {

書き込みの記録は、B+木を更新する前に行っています。B+木への挿入が失敗した場合でも、ヒープに入ったレコードはロールバックで確実に取り消されます。

テストで使っているヘルパーも、SQL文ごとにトランザクションを開始し、成功したらコミット、失敗したらロールバックするように変えておきます。

// sql/executor/executor_test.go（変更部分）
// try はSQLを1文だけのトランザクションとして実行する（成功したらコミット、失敗したらロールバック）
func (db *testDB) try(sql string) (*executor.Result, error) {
    stmt, err := parser.Parse(sql)
    if err != nil {
        return nil, err
    }
    tx := db.tm.Begin()
    res, err := db.exec.Execute(tx, stmt)
    if err != nil {
        db.tm.Rollback(tx)
        return nil, err
    }
    return res, db.tm.Commit(tx)
}

テストを書く

コミット、ロールバック、そして「ロックがないと何が起きるか」をテストします。

// transaction/transaction_manager_test.go
package transaction_test

import (
    "path/filepath"
    "testing"

    "github.com/yourname/go-rdbms/sql/catalog"
    "github.com/yourname/go-rdbms/sql/executor"
    "github.com/yourname/go-rdbms/sql/parser"
    "github.com/yourname/go-rdbms/storage/buffer"
    "github.com/yourname/go-rdbms/storage/disk"
    "github.com/yourname/go-rdbms/transaction"
)

type env struct {
    t        *testing.T
    tm       *transaction.TransactionManager
    executor *executor.Executor
}

func setup(t *testing.T) *env {
    t.Helper()
    dir := t.TempDir()
    dm, err := disk.NewFileDiskManager(filepath.Join(dir, "test.db"))
    if err != nil {
        t.Fatal(err)
    }
    t.Cleanup(func() { dm.Close() })
    bpm := buffer.NewBufferPoolManager(dm, 32)
    cat, err := catalog.NewCatalog(bpm, filepath.Join(dir, "catalog.json"))
    if err != nil {
        t.Fatal(err)
    }
    e := &env{t: t, tm: transaction.NewTransactionManager(), executor: executor.New(cat)}

    tx := e.tm.Begin()
    e.exec(tx, "CREATE TABLE items (id INT PRIMARY KEY, name VARCHAR(10))")
    e.tm.Commit(tx)
    return e
}

// exec はトランザクションtxの中でSQLを実行する
func (e *env) exec(tx *transaction.Transaction, sql string) (*executor.Result, error) {
    stmt, err := parser.Parse(sql)
    if err != nil {
        e.t.Fatal(err)
    }
    return e.executor.Execute(tx, stmt)
}

// count は新しいトランザクションで items の行数を数える
func (e *env) count() int {
    tx := e.tm.Begin()
    defer e.tm.Commit(tx)
    res, err := e.exec(tx, "SELECT * FROM items")
    if err != nil {
        e.t.Fatal(err)
    }
    return len(res.Rows)
}

func TestTransaction_Commit(t *testing.T) {
    e := setup(t)

    tx := e.tm.Begin()
    e.exec(tx, "INSERT INTO items VALUES (1, 'apple')")
    e.exec(tx, "INSERT INTO items VALUES (2, 'banana')")
    if err := e.tm.Commit(tx); err != nil {
        t.Fatal(err)
    }
    if got := e.count(); got != 2 {
        t.Errorf("コミット後の件数: want 2, got %d", got)
    }

    // 終了したトランザクションは再びコミットもロールバックもできない
    if err := e.tm.Commit(tx); err == nil {
        t.Error("2回目のCommitがエラーにならなかった")
    }
    if err := e.tm.Rollback(tx); err == nil {
        t.Error("コミット済みトランザクションのRollbackがエラーにならなかった")
    }
}

func TestTransaction_Rollback(t *testing.T) {
    e := setup(t)

    tx := e.tm.Begin()
    e.exec(tx, "INSERT INTO items VALUES (1, 'apple')")
    e.exec(tx, "INSERT INTO items VALUES (2, 'banana')")
    if err := e.tm.Rollback(tx); err != nil {
        t.Fatal(err)
    }
    if tx.Status() != transaction.StatusAborted {
        t.Errorf("状態がAbortedになっていない: %v", tx.Status())
    }
    if got := e.count(); got != 0 {
        t.Errorf("ロールバック後の件数: want 0, got %d", got)
    }

    // インデックスからもキーが消えているので、同じ主キーをもう一度挿入できる
    tx2 := e.tm.Begin()
    if _, err := e.exec(tx2, "INSERT INTO items VALUES (1, 'avocado')"); err != nil {
        t.Errorf("ロールバックした主キーを再利用できない: %v", err)
    }
    e.tm.Commit(tx2)
}

func TestTransaction_DirtyReadWithoutLocks(t *testing.T) {
    e := setup(t)

    // TX1がINSERTしてまだコミットしていない
    tx1 := e.tm.Begin()
    e.exec(tx1, "INSERT INTO items VALUES (1, 'apple')")

    // TX2からTX1の未コミットの行が見えてしまう（ダーティリード）
    if got := e.count(); got != 1 {
        t.Fatalf("この時点ではロックがないので未コミットの行が見える: got %d", got)
    }
    t.Log("ロックがないため、未コミットの行が他のトランザクションから見えた（第11章で解決する）")

    e.tm.Rollback(tx1)
    if got := e.count(); got != 0 {
        t.Errorf("ロールバック後の件数: want 0, got %d", got)
    }
}

go test ./transaction/... -v

期待される出力（=== RUN の行は省略）：

--- PASS: TestTransaction_Commit (0.02s)
--- PASS: TestTransaction_Rollback (0.01s)
    transaction_manager_test.go:120: ロックがないため、未コミットの行が他のトランザクションから見えた（第11章で解決する）
--- PASS: TestTransaction_DirtyReadWithoutLocks (0.02s)
PASS
ok      github.com/yourname/go-rdbms/transaction   0.277s

TestTransaction_Rollback では、ロールバックした後に同じ主キーでもう一度INSERTできることも確かめています。ヒープのレコードだけでなく、B+木のキーもきちんと取り消されている証拠です。

一方、TestTransaction_DirtyReadWithoutLocks は、現時点の実装の問題点を示しています。TX1がまだコミットしていないINSERTの結果が、別のトランザクションのSELECTから見えてしまいます。TX1がこの後ロールバックすれば、TX2は「存在しなかったデータ」を読んだことになります。これがダーティリードで、分離性（I）が守られていない状態です。次章でロックを導入して解決します。

まだ残っている問題

本章のロールバックは、UNDO情報（書き込みの記録）をメモリ上に持っています。そのため次の問題があります。

• クラッシュすると取り消せない: トランザクションの途中でプロセスが落ちると書き込みの記録が失われます。バッファプールから追い出されてディスクに書かれていた未コミットの変更は、そのまま残ってしまいます
• コミットしても永続化されない: コミットした変更がバッファプールにしかない状態でクラッシュすると、変更が失われます

どちらも第12章のWALで解決します。

ここまでのコード全体像

go-rdbms/
├── storage/
├── sql/
│   └── executor/
│       └── executor.go            ← トランザクションを受け取るように変更（本章）
└── transaction/
    ├── txid/
    │   └── txid.go                ← トランザクションIDの型（本章）
    ├── transaction.go             ← Transaction・WriteRecord（本章）
    └── transaction_manager.go     ← 開始・コミット・ロールバック（本章）

TransactionManagerの設計で意識したポイント

① 操作を記録し、逆順に取り消す

ロールバックは「記録した操作の逆を、後ろから順に実行する」だけです。INSERTの逆はDELETEなので、取り消し方も単純です。

② インデックスを更新する前に記録する

途中で失敗しても取り消し漏れがないように、記録を先に行います。取り消し時には、実行されなかった操作（B+木への挿入）を安全に無視します。

③ TxIDを独立したパッケージに置く

後の章で lock と log から参照されることを見越して、循環importを避ける配置にしています。

④ 状態の遷移を1か所にまとめる

ACTIVE → COMMITTED / ABORTED の遷移とチェックを finish にまとめることで、二重コミットのような誤った操作を確実に防ぎます。

まとめ

• トランザクションは、複数の操作を「全部反映するか、何も反映しないか」の単位にまとめる
• ACIDはAtomicity・Consistency・Isolation・Durabilityの4つの性質で、それぞれUNDO情報・制約チェック・ロック・WALで実現する
• 本章では、INSERTのたびに書き込みを記録し、ロールバック時に逆順で取り消すTransactionManagerを実装した
• TxIDは、後の章で lock と log からも使うので、独立したパッケージに置いた
• 現時点ではロックがないのでダーティリードが起き、UNDO情報がメモリ上にしかないのでクラッシュに耐えられない

次章では、複数のトランザクションが同時に動いても互いに干渉しないように、ロックマネージャと**2相ロック（2PL）**を実装します。デッドロックの検出も行います。