データベースインデックスの基礎——B ツリー、ハッシュ、カバーリングインデックスの仕組み

import time
import random

def simulate_full_table_scan(num_rows, target_id):
    """フルテーブルスキャンのシミュレーション"""
    start_time = time.time()

    # 全行を順にスキャン
    for row_id in range(num_rows):
        if row_id == target_id:
            break

    elapsed = time.time() - start_time
    return elapsed

def simulate_index_scan(num_rows, target_id, index_type='btree'):
    """インデックススキャンのシミュレーション（対数時間）"""
    import math
    start_time = time.time()

    # B ツリー：log2(N) 回のアクセス
    if index_type == 'btree':
        accesses = math.log2(num_rows) if num_rows > 0 else 0
    # ハッシュ：O(1)
    elif index_type == 'hash':
        accesses = 1
    else:
        accesses = num_rows

    # 各アクセスに 0.001ms と仮定
    time.sleep(accesses * 0.000001)

    elapsed = time.time() - start_time
    return elapsed

# 比較：100 万行のデータから特定の 1 行を検索
num_rows = 1_000_000
target_id = 500_000

print("【100 万行からの検索パフォーマンス比較】\n")

full_scan_time = simulate_full_table_scan(num_rows, target_id)
print(f"フルテーブルスキャン：{full_scan_time*1000:.2f}ms（O(N)）")

btree_time = simulate_index_scan(num_rows, target_id, 'btree')
print(f"B ツリーインデックス：{btree_time*1000:.2f}ms（O(log N)）")

hash_time = simulate_index_scan(num_rows, target_id, 'hash')
print(f"ハッシュインデックス：{hash_time*1000:.2f}ms（O(1)）")

print(f"\n高速化倍率（B ツリー）：{full_scan_time/btree_time:.0f}倍")
print(f"高速化倍率（ハッシュ）：{full_scan_time/hash_time:.0f}倍")

def analyze_index_tradeoffs():
    """インデックスのメリット・デメリット分析"""

    tradeoffs = {
        'メリット': [
            '検索クエリの劇的な高速化（O(N) → O(log N)）',
            'ORDER BY ソートの不要化（インデックスがソート済み）',
            'ユニーク制約の強制（重複チェックが高速）',
            '結合クエリの最適化（結合キーにインデックス）',
            'カバーリングインデックスによる I/O 削減'
        ],
        'デメリット': [
            'ディスク容量の増加（データサイズの 10-30%）',
            'INSERT/UPDATE/DELETE の低速化（インデックス更新が必要）',
            'インデックスメンテナンスのオーバーヘッド（再構成）',
            '誤ったインデックス設計による逆効果（選択度の低い列）'
        ]
    }

    print("【インデックスのトレードオフ】\n")
    print("### メリット")
    for i, item in enumerate(tradeoffs['メリット'], 1):
        print(f"{i}. {item}")

    print("\n### デメリット")
    for i, item in enumerate(tradeoffs['デメリット'], 1):
        print(f"{i}. {item}")

    print("\n【適切なインデックス設計の指針】")
    print("・WHERE 句で使用される列に優先的に貼る")
    print("・選択度（カーディナリティ）の高い列を選ぶ")
    print("・頻繁に更新されるテーブルはインデックス数を絞る")
    print("・複合インデックスは最左接頭辞の原則を守る")

analyze_index_tradeoffs()

class BTreeNode:
    """B ツリーのノード（簡易実装）"""

    def __init__(self, order=4):
        self.order = order  # 最大子ノード数
        self.keys = []      # キーのリスト
        self.children = []  # 子ノードのリスト
        self.is_leaf = False

    def search(self, key):
        """キーを検索"""
        # キーを見つける
        i = 0
        while i < len(self.keys) and key > self.keys[i]:
            i += 1

        # 一致したキーが見つかった
        if i < len(self.keys) and key == self.keys[i]:
            return True

        # 葉ノードの場合、見つからない
        if self.is_leaf:
            return False

        # 子ノードを再帰的に検索
        return self.children[i].search(key)

    def insert(self, key):
        """キーを挿入（簡易版）"""
        # 葉ノードの場合
        if self.is_leaf:
            if key not in self.keys:
                self.keys.append(key)
                self.keys.sort()
            return

        # 適切な子ノードを見つける
        i = 0
        while i < len(self.keys) and key > self.keys[i]:
            i += 1

        # 子ノードに挿入
        self.children[i].insert(key)

# B ツリーの動作例
def demonstrate_btree():
    """B ツリーの動作デモ"""

    print("【B ツリーの挿入と検索】\n")

    # 次数 4 の B ツリーを作成（各ノード最大 3 キー）
    root = BTreeNode(order=4)
    root.is_leaf = True  # 最初は葉ノード

    # キーを挿入
    keys_to_insert = [50, 30, 70, 20, 40, 60, 80, 10, 25, 35, 55]

    for key in keys_to_insert:
        root.insert(key)
        print(f"{key}を挿入 → キー：{root.keys}")

    # 検索テスト
    print("\n【検索テスト】")
    for search_key in [30, 55, 100]:
        found = root.search(search_key)
        status = "発見" if found else "不在"
        print(f"キー{search_key}: {status}")

demonstrate_btree()

import math

def calculate_btree_height(num_entries, order=100):
    """B ツリーの高度を計算"""
    # 最悪ケース：各ノードが最小限のキーしか持たない
    # 高さ h = log_ceil(order/2)(num_entries)
    min_children = math.ceil(order / 2)
    height = math.ceil(math.log(num_entries, min_children)) if num_entries > 0 else 1
    return height

def analyze_btree_disk_accesses():
    """B ツリーのディスクアクセス回数を分析"""

    print("【B ツリーのディスクアクセス解析】\n")
    print(f"{'データ行数':>12} | {'ツリー高度':>10} | {'アクセス回数':>12}")
    print("-" * 40)

    for num_rows in [1_000, 10_000, 100_000, 1_000_000, 10_000_000]:
        # 次数 100（実用的な B ツリー）を仮定
        height = calculate_btree_height(num_rows, order=100)

        # ルートノードはメモリ常駐と仮定（+1 はルートアクセス）
        disk_accesses = height

        print(f"{num_rows:>12,} | {height:>10} | {disk_accesses:>12}回")

    print("\n【考察】")
    print("・100 万行でもツリー高度は 3-4 層")
    print("・ルートノードがメモリにあれば、ディスクアクセスは 2-3 回")
    print("・これが O(N) のフルスキャンとの決定的な差")

analyze_btree_disk_accesses()

def compare_btree_vs_bplus_tree():
    """B ツリーと B+ ツリーの比較"""

    comparison = {
        'B ツリー': {
            '構造': '各ノードにキーと値を格納',
            '検索': 'ルートから葉まで辿る（平均 O(log N)）',
            '範囲検索': '苦手（値が散在）',
            '特徴': '早期ヒットで戻れる（ルートで一致すれば OK）',
            '使用例': 'Redis、一部の RDBMS'
        },
        'B+ ツリー': {
            '構造': '値は葉ノードのみ、内部ノードは索引',
            '検索': '必ず葉まで到達（O(log N)）',
            '範囲検索': '得意（葉ノードがリンクリスト）',
            '特徴': '全走査が高速、内部ノードに余裕',
            '使用例': 'MySQL InnoDB、PostgreSQL、Oracle'
        }
    }

    print("【B ツリー vs B+ ツリー】\n")

    for tree_type, props in comparison.items():
        print(f"### {tree_type}")
        for key, value in props.items():
            print(f"・{key}: {value}")
        print()

compare_btree_vs_bplus_tree()

def simple_hash_function(key, bucket_size):
    """簡易ハッシュ関数"""
    # 文字列キーの場合
    if isinstance(key, str):
        hash_value = sum(ord(c) for c in key)
    # 数値キーの場合
    elif isinstance(key, int):
        hash_value = key
    else:
        hash_value = hash(key)

    return hash_value % bucket_size

def demonstrate_hash_index():
    """ハッシュインデックスの動作デモ"""

    bucket_size = 10
    hash_table = {i: [] for i in range(bucket_size)}

    # データの挿入
    data = [
        (1001, 'Alice'),
        (2003, 'Bob'),
        (3007, 'Charlie'),
        (1504, 'David'),
        (2501, 'Eve'),
        (3509, 'Frank'),
    ]

    print("【ハッシュインデックスへのデータ登録】\n")
    for key, value in data:
        bucket = simple_hash_function(key, bucket_size)
        hash_table[bucket].append((key, value))
        print(f"キー{key} → バケット{bucket} に格納")

    print("\n【ハッシュテーブルの状態】")
    for bucket, items in hash_table.items():
        if items:
            print(f"バケット{bucket}: {items}")
        else:
            print(f"バケット{bucket}: (空)")

    # 検索
    print("\n【検索デモ】")
    search_key = 2003
    bucket = simple_hash_function(search_key, bucket_size)
    print(f"キー{search_key}を検索 → バケット{bucket}")

    for key, value in hash_table[bucket]:
        if key == search_key:
            print(f"発見：{value}")
            break
    else:
        print("該当データなし")

demonstrate_hash_index()

def analyze_hash_index_characteristics():
    """ハッシュインデックスの特性分析"""

    characteristics = {
        '得意なクエリ': [
            '等価検索（=, IN）',
            '一意制約のチェック',
            '完全一致の結合'
        ],
        '苦手なクエリ': [
            '範囲検索（>, <, BETWEEN）',
            'ORDER BY ソート',
            '前方一致検索（LIKE \'abc%\'）',
            '部分一致検索（LIKE \'%abc%\'）'
        ],
        'ハッシュ衝突対策': [
            'チェイン法（連結リスト）',
            'オープンアドレス法',
            '再ハッシュ法'
        ],
        '実装例': [
            'MySQL MEMORY エンジン',
            'PostgreSQL ハッシュインデックス',
            'Oracle ハッシュクラスタ'
        ]
    }

    print("【ハッシュインデックスの特性】\n")

    for category, items in characteristics.items():
        print(f"### {category}")
        for item in items:
            print(f"・{item}")
        print()

analyze_hash_index_characteristics()

def demonstrate_composite_index():
    """複合インデックスの動作デモ"""

    # 従業員のデータ
    employees = [
        {'dept': '営業', 'rank': '課長', 'name': '山田'},
        {'dept': '営業', 'rank': '係長', 'name': '鈴木'},
        {'dept': '営業', 'rank': '平社員', 'name': '田中'},
        {'dept': '技術', 'rank': '課長', 'name': '佐藤'},
        {'dept': '技術', 'rank': '係長', 'name': '高橋'},
        {'dept': '技術', 'rank': '平社員', 'name': '伊藤'},
        {'dept': '人事', 'rank': '課長', 'name': '中村'},
        {'dept': '人事', 'rank': '係長', 'name': '小林'},
    ]

    # 複合インデックス (dept, rank) を作成
    composite_index = {}
    for i, emp in enumerate(employees):
        key = (emp['dept'], emp['rank'])
        composite_index[key] = i

    print("【複合インデックス (dept, rank) の構造】\n")
    print("インデックスキー → 行ポインタ")
    for key, ptr in sorted(composite_index.items()):
        print(f"  {key} → {ptr}")

    print("\n【最左接頭辞の原則】")
    print("\n○ インデックス使えるクエリ:")
    print("  WHERE dept = '営業'                    (左端 1 列)")
    print("  WHERE dept = '営業' AND rank = '課長'  (左端 2 列)")
    print("  WHERE dept = '営業' ORDER BY rank      (左端 + ソート)")

    print("\n× インデックス使えないクエリ:")
    print("  WHERE rank = '課長'                    (左端ではない)")
    print("  WHERE rank = '課長' AND dept = '営業'  (順序逆)")

    print("\n【理由】")
    print("複合インデックスは (dept, rank) の順でソートされているため、")
    print("rank だけでは連続した範囲にならない（インデックスが使えない）")

demonstrate_composite_index()

def calculate_selectivity(column_data, total_rows):
    """選択度（カーディナリティ）の計算"""
    unique_values = len(set(column_data))
    selectivity = unique_values / total_rows
    return selectivity, unique_values

def analyze_index_selectivity():
    """インデックス選択度の分析"""

    # サンプルデータ（1000 行）
    total_rows = 1000

    # 各列のデータ例
    columns = {
        'user_id': list(range(1000)),  # 一意
        'email': [f"user{i}@example.com" for i in range(1000)],  # 一意
        'dept_id': [i % 10 for i in range(1000)],  # 10 種類
        'gender': ['M', 'F'] * 500,  # 2 種類
        'is_active': [True] * 950 + [False] * 50,  # 偏りあり
    }

    print("【インデックス選択度の分析】\n")
    print(f"{'列名':>12} | {'一意の数':>10} | {'選択度':>8} | {'インデックス効果':>15}")
    print("-" * 55)

    for col_name, data in columns.items():
        selectivity, unique_count = calculate_selectivity(data, total_rows)

        # 効果判定
        if selectivity > 0.1:
            effect = "大（推奨）"
        elif selectivity > 0.01:
            effect = "中（状況次第）"
        else:
            effect = "小（非推奨）"

        print(f"{col_name:>12} | {unique_count:>10} | {selectivity:>8.4f} | {effect:>15}")

    print("\n【選択度の基準】")
    print("・選択度 = 一意の値の数 / 総行数")
    print("・1.0 に近いほど選択度が高い（インデックス効果大）")
    print("・0.01 以下（1% 未満）の列はインデックス効果が小さい")
    print("・gender（0.002）のような列はインデックス不要")

analyze_index_selectivity()

def demonstrate_covering_index():
    """カバーリングインデックスのデモ"""

    # 通常のテーブル（ヒープ）
    heap_table = [
        {'id': 1, 'name': 'Alice', 'email': 'alice@example.com', 'dept': '営業'},
        {'id': 2, 'name': 'Bob', 'email': 'bob@example.com', 'dept': '技術'},
        {'id': 3, 'name': 'Charlie', 'email': 'charlie@example.com', 'dept': '営業'},
        {'id': 4, 'name': 'David', 'email': 'david@example.com', 'dept': '人事'},
        {'id': 5, 'name': 'Eve', 'email': 'eve@example.com', 'dept': '技術'},
    ]

    # 通常インデックス（name だけ）
    name_index = {
        'Alice': 0,
        'Bob': 1,
        'Charlie': 2,
        'David': 3,
        'Eve': 4,
    }

    # カバーリングインデックス（name, email）
    covering_index = {
        'Alice': {'name': 'Alice', 'email': 'alice@example.com'},
        'Bob': {'name': 'Bob', 'email': 'bob@example.com'},
        'Charlie': {'name': 'Charlie', 'email': 'charlie@example.com'},
        'David': {'name': 'David', 'email': 'david@example.com'},
        'Eve': {'name': 'Eve', 'email': 'eve@example.com'},
    }

    print("【カバーリングインデックスの比較】\n")

    # クエリ：SELECT name, email FROM employees WHERE name = 'Alice'

    print("### 通常インデックスの場合")
    print("1. name_index から行ポインタ（0）を取得")
    print("2. heap_table[0] にアクセス（ランダム I/O）")
    print("3. name と email を読み取り")
    print("→ データページへのアクセスが必要")

    print("\n### カバーリングインデックスの場合")
    print("1. covering_index['Alice'] から直接 name, email を取得")
    print("2. テーブルアクセス不要")
    print("→ インデックスのみでクエリ完了（I/O 削減）")

    print("\n【効果】")
    print("・テーブル（ヒープ）アクセスが不要")
    print("・ランダム I/O が削減される")
    print("・特に大量データの集計クエリで効果的")

demonstrate_covering_index()

def analyze_index_only_scan_benefit():
    """インデックスオンリースキャンのメリット分析"""

    scenarios = {
        '集計クエリ': {
            '例': 'SELECT COUNT(*) FROM orders WHERE status = 1',
            '効果': 'status のインデックスのみで完了'
        },
        'ソートクエリ': {
            '例': 'SELECT name FROM users ORDER BY name',
            '効果': 'name インデックスがソート済みなのでソート不要'
        },
        '範囲クエリ': {
            '例': 'SELECT id, created_at FROM logs WHERE created_at BETWEEN ?',
            '効果': '(created_at, id) のカバーリングインデックスで対応'
        },
        '結合クエリ': {
            '例': 'SELECT u.name, o.total FROM users u JOIN orders o ON u.id = o.user_id',
            '効果': 'user_id を含むカバーリングインデックスでテーブルアクセス削減'
        }
    }

    print("【インデックスオンリースキャンの適用シナリオ】\n")

    for scenario, details in scenarios.items():
        print(f"### {scenario}")
        print(f"クエリ例：{details['例']}")
        print(f"効果：{details['効果']}\n")

analyze_index_only_scan_benefit()

def design_indexes_for_queries():
    """クエリパターンに基づくインデックス設計"""

    # 想定クエリ
    queries = [
        "SELECT * FROM users WHERE email = ?",
        "SELECT * FROM orders WHERE user_id = ? AND status = ?",
        "SELECT * FROM products WHERE category_id = ? ORDER BY price",
        "SELECT COUNT(*) FROM logs WHERE created_at BETWEEN ? AND ?",
        "SELECT * FROM articles WHERE published_at >= ? ORDER BY published_at DESC LIMIT 10"
    ]

    # 推奨インデックス
    recommended_indexes = [
        "CREATE UNIQUE INDEX idx_users_email ON users(email)",
        "CREATE INDEX idx_orders_user_status ON orders(user_id, status)",
        "CREATE INDEX idx_products_category_price ON products(category_id, price)",
        "CREATE INDEX idx_logs_created_at ON logs(created_at)",
        "CREATE INDEX idx_articles_published_at ON articles(published_at DESC)"
    ]

    print("【クエリパターンと推奨インデックス】\n")

    for query, index in zip(queries, recommended_indexes):
        print(f"クエリ：{query}")
        print(f"推奨：  {index}\n")

    print("【設計のポイント】")
    print("1. WHERE 句の列を左端に")
    print("2. ORDER BY の列を WHERE の後に")
    print("3. 一意制約は UNIQUE INDEX で")
    print("4. 降順ソートは DESC インデックス（DB による）")

design_indexes_for_queries()

def monitor_index_usage():
    """インデックス使用状況のモニタリング"""

    # 架空のインデックス使用状況
    index_stats = [
        {'table': 'users', 'index': 'idx_email', 'reads': 150000, 'writes': 500, 'size_mb': 50},
        {'table': 'orders', 'index': 'idx_user_id', 'reads': 500000, 'writes': 2000, 'size_mb': 200},
        {'table': 'orders', 'index': 'idx_status', 'reads': 100, 'writes': 2000, 'size_mb': 80},
        {'table': 'logs', 'index': 'idx_created_at', 'reads': 50000, 'writes': 10000, 'size_mb': 500},
        {'table': 'products', 'index': 'idx_category', 'reads': 10, 'writes': 100, 'size_mb': 30},
    ]

    print("【インデックス使用状況レポート】\n")
    print(f"{'テーブル':>10} | {'インデックス':>20} | {'リード':>8} | {'ライト':>8} | {'サイズ':>8} | {'判断':>10}")
    print("-" * 80)

    for stat in index_stats:
        # 使用頻度の判断
        read_write_ratio = stat['reads'] / stat['writes'] if stat['writes'] > 0 else float('inf')

        if stat['reads'] > 10000 and read_write_ratio > 10:
            judgment = "維持推奨"
        elif stat['reads'] < 1000 and read_write_ratio < 1:
            judgment = "削除検討"
        elif stat['reads'] < 100:
            judgment = "-unused-"
        else:
            judgment = "現状維持"

        print(f"{stat['table']:>10} | {stat['index']:>20} | {stat['reads']:>8,} | {stat['writes']:>8,} | {stat['size_mb']:>6}MB | {judgment:>10}")

    print("\n【モニタリングの指針】")
    print("・リード/ライト比が極端に低いインデックスは削除を検討")
    print("・未使用インデックスは書き込みコストだけがかかる")
    print("・ただし、バッチ処理用のインデックスは例外")

monitor_index_usage()

def explain_mysql_innodb_indexes():
    """MySQL InnoDB のインデックス実装"""

    innodb_features = {
        'クラスタインデックス': 'プライマリキーがデータ自体を格納（ヒープ不要）',
        'セカンダリインデックス': 'リーフにプライマリキーを格納（2 回アクセス）',
        'アダプティブハッシュインデックス': '頻繁アクセスページを自動ハッシュ化',
        'フルテキストインデックス': 'Ver5.6 以降で利用可能'
    }

    print("【MySQL InnoDB のインデックス実装】\n")

    for feature, description in innodb_features.items():
        print(f"### {feature}")
        print(f"{description}\n")

    print("【クエリ例】")
    print("""
-- プライマリキー（クラスタインデックス）
CREATE TABLE users (
    id INT PRIMARY KEY,  -- これがクラスタインデックス
    name VARCHAR(100),
    email VARCHAR(255)
);

-- セカンダリインデックス
CREATE INDEX idx_email ON users(email);
-- 内部構造：email → id（プライマリキー）→ データ

-- 複合インデックス
CREATE INDEX idx_name_email ON users(name, email);
-- 最左接頭辞：name 単独、name+email で使用可能
""")

explain_mysql_innodb_indexes()

def explain_postgresql_indexes():
    """PostgreSQL のインデックス実装"""

    pg_index_types = {
        'B-tree': 'デフォルト。等価・範囲・ソートに対応',
        'Hash': '等価検索のみ。Ver10 で WAL ロギング対応',
        'GiST': '幾何データ型、全文検索、R ツリー',
        'SP-GiST': 'パーティション検索（四角形、ツリー）',
        'GIN': '配列、JSONB、全文検索（逆引き索引）',
        'BRIN': '超大規模テーブル（ブロック範囲）'
    }

    print("【PostgreSQL のインデックスタイプ】\n")

    for index_type, description in pg_index_types.items():
        print(f"### {index_type}")
        print(f"{description}\n")

    print("【使い分け指針】")
    print("""
・通常クエリ → B-tree（デフォルト）
・等価検索のみ → Hash（B-tree より高速な場合も）
・配列・JSONB → GIN（@> 演算子など）
・全文検索 → GIN（tsvector）
・時系列超大規模 → BRIN（時系列ログなど）
・地理情報 → GiST（PostGIS）
""")

explain_postgresql_indexes()