PostgreSQL 实战：从入门到生产环境

-- 基本查询
SELECT id, name, email FROM users WHERE status = 'active';

-- 别名
SELECT 
  u.id AS user_id,
  u.name AS user_name,
  COUNT(o.id) AS order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name;

-- 条件表达式
SELECT 
  name,
  CASE 
    WHEN age < 18 THEN '未成年'
    WHEN age < 35 THEN '青年'
    WHEN age < 60 THEN '中年'
    ELSE '老年'
  END AS age_group
FROM users;

-- JSON 字段查询（PostgreSQL 特色）
SELECT data->>'name' FROM users WHERE data->>'role' = 'admin';

-- 递归查询（组织架构、分类树）
WITH RECURSIVE category_tree AS (
  -- 基础查询：顶级分类
  SELECT id, name, parent_id, 0 AS level
  FROM categories
  WHERE parent_id IS NULL
  
  UNION ALL
  
  -- 递归部分：子分类
  SELECT c.id, c.name, c.parent_id, ct.level + 1
  FROM categories c
  JOIN category_tree ct ON c.parent_id = ct.id
)
SELECT * FROM category_tree;

-- 窗口函数（排名、分组统计）
SELECT 
  name,
  department,
  salary,
  RANK() OVER (PARTITION BY department ORDER BY salary DESC) AS dept_rank,
  AVG(salary) OVER (PARTITION BY department) AS dept_avg_salary
FROM employees;

-- Lateral 子查询
SELECT u.*, latest_order.*
FROM users u
CROSS JOIN LATERAL (
  SELECT * FROM orders 
  WHERE user_id = u.id 
  ORDER BY created_at DESC 
  LIMIT 1
) AS latest_order;

-- 事务基本用法
BEGIN;

UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;

-- 检查余额是否足够
SELECT balance FROM accounts WHERE user_id = 1 FOR UPDATE; -- 行锁

IF (SELECT balance FROM accounts WHERE user_id = 1) < 100 THEN
  ROLLBACK;
  RAISE EXCEPTION '余额不足';
END IF;

COMMIT;

-- 悲观锁 vs 乐观锁
-- 悲观锁：SELECT ... FOR UPDATE
BEGIN;
SELECT * FROM products WHERE id = 1 FOR UPDATE;
-- 更新时检查
UPDATE products SET stock = stock - 1 WHERE id = 1 AND stock > 0;
COMMIT;

-- 乐观锁：版本号
-- 表结构: ALTER TABLE products ADD COLUMN version INT DEFAULT 1;
BEGIN;
UPDATE products 
SET stock = stock - 1, version = version + 1 
WHERE id = 1 AND version = 1;
-- 如果 version 不匹配，说明被其他事务修改了
COMMIT;

-- B-tree 索引（默认，最常用）
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_status_created ON users(status, created_at);

-- 唯一索引
CREATE UNIQUE INDEX idx_users_email_unique ON users(email);

-- 全文索引（中文需要分词器）
CREATE INDEX idx_posts_content ON posts USING gin(to_tsvector('chinese', content));

-- 表达式索引
CREATE INDEX idx_users_lower_email ON users(LOWER(email));

-- 部分索引（只索引满足条件的行）
CREATE INDEX idx_active_users ON users(email) WHERE status = 'active';

-- 复合索引设计原则
-- 1. 等值查询的列放前面
-- 2. 范围查询的列放后面
-- 示例：WHERE status = 'active' AND created_at > '2026-01-01'
-- 最佳索引：(status, created_at)
CREATE INDEX idx_users_status_created ON users(status, created_at);

EXPLAIN ANALYZE 
SELECT u.name, COUNT(o.id) AS order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.status = 'active' AND o.created_at > '2026-01-01'
GROUP BY u.id;

-- 分析结果解读
-- Seq Scan: 顺序扫描（全表扫描）
-- Index Scan: 索引扫描
-- Index Only Scan: 索引覆盖扫描（更快）
-- Nested Loop: 嵌套循环 join
-- Hash Join: 哈希 join
-- Sort: 排序操作

-- 优化提示
-- 1. 查看是否有 Seq Scan（全表扫描）
-- 2. 检查是否有合适的索引
-- 3. 评估 join 方式是否最优

-- 问题：COUNT(*) 慢
-- 解决：使用近似值或物化视图
CREATE MATERIALIZED VIEW user_stats AS
SELECT 
  status,
  COUNT(*) AS count
FROM users
GROUP BY status;

-- 定期刷新
REFRESH MATERIALIZED VIEW CONCURRENTLY user_stats;

-- 问题：分页 OFFSET 大时慢
-- 解决：使用游标分页
-- ❌ 传统分页（OFFSET 大时慢）
SELECT * FROM users ORDER BY id LIMIT 20 OFFSET 100000;

-- ✅ 游标分页（高效）
SELECT * FROM users WHERE id > 100000 ORDER BY id LIMIT 20;

-- 问题：JOIN 过多
-- 解决：考虑反范式化设计
ALTER TABLE orders ADD COLUMN user_name VARCHAR(100);
ALTER TABLE orders ADD COLUMN user_email VARCHAR(255);

-- 创建带数组字段的表
CREATE TABLE products (
  id SERIAL PRIMARY KEY,
  name VARCHAR(255),
  tags TEXT[],  -- 数组类型
  prices DECIMAL[]  -- 价格数组
);

-- 数组查询
INSERT INTO products (name, tags) 
VALUES ('iPhone', ARRAY['手机', '苹果', '电子产品']);

SELECT * FROM products WHERE '苹果' = ANY(tags);
SELECT * FROM products WHERE tags && ARRAY['手机', '电脑'];  -- 交集

-- 创建 JSON 字段
CREATE TABLE events (
  id SERIAL PRIMARY KEY,
  data JSONB,
  created_at TIMESTAMP DEFAULT NOW()
);

-- 插入 JSON 数据
INSERT INTO events (data) VALUES 
('{"type": "click", "page": "/home", "user_id": 123}'),
('{"type": "purchase", "amount": 99.99, "items": ["a", "b"]}');

-- 查询 JSON 字段
SELECT data->>'type' AS event_type FROM events;
SELECT data->'items'->>0 AS first_item FROM events WHERE data->>'type' = 'purchase';

-- 索引 JSON 字段
CREATE INDEX idx_events_data ON events USING gin(data);

-- 查询优化
SELECT * FROM events WHERE data @> '{"type": "click"}';  -- 包含
SELECT * FROM events WHERE data ? 'user_id';            -- 存在
SELECT * FROM events WHERE data ->> 'type' IN ('click', 'purchase');

-- 自动记录更新时间
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
  NEW.updated_at = NOW();
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER users_updated_at
  BEFORE UPDATE ON users
  FOR EACH ROW
  EXECUTE FUNCTION update_updated_at();

-- 审计日志
CREATE TABLE audit_log (
  id SERIAL PRIMARY KEY,
  table_name VARCHAR(100),
  action VARCHAR(10),
  old_data JSONB,
  new_data JSONB,
  changed_by VARCHAR(100),
  changed_at TIMESTAMP DEFAULT NOW()
);

CREATE OR REPLACE FUNCTION audit_trigger()
RETURNS TRIGGER AS $$
BEGIN
  INSERT INTO audit_log (table_name, action, old_data, new_data, changed_by)
  VALUES (
    TG_TABLE_NAME,
    TG_OP,
    CASE WHEN TG_OP = 'DELETE' THEN to_jsonb(OLD) ELSE NULL END,
    CASE WHEN TG_OP IN ('INSERT', 'UPDATE') THEN to_jsonb(NEW) ELSE NULL END,
    current_user
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER users_audit
  AFTER INSERT OR UPDATE OR DELETE ON users
  FOR EACH ROW
  EXECUTE FUNCTION audit_trigger();

# postgresql.conf

# 内存配置
shared_buffers = 256MB              # 建议 25% 物理内存
effective_cache_size = 1GB          # 建议 75% 物理内存
work_mem = 16MB                     # 每个排序操作内存
maintenance_work_mem = 128MB        # 维护操作内存

# 并行查询
max_worker_processes = 8            # CPU 核心数
max_parallel_workers_per_gather = 4 # 并行度

# 日志
log_destination = 'csvlog'
logging_collector = on
log_directory = 'log'
log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log'
log_statement = 'ddl'               # 只记录 DDL
log_min_duration_statement = 1000   # 记录超过 1 秒的查询

# 连接
max_connections = 200

# 写入优化
wal_buffers = 16MB
checkpoint_completion_target = 0.9

# 全量备份
pg_dump -h localhost -U postgres -Fc mydb > backup.dump

# 增量备份（基于 WAL）
# 配置wal_archiving
archive_mode = on
archive_command = 'cp %p /backup/wal/%f'

# 恢复
pg_restore -h localhost -U postgres -d mydb backup.dump

# _point_in_time 恢复
# 1. 停止 PostgreSQL
# 2. 清空数据目录
# 3. 初始化
# 4. 修改 postgresql.conf:
#    restore_command = 'cp /backup/wal/%f %p'
#    recovery_target_time = '2026-03-29 10:00:00'
# 5. 创建 recovery.signal 文件
# 6. 启动 PostgreSQL

# Patroni + etcd 高可用配置示例
patroni:
  name: node1
  scope: mycluster
  etcd:
    hosts: etcd1:2379,etcd2:2379,etcd3:2379
  postgres:
    data_dir: /data/postgresql
    parameters:
      max_connections: 200
      shared_buffers: 256MB
  bootstrap:
    dcs:
      ttl: 30
      loop_wait: 10
      retry_timeout: 10
      maximum_lag_on_failover: 1048576
    users:
      replication:
        username: replicator
        password: replicator_password
      postgres:
        password: postgres_password
  watchdog:
    mode: required
    device: /dev/watchdog

// pgbouncer 连接池配置
[databases]
mydb = host=127.0.0.1 port=5432 dbname=mydb

[pgbouncer]
listen_addr = 127.0.0.1
listen_port = 6432
auth_type = md5
auth_file = users.txt
pool_mode = transaction  # 事务级连接池
max_client_conn = 1000
default_pool_size = 25
min_pool_size = 10
reserve_pool_size = 5

// Node.js 使用
const { Pool } = require('pg');
const pool = new Pool({
  // 连接 pgbouncer 端口 6432
  host: '127.0.0.1',
  port: 6432,
  database: 'mydb',
  max: 20  // 客户端连接数
});

// schema.prisma
model User {
  id        Int       @id @default(autoincrement())
  email     String    @unique
  name      String?
  orders    Order[]
  createdAt DateTime  @default(now())
  updatedAt DateTime  @updatedAt
}

model Order {
  id        Int      @id @default(autoincrement())
  userId    Int
  user      User     @relation(fields: [userId], references: [id])
  amount    Decimal
  status    String
  createdAt DateTime @default(now())
}

// 查询示例
const users = await prisma.user.findMany({
  where: { status: 'active' },
  include: { orders: true },
  orderBy: { createdAt: 'desc' },
  take: 20
});

// 事务
const result = await prisma.$transaction(async (tx) => {
  const user = await tx.user.update({
    where: { id: userId },
    data: { balance: { decrement: amount } }
  });
  
  await tx.order.create({
    data: { userId, amount, status: 'pending' }
  });
  
  return user;
});