RDBMSを自作して学ぶ：第9章 カタログ管理

前回は、ASTから実行計画を作るPlannerと、イテレータモデルで動くExecutorを実装し、SQLでデータを読み書きできるようになりました。ただし、テーブル定義を管理するカタログはメモリ上にしかないので、プログラムを終了するとテーブルの存在ごと忘れてしまいます。データファイルにはレコードが残っていても、どのページがどのテーブルのものかがわからなければ読めません。今回は、カタログをファイルに永続化して、データベースを開き直してもテーブルを使い続けられるようにします。

カタログとは

カタログは、データベースの中にどんなテーブルがあり、それぞれがどんな構造をしているかという「データについてのデータ（メタデータ）」を管理するコンポーネントです。PostgreSQLの pg_catalog スキーマにある pg_class（テーブルの一覧）や pg_attribute（カラムの一覧）、MySQLの information_schema で見られる情報がこれにあたります。

クエリを実行するとき、PlannerとExecutorはカタログから次の情報を得ます。

テーブル定義をどこに持つか

カタログをどこに保存するかには、大きく2つの方式があります。

システムテーブル方式では、カタログの読み書きにも普段のテーブルと同じ仕組み（ヒープ、インデックス、トランザクション）が使えます。CREATE TABLE をトランザクションの中で実行してロールバックできる、といった性質も自然に得られます。ただし「カタログを読むためにカタログが必要」という鶏と卵の問題があるので、カタログ自体の置き場所は決め打ちにしておく必要があります。例えばSQLiteでは、sqlite_schema テーブルのルートは必ずデータベースファイルの1ページ目にあります。

本シリーズでは実装の簡単さを優先して、カタログをJSONファイルに保存する方式を取ります。人が中身を読んで確かめられるのも利点です。その代わり、テーブルの作成（DDL）はトランザクションの対象外になります。

何を保存するか

JSONに保存するのは、テーブル名・スキーマ、そしてヒープファイルとインデックスの入口のページIDだけです。ヒープファイルは第5章で先頭ページからリストをたどれるようにし、B+木は第6章でルートのPageIDが変わらないようにしたので、入口さえわかれば残りはデータファイルから復元できます。

実際に保存されるカタログファイルは、次のようになります（後で示すテストの出力です）。

{
  "tables": [
    {
      "name": "USERS",
      "schema": {
        "columns": [
          {
            "name": "ID",
            "type": "INT",
            "primary_key": true
          },
          {
            "name": "NAME",
            "type": "VARCHAR",
            "length": 50
          }
        ]
      },
      "first_page_id": 0,
      "index_root_page_id": 1
    }
  ]
}

主キーのないテーブルでは、index_root_page_id に無効なページID（InvalidPageID = 4294967295）を入れます。ページ0も有効なページIDなので、「0なら無し」とはできないからです。

実装：スキーマのJSON化

前章の schema.go にJSONのタグを付けます。DataType は整数のままだとJSONに 0 や 1 と書かれてしまうので、MarshalText / UnmarshalText を実装して "INT" / "VARCHAR" という文字列で保存します。encoding/json は、型がこれらのメソッドを持っていれば自動的に使ってくれます。

// sql/catalog/schema.go
package catalog

import "fmt"

// DataType はカラムのデータ型
type DataType int

const (
    TypeInt     DataType = iota // 32ビット符号付き整数
    TypeVarchar                 // 可変長文字列
)

func (t DataType) String() string {
    if t == TypeInt {
        return "INT"
    }
    return "VARCHAR"
}

// MarshalText はJSONに "INT" / "VARCHAR" という文字列で保存するためのメソッド
func (t DataType) MarshalText() ([]byte, error) {
    return []byte(t.String()), nil
}

// UnmarshalText はJSONの "INT" / "VARCHAR" をDataTypeに戻す
func (t *DataType) UnmarshalText(b []byte) error {
    switch string(b) {
    case "INT":
        *t = TypeInt
    case "VARCHAR":
        *t = TypeVarchar
    default:
        return fmt.Errorf("catalog: 未知の型です: %s", b)
    }
    return nil
}

// Column はカラムの定義
type Column struct {
    Name       string   `json:"name"`
    Type       DataType `json:"type"`
    Length     int      `json:"length,omitempty"` // VARCHARの最大文字数
    PrimaryKey bool     `json:"primary_key,omitempty"`
}

// Schema はテーブルのカラム定義の並び
type Schema struct {
    Columns []Column `json:"columns"`
}

// PrimaryKeyIndex は主キーのカラム位置を返す（主キーがなければ-1）
func (s *Schema) PrimaryKeyIndex() int {
    for i, c := range s.Columns {
        if c.PrimaryKey {
            return i
        }
    }
    return -1
}

実装：永続化するカタログ

TableInfo には、JSONに保存するページIDと、実行時にそこから復元するヒープファイル・B+木の両方を持たせます。後者には json:"-" を付けて、保存の対象から外します。

// sql/catalog/catalog.go
package catalog

import (
    "encoding/json"
    "errors"
    "fmt"
    "io/fs"
    "os"
    "sort"
    "sync"

    "github.com/yourname/go-rdbms/storage/btree"
    "github.com/yourname/go-rdbms/storage/buffer"
    "github.com/yourname/go-rdbms/storage/heap"
    "github.com/yourname/go-rdbms/storage/page"
)

// TableInfo はテーブルの定義と、実データ・インデックスへの参照を持つ
type TableInfo struct {
    Name            string      `json:"name"`
    Schema          *Schema     `json:"schema"`
    FirstPageID     page.PageID `json:"first_page_id"`      // ヒープファイルの先頭ページ
    IndexRootPageID page.PageID `json:"index_root_page_id"` // 主キーのB+木のルート（なければInvalidPageID）

    // 以下は起動時にページIDから復元するので、JSONには保存しない
    Heap  *heap.HeapFile   `json:"-"`
    Index *btree.BPlusTree `json:"-"` // 主キーがなければnil
}

// catalogFile はカタログファイル（JSON）の中身
type catalogFile struct {
    Tables []*TableInfo `json:"tables"`
}

// Catalog はテーブル定義を管理し、JSONファイルに永続化する
type Catalog struct {
    mu     sync.RWMutex
    bpm    *buffer.BufferPoolManager
    path   string
    tables map[string]*TableInfo
}

// NewCatalog はカタログファイルを読み込んでカタログを作る
// ファイルがまだなければ、空のカタログとして始める
func NewCatalog(bpm *buffer.BufferPoolManager, path string) (*Catalog, error) {
    c := &Catalog{bpm: bpm, path: path, tables: make(map[string]*TableInfo)}
    if err := c.load(); err != nil {
        return nil, fmt.Errorf("catalog: %s を読み込めません: %w", path, err)
    }
    return c, nil
}

// load はカタログファイルを読み、各テーブルのヒープファイルとインデックスを開き直す
func (c *Catalog) load() error {
    data, err := os.ReadFile(c.path)
    if errors.Is(err, fs.ErrNotExist) {
        return nil
    }
    if err != nil {
        return err
    }

    var cf catalogFile
    if err := json.Unmarshal(data, &cf); err != nil {
        return err
    }
    for _, info := range cf.Tables {
        if info.Heap, err = heap.OpenHeapFile(c.bpm, info.FirstPageID); err != nil {
            return err
        }
        if info.IndexRootPageID != page.InvalidPageID {
            info.Index = btree.OpenBPlusTree(c.bpm, info.IndexRootPageID)
        }
        c.tables[info.Name] = info
    }
    return nil
}

// save はカタログ全体をJSONファイルに書き出す
// 一時ファイルに書いてfsyncしてからリネームするので、書き込み途中で落ちても
// 「古いカタログ」か「新しいカタログ」のどちらかが必ず残る
func (c *Catalog) save() error {
    cf := catalogFile{}
    for _, name := range c.tableNames() {
        cf.Tables = append(cf.Tables, c.tables[name])
    }
    data, err := json.MarshalIndent(cf, "", "  ")
    if err != nil {
        return err
    }

    tmp := c.path + ".tmp"
    f, err := os.Create(tmp)
    if err != nil {
        return err
    }
    if _, err := f.Write(data); err != nil {
        f.Close()
        return err
    }
    if err := f.Sync(); err != nil {
        f.Close()
        return err
    }
    if err := f.Close(); err != nil {
        return err
    }
    return os.Rename(tmp, c.path)
}

// CreateTable はテーブルを作成し、ヒープファイルと主キーのインデックスを用意する
func (c *Catalog) CreateTable(name string, columns []Column) (*TableInfo, error) {
    c.mu.Lock()
    defer c.mu.Unlock()

    if _, exists := c.tables[name]; exists {
        return nil, fmt.Errorf("catalog: テーブル %s は既に存在します", name)
    }
    schema := &Schema{Columns: columns}
    if err := validateSchema(schema); err != nil {
        return nil, err
    }

    hf, err := heap.NewHeapFile(c.bpm)
    if err != nil {
        return nil, err
    }
    info := &TableInfo{
        Name:            name,
        Schema:          schema,
        FirstPageID:     hf.FirstPageID(),
        IndexRootPageID: page.InvalidPageID,
        Heap:            hf,
    }
    if schema.PrimaryKeyIndex() >= 0 {
        if info.Index, err = btree.NewBPlusTree(c.bpm); err != nil {
            return nil, err
        }
        info.IndexRootPageID = info.Index.RootPageID()
    }

    // カタログが指すページを先にディスクへ書いてから、カタログを保存する。
    // 逆の順序だと、カタログだけ残って中身が初期化されていないページを指す恐れがある
    if err := c.bpm.FlushAllPages(); err != nil {
        return nil, err
    }
    c.tables[name] = info
    if err := c.save(); err != nil {
        delete(c.tables, name)
        return nil, err
    }
    return info, nil
}

// GetTable はテーブル名からテーブル情報を返す
func (c *Catalog) GetTable(name string) (*TableInfo, error) {
    c.mu.RLock()
    defer c.mu.RUnlock()

    info, ok := c.tables[name]
    if !ok {
        return nil, fmt.Errorf("catalog: テーブル %s は存在しません", name)
    }
    return info, nil
}

// ListTables はテーブル名の一覧を名前順で返す
func (c *Catalog) ListTables() []string {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.tableNames()
}

func (c *Catalog) tableNames() []string {
    names := make([]string, 0, len(c.tables))
    for name := range c.tables {
        names = append(names, name)
    }
    sort.Strings(names)
    return names
}

// validateSchema はカラム定義の妥当性を確認する
func validateSchema(s *Schema) error {
    if len(s.Columns) == 0 {
        return fmt.Errorf("catalog: カラムが1つもありません")
    }
    seen := map[string]bool{}
    pkCount := 0
    for _, col := range s.Columns {
        if seen[col.Name] {
            return fmt.Errorf("catalog: カラム %s が重複しています", col.Name)
        }
        seen[col.Name] = true
        if col.PrimaryKey {
            pkCount++
            if col.Type != TypeInt {
                return fmt.Errorf("catalog: PRIMARY KEY は INT 型のカラムにだけ指定できます")
            }
        }
    }
    if pkCount > 1 {
        return fmt.Errorf("catalog: PRIMARY KEY は1つのカラムにだけ指定できます")
    }
    return nil
}

ポイントを3つ見ていきます。

起動時の復元

load はカタログファイルを読み、各テーブルについて heap.OpenHeapFile と btree.OpenBPlusTree でヒープファイルとB+木を開き直します。ファイルがまだなければ（初回起動）、空のカタログとして始めます。

書き込みの順序

CreateTable では、ヒープファイルとB+木のページをディスクに書き出してからカタログを保存しています。逆の順序だと、カタログを保存した直後にクラッシュしたとき、カタログが「初期化されていないページ」を指したまま残ってしまいます。この順序なら、途中でクラッシュしても、最悪でもどこからも参照されないページが残るだけです。

ページの書き出しには、バッファプールに新しく追加する FlushAllPages を使います。

// storage/buffer/buffer_pool_manager.go（追加）
// FlushAllPages はダーティなページをすべてディスクに書き戻し、fsyncで永続化する
// ピン中のページは他のゴルーチンが書き換えている最中かもしれないので対象外にする
func (bpm *BufferPoolManager) FlushAllPages() error {
    bpm.mu.Lock()
    defer bpm.mu.Unlock()

    for pageID, frameID := range bpm.pageTable {
        if bpm.frames[frameID].pinCount > 0 {
            continue
        }
        if err := bpm.flushFrame(pageID); err != nil {
            return err
        }
    }
    return bpm.diskManager.Sync()
}

ダーティなページをすべて書き戻した後、Sync（fsync）でOSのバッファからディスクへ確実に書き込みます。ピン中のページを対象外にしているのは、他のゴルーチンがそのページを書き換えている最中かもしれないからです。書き換えの途中の中途半端な状態をディスクに書いてしまうことになります。新しく作ったヒープファイルとB+木のページは、作成が終わった時点でアンピン済みなので、確実に書き出されます。

一時ファイルとリネームによる保存

save は、カタログファイルを直接上書きしません。いったん一時ファイル（catalog.json.tmp）に全体を書いてfsyncし、その後 os.Rename で本来のファイル名に置き換えます。

① catalog.json.tmp に新しい内容を書く → fsync
② catalog.json.tmp を catalog.json にリネーム（置き換え）

直接上書きすると、書き込みの途中でクラッシュしたときに、前半だけが新しい中途半端なJSONが残ってしまいます。POSIXのファイルシステムでは同じディレクトリ内のリネームは原子的なので、この方法なら、どの時点でクラッシュしても「古いカタログ」か「新しいカタログ」のどちらかが完全な形で残ります。

厳密には、リネームしたこと自体を電源断に耐えさせるには、ディレクトリにもfsyncが必要です。本シリーズでは省略していますが、SQLiteやPostgreSQLのような実際のデータベースは、こうしたファイルシステムの細かな挙動まで考慮して作られています。

テストを書く

カタログのテストでは、テーブルの作成と取得、不正な定義、そして開き直したときに定義・ヒープファイル・インデックスが復元されることを確かめます。

// sql/catalog/catalog_test.go
package catalog_test

import (
    "os"
    "path/filepath"
    "strings"
    "testing"

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

// openCatalog はdir内のデータファイルとカタログファイルを開く
func openCatalog(t *testing.T, dir string) (*catalog.Catalog, *buffer.BufferPoolManager, disk.DiskManager) {
    t.Helper()
    dm, err := disk.NewFileDiskManager(filepath.Join(dir, "test.db"))
    if err != nil {
        t.Fatal(err)
    }
    bpm := buffer.NewBufferPoolManager(dm, 32)
    cat, err := catalog.NewCatalog(bpm, filepath.Join(dir, "catalog.json"))
    if err != nil {
        t.Fatal(err)
    }
    return cat, bpm, dm
}

var userColumns = []catalog.Column{
    {Name: "ID", Type: catalog.TypeInt, PrimaryKey: true},
    {Name: "NAME", Type: catalog.TypeVarchar, Length: 50},
}

func TestCatalog_CreateAndGet(t *testing.T) {
    cat, _, dm := openCatalog(t, t.TempDir())
    defer dm.Close()

    if _, err := cat.CreateTable("USERS", userColumns); err != nil {
        t.Fatal(err)
    }
    info, err := cat.GetTable("USERS")
    if err != nil {
        t.Fatal(err)
    }
    if len(info.Schema.Columns) != 2 || info.Heap == nil || info.Index == nil {
        t.Errorf("テーブル情報が不完全: %+v", info)
    }
    if _, err := cat.GetTable("NOTHING"); err == nil {
        t.Error("存在しないテーブルでエラーにならなかった")
    }
}

func TestCatalog_InvalidDefinitions(t *testing.T) {
    cat, _, dm := openCatalog(t, t.TempDir())
    defer dm.Close()

    cat.CreateTable("USERS", userColumns)
    tests := []struct {
        table string
        cols  []catalog.Column
    }{
        {"USERS", userColumns}, // 同名のテーブル
        {"T1", []catalog.Column{{Name: "A", Type: catalog.TypeInt}, {Name: "A", Type: catalog.TypeInt}}},
        {"T2", []catalog.Column{{Name: "A", Type: catalog.TypeVarchar, Length: 5, PrimaryKey: true}}},
        {"T3", []catalog.Column{
            {Name: "A", Type: catalog.TypeInt, PrimaryKey: true},
            {Name: "B", Type: catalog.TypeInt, PrimaryKey: true},
        }},
    }
    for _, tt := range tests {
        if _, err := cat.CreateTable(tt.table, tt.cols); err == nil {
            t.Errorf("%s: エラーになるはず", tt.table)
        }
    }
}

func TestCatalog_Persistence(t *testing.T) {
    dir := t.TempDir()

    // 1回目: テーブルを作ってデータを入れ、ページを書き出して閉じる
    {
        cat, bpm, dm := openCatalog(t, dir)
        info, err := cat.CreateTable("USERS", userColumns)
        if err != nil {
            t.Fatal(err)
        }
        for i := int64(1); i <= 300; i++ {
            rid, err := info.Heap.InsertRecord([]byte(strings.Repeat("x", 50)))
            if err != nil {
                t.Fatal(err)
            }
            if err := info.Index.Insert(i, rid); err != nil {
                t.Fatal(err)
            }
        }
        if err := bpm.FlushAllPages(); err != nil {
            t.Fatal(err)
        }
        dm.Close()
    }

    // カタログファイルの中身はJSON
    data, _ := os.ReadFile(filepath.Join(dir, "catalog.json"))
    t.Logf("catalog.json:\n%s", data)

    // 2回目: 開き直すと、テーブル定義・ヒープ・インデックスが復元される
    cat, _, dm := openCatalog(t, dir)
    defer dm.Close()

    info, err := cat.GetTable("USERS")
    if err != nil {
        t.Fatalf("再オープン後にテーブルが見つからない: %v", err)
    }
    if info.Schema.Columns[1].Type != catalog.TypeVarchar || info.Schema.Columns[1].Length != 50 {
        t.Errorf("スキーマが復元されていない: %+v", info.Schema.Columns[1])
    }

    count := 0
    for scanner := info.Heap.Scan(); ; count++ {
        _, _, ok, err := scanner.Next()
        if err != nil {
            t.Fatal(err)
        }
        if !ok {
            break
        }
    }
    if count != 300 {
        t.Errorf("ヒープのレコード数: want 300, got %d", count)
    }
    if _, found, _ := info.Index.Search(150); !found {
        t.Error("インデックスからキー150が引けない")
    }

    // 開き直した後も、ヒープの末尾に追記できる
    if _, err := info.Heap.InsertRecord([]byte("after reopen")); err != nil {
        t.Fatal(err)
    }
}

Executorのテストも、テンポラリディレクトリにデータファイルとカタログファイルを置く形に変え、SQLレベルで開き直しを確かめるテストを追加します。

// sql/executor/executor_test.go（変更・追加部分）
func setup(t *testing.T) *testDB {
    t.Helper()
    db, _ := openDB(t, t.TempDir())
    return db
}

// openDB はdir内のデータファイルとカタログファイルを開く
func openDB(t *testing.T, dir string) (*testDB, *buffer.BufferPoolManager) {
    t.Helper()
    dm, err := disk.NewFileDiskManager(filepath.Join(dir, "test.db"))
    if err != nil {
        t.Fatal(err)
    }
    t.Cleanup(func() { dm.Close() })
    bpm := buffer.NewBufferPoolManager(dm, 64)
    cat, err := catalog.NewCatalog(bpm, filepath.Join(dir, "catalog.json"))
    if err != nil {
        t.Fatal(err)
    }
    return &testDB{t: t, cat: cat, exec: executor.New(cat)}, bpm
}

func TestExecutor_Reopen(t *testing.T) {
    dir := t.TempDir()

    // 1回目: テーブルを作って1000行入れ、ページを書き出す
    db, bpm := openDB(t, dir)
    db.run("CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(20))")
    for i := 1; i <= 1000; i++ {
        db.run(fmt.Sprintf("INSERT INTO users VALUES (%d, 'user%d')", i, i))
    }
    if err := bpm.FlushAllPages(); err != nil {
        t.Fatal(err)
    }

    // 2回目: 開き直してもSQLで同じデータが読める（インデックスも使える）
    db2, _ := openDB(t, dir)
    if got := rows(db2.run("SELECT name FROM users WHERE id = 777")); len(got) != 1 || got[0] != "user777" {
        t.Errorf("再オープン後の検索結果が不正: %v", got)
    }
    if got := len(db2.run("SELECT * FROM users").Rows); got != 1000 {
        t.Errorf("再オープン後の件数: want 1000, got %d", got)
    }
}

go test ./sql/... -v

期待される出力（抜粋）：

--- PASS: TestCatalog_CreateAndGet (0.02s)
--- PASS: TestCatalog_InvalidDefinitions (0.01s)
--- PASS: TestCatalog_Persistence (0.01s)
PASS
ok      github.com/yourname/go-rdbms/sql/catalog    0.260s
--- PASS: TestExecutor_SelectWhere (0.01s)
--- PASS: TestExecutor_IndexScan (0.02s)
--- PASS: TestExecutor_Join (0.02s)
--- PASS: TestExecutor_InsertErrors (0.01s)
--- PASS: TestPlanner_Explain (0.01s)
--- PASS: TestExecutor_Reopen (0.03s)
PASS
ok      github.com/yourname/go-rdbms/sql/executor   0.493s

TestExecutor_Reopen では、1000行を挿入して FlushAllPages した後、別のバッファプールでデータベースを開き直し、WHERE id = 777（IndexScan）と全件検索の両方で、同じデータが読めることを確かめています。

なお、このテストでは開き直す前に FlushAllPages を明示的に呼んでいます。これを呼ばずに終了すると、バッファプールにしかない変更は失われます。「コミットした変更は、クラッシュしても失われない」ことを保証する仕組みは、第12章のWALで実装します。

ここまでのコード全体像

go-rdbms/
├── storage/
│   ├── page/  disk/  heap/  btree/
│   └── buffer/
│       └── buffer_pool_manager.go ← FlushAllPagesを追加（本章）
└── sql/
    ├── lexer/  parser/  planner/  executor/
    └── catalog/
        ├── schema.go              ← JSONのタグとDataTypeの文字列化（本章）
        └── catalog.go             ← カタログの永続化（本章）

カタログの設計で意識したポイント

① 入口のページIDだけを保存する

ヒープファイルは先頭ページから、B+木はルートからたどれるように設計してきたので、カタログに保存するのは入口のページIDだけで済みます。ページの中身はデータファイルが正です。

② ページを書いてからカタログを書く

参照される側（ページ）を先に永続化し、参照する側（カタログ）を後に書くことで、どこでクラッシュしてもカタログが壊れたページを指さないようにしています。

③ 一時ファイルとリネームで原子的に置き換える

ファイルの上書きは原子的ではありませんが、リネームは原子的です。「全部書けたら差し替える」ことで、中途半端な状態を残しません。

④ 人が読める形式にする

JSONにしたことで、カタログの中身をエディタで直接確かめられます。自作のデータベースをデバッグするときには大きな助けになります。

この方式の限界

• テーブルの作成（DDL）はトランザクションの対象外で、ロールバックできません
• カタログファイルとデータファイルは別々に書かれるので、両方を1つの操作として同時に更新することはできません（上記の書き込み順序で、矛盾が起きないようにしています）
• DROP TABLE などのテーブル削除は実装していません

まとめ

• カタログは、テーブル名・スキーマ・ヒープファイルの先頭ページID・インデックスのルートページIDを管理する
• カタログの保存方式には、システムテーブル（PostgreSQL・SQLite）と別ファイル（本シリーズ）がある
• 本シリーズではカタログをJSONに保存し、起動時にページIDからヒープファイルとB+木を開き直す
• ページを先に書き出してからカタログを保存し、カタログは一時ファイルに書いてからリネームで置き換える
• ここまでで、SQLでテーブルを作り、データを入れ、開き直しても同じデータを検索できるようになった

次章からはトランザクション管理の実装に入ります。まずは、業務でもよく耳にするACID特性が実装上どういう意味を持つのかを整理し、TransactionManagerの骨格を作ります。