RDBMSを自作して学ぶ：第5章 ヒープファイルとスロット管理

前回はディスクI/Oを減らすメモリキャッシュ層であるバッファプールマネージャを実装しました。ここまでで「4KBのページを効率よく読み書きする」仕組みはできましたが、ページはまだただのバイト列です。今回はその上に、テーブルの1行（レコード）を格納する仕組みを作ります。ページの中でレコードを管理する**スロット付きページ（Slotted Page）と、複数のページを束ねてテーブル全体を表すヒープファイル（Heap File）**を実装します。

レコードをページに詰め込む

レコードとは

テーブルの1行分のデータをレコードと呼びます。たとえば users(id INT, name VARCHAR(20)) というテーブルなら、1件のレコードは「4バイトの整数」と「最大20文字の文字列」の組です。これをバイト列にして、4096バイトのページに詰め込む方法を考えます。

先頭から順に詰める方法の問題点

最も単純なのは、レコードを固定長にしてページの先頭から順に並べる方法です。

Page (4096バイト)
┌──────────┬──────────┬──────────┬──────────┐
│ Record 0 │ Record 1 │ Record 2 │ ........ │
│ (固定長) │ (固定長) │ (固定長) │          │
└──────────┴──────────┴──────────┴──────────┘

この方法には次の問題があります。

1. 可変長のデータを扱いにくい: VARCHAR(20) を常に20文字ぶん確保すると、短い文字列ばかりのときに領域が無駄になる
2. 削除で穴が空く: 途中のレコードを消すと、そこが使われない隙間になる
3. レコードの位置が変わると困る: 隙間を詰めるためにレコードを動かすと、「3番目のレコード」のように位置で参照していた上位層（インデックスなど）の参照が壊れる

これらをまとめて解決するのがスロット付きページです。

スロット付きページ（Slotted Page）

ページのレイアウト

スロット付きページでは、ページの前方にスロット配列を置き、レコード本体はページの末尾から前に向かって詰めていきます。スロットは「レコードがページ内のどこに、何バイトあるか」を指す小さな索引です。

Page (4096バイト)
┌────────────────────────────────────────────────────┐
│ [0:16]  共通ページヘッダ（第3章）                  │
│ [16:24] ページLSN（第12章のために予約）            │
│ [24:28] 次のヒープページのPageID                   │
├──────────┬──────────┬──────────┬───────────────────┤
│ Slot 0   │ Slot 1   │ Slot 2   │ → スロット配列は  │
│(off,len) │(off,len) │ (削除済) │   後ろへ伸びる    │
├──────────┴──────────┴──────────┴───────────────────┤
│                    空き領域                        │
├────────────────────────────────────────────────────┤
│   レコードは前へ伸びる ←     │ Record 2 │ Record 1 │
├──────────────────────────────┴──────────┴──────────┤
│                                         │ Record 0 │
└────────────────────────────────────────────────────┘

スロット配列とレコード領域が両側から空き領域に向かって伸びていき、ぶつかったらページが満杯です。

第3章で定義した共通ページヘッダ（16バイト）の各フィールドは、次のように使います。

共通ヘッダの直後の8バイトは、ページLSNのために空けておきます。これは第12章でWAL（Write-Ahead Logging）を実装するときに、「このページにどこまでのログが反映されているか」を記録するための領域です。PostgreSQLのページヘッダにも同じ役割の pd_lsn があります。後からヘッダの形を変えると既存のページが読めなくなるので、先に場所だけ確保しておきます。

スロットの形式

1つのスロットは4バイトです。

ページサイズは4096バイトなので、位置も長さも2バイト（最大65535）で表せます。長さ0のレコードは格納しない決まりにして、length == 0 を「削除済み」の印に使います。

この設計の利点

① 可変長レコードをそのまま格納できる

スロットが位置と長さを持つので、レコード本体は必要な長さだけ使えば済みます。

② 削除はスロットに印を付けるだけ

レコードを削除するときは、スロットの長さを0にするだけです。ページ全体を書き換える必要はありません。

③ レコードの参照が安定する

上位層はレコードを「ページID + スロット番号」で参照します。仮にページ内でレコード本体を詰め直したとしても、スロットの中身（位置）を書き換えればよく、スロット番号は変わりません。

本シリーズでの割り切り

実装を小さく保つため、本シリーズでは次の2つを行いません。

• 削除した領域の詰め直し（コンパクション）: 削除したレコードが使っていた領域は、そのページでは再利用されません。PostgreSQLは、ページ内の不要な領域を詰め直す処理を持っていますが、そのときも行ポインタ（スロットにあたるもの）の番号は変えません
• 削除済みスロットの再利用: スロット番号は常に末尾に追加し、削除済みのスロットを別のレコードに割り当て直しません。スロットを使い回すと、「削除したはずのRIDが、いつの間にか別のレコードを指している」という状況が生まれます。これは第10章のロールバックや第12章のリカバリを複雑にします

実装：SlottedPage

第2章で決めたディレクトリ構成どおり、スロット付きページは storage/page パッケージに置きます。SlottedPage は *Page を包んで、スロット付きページとして読み書きするメソッドを提供するラッパーです。

// storage/page/slotted_page.go
package page

import (
    "encoding/binary"
    "errors"
    "fmt"
)

// スロット付きページのレイアウト
//
//    [0:16]  共通ページヘッダ（第3章）
//    [16:24] ページLSN（第12章のWALで使うため予約）
//    [24:28] 次のヒープページのPageID
//    [28:]   スロット配列（1スロット4バイト、先頭から後ろへ伸びる）
//      ...   空き領域
//    [..:4096] レコードデータ（末尾から前へ伸びる）
const (
    offsetNextPageID  = 24
    slottedHeaderSize = 28
    slotSize          = 4 // offset(2B) + length(2B)

    // MaxRecordSize は1ページに格納できるレコードの最大サイズ
    MaxRecordSize = PageSize - slottedHeaderSize - slotSize
)

var (
    // ErrNoSpace はページに空き領域が足りないことを表す
    ErrNoSpace = errors.New("slotted_page: ページに空き領域がありません")
    // ErrRecordDeleted は削除済みのスロットを参照したことを表す
    ErrRecordDeleted = errors.New("slotted_page: レコードは削除済みです")
)

// SlottedPage はPageをスロット付きページとして操作するためのラッパー
type SlottedPage struct {
    p *Page
}

// NewSlottedPage はpをスロット付きページとして扱うSlottedPageを返す
func NewSlottedPage(p *Page) *SlottedPage {
    return &SlottedPage{p: p}
}

// Init はページを空のスロット付きページとして初期化する
func (sp *SlottedPage) Init() {
    sp.p.SetPageType(PageTypeHeap)
    sp.p.SetNumSlots(0)
    sp.p.SetFreeSpaceOffset(PageSize) // レコード領域の先頭 = ページ末尾
    sp.SetNextPageID(InvalidPageID)
}

// NumSlots はスロット数（削除済みを含む）を返す
func (sp *SlottedPage) NumSlots() uint16 {
    return uint16(sp.p.GetNumSlots())
}

// FreeSpace はスロット配列とレコード領域の間の空きバイト数を返す
func (sp *SlottedPage) FreeSpace() int {
    slotArrayEnd := slottedHeaderSize + int(sp.NumSlots())*slotSize
    return int(sp.p.GetFreeSpaceOffset()) - slotArrayEnd
}

// InsertRecord はレコードを格納し、割り当てたスロット番号を返す
func (sp *SlottedPage) InsertRecord(data []byte) (uint16, error) {
    if len(data) == 0 {
        return 0, fmt.Errorf("slotted_page: 空のレコードは格納できません")
    }
    // レコード本体に加えて、新しいスロット1つ分の領域が必要
    if sp.FreeSpace() < len(data)+slotSize {
        return 0, ErrNoSpace
    }

    // レコードをレコード領域の先頭（末尾側から詰める）に書く
    offset := int(sp.p.GetFreeSpaceOffset()) - len(data)
    copy(sp.p[offset:], data)
    sp.p.SetFreeSpaceOffset(uint32(offset))

    // スロット配列の末尾にスロットを追加する
    slotID := sp.NumSlots()
    sp.setSlot(slotID, uint16(offset), uint16(len(data)))
    sp.p.SetNumSlots(uint32(slotID) + 1)
    return slotID, nil
}

// GetRecord はスロット番号のレコードのコピーを返す
func (sp *SlottedPage) GetRecord(slotID uint16) ([]byte, error) {
    if slotID >= sp.NumSlots() {
        return nil, fmt.Errorf("slotted_page: スロット %d は範囲外です", slotID)
    }
    offset, length := sp.getSlot(slotID)
    if length == 0 {
        return nil, ErrRecordDeleted
    }
    data := make([]byte, length)
    copy(data, sp.p[offset:int(offset)+int(length)])
    return data, nil
}

// DeleteRecord はスロットを削除済みにする
// レコード本体の領域は詰め直さず、スロット番号も再利用しない
func (sp *SlottedPage) DeleteRecord(slotID uint16) error {
    if slotID >= sp.NumSlots() {
        return fmt.Errorf("slotted_page: スロット %d は範囲外です", slotID)
    }
    if _, length := sp.getSlot(slotID); length == 0 {
        return ErrRecordDeleted
    }
    sp.setSlot(slotID, 0, 0) // length=0 が削除済みの印
    return nil
}

// NextPageID は次のヒープページのPageIDを返す
func (sp *SlottedPage) NextPageID() PageID {
    return PageID(binary.BigEndian.Uint32(sp.p[offsetNextPageID:]))
}

// SetNextPageID は次のヒープページのPageIDを設定する
func (sp *SlottedPage) SetNextPageID(id PageID) {
    binary.BigEndian.PutUint32(sp.p[offsetNextPageID:], uint32(id))
}

func (sp *SlottedPage) slotPos(slotID uint16) int {
    return slottedHeaderSize + int(slotID)*slotSize
}

func (sp *SlottedPage) getSlot(slotID uint16) (offset, length uint16) {
    pos := sp.slotPos(slotID)
    offset = binary.BigEndian.Uint16(sp.p[pos:])
    length = binary.BigEndian.Uint16(sp.p[pos+2:])
    return offset, length
}

func (sp *SlottedPage) setSlot(slotID uint16, offset, length uint16) {
    pos := sp.slotPos(slotID)
    binary.BigEndian.PutUint16(sp.p[pos:], offset)
    binary.BigEndian.PutUint16(sp.p[pos+2:], length)
}

いくつかポイントを補足します。

• InsertRecord は、レコード本体に加えて**新しいスロット1つ分（4バイト）**の空きも必要です。FreeSpace() がスロット配列の末尾からレコード領域の先頭までの空きを返すので、それと比べています
• MaxRecordSize は空のページに入る最大のレコード長です（4096 − 28 − 4 = 4064バイト）。これより大きいレコードは、どのページにも入りません
• 「空きがない」「削除済み」は呼び出し側で区別したいので、ErrNoSpace と ErrRecordDeleted をセンチネルエラーとして公開し、errors.Is で判定できるようにしています

Record ID（RID）

テーブル内のレコードの位置は「ページID + スロット番号」で一意に決まります。これを RID（Record ID） と呼びます。次章のB+木インデックスは、キーからこのRIDを引く仕組みです。

// storage/heap/rid.go
package heap

import (
    "fmt"

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

// RID (Record ID) はテーブル内のレコードの位置を一意に表す
type RID struct {
    PageID page.PageID
    SlotID uint16
}

func (r RID) String() string {
    return fmt.Sprintf("(%d,%d)", r.PageID, r.SlotID)
}

可変長レコードのシリアライズ

1行の値をどうバイト列にするかも決めておきます。本シリーズでは、各フィールドを「長さ（2バイト）+ データ」の形で順に並べます。

[フィールド0の長さ(2B)][フィールド0のデータ][フィールド1の長さ(2B)][フィールド1のデータ]...

INT型のように固定長の型でも、形式をそろえるために長さを付けます。このパッケージは各フィールドを単なるバイト列として扱い、「INTを4バイトの整数にする」といった型ごとの変換は、第8章の実行エンジンで行います。なお第1章で決めたとおり、本シリーズではNULLを扱いません。

// storage/heap/record.go
package heap

import (
    "encoding/binary"
    "fmt"
    "math"
)

// EncodeRecord は各フィールドのバイト列を「長さ(2バイト) + データ」の形式で連結する
func EncodeRecord(fields [][]byte) ([]byte, error) {
    size := 0
    for _, f := range fields {
        if len(f) > math.MaxUint16 {
            return nil, fmt.Errorf("record: フィールドが長すぎます（%dバイト）", len(f))
        }
        size += 2 + len(f)
    }

    buf := make([]byte, 0, size)
    for _, f := range fields {
        buf = binary.BigEndian.AppendUint16(buf, uint16(len(f)))
        buf = append(buf, f...)
    }
    return buf, nil
}

// DecodeRecord はEncodeRecordで作ったバイト列をフィールドごとに分解する
func DecodeRecord(data []byte) ([][]byte, error) {
    var fields [][]byte
    offset := 0
    for offset < len(data) {
        if offset+2 > len(data) {
            return nil, fmt.Errorf("record: 長さフィールドが途中で切れています")
        }
        length := int(binary.BigEndian.Uint16(data[offset:]))
        offset += 2
        if offset+length > len(data) {
            return nil, fmt.Errorf("record: フィールドのデータが途中で切れています")
        }
        fields = append(fields, data[offset:offset+length])
        offset += length
    }
    return fields, nil
}

実装：HeapFile

ページを連結リストでつなぐ

1つのページに入るレコードはせいぜい数十〜数百件なので、テーブル全体は複数のページにまたがります。ヒープファイルは、テーブルに属するページを連結リストでつないで管理します。各ページのヘッダにある「次のヒープページのPageID」が、リストのポインタです。

HeapFile (usersテーブル)
  firstPageID                                 lastPageID
      │                                           │
      ▼                                           ▼
  ┌────────┐  next  ┌────────┐  next  ┌────────┐
  │ Page 0 │ ─────▶ │ Page 5 │ ─────▶ │ Page 9 │ ─▶ InvalidPageID
  └────────┘        └────────┘        └────────┘

データベースファイルには複数のテーブルのページが混在するので、あるテーブルのページがファイル上で連続しているとは限りません。連結リストにしておけば、ページがどこにあってもたどれます。「ヒープ（heap）」という名前は、レコードを特定の順序に並べず、空いている場所に積み上げていくことに由来します。

挿入は、常に末尾のページに対して試みます。末尾のページが満杯なら新しいページを確保し、末尾につないでから挿入します。末尾のページIDは lastPageID としてメモリに持っておき、既存のヒープファイルを開くときは、先頭ページからリストをたどって見つけます。

構造体と生成

// HeapFile は1つのテーブルのレコードを、連結されたスロット付きページの列として管理する
type HeapFile struct {
    mu          sync.Mutex
    bpm         *buffer.BufferPoolManager
    firstPageID page.PageID
    lastPageID  page.PageID
}

// NewHeapFile は先頭ページを1つ割り当てて、空のHeapFileを作る
func NewHeapFile(bpm *buffer.BufferPoolManager) (*HeapFile, error) {
    pageID, p, err := bpm.NewPage()
    if err != nil {
        return nil, fmt.Errorf("heap: 先頭ページの割り当てに失敗しました: %w", err)
    }
    page.NewSlottedPage(p).Init()
    if err := bpm.UnpinPage(pageID, true); err != nil {
        return nil, err
    }
    return &HeapFile{bpm: bpm, firstPageID: pageID, lastPageID: pageID}, nil
}

// OpenHeapFile は先頭ページIDから既存のHeapFileを開く
// ページのつながりをたどって末尾ページを見つけておく
func OpenHeapFile(bpm *buffer.BufferPoolManager, firstPageID page.PageID) (*HeapFile, error) {
    last := firstPageID
    for {
        p, err := bpm.FetchPage(last)
        if err != nil {
            return nil, fmt.Errorf("heap: ページ %d を読めません: %w", last, err)
        }
        next := page.NewSlottedPage(p).NextPageID()
        if err := bpm.UnpinPage(last, false); err != nil {
            return nil, err
        }
        if next == page.InvalidPageID {
            break
        }
        last = next
    }
    return &HeapFile{bpm: bpm, firstPageID: firstPageID, lastPageID: last}, nil
}

// FirstPageID は先頭ページのPageIDを返す
func (hf *HeapFile) FirstPageID() page.PageID {
    return hf.firstPageID
}

mu は、同じテーブルへの操作を1つずつ順番に実行するためのミューテックスです。複数のゴルーチンが同じページのバイト列を同時に書き換えると、データが壊れてしまいます。

レコードの挿入

// InsertRecord はレコードを格納してRIDを返す
func (hf *HeapFile) InsertRecord(data []byte) (RID, error) {
    if len(data) > page.MaxRecordSize {
        return RID{}, fmt.Errorf("heap: レコードが大きすぎます（%dバイト、上限%dバイト）",
            len(data), page.MaxRecordSize)
    }

    hf.mu.Lock()
    defer hf.mu.Unlock()

    // まず末尾ページに入れてみる
    lastID := hf.lastPageID
    p, err := hf.bpm.FetchPage(lastID)
    if err != nil {
        return RID{}, err
    }
    sp := page.NewSlottedPage(p)
    slotID, err := sp.InsertRecord(data)
    if err == nil {
        return RID{PageID: lastID, SlotID: slotID}, hf.bpm.UnpinPage(lastID, true)
    }
    if !errors.Is(err, page.ErrNoSpace) {
        hf.bpm.UnpinPage(lastID, false)
        return RID{}, err
    }

    // 末尾ページが満杯なら、新しいページを確保して末尾につなぐ
    newID, newP, err := hf.bpm.NewPage()
    if err != nil {
        hf.bpm.UnpinPage(lastID, false)
        return RID{}, fmt.Errorf("heap: 新規ページの割り当てに失敗しました: %w", err)
    }
    newSP := page.NewSlottedPage(newP)
    newSP.Init()
    sp.SetNextPageID(newID)
    if err := hf.bpm.UnpinPage(lastID, true); err != nil {
        return RID{}, err
    }
    hf.lastPageID = newID

    slotID, err = newSP.InsertRecord(data)
    if err != nil {
        hf.bpm.UnpinPage(newID, true)
        return RID{}, err
    }
    return RID{PageID: newID, SlotID: slotID}, hf.bpm.UnpinPage(newID, true)
}

新しいページをつなぐときは、元の末尾ページ（next を書き換える）と新しいページ（初期化する）の両方がダーティになります。どちらも UnpinPage(..., true) で変更を伝えている点に注意してください。

取得・削除

// GetRecord はRIDのレコードを返す
func (hf *HeapFile) GetRecord(rid RID) ([]byte, error) {
    hf.mu.Lock()
    defer hf.mu.Unlock()

    p, err := hf.bpm.FetchPage(rid.PageID)
    if err != nil {
        return nil, err
    }
    defer hf.bpm.UnpinPage(rid.PageID, false)
    return page.NewSlottedPage(p).GetRecord(rid.SlotID)
}

// DeleteRecord はRIDのレコードを削除する
func (hf *HeapFile) DeleteRecord(rid RID) error {
    hf.mu.Lock()
    defer hf.mu.Unlock()

    p, err := hf.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)
        return err
    }
    return hf.bpm.UnpinPage(rid.PageID, true)
}

GetRecord は、ページ上のバイト列ではなくコピーを返します（SlottedPage.GetRecord の中でコピーしています）。ページのバイト列を直接返すと、アンピンした後でそのフレームに別のページが読み込まれたとき、呼び出し側が持っているスライスの中身が勝手に変わってしまうからです。

全件走査（スキャン）

テーブルの全レコードを先頭から順に返すイテレータも用意します。第8章の SeqScan（全件走査）オペレータは、これを使って実装します。

// Scan は全レコードを先頭から順に返すスキャナを作る
func (hf *HeapFile) Scan() *HeapScanner {
    return &HeapScanner{hf: hf, pageID: hf.firstPageID}
}

// HeapScanner はHeapFileのレコードを順番にたどるイテレータ
type HeapScanner struct {
    hf     *HeapFile
    pageID page.PageID // 次に読むページ
    slotID uint16      // 次に読むスロット
}

// Next は次の有効なレコードを返す。最後まで読み終えたら ok=false を返す
func (s *HeapScanner) Next() (rid RID, data []byte, ok bool, err error) {
    s.hf.mu.Lock()
    defer s.hf.mu.Unlock()

    for s.pageID != page.InvalidPageID {
        p, err := s.hf.bpm.FetchPage(s.pageID)
        if err != nil {
            return RID{}, nil, false, err
        }
        sp := page.NewSlottedPage(p)

        for s.slotID < sp.NumSlots() {
            slotID := s.slotID
            s.slotID++
            data, err := sp.GetRecord(slotID)
            if errors.Is(err, page.ErrRecordDeleted) {
                continue // 削除済みスロットは読み飛ばす
            }
            if err != nil {
                s.hf.bpm.UnpinPage(s.pageID, false)
                return RID{}, nil, false, err
            }
            rid := RID{PageID: s.pageID, SlotID: slotID}
            return rid, data, true, s.hf.bpm.UnpinPage(s.pageID, false)
        }

        // このページを読み終えたら次のページへ
        next := sp.NextPageID()
        if err := s.hf.bpm.UnpinPage(s.pageID, false); err != nil {
            return RID{}, nil, false, err
        }
        s.pageID = next
        s.slotID = 0
    }
    return RID{}, nil, false, nil
}

Next は、呼ばれるたびに1件を返し、返す前にページをアンピンしています。スキャンの途中でページをピンしたままにすると、呼び出し側が途中でスキャンをやめたときにピンが残り続けてしまうからです。そのぶん、同じページを何度もフェッチすることになりますが、2回目以降はバッファプールに載っているのでディスクI/Oは発生しません（このコストは第13章のベンチマークで改めて見ます）。

テストを書く

複数ページにまたがる挿入とスキャン、削除、開き直し、大きすぎるレコードをテストします。スキャンのテストでは、バッファプールのサイズを4フレームに絞り、スキャンの途中でページの追い出しが起きるようにしています。

// storage/heap/heap_file_test.go
package heap_test

import (
    "bytes"
    "fmt"
    "os"
    "testing"

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

// setupHeapFile はテスト用の一時ファイル上にHeapFileを作る
func setupHeapFile(t *testing.T, poolSize int) (*heap.HeapFile, *buffer.BufferPoolManager, func()) {
    t.Helper()

    f, err := os.CreateTemp("", "test-heap-*.db")
    if err != nil {
        t.Fatal(err)
    }
    f.Close()

    dm, err := disk.NewFileDiskManager(f.Name())
    if err != nil {
        os.Remove(f.Name())
        t.Fatal(err)
    }
    bpm := buffer.NewBufferPoolManager(dm, poolSize)
    hf, err := heap.NewHeapFile(bpm)
    if err != nil {
        t.Fatal(err)
    }
    cleanup := func() {
        dm.Close()
        os.Remove(f.Name())
    }
    return hf, bpm, cleanup
}

func TestHeapFile_InsertAndGet(t *testing.T) {
    hf, _, cleanup := setupHeapFile(t, 10)
    defer cleanup()

    data := []byte("hello, heap file!")
    rid, err := hf.InsertRecord(data)
    if err != nil {
        t.Fatal(err)
    }

    got, err := hf.GetRecord(rid)
    if err != nil {
        t.Fatal(err)
    }
    if !bytes.Equal(got, data) {
        t.Errorf("データ不一致: want %q, got %q", data, got)
    }
}

func TestHeapFile_ScanAcrossPages(t *testing.T) {
    // プールサイズを小さくして、スキャン中にページの追い出しも起きるようにする
    hf, _, cleanup := setupHeapFile(t, 4)
    defer cleanup()

    const n = 1000
    for i := 0; i < n; i++ {
        rec := []byte(fmt.Sprintf("record-%04d-%s", i, bytes.Repeat([]byte("x"), 80)))
        if _, err := hf.InsertRecord(rec); err != nil {
            t.Fatal(err)
        }
    }

    scanner := hf.Scan()
    count := 0
    pages := map[page.PageID]bool{}
    for {
        rid, data, ok, err := scanner.Next()
        if err != nil {
            t.Fatal(err)
        }
        if !ok {
            break
        }
        want := fmt.Sprintf("record-%04d-", count)
        if !bytes.HasPrefix(data, []byte(want)) {
            t.Fatalf("%d件目のデータ不一致: got %q", count, data[:12])
        }
        pages[rid.PageID] = true
        count++
    }
    if count != n {
        t.Errorf("件数不一致: want %d, got %d", n, count)
    }
    if len(pages) < 2 {
        t.Errorf("複数ページにまたがっていない: %dページ", len(pages))
    }
}

func TestHeapFile_Delete(t *testing.T) {
    hf, _, cleanup := setupHeapFile(t, 10)
    defer cleanup()

    rid1, _ := hf.InsertRecord([]byte("Alice"))
    rid2, _ := hf.InsertRecord([]byte("Bob"))

    if err := hf.DeleteRecord(rid1); err != nil {
        t.Fatal(err)
    }
    if _, err := hf.GetRecord(rid1); err == nil {
        t.Error("削除済みレコードを取得できてしまった")
    }

    // スキャンでは削除済みレコードが読み飛ばされる
    rid, data, ok, err := hf.Scan().Next()
    if err != nil || !ok {
        t.Fatalf("スキャン失敗: ok=%v, err=%v", ok, err)
    }
    if rid != rid2 || string(data) != "Bob" {
        t.Errorf("削除されていないBobが返るはず: rid=%v, data=%q", rid, data)
    }
}

func TestHeapFile_Reopen(t *testing.T) {
    hf, bpm, cleanup := setupHeapFile(t, 10)
    defer cleanup()

    for i := 0; i < 200; i++ {
        if _, err := hf.InsertRecord(bytes.Repeat([]byte("y"), 100)); err != nil {
            t.Fatal(err)
        }
    }

    // 先頭ページIDだけを手がかりに開き直すと、末尾ページが復元される
    reopened, err := heap.OpenHeapFile(bpm, hf.FirstPageID())
    if err != nil {
        t.Fatal(err)
    }
    rid, err := reopened.InsertRecord([]byte("after reopen"))
    if err != nil {
        t.Fatal(err)
    }
    if rid.PageID == hf.FirstPageID() {
        t.Errorf("末尾ページに追記されるはずが先頭ページに入った: %v", rid)
    }
}

func TestHeapFile_TooLargeRecord(t *testing.T) {
    hf, _, cleanup := setupHeapFile(t, 10)
    defer cleanup()

    if _, err := hf.InsertRecord(make([]byte, page.MaxRecordSize+1)); err == nil {
        t.Error("1ページに収まらないレコードがエラーにならなかった")
    }
    if _, err := hf.InsertRecord(make([]byte, page.MaxRecordSize)); err != nil {
        t.Errorf("上限ちょうどのレコードは格納できるはず: %v", err)
    }
}

func TestRecord_EncodeDecode(t *testing.T) {
    fields := [][]byte{{0, 0, 0, 42}, []byte("Alice"), {}}
    data, err := heap.EncodeRecord(fields)
    if err != nil {
        t.Fatal(err)
    }
    got, err := heap.DecodeRecord(data)
    if err != nil {
        t.Fatal(err)
    }
    if len(got) != len(fields) {
        t.Fatalf("フィールド数不一致: want %d, got %d", len(fields), len(got))
    }
    for i := range fields {
        if !bytes.Equal(got[i], fields[i]) {
            t.Errorf("フィールド%d不一致: want %v, got %v", i, fields[i], got[i])
        }
    }
}

go test ./storage/heap/... -v

期待される出力：

=== RUN   TestHeapFile_InsertAndGet
--- PASS: TestHeapFile_InsertAndGet (0.00s)
=== RUN   TestHeapFile_ScanAcrossPages
--- PASS: TestHeapFile_ScanAcrossPages (0.00s)
=== RUN   TestHeapFile_Delete
--- PASS: TestHeapFile_Delete (0.00s)
=== RUN   TestHeapFile_Reopen
--- PASS: TestHeapFile_Reopen (0.00s)
=== RUN   TestHeapFile_TooLargeRecord
--- PASS: TestHeapFile_TooLargeRecord (0.00s)
=== RUN   TestRecord_EncodeDecode
--- PASS: TestRecord_EncodeDecode (0.00s)
PASS
ok      github.com/yourname/go-rdbms/storage/heap   0.229s

ここまでのコード全体像

本章が完了した時点のパッケージ構成です。

storage/
├── page/
│   ├── page.go                    ← Page型・PageID型・ページヘッダ操作
│   └── slotted_page.go            ← スロット付きページ（本章）
├── disk/
│   ├── disk_manager.go
│   └── file_disk_manager.go
├── buffer/
│   ├── replacer.go
│   ├── lru_replacer.go
│   └── buffer_pool_manager.go
└── heap/
    ├── rid.go                     ← Record ID（本章）
    ├── record.go                  ← レコードのシリアライズ（本章）
    └── heap_file.go               ← ヒープファイルとスキャナ（本章）

ヒープファイルの設計で意識したポイント

① スロットによる間接参照

上位層が持つのは「ページID + スロット番号」だけで、ページ内の物理的な位置はスロットの中に隠れています。この一段の間接参照のおかげで、レコード本体の配置を変えてもRIDは変わりません。

② ページを連結リストでつなぐ

テーブルのページがファイル上のどこにあっても、先頭ページIDさえ覚えておけば全ページをたどれます。第9章のカタログには、テーブルごとにこの先頭ページIDを保存します。

③ 削除済みスロットを再利用しない

RIDが指すレコードが入れ替わらないことを保証するためです。領域の無駄は増えますが、ロールバックとリカバリの実装が単純になります。

④ 呼び出し側が判別したいエラーはセンチネルエラーにする

「ページが満杯」はヒープファイルにとって、エラーではなく「次のページへ進む」合図です。errors.Is(err, page.ErrNoSpace) で判定できるようにして、制御の流れを明確にしています。

まとめ

• スロット付きページは、前方のスロット配列と後方のレコード領域が向かい合う構造で、可変長レコードを格納できる
• スロットの長さを0にするだけでレコードを削除できる。本シリーズでは削除した領域の詰め直しとスロットの再利用は行わない
• レコードの位置は RID（ページID + スロット番号）で一意に表す
• ヒープファイルはテーブルのページを連結リストでつなぎ、末尾のページにレコードを追加していく
• スキャナは1件ずつページをフェッチ・アンピンしながら、全レコードを順に返す

次章では、キーからRIDを高速に引くためのB+木インデックスを実装します。内部ノードをたどる検索と、ノードの分割によって木が成長していく様子を見ていきます。