Database Design Expert

データベース設計エキスパート

0. 必須の参照手順

重要：データベーススキーマを実装する前に、必ず関連する参照ファイルを読んでください。

参照ファイルを読む条件

次の場合は references/advanced-patterns.md を読んでください：

• 新機能のスキーマを設計する場合
• 複雑なリレーションシップ（多対多、多態性）を実装する場合
• 継承パターンを設定する場合
• 高パフォーマンスのクエリ向けに設計する場合

次の場合は references/security-examples.md を読んでください：

• 機密性の高いユーザーデータを保存する場合
• 監査証跡を設計する場合
• データベースレベルでアクセス制御を実装する場合
• PII や金融データを扱う場合

---

1. 概要

リスクレベル：中

理由：データベーススキーマの設計は、データの整合性、クエリのパフォーマンス、アプリケーションのセキュリティに影響します。設計が不適切だと、データの破損、パフォーマンスのボトルネック、データの一貫性の維持が困難になるといった問題につながる可能性があります。本番環境でのスキーマ変更には、慎重なマイグレーション計画が必要です。

データベーススキーマ設計のエキスパートとして、次の分野を専門としています。

• 正規化と、パフォーマンス向上のための適切な非正規化
• クエリを最適化するためのインデックス戦略
• **全文検索（FTS5）**のスキーマ設計
• データの整合性を確保するための制約設計
• 安全に進化できる、マイグレーションに適したスキーマ

基本原則

1. まず TDD - 実装前に、スキーマとクエリのテストを作成します
2. パフォーマンスを考慮 - クエリパターンに合わせて設計し、インデックスを最適化して、定期的にプロファイリングします
3. 正規化してから非正規化 - まず 3NF で設計し、測定したニーズに基づいて非正規化します
4. すべてに制約を適用 - 最後の防衛線としてデータベース制約を使用します
5. 安全なマイグレーション - すべてのスキーマ変更は元に戻せるようにし、テストします

主なユースケース

• デスクトップアプリケーションのデータモデリング
• ローカルファーストのアプリケーションアーキテクチャ
• 効率的な検索と取得のパターン
• 監査と履歴の追跡
• 設定と環境設定の保存

---

2. 主な責務

2.1 データ整合性の原則

1. 冗長性を排除するために正規化する - その後、パフォーマンス向上のために計画的に非正規化します
2. 適切な制約を使用する - 主キー、外部キー、一意制約、チェック制約
3. 参照整合性を考慮して設計する - 適切なカスケードルールを設定した外部キーを使用します
4. スキーマの進化に備える - データを保持するマイグレーションを設計します

2.2 パフォーマンス設計の原則

1. クエリに合わせてインデックスを作成する - インデックスを作成する前にクエリパターンを分析します
2. インデックスの過剰作成を避ける - インデックスはそれぞれ書き込みを遅くします
3. カバリングインデックスを使用する - テーブル参照を避けるため、インデックスに列を含めます
4. データの局所性を考慮して設計する - 関連するデータをまとめて保持します

---

3. 技術的基盤

3.1 SQLite データ型

SQLite 型	用途	備考
INTEGER	ID、カウント、ブール値	自動インクリメントには PRIMARY KEY を使用
TEXT	文字列、JSON、UUID	長さの制限なし
REAL	浮動小数点数	8 バイト IEEE 浮動小数点数
BLOB	バイナリデータ	ファイル、暗号化データ
NUMERIC	日付、小数	最も効率的な型で保存

3.2 正規化の段階

形式	説明	使用する場面
1NF	原子値、繰り返しグループなし	常に
2NF	1NF ＋ 部分関数従属なし	ほとんどのテーブル
3NF	2NF ＋ 推移的関数従属なし	デフォルトの選択
BCNF	3NF ＋ すべての決定子がキー	複雑なリレーションシップ

---

4. 実装パターン

4.1 基本テーブルのテンプレート

CREATE TABLE entities (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL CHECK(length(name) BETWEEN 1 AND 255),
    email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'),
    status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active', 'inactive', 'deleted')),
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    deleted_at TEXT
);

CREATE INDEX idx_entities_status ON entities(status) WHERE deleted_at IS NULL;

4.2 リレーションシップのパターン

一対多

CREATE TABLE documents (
    id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, title TEXT NOT NULL,
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE INDEX idx_documents_user ON documents(user_id);

多対多

CREATE TABLE document_tags (
    document_id INTEGER NOT NULL, tag_id INTEGER NOT NULL,
    PRIMARY KEY (document_id, tag_id),
    FOREIGN KEY (document_id) REFERENCES documents(id) ON DELETE CASCADE,
    FOREIGN KEY (tag_id) REFERENCES tags(id) ON DELETE CASCADE
);
CREATE INDEX idx_doctags_tag ON document_tags(tag_id);

自己参照型（階層構造）

-- Tree structure (adjacency list)
CREATE TABLE categories (
    id INTEGER PRIMARY KEY,
    parent_id INTEGER REFERENCES categories(id) ON DELETE CASCADE,
    name TEXT NOT NULL
);
CREATE INDEX idx_categories_parent ON categories(parent_id);

4.3 全文検索スキーマ

-- Content table
CREATE TABLE articles (
    id INTEGER PRIMARY KEY, title TEXT NOT NULL, body TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now'))
);

-- FTS5 virtual table
CREATE VIRTUAL TABLE articles_fts USING fts5(
    title, body, content=articles, content_rowid=id,
    tokenize='porter unicode61', prefix='2,3'
);

-- Sync triggers (INSERT, UPDATE, DELETE)
CREATE TRIGGER articles_ai AFTER INSERT ON articles BEGIN
    INSERT INTO articles_fts(rowid, title, body) VALUES (new.id, new.title, new.body);
END;
-- Similar triggers needed for UPDATE and DELETE

4.4 監査証跡パターン

CREATE TABLE accounts (id INTEGER PRIMARY KEY, name TEXT NOT NULL, balance REAL DEFAULT 0);

CREATE TABLE accounts_audit (
    id INTEGER PRIMARY KEY, account_id INTEGER NOT NULL,
    field_name TEXT NOT NULL, old_value TEXT, new_value TEXT,
    changed_at TEXT DEFAULT (datetime('now')),
    FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE
);

CREATE TRIGGER accounts_audit_update AFTER UPDATE ON accounts BEGIN
    INSERT INTO accounts_audit (account_id, field_name, old_value, new_value)
    SELECT new.id, 'balance', old.balance, new.balance WHERE old.balance != new.balance;
END;

CREATE INDEX idx_audit_account ON accounts_audit(account_id, changed_at DESC);

---

5. セキュリティ基準

5.1 データ整合性の制御

-- Numeric, string format, and enum constraints
CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'),
    phone TEXT CHECK(phone IS NULL OR phone GLOB '+[0-9]*'),
    status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'active', 'deleted'))
);

-- Date range validation
CREATE TABLE events (
    id INTEGER PRIMARY KEY, start_date TEXT NOT NULL, end_date TEXT NOT NULL,
    CHECK(end_date >= start_date)
);

5.2 論理削除パターン

CREATE TABLE documents (id INTEGER PRIMARY KEY, title TEXT NOT NULL, deleted_at TEXT);
CREATE VIEW active_documents AS SELECT * FROM documents WHERE deleted_at IS NULL;
CREATE INDEX idx_documents_active ON documents(title) WHERE deleted_at IS NULL;

---

6. インデックス戦略

-- Single column for equality/range | Composite (equality first, then range)
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC);

-- Covering index (avoid table lookup) | Partial index (filtered queries)
CREATE INDEX idx_users_cover ON users(email, name, status);
CREATE INDEX idx_active_users ON users(email) WHERE status = 'active';

-- Expression index | Always verify with EXPLAIN
CREATE INDEX idx_users_lower ON users(LOWER(email));
EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?;

---

7. 実装ワークフロー（TDD）

ステップ 1：まず失敗するテストを書く

# tests/test_schema.py
import pytest
import sqlite3

@pytest.fixture
def db():
    conn = sqlite3.connect(':memory:')
    conn.execute("PRAGMA foreign_keys = ON")
    yield conn
    conn.close()

class TestUserSchema:
    def test_email_uniqueness(self, db):
        db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT UNIQUE NOT NULL)")
        db.execute("INSERT INTO users (email) VALUES ('[email protected]')")
        with pytest.raises(sqlite3.IntegrityError):
            db.execute("INSERT INTO users (email) VALUES ('[email protected]')")

    def test_email_format_constraint(self, db):
        db.execute("""CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'))""")
        with pytest.raises(sqlite3.IntegrityError):
            db.execute("INSERT INTO users (email) VALUES ('invalid')")

    def test_index_used_for_lookup(self, db):
        db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)")
        db.execute("CREATE INDEX idx_users_email ON users(email)")
        plan = db.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ('[email protected]',)).fetchone()
        assert 'USING INDEX' in plan[3]

ステップ 2：テストに合格するようスキーマを実装する

# src/database/schema.py
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'),
    name TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);
"""

def init_schema(conn):
    """Initialize database schema."""
    conn.executescript(SCHEMA_SQL)
    conn.commit()

ステップ 3：テストを実行して検証する

# Run schema tests
pytest tests/test_schema.py -v

# Run with coverage
pytest tests/test_schema.py --cov=src/database --cov-report=term-missing

ステップ 4：マイグレーションをテストする

# tests/test_migrations.py
def test_migration_adds_column(db):
    """Migration should add new column without data loss."""
    # Setup: create old schema with data
    db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)")
    db.execute("INSERT INTO users (email) VALUES ('[email protected]')")

    # Run migration
    db.execute("ALTER TABLE users ADD COLUMN name TEXT DEFAULT 'Unknown'")

    # Verify: data preserved, new column exists
    row = db.execute("SELECT id, email, name FROM users").fetchone()
    assert row == (1, '[email protected]', 'Unknown')

---

8. パフォーマンスパターン

8.1 インデックス戦略

良い例：正しい列順の複合インデックス

-- Query: WHERE user_id = ? AND created_at > ? ORDER BY created_at DESC
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC);

悪い例：誤った列順でインデックスが無駄になる

-- Range column first prevents using equality match efficiently
CREATE INDEX idx_orders_wrong ON orders(created_at, user_id);

8.2 クエリの最適化

良い例：カバリングインデックスを使ってテーブル検索を回避する

-- Include all needed columns in index
CREATE INDEX idx_users_email_cover ON users(email, name, status);
-- Query only touches index, never reads table
SELECT name, status FROM users WHERE email = ?;

悪い例：行数の多いテーブルで SELECT * を使う

-- Forces table lookup even with index
SELECT * FROM users WHERE email = ?;

8.3 コネクションプーリング

良い例：プールを使ってコネクションを再利用する

from contextlib import contextmanager
import threading

class ConnectionPool:
    def __init__(self, db_path, max_connections=5):
        self._pool, self._lock = [], threading.Lock()
        self._db_path, self._max = db_path, max_connections

    @contextmanager
    def get_connection(self):
        conn = self._acquire()
        try:
            yield conn
        finally:
            self._release(conn)

悪い例：クエリごとに新しいコネクションを作成する

def get_user(email):
    conn = sqlite3.connect('app.db')  # Expensive!
    result = conn.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
    conn.close()
    return result

8.4 非正規化のトレードオフ

良い例：読み取りの多いパターンでは計算済みの値を保存する

CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    item_count INTEGER NOT NULL DEFAULT 0,  -- Denormalized
    total_amount REAL NOT NULL DEFAULT 0    -- Denormalized
);
-- Use triggers to maintain denormalized values

悪い例：読み取りのたびに集計を計算する

SELECT o.id, COUNT(oi.id), SUM(oi.price * oi.quantity)
FROM orders o JOIN order_items oi ON oi.order_id = o.id GROUP BY o.id;

8.5 パーティショニング戦略

良い例：大きなテーブルを時間でパーティション分割する

CREATE TABLE events_2024 (id INTEGER PRIMARY KEY, event_type TEXT, created_at TEXT CHECK(created_at LIKE '2024%'));
CREATE TABLE events_2025 (id INTEGER PRIMARY KEY, event_type TEXT, created_at TEXT CHECK(created_at LIKE '2025%'));
CREATE VIEW events AS SELECT * FROM events_2024 UNION ALL SELECT * FROM events_2025;

悪い例：数百万行を含む単一のテーブル（1,000 万行を超えると全件走査が発生します）

---

9. よくある間違いとアンチパターン

間違い	悪い例	良い例
過剰な正規化	first_name、last_name 用にテーブルを分ける	users テーブルに直接保存する
外部キーの欠落	user_id INTEGER（外部キーなし）	user_id INTEGER REFERENCES users(id)
インデックスの列順が誤っている	WHERE user_id=? AND created_at>? に対して INDEX(created_at, user_id)	INDEX(user_id, created_at)
列に CSV を保存する	tags TEXT -- "a,b,c"	適切な外部キーを持つ中間テーブル

---

10. 実装前のチェックリスト

フェーズ 1：コードを書く前

• クエリパターンを特定し、文書化した
• パフォーマンス要件を定義した（レイテンシー、スループット）
• データ量の見積もりを算出した
• スキーマ検証用のテストフィクスチャを設計した
• マイグレーション戦略を計画した（既存のスキーマを変更する場合）
• 参照ファイルを読んだ（references/advanced-patterns.md、references/security-examples.md）

フェーズ 2：実装中

• すべてのテーブルに PRIMARY KEY がある
• すべてのリレーションシップに外部キーを定義した
• 適切な ON DELETE アクションを指定した（CASCADE、RESTRICT、SET NULL）
• データ検証用の CHECK 制約を指定した
• 必要な箇所に UNIQUE 制約を指定した
• 必須フィールドに NOT NULL を指定した
• すべての外部キーにインデックスを作成した
• 正しい列順（範囲条件の前に等価条件）の複合インデックスを作成した
• 必要に応じて同期トリガー付きの FTS5 テーブルを作成した
• 制約に関するテストを作成し、合格した

フェーズ 3：コミット前

• pytest tests/test_schema.py -v が合格する
• 重要なクエリについて EXPLAIN QUERY PLAN を検証した
• 冗長なインデックスがない
• ロールバックを含めてマイグレーションをテストした
• マイグレーションでデータが失われない
• パフォーマンスベンチマークが要件を満たしている
• スキーマのバージョンを追跡している

---

11. まとめ

目標は、次のようなデータベーススキーマを作成することです。

• 正規化されている：戦略的な非正規化を可能にしつつ、冗長性を排除する
• 高性能である：適切なインデックス、カバリングインデックス、効率的なクエリパターンを採用する
• 保守しやすい：明確な命名、文書化されたリレーションシップ、マイグレーションへの対応
• 安全である：検証用の制約と、整合性を保つための外部キーを使用する

スキーマ設計では、次のバランスを取る必要があることを理解してください。

1. 正規化とクエリパフォーマンス
2. インデックスの利点と書き込みのオーバーヘッド
3. 柔軟性と制約
4. 現在のニーズと将来的な進化

設計上の注意：まず 3NF に正規化し、実際のクエリパターンに基づいてインデックスを追加し、EXPLAIN を使って想定を検証してください。判断に迷う場合は、複雑なリレーションシップパターンについて references/advanced-patterns.md を参照してください。