# Python — SQLAlchemy
from sqlalchemy import create_engine
 
engine = create_engine(
    "postgresql://user:pass@localhost:5432/mydb",
    pool_size=10,           # 常時維持する接続数
    max_overflow=5,         # pool_size 超過時の追加接続数
    pool_timeout=30,        # 接続取得のタイムアウト(秒)
    pool_recycle=1800,      # 接続の再作成間隔(秒) ← コネクションリーク対策
    pool_pre_ping=True,     # 使用前に接続の生存確認
)

// Node.js — pg (node-postgres)
import { Pool } from 'pg';
 
const pool = new Pool({
  host: 'localhost',
  port: 5432,
  database: 'mydb',
  user: 'user',
  password: 'pass',
  max: 10,                    // 最大接続数
  idleTimeoutMillis: 30000,   // アイドル接続のタイムアウト
  connectionTimeoutMillis: 5000, // 接続取得のタイムアウト
});
 
// 使用例
const result = await pool.query('SELECT * FROM users WHERE id = $1', [userId]);

// Go — database/sql
import (
    "database/sql"
    _ "github.com/lib/pq"
    "time"
)
 
db, err := sql.Open("postgres", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
if err != nil {
    log.Fatal(err)
}
 
db.SetMaxOpenConns(10)                  // 最大接続数
db.SetMaxIdleConns(5)                   // 最大アイドル接続数
db.SetConnMaxLifetime(30 * time.Minute) // 接続の最大生存時間
db.SetConnMaxIdleTime(5 * time.Minute)  // アイドル接続の最大生存時間

// Java — HikariCP (Spring Boot デフォルト)
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
 
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:postgresql://localhost:5432/mydb");
config.setUsername("user");
config.setPassword("pass");
config.setMaximumPoolSize(10);          // 最大プールサイズ
config.setMinimumIdle(5);               // 最小アイドル接続数
config.setIdleTimeout(300000);          // アイドルタイムアウト(ms)
config.setConnectionTimeout(30000);     // 接続取得タイムアウト(ms)
config.setMaxLifetime(1800000);         // 接続の最大生存時間(ms)
config.setLeakDetectionThreshold(60000); // リーク検出閾値(ms)
 
HikariDataSource ds = new HikariDataSource(config);

; pgbouncer.ini の設定例
[databases]
mydb = host=localhost port=5432 dbname=mydb
 
[pgbouncer]
listen_addr = 0.0.0.0
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
 
; プールモード（transaction が一般的）
pool_mode = transaction
 
; プールサイズ
default_pool_size = 20
min_pool_size = 5
max_client_conn = 200        ; クライアント最大接続数
max_db_connections = 20      ; DB への最大接続数
 
; タイムアウト
server_idle_timeout = 600
client_idle_timeout = 0      ; 0 = 無制限
query_timeout = 0

-- pgBouncer の統計確認
-- pgBouncer 管理コンソールに接続
-- psql -p 6432 -U admin pgbouncer
 
SHOW POOLS;
-- database | user | cl_active | cl_waiting | sv_active | sv_idle | ...
 
SHOW STATS;
-- database | total_xact_count | total_query_count | avg_xact_time | ...
 
SHOW CLIENTS;
-- type | user | database | state | addr | port | ...

-- PostgreSQL: 現在の接続状態を確認
SELECT
    state,
    COUNT(*) AS connections,
    ROUND(100.0 * COUNT(*) / (SELECT setting::INT FROM pg_settings
        WHERE name = 'max_connections'), 1) AS pct_of_max
FROM pg_stat_activity
GROUP BY state
ORDER BY connections DESC;
 
-- アイドル接続の詳細（長時間アイドルは問題の兆候）
SELECT
    pid,
    usename,
    application_name,
    client_addr,
    state,
    NOW() - state_change AS idle_duration,
    query
FROM pg_stat_activity
WHERE state = 'idle'
  AND NOW() - state_change > INTERVAL '5 minutes'
ORDER BY idle_duration DESC;
 
-- 接続待ちの検出
SELECT
    COUNT(*) FILTER (WHERE state = 'active') AS active,
    COUNT(*) FILTER (WHERE state = 'idle') AS idle,
    COUNT(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_tx,
    COUNT(*) FILTER (WHERE wait_event IS NOT NULL) AS waiting,
    (SELECT setting::INT FROM pg_settings WHERE name = 'max_connections') AS max_conn
FROM pg_stat_activity
WHERE backend_type = 'client backend';

import redis
import json
from sqlalchemy.orm import Session
 
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
 
def get_user(db: Session, user_id: str) -> dict:
    """Cache-Aside パターンでユーザーを取得"""
    cache_key = f"user:{user_id}"
 
    # 1. キャッシュを確認
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache Hit
 
    # 2. キャッシュミス → DB から取得
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        return None
 
    user_dict = {"id": str(user.id), "name": user.name, "email": user.email}
 
    # 3. キャッシュに保存 (TTL: 5分)
    r.setex(cache_key, 300, json.dumps(user_dict))
 
    return user_dict
 
def update_user(db: Session, user_id: str, name: str) -> dict:
    """更新時はキャッシュを削除（Write-Invalidate）"""
    user = db.query(User).filter(User.id == user_id).first()
    user.name = name
    db.commit()
 
    # キャッシュ削除（次回読み取り時に再キャッシュ）
    r.delete(f"user:{user_id}")
 
    return {"id": str(user.id), "name": user.name}

def update_user_write_through(db: Session, user_id: str, name: str) -> dict:
    """Write-Through: DB とキャッシュを同時更新"""
    user = db.query(User).filter(User.id == user_id).first()
    user.name = name
    db.commit()
 
    # キャッシュも同時に更新（削除ではなく上書き）
    user_dict = {"id": str(user.id), "name": user.name, "email": user.email}
    r.setex(f"user:{user_id}", 300, json.dumps(user_dict))
 
    return user_dict

import asyncio
from collections import defaultdict
 
class WriteBehindCache:
    """Write-Behind: キャッシュに書き込み、非同期でDBに反映"""
 
    def __init__(self, redis_client, db_session_factory, flush_interval=5):
        self.redis = redis_client
        self.db_factory = db_session_factory
        self.flush_interval = flush_interval
        self.pending_writes = defaultdict(dict)
 
    async def set(self, key: str, value: dict):
        """キャッシュに即座に書き込み"""
        self.redis.setex(key, 600, json.dumps(value))
        self.pending_writes[key] = value
 
    async def flush_to_db(self):
        """定期的にDBに書き込み（バッチ処理）"""
        while True:
            await asyncio.sleep(self.flush_interval)
            if not self.pending_writes:
                continue
 
            writes = dict(self.pending_writes)
            self.pending_writes.clear()
 
            db = self.db_factory()
            try:
                for key, value in writes.items():
                    # バルクUPSERTでDB負荷を軽減
                    db.execute(
                        """INSERT INTO users (id, name, email)
                           VALUES (:id, :name, :email)
                           ON CONFLICT (id) DO UPDATE
                           SET name = EXCLUDED.name, email = EXCLUDED.email""",
                        value
                    )
                db.commit()
            except Exception as e:
                db.rollback()
                # 失敗した書き込みを戻す
                self.pending_writes.update(writes)
                raise
            finally:
                db.close()

# ===========================
# Lua スクリプトによるアトミック操作
# ===========================
RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
 
local current = redis.call('INCR', key)
if current == 1 then
    redis.call('EXPIRE', key, window)
end
 
if current > limit then
    return 0  -- 拒否
else
    return 1  -- 許可
end
"""
 
def check_rate_limit(user_id: str, limit: int = 100, window: int = 60) -> bool:
    """スライディングウィンドウによるレート制限"""
    key = f"rate:{user_id}:{int(time.time()) // window}"
    result = r.eval(RATE_LIMIT_SCRIPT, 1, key, limit, window)
    return bool(result)
 
 
# ===========================
# Redis Pipeline でバッチ取得
# ===========================
def get_users_batch(user_ids: list[str]) -> list[dict]:
    """パイプラインで複数キーを一括取得（N+1問題の回避）"""
    cache_keys = [f"user:{uid}" for uid in user_ids]
 
    # 1. パイプラインで一括取得
    pipe = r.pipeline(transaction=False)
    for key in cache_keys:
        pipe.get(key)
    results = pipe.execute()
 
    users = []
    missing_ids = []
 
    for uid, cached in zip(user_ids, results):
        if cached:
            users.append(json.loads(cached))
        else:
            missing_ids.append(uid)
 
    # 2. キャッシュミス分をDBから取得
    if missing_ids:
        db_users = db.query(User).filter(User.id.in_(missing_ids)).all()
        pipe = r.pipeline(transaction=False)
        for user in db_users:
            user_dict = {"id": str(user.id), "name": user.name, "email": user.email}
            users.append(user_dict)
            pipe.setex(f"user:{user.id}", 300, json.dumps(user_dict))
        pipe.execute()
 
    return users
 
 
# ===========================
# キャッシュウォーミング
# ===========================
def warm_cache_on_deploy():
    """デプロイ時にホットデータをプリロード"""
    # アクセス頻度の高いデータをロード
    hot_users = db.query(User).order_by(User.last_login_at.desc()).limit(1000).all()
    pipe = r.pipeline(transaction=False)
    for user in hot_users:
        user_dict = {"id": str(user.id), "name": user.name, "email": user.email}
        pipe.setex(f"user:{user.id}", 600, json.dumps(user_dict))
    pipe.execute()
    print(f"Warmed cache with {len(hot_users)} users")

-- スロークエリの例
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON o.user_id = u.id
WHERE u.created_at > '2025-01-01'
GROUP BY u.name
ORDER BY order_count DESC
LIMIT 10;
 
-- 出力例:
-- Limit  (cost=1234.56..1234.58 rows=10 width=40)
--        (actual time=45.123..45.130 rows=10 loops=1)
--   -> Sort  (cost=1234.56..1237.89 rows=1000 width=40)
--            (actual time=45.120..45.125 rows=10 loops=1)
--     Sort Key: (count(o.id)) DESC
--     Sort Method: top-N heapsort  Memory: 26kB
--       -> HashAggregate  (cost=1200.00..1210.00 rows=1000 width=40)
--                         (actual time=44.000..44.500 rows=1000 loops=1)
--           -> Hash Left Join  (cost=100.00..900.00 rows=50000 width=36)
--                              (actual time=5.000..35.000 rows=50000 loops=1)
--               Hash Cond: (o.user_id = u.id)
--               -> Seq Scan on orders o  ← 全件スキャン! 改善ポイント
--                  (actual time=0.010..15.000 rows=100000 loops=1)
--               -> Hash  (cost=80.00..80.00 rows=1000 width=20)
--                  -> Index Scan using idx_users_created_at on users u
--                     Filter: (created_at > '2025-01-01')
--                     (actual time=0.020..2.000 rows=1000 loops=1)
-- Buffers: shared hit=5000 read=200
-- Planning Time: 0.500 ms
-- Execution Time: 45.200 ms

-- 1. 単一カラムインデックス
CREATE INDEX idx_users_email ON users (email);
 
-- 2. 複合インデックス（カーディナリティの高いカラムを先に）
CREATE INDEX idx_orders_user_status ON orders (user_id, status);
 
-- 3. 部分インデックス（条件を絞ってサイズ削減）
CREATE INDEX idx_orders_active ON orders (user_id)
WHERE status = 'active';
 
-- 4. カバリングインデックス（INCLUDE でインデックスオンリースキャン）
CREATE INDEX idx_users_email_covering ON users (email)
INCLUDE (name, created_at);
 
-- 5. 式インデックス
CREATE INDEX idx_users_lower_email ON users (LOWER(email));
 
-- 6. GINインデックス（全文検索、JSONB、配列）
CREATE INDEX idx_products_tags ON products USING GIN (tags);
 
-- 7. GiSTインデックス（地理空間、範囲型）
CREATE INDEX idx_events_period ON events USING GIST (
    tstzrange(start_at, end_at)
);
 
-- 8. BRINインデックス（大規模テーブルのタイムスタンプ）
CREATE INDEX idx_logs_created ON logs USING BRIN (created_at)
WITH (pages_per_range = 32);
-- → B-Tree の1/100以下のサイズで、時系列データに有効

-- PostgreSQL: スロークエリログの有効化
ALTER SYSTEM SET log_min_duration_statement = 100;  -- 100ms以上をログ
ALTER SYSTEM SET log_statement = 'none';             -- 通常クエリはログしない
SELECT pg_reload_conf();
 
-- 実行中の遅いクエリを確認
SELECT pid, now() - pg_stat_activity.query_start AS duration,
       query, state
FROM pg_stat_activity
WHERE (now() - pg_stat_activity.query_start) > interval '5 seconds'
  AND state != 'idle'
ORDER BY duration DESC;
 
-- pg_stat_statements による統計分析（要拡張インストール）
-- CREATE EXTENSION pg_stat_statements;
SELECT
    queryid,
    LEFT(query, 80) AS query_preview,
    calls,
    ROUND(total_exec_time::NUMERIC, 2) AS total_time_ms,
    ROUND(mean_exec_time::NUMERIC, 2) AS avg_time_ms,
    ROUND(stddev_exec_time::NUMERIC, 2) AS stddev_ms,
    rows,
    shared_blks_hit + shared_blks_read AS total_blocks,
    CASE WHEN shared_blks_hit + shared_blks_read > 0
         THEN ROUND(100.0 * shared_blks_hit /
              (shared_blks_hit + shared_blks_read), 1)
         ELSE 100
    END AS cache_hit_pct
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 20;
 
-- インデックス使用率の確認
SELECT schemaname, tablename, indexname,
       idx_scan as times_used,
       pg_size_pretty(pg_relation_size(indexrelid)) as index_size
FROM pg_stat_user_indexes
ORDER BY idx_scan ASC;
-- idx_scan = 0 のインデックスは不要な可能性
 
-- テーブル統計の更新
ANALYZE users;
ANALYZE orders;

-- テーブルの統計情報を確認
SELECT
    attname,
    n_distinct,          -- ユニーク値の推定数（負値 = 行数に対する割合）
    most_common_vals,    -- 最頻値
    most_common_freqs,   -- 最頻値の出現頻度
    histogram_bounds     -- ヒストグラムの区切り値
FROM pg_stats
WHERE tablename = 'orders' AND attname = 'status';
 
-- 拡張統計（相関のあるカラム群）
-- PostgreSQL 10+
CREATE STATISTICS stat_orders_user_status (dependencies, ndistinct, mcv)
ON user_id, status FROM orders;
ANALYZE orders;
 
-- 統計情報の精度を上げる（デフォルト=100、最大=10000）
ALTER TABLE orders ALTER COLUMN status SET STATISTICS 500;
ANALYZE orders;

-- =============================================
-- パターン1: N+1 問題の解決
-- =============================================
-- NG: アプリケーション側のループ（N+1クエリ）
-- for user in users:
--     orders = SELECT * FROM orders WHERE user_id = user.id
 
-- OK: JOINで一括取得
SELECT u.*, o.id AS order_id, o.total_amount
FROM users u
LEFT JOIN orders o ON o.user_id = u.id
WHERE u.created_at > '2025-01-01';
 
-- OK: LATERAL JOIN で「各ユーザーの最新3件」
SELECT u.name, recent_orders.*
FROM users u
CROSS JOIN LATERAL (
    SELECT o.id, o.total_amount, o.created_at
    FROM orders o
    WHERE o.user_id = u.id
    ORDER BY o.created_at DESC
    LIMIT 3
) recent_orders
WHERE u.created_at > '2025-01-01';
 
-- =============================================
-- パターン2: カーソルベースページネーション
-- =============================================
-- NG: OFFSET が大きくなるほど遅くなる
SELECT * FROM orders ORDER BY created_at DESC LIMIT 20 OFFSET 100000;
-- → 100,020行を読んで100,000行を捨てる
 
-- OK: カーソルベースページネーション
SELECT * FROM orders
WHERE created_at < '2026-01-15T10:30:00Z'  -- 前ページ最後のタイムスタンプ
ORDER BY created_at DESC
LIMIT 20;
-- → インデックスを使って20行だけ読む
 
-- OK: カーソル + ID（同一タイムスタンプ対応）
SELECT * FROM orders
WHERE (created_at, id) < ('2026-01-15T10:30:00Z', 12345)
ORDER BY created_at DESC, id DESC
LIMIT 20;
 
-- =============================================
-- パターン3: EXISTS vs IN vs JOIN
-- =============================================
-- 外側が小さく内側が大きい場合: EXISTS が有利
SELECT * FROM users u
WHERE EXISTS (
    SELECT 1 FROM orders o WHERE o.user_id = u.id AND o.status = 'active'
);
 
-- 内側が小さい場合: IN が有利
SELECT * FROM orders
WHERE user_id IN (SELECT id FROM users WHERE role = 'admin');
 
-- =============================================
-- パターン4: DISTINCT を最適化
-- =============================================
-- NG: 全行ソート後に重複排除
SELECT DISTINCT category FROM products;
 
-- OK: Loose Index Scan（PostgreSQL 14+ で自動適用される場合あり）
-- 手動で再現する場合:
WITH RECURSIVE categories AS (
    (SELECT category FROM products ORDER BY category LIMIT 1)
    UNION ALL
    SELECT (SELECT p.category FROM products p
            WHERE p.category > c.category
            ORDER BY p.category LIMIT 1)
    FROM categories c
    WHERE c.category IS NOT NULL
)
SELECT category FROM categories WHERE category IS NOT NULL;
 
-- =============================================
-- パターン5: バッチ UPDATE
-- =============================================
-- NG: 一括で大量更新（ロック時間が長い）
UPDATE orders SET status = 'archived'
WHERE created_at < '2024-01-01';
-- → 100万行ロック、他のトランザクションがブロックされる
 
-- OK: バッチに分割
DO $$
DECLARE
    batch_size INT := 10000;
    updated INT;
BEGIN
    LOOP
        UPDATE orders SET status = 'archived'
        WHERE id IN (
            SELECT id FROM orders
            WHERE created_at < '2024-01-01'
              AND status != 'archived'
            LIMIT batch_size
            FOR UPDATE SKIP LOCKED  -- ロック競合を回避
        );
        GET DIAGNOSTICS updated = ROW_COUNT;
        EXIT WHEN updated = 0;
        COMMIT;
        PERFORM pg_sleep(0.1);  -- 他のトランザクションに実行機会を与える
    END LOOP;
END $$;

-- バッファキャッシュのヒット率
SELECT
    datname,
    blks_hit,
    blks_read,
    CASE WHEN blks_hit + blks_read > 0
         THEN ROUND(100.0 * blks_hit / (blks_hit + blks_read), 2)
         ELSE 100
    END AS cache_hit_ratio
FROM pg_stat_database
WHERE datname = current_database();
-- 目標: 99% 以上
 
-- テーブル別のキャッシュヒット率
SELECT
    schemaname, relname,
    heap_blks_hit,
    heap_blks_read,
    CASE WHEN heap_blks_hit + heap_blks_read > 0
         THEN ROUND(100.0 * heap_blks_hit / (heap_blks_hit + heap_blks_read), 2)
         ELSE 100
    END AS cache_hit_ratio
FROM pg_statio_user_tables
ORDER BY heap_blks_read DESC
LIMIT 20;
 
-- pg_buffercache 拡張で詳細確認
-- CREATE EXTENSION pg_buffercache;
SELECT
    c.relname,
    COUNT(*) AS buffers,
    pg_size_pretty(COUNT(*) * 8192) AS cached_size,
    ROUND(100.0 * COUNT(*) /
        (SELECT setting::INT FROM pg_settings WHERE name = 'shared_buffers'), 1)
        AS pct_of_shared_buffers
FROM pg_buffercache b
    JOIN pg_class c ON b.relfilenode = pg_relation_filenode(c.oid)
WHERE b.reldatabase = (SELECT oid FROM pg_database WHERE datname = current_database())
GROUP BY c.relname
ORDER BY buffers DESC
LIMIT 20;

-- =============================================
-- RANGE パーティション（最も一般的: 日付ベース）
-- =============================================
CREATE TABLE events (
    id          BIGSERIAL,
    event_type  VARCHAR(50),
    payload     JSONB,
    created_at  TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);
 
-- 月別パーティション作成
CREATE TABLE events_2024_01 PARTITION OF events
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE events_2024_02 PARTITION OF events
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
 
-- パーティションの自動作成（pg_partman 拡張）
-- CREATE EXTENSION pg_partman;
-- SELECT partman.create_parent(
--     p_parent_table := 'public.events',
--     p_control := 'created_at',
--     p_type := 'range',
--     p_interval := '1 month',
--     p_premake := 3  -- 3ヶ月先まで事前作成
-- );
 
-- =============================================
-- LIST パーティション（カテゴリベース）
-- =============================================
CREATE TABLE orders (
    id          BIGSERIAL,
    region      VARCHAR(20) NOT NULL,
    total       DECIMAL(12,2),
    created_at  TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (id, region)
) PARTITION BY LIST (region);
 
CREATE TABLE orders_japan PARTITION OF orders
    FOR VALUES IN ('tokyo', 'osaka', 'nagoya');
CREATE TABLE orders_us PARTITION OF orders
    FOR VALUES IN ('new_york', 'san_francisco', 'chicago');
CREATE TABLE orders_eu PARTITION OF orders
    FOR VALUES IN ('london', 'paris', 'berlin');
 
-- =============================================
-- HASH パーティション（均等分散）
-- =============================================
CREATE TABLE sessions (
    id          UUID NOT NULL,
    user_id     BIGINT NOT NULL,
    data        JSONB,
    expires_at  TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (id)
) PARTITION BY HASH (id);
 
CREATE TABLE sessions_0 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE sessions_1 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE sessions_2 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE sessions_3 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 3);

-- Partition Pruning の動作確認
EXPLAIN (ANALYZE)
SELECT * FROM events
WHERE created_at BETWEEN '2024-03-01' AND '2024-03-31';
 
-- 出力例:
-- Append (actual rows=50000)
--   Subplans Removed: 11        ← 12パーティション中11個が除外された
--   -> Seq Scan on events_2024_03 (actual rows=50000)
--        Filter: (created_at >= '2024-03-01' AND created_at <= '2024-03-31')

-- =============================================
-- COPY による高速バルクロード（INSERT より 10-100倍高速）
-- =============================================
COPY users (name, email, created_at)
FROM '/tmp/users.csv'
WITH (FORMAT csv, HEADER true, DELIMITER ',');
 
-- プログラムからの COPY
-- Python (psycopg2)
-- with cursor.copy("COPY users (name, email) FROM STDIN") as copy:
--     for row in data:
--         copy.write_row(row)
 
-- =============================================
-- バルクINSERT の最適化
-- =============================================
-- NG: 1行ずつ INSERT
INSERT INTO users (name, email) VALUES ('A', 'a@x.com');
INSERT INTO users (name, email) VALUES ('B', 'b@x.com');
-- ...10000回
 
-- OK: バルク INSERT (1文で複数行)
INSERT INTO users (name, email) VALUES
    ('A', 'a@x.com'),
    ('B', 'b@x.com'),
    ('C', 'c@x.com');
-- → ネットワークラウンドトリップを削減
 
-- =============================================
-- 大量ロード時の最適化テクニック
-- =============================================
-- 1. インデックスを一時的に削除
DROP INDEX idx_users_email;
 
-- 2. 制約チェックを遅延
SET session_replication_role = replica;  -- トリガー無効化
 
-- 3. バルクロード実行
COPY users FROM '/tmp/bulk_users.csv' WITH (FORMAT csv);
 
-- 4. インデックスを再作成（CONCURRENTLY で無停止）
CREATE INDEX CONCURRENTLY idx_users_email ON users (email);
 
-- 5. 制約チェックを復元
SET session_replication_role = DEFAULT;
 
-- 6. 統計情報を更新
ANALYZE users;
 
-- =============================================
-- UPSERT（INSERT ON CONFLICT）
-- =============================================
INSERT INTO products (sku, name, price, updated_at)
VALUES ('SKU001', 'Widget', 19.99, NOW()),
       ('SKU002', 'Gadget', 29.99, NOW()),
       ('SKU003', 'Doohickey', 39.99, NOW())
ON CONFLICT (sku)
DO UPDATE SET
    name = EXCLUDED.name,
    price = EXCLUDED.price,
    updated_at = EXCLUDED.updated_at
WHERE products.price != EXCLUDED.price;  -- 実際に変更がある場合のみ更新

-- NG: OFFSET が大きくなるほど遅くなる
SELECT * FROM orders ORDER BY created_at DESC LIMIT 20 OFFSET 100000;
-- → 100,020行を読んで100,000行を捨てる
 
-- OK: カーソルベースページネーション
SELECT * FROM orders
WHERE created_at < '2026-01-15T10:30:00Z'  -- 前ページ最後のタイムスタンプ
ORDER BY created_at DESC
LIMIT 20;
-- → インデックスを使って20行だけ読む

# NG: 全てのキャッシュが同時に期限切れ → DB に大量リクエスト
def bad_cache_set(key, value):
    r.setex(key, 3600, value)  # 全て TTL=1時間
 
# OK: TTL にジッター（ランダム変動）を追加
import random
 
def good_cache_set(key, value, base_ttl=3600):
    jitter = random.randint(0, 600)  # 0-10分のランダム
    r.setex(key, base_ttl + jitter, value)
 
# OK: ロックで同時再構築を防止（Mutex パターン）
def get_with_lock(key, rebuild_fn, ttl=3600):
    value = r.get(key)
    if value:
        return json.loads(value)
 
    lock_key = f"lock:{key}"
    if r.set(lock_key, "1", nx=True, ex=10):  # 10秒ロック
        try:
            value = rebuild_fn()
            r.setex(key, ttl + random.randint(0, 600), json.dumps(value))
            return value
        finally:
            r.delete(lock_key)
    else:
        # 他のプロセスが構築中 → 短時間待って再試行
        import time
        time.sleep(0.1)
        return get_with_lock(key, rebuild_fn, ttl)
 
# OK: 論理的期限切れ（Probabilistic Early Expiration）
import math
 
def get_with_early_expiry(key, rebuild_fn, ttl=3600, beta=1.0):
    """値がまだ有効でも、期限切れが近づくと確率的に再構築"""
    data = r.get(key)
    if data:
        cached = json.loads(data)
        remaining_ttl = r.ttl(key)
        # 残りTTLが短いほど再構築の確率が高くなる
        if remaining_ttl > 0:
            delta = ttl - remaining_ttl
            threshold = delta * beta * math.log(random.random())
            if threshold < remaining_ttl:
                return cached["value"]
 
    # 再構築
    value = rebuild_fn()
    r.setex(key, ttl, json.dumps({"value": value}))
    return value

-- NG: 不要なカラムも全て取得
SELECT * FROM users WHERE id = 1;
-- → LOBカラムや大きなJSONBも読み込む
-- → Index Only Scan が使えない
 
-- OK: 必要なカラムのみ
SELECT id, name, email FROM users WHERE id = 1;
-- → カバリングインデックスがあれば Index Only Scan

-- NG: 全カラムにインデックス
CREATE INDEX idx_users_name ON users (name);
CREATE INDEX idx_users_email ON users (email);
CREATE INDEX idx_users_age ON users (age);
CREATE INDEX idx_users_city ON users (city);
CREATE INDEX idx_users_name_email ON users (name, email);
CREATE INDEX idx_users_email_name ON users (email, name);
-- → INSERT/UPDATE が遅くなる、ストレージ浪費
 
-- OK: 実際のクエリパターンに基づいて必要最小限のインデックスを作成
-- 1. 最も頻繁なWHERE句のカラム
-- 2. JOIN条件のカラム（FK）
-- 3. ORDER BY のカラム
-- 4. pg_stat_user_indexes でidx_scan=0のインデックスは定期的に削除

# 問題: 長時間トランザクションがプールを占有
async def process_batch_bad(items):
    """NG: 1つのコネクションで長時間処理"""
    async with pool.acquire() as conn:
        for item in items:  # 10,000件を1接続で逐次処理
            await conn.execute("INSERT INTO results VALUES ($1)", item)
            await external_api_call(item)  # 100ms/call = 合計1000秒
 
# 解決: バッチを分割し、接続を早期に返却
async def process_batch_good(items):
    """OK: 小バッチで接続を使い回し"""
    for batch in chunks(items, 100):
        async with pool.acquire() as conn:
            async with conn.transaction():
                for item in batch:
                    await conn.execute("INSERT INTO results VALUES ($1)", item)
        # ここで接続がプールに返却される
        await asyncio.gather(*[external_api_call(item) for item in batch])

# 問題: DB更新とキャッシュ削除の間にクラッシュ
def update_user_bad(db, user_id, name):
    user = db.query(User).filter(User.id == user_id).first()
    user.name = name
    db.commit()          # ← DBは更新された
    # ↓ ここでクラッシュするとキャッシュが古いまま
    r.delete(f"user:{user_id}")
 
# 解決: Outbox パターン（DB トランザクションに含める）
def update_user_good(db, user_id, name):
    user = db.query(User).filter(User.id == user_id).first()
    user.name = name
    # キャッシュ無効化イベントもDBに記録
    db.execute(
        "INSERT INTO outbox (event_type, payload) VALUES (:type, :payload)",
        {"type": "cache_invalidate", "payload": json.dumps({"key": f"user:{user_id}"})}
    )
    db.commit()
    # バックグラウンドワーカーがoutboxを処理してキャッシュを削除

-- PostgreSQL の HOT (Heap-Only Tuple) UPDATE
-- 条件: 更新カラムがインデックスに含まれていない
-- 効果: インデックスの更新をスキップし、テーブルブロック内で完結
 
-- HOT UPDATE が効くケース
CREATE TABLE counters (
    id    SERIAL PRIMARY KEY,
    name  VARCHAR(50),
    count INTEGER DEFAULT 0
);
CREATE INDEX idx_counters_name ON counters (name);
 
-- name は変更しない → HOT UPDATE が発動
UPDATE counters SET count = count + 1 WHERE id = 1;
 
-- HOT UPDATE の状況を確認
SELECT
    relname,
    n_tup_upd,
    n_tup_hot_upd,
    CASE WHEN n_tup_upd > 0
         THEN ROUND(100.0 * n_tup_hot_upd / n_tup_upd, 1)
         ELSE 0
    END AS hot_update_pct
FROM pg_stat_user_tables
WHERE relname = 'counters';
-- hot_update_pct が高いほど効率的

EXPLAIN (ANALYZE, BUFFERS)
SELECT o.id, o.total_amount, u.name
FROM orders o
JOIN users u ON u.id = o.user_id
WHERE o.status = 'pending'
  AND o.created_at > '2025-06-01'
ORDER BY o.created_at DESC
LIMIT 50;
 
-- Sort  (cost=15000.00..15000.13 rows=50 width=52)
--       (actual time=450.123..450.130 rows=50 loops=1)
--   Sort Key: o.created_at DESC
--   Sort Method: top-N heapsort  Memory: 32kB
--   ->  Hash Join  (cost=200.00..14500.00 rows=5000 width=52)
--                  (actual time=3.000..448.000 rows=5000 loops=1)
--         Hash Cond: (o.user_id = u.id)
--         ->  Seq Scan on orders o  (actual time=0.030..440.000 rows=5000 loops=1)
--               Filter: (status = 'pending' AND created_at > '2025-06-01')
--               Rows Removed by Filter: 995000
--               Buffers: shared read=25000
--         ->  Hash  (cost=150.00..150.00 rows=10000 width=20)
--               ->  Seq Scan on users u  (actual time=0.010..2.000 rows=10000 loops=1)
--               Buffers: shared hit=100
-- Buffers: shared hit=100 read=25000
-- Execution Time: 450.200 ms

-- 複合インデックス（部分インデックス）
CREATE INDEX idx_orders_pending_created
ON orders (created_at DESC)
WHERE status = 'pending';
-- → Seq Scan → Index Scan に改善
-- → Rows Removed が 0 に近づく
-- → 期待される改善: 450ms → 5ms 以下
 
-- さらに最適化: カバリングインデックス
CREATE INDEX idx_orders_pending_covering
ON orders (created_at DESC)
INCLUDE (total_amount, user_id)
WHERE status = 'pending';
-- → Index Only Scan が可能に

# 商品情報: Cache-Aside + TTL 30分 + Write-Invalidate
def get_product(product_id: str) -> dict:
    key = f"product:{product_id}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
 
    product = db.query(Product).filter(Product.id == product_id).first()
    data = product.to_dict()
    r.setex(key, 1800, json.dumps(data))  # TTL 30分
    return data
 
# 在庫数: TTL 短め（10秒）+ 更新時即時反映
def get_stock(product_id: str) -> int:
    key = f"stock:{product_id}"
    cached = r.get(key)
    if cached:
        return int(cached)
 
    stock = db.query(Inventory.quantity).filter(
        Inventory.product_id == product_id
    ).scalar()
    r.setex(key, 10, str(stock))  # TTL 10秒（ほぼリアルタイム）
    return stock
 
def purchase_product(product_id: str, quantity: int):
    # DB更新
    db.execute(
        "UPDATE inventory SET quantity = quantity - :qty WHERE product_id = :pid",
        {"qty": quantity, "pid": product_id}
    )
    db.commit()
    # 在庫キャッシュは即座に削除（リアルタイム性）
    r.delete(f"stock:{product_id}")
 
# 画像URL: 長期キャッシュ（変更されないため）
# → CDN + Cache-Control: max-age=31536000 で対応

# パフォーマンス改善の優先順位（効果が大きい順）
 
optimization_checklist = [
    {
        "priority": 1,
        "category": "インデックス",
        "actions": [
            "WHERE句のカラムにインデックスがあるか確認",
            "JOIN条件のカラムにインデックスがあるか確認",
            "ORDER BY のカラムにインデックスがあるか確認",
            "複合インデックスのカラム順序を確認",
            "不要なインデックスの削除（idx_scan=0）",
        ],
    },
    {
        "priority": 2,
        "category": "クエリ書き換え",
        "actions": [
            "SELECT * を必要なカラムのみに変更",
            "サブクエリを JOIN に書き換え",
            "DISTINCT を GROUP BY に書き換え",
            "OFFSET ページネーションをカーソルベースに変更",
            "N+1 クエリを JOIN またはバッチ取得に変更",
        ],
    },
    {
        "priority": 3,
        "category": "テーブル設計",
        "actions": [
            "正規化の見直し（読み取り重視なら非正規化）",
            "パーティショニングの検討（大テーブル）",
            "マテリアライズドビューの利用（集約クエリ）",
            "BRIN インデックスの検討（時系列データ）",
        ],
    },
    {
        "priority": 4,
        "category": "キャッシュ",
        "actions": [
            "Cache-Aside + TTL + Write-Invalidate の導入",
            "キャッシュヒット率の監視",
            "Cache Stampede 対策（TTL ジッター + ロック）",
            "ホットデータのプリウォーミング",
        ],
    },
    {
        "priority": 5,
        "category": "インフラ",
        "actions": [
            "接続プールの最適化（実測ベース）",
            "pgBouncer 導入（多数アプリインスタンス）",
            "リードレプリカの導入（読み取り分散）",
            "SSD 移行と random_page_cost の調整",
        ],
    },
]

# NG: 文字列結合
query = f"SELECT * FROM users WHERE email = '{email}'"
 
# OK: パラメータバインド
query = "SELECT * FROM users WHERE email = :email"
db.execute(query, {"email": email})