sql-database-assistant

SQL 数据库助手 - POWERFUL 等级技能

概述

数据库设计的日常运维配套工具。database-designer 专注于 schema 架构，database-schema-designer 负责 ERD 建模，而本技能覆盖日常工作：编写查询、优化性能、生成迁移，并衔接应用代码与数据库引擎。

核心能力

• 自然语言转 SQL：将需求转化为正确且高性能的查询
• Schema 探查：内省 PostgreSQL、MySQL、SQLite、SQL Server 的在线数据库
• 查询优化：EXPLAIN 分析、索引建议、N+1 检测、改写模式
• 迁移生成：升级/回滚脚本、零停机策略、回滚方案
• ORM 集成：Prisma、Drizzle、TypeORM、SQLAlchemy 的使用模式与逃生口
• 多数据库支持：感知方言的 SQL，并提供兼容性指导

工具

脚本	用途
scripts/query_optimizer.py	对 SQL 查询进行静态分析，查找性能问题
scripts/migration_generator.py	根据变更描述生成迁移文件模板
scripts/schema_explorer.py	基于内省查询生成 schema 文档

---

自然语言转 SQL

转换模式

将需求转换为 SQL 时，请遵循以下步骤：

1. 识别实体：将名词映射为表
2. 识别关系：将动词映射为 JOIN 或子查询
3. 识别过滤条件：将形容词或条件映射为 WHERE 子句
4. 识别聚合：将“总计”“平均”“计数”映射为 GROUP BY
5. 识别排序：将“前 N”“最新”“最高”映射为 ORDER BY + LIMIT

常用查询模板

分组取前 N 条（窗口函数）

SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY department_id ORDER BY salary DESC) AS rn
  FROM employees
) ranked WHERE rn <= 3;

累计求和

SELECT date, amount,
  SUM(amount) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_total
FROM transactions;

间隙检测

SELECT curr.id, curr.seq_num, prev.seq_num AS prev_seq
FROM records curr
LEFT JOIN records prev ON prev.seq_num = curr.seq_num - 1
WHERE prev.id IS NULL AND curr.seq_num > 1;

UPSERT（PostgreSQL）

INSERT INTO settings (key, value, updated_at)
VALUES ('theme', 'dark', NOW())
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = EXCLUDED.updated_at;

UPSERT（MySQL）

INSERT INTO settings (key_name, value, updated_at)
VALUES ('theme', 'dark', NOW())
ON DUPLICATE KEY UPDATE value = VALUES(value), updated_at = VALUES(updated_at);

JOIN、CTE、窗口函数、JSON 操作等更多内容，请参阅 references/query_patterns.md。

---

Schema 探查

内省查询

PostgreSQL：列出表和列

SELECT table_name, column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position;

PostgreSQL：外键

SELECT tc.table_name, kcu.column_name,
  ccu.table_name AS foreign_table, ccu.column_name AS foreign_column
FROM information_schema.table_constraints tc
JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name
JOIN information_schema.constraint_column_usage ccu ON tc.constraint_name = ccu.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY';

MySQL：表大小

SELECT table_name, table_rows,
  ROUND(data_length / 1024 / 1024, 2) AS data_mb,
  ROUND(index_length / 1024 / 1024, 2) AS index_mb
FROM information_schema.tables
WHERE table_schema = DATABASE()
ORDER BY data_length DESC;

SQLite：schema 导出

SELECT name, sql FROM sqlite_master WHERE type = 'table' ORDER BY name;

SQL Server：带类型的列

SELECT t.name AS table_name, c.name AS column_name,
  ty.name AS data_type, c.max_length, c.is_nullable
FROM sys.columns c
JOIN sys.tables t ON c.object_id = t.object_id
JOIN sys.types ty ON c.user_type_id = ty.user_type_id
ORDER BY t.name, c.column_id;

基于 Schema 生成文档

使用 scripts/schema_explorer.py 生成 Markdown 或 JSON 格式的文档：

python scripts/schema_explorer.py --dialect postgres --tables all --format md
python scripts/schema_explorer.py --dialect mysql --tables users,orders --format json --json

---

查询优化

EXPLAIN 分析流程

1. 运行 EXPLAIN ANALYZE（PostgreSQL）或 EXPLAIN FORMAT=JSON（MySQL）
2. 找出开销最大的节点：大表上的 Seq Scan、预估行数很高的 Nested Loop
3. 检查缺失的索引：对过滤列进行顺序扫描
4. 留意估算误差：计划行数与实际行数的偏差表明统计信息已过期
5. 评估 JOIN 顺序：确保结果集最小的表驱动连接

索引建议清单

• WHERE 子句中选择性高的列
• JOIN 条件中的列（外键）
• 与 LIMIT 组合使用的 ORDER BY 列
• 匹配多列 WHERE 谓词的复合索引（选择性最高的列放在最前）
• 针对常量过滤条件的查询使用部分索引（例如 WHERE status = 'active'）
• 对读密集型查询使用覆盖索引，避免回表

查询改写模式

反模式	改写
SELECT * FROM orders	SELECT id, status, total FROM orders（显式列出列名）
WHERE YEAR(created_at) = 2025	WHERE created_at >= '2025-01-01' AND created_at < '2026-01-01'（可走索引的 sargable 写法）
SELECT 中的相关子查询	带聚合的 LEFT JOIN
含 NULL 的 NOT IN (SELECT ...)	NOT EXISTS (SELECT 1 ...)
不需要去重时使用 UNION	UNION ALL
LIKE '%search%'	全文搜索索引（GIN/FULLTEXT）
ORDER BY RAND()	应用层随机抽样或 TABLESAMPLE

N+1 检测

症状：

• 应用循环中每个父级行执行一次查询
• ORM 在循环内延迟加载关联实体
• 查询日志中出现数百条相同的 SELECT 模式，仅 ID 不同

修复方法：

• 使用预加载（Prisma 中的 include，SQLAlchemy 中的 joinedload）
• 使用 WHERE id IN (...) 批量查询
• 在 GraphQL 解析器中使用 DataLoader 模式

静态分析工具

python scripts/query_optimizer.py --query "SELECT * FROM orders WHERE status = 'pending'" --dialect postgres
python scripts/query_optimizer.py --query queries.sql --dialect mysql --json

有关 EXPLAIN 执行计划解读、索引类型和连接池的内容，请参阅 references/optimization_guide.md。

---

迁移生成

零停机迁移模式

新增列（安全）

-- Up
ALTER TABLE users ADD COLUMN phone VARCHAR(20);

-- Down
ALTER TABLE users DROP COLUMN phone;

重命名列（扩展-收缩）

-- Step 1: Add new column
ALTER TABLE users ADD COLUMN full_name VARCHAR(255);
-- Step 2: Backfill
UPDATE users SET full_name = name;
-- Step 3: Deploy app reading both columns
-- Step 4: Deploy app writing only new column
-- Step 5: Drop old column
ALTER TABLE users DROP COLUMN name;

新增 NOT NULL 列（安全顺序）

-- Step 1: Add nullable
ALTER TABLE orders ADD COLUMN region VARCHAR(50);
-- Step 2: Backfill with default
UPDATE orders SET region = 'unknown' WHERE region IS NULL;
-- Step 3: Add constraint
ALTER TABLE orders ALTER COLUMN region SET NOT NULL;
ALTER TABLE orders ALTER COLUMN region SET DEFAULT 'unknown';

创建索引（非阻塞，PostgreSQL）

CREATE INDEX CONCURRENTLY idx_orders_status ON orders (status);

数据回填策略

• 批量更新：每批处理 1000-10000 行，以避免锁竞争
• 后台任务：异步运行回填并跟踪进度
• 双写：在过渡期间同时写入旧列和新列
• 验证查询：每批处理后核对行数和数据完整性

回滚策略

每个迁移都必须有可逆的 down 脚本。对于不可逆的变更：

1. 执行前备份：对受影响的表执行 pg_dump
2. 功能开关：应用可在旧、新 Schema 读取之间切换
3. 影子表：在迁移窗口期内保留原始表的副本

迁移生成工具

python scripts/migration_generator.py --change "add email_verified boolean to users" --dialect postgres --format sql
python scripts/migration_generator.py --change "rename column name to full_name in customers" --dialect mysql --format alembic --json

---

多数据库支持

方言差异

特性	PostgreSQL	MySQL	SQLite	SQL Server
UPSERT	ON CONFLICT DO UPDATE	ON DUPLICATE KEY UPDATE	ON CONFLICT DO UPDATE	MERGE
布尔值	原生 BOOLEAN	TINYINT(1)	INTEGER	BIT
自增	SERIAL / GENERATED	AUTO_INCREMENT	INTEGER PRIMARY KEY	IDENTITY
JSON	JSONB（可索引）	JSON	文本（扩展）	NVARCHAR(MAX)
数组	原生 ARRAY	不支持	不支持	不支持
CTE（递归）	完全支持	8.0+	3.8.3+	完全支持
窗口函数	完全支持	8.0+	3.25.0+	完全支持
全文检索	tsvector + GIN	FULLTEXT 索引	FTS5 扩展	全文目录
LIMIT/OFFSET	LIMIT n OFFSET m	LIMIT n OFFSET m	LIMIT n OFFSET m	OFFSET m ROWS FETCH NEXT n ROWS ONLY

兼容性建议

• 始终使用参数化查询：在所有方言中防止 SQL 注入
• 避免在共享代码中使用方言特定函数：将其封装在适配层中
• 在目标引擎上测试迁移：不同引擎的 information_schema 存在差异
• 使用 ISO 日期格式：'YYYY-MM-DD' 在所有引擎中通用
• 为标识符加引号：使用双引号（SQL 标准）或反引号（MySQL）

---

ORM 模式

Prisma

Schema 定义

model User {
  id        Int      @id @default(autoincrement())
  email     String   @unique
  name      String?
  posts     Post[]
  createdAt DateTime @default(now())
}

model Post {
  id       Int    @id @default(autoincrement())
  title    String
  author   User   @relation(fields: [authorId], references: [id])
  authorId Int
}

迁移：npx prisma migrate dev --name add_user_email 查询 API：prisma.user.findMany({ where: { email: { contains: '@' } }, include: { posts: true } }) 原生 SQL 逃生口：prisma.$queryRaw\SELECT * FROM users WHERE id = ${userId}``

Drizzle

Schema 优先定义

export const users = pgTable('users', {
  id: serial('id').primaryKey(),
  email: varchar('email', { length: 255 }).notNull().unique(),
  name: text('name'),
  createdAt: timestamp('created_at').defaultNow(),
});

查询构建器：db.select().from(users).where(eq(users.email, email)) 迁移：npx drizzle-kit generate:pg 后执行 npx drizzle-kit push:pg

TypeORM

实体装饰器

@Entity()
export class User {
  @PrimaryGeneratedColumn()
  id: number;

  @Column({ unique: true })
  email: string;

  @OneToMany(() => Post, post => post.author)
  posts: Post[];
}

Repository 模式：userRepo.find({ where: { email }, relations: ['posts'] }) 迁移：npx typeorm migration:generate -n AddUserEmail

SQLAlchemy

声明式模型

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    name = Column(String(255))
    posts = relationship('Post', back_populates='author')

会话管理：始终使用 with Session() as session: 上下文管理器 Alembic 迁移：alembic revision --autogenerate -m "add user email"

有关各 ORM 的并列对比和迁移工作流，请参阅 references/orm_patterns.md。

---

数据完整性

约束策略

• 主键：每张表都必须有主键；优先使用代理键（自增序列或 UUID）
• 外键：强制引用完整性；显式定义 ON DELETE 行为
• UNIQUE 约束：用于业务层面的唯一性（邮箱、slug、API 密钥）
• CHECK 约束：在数据库层面校验取值范围、枚举值和业务规则
• NOT NULL：默认使用 NOT NULL；只有确实可选时才允许为空

事务隔离级别

级别	脏读	不可重复读	幻读	适用场景
READ UNCOMMITTED	是	是	是	从不推荐
READ COMMITTED	否	是	是	PostgreSQL 默认级别，通用 OLTP
REPEATABLE READ	否	否	是（InnoDB：否）	金融计算
SERIALIZABLE	否	否	否	关键一致性场景（计费、库存）

死锁预防

1. 一致的加锁顺序：始终按相同的表/行顺序获取锁
2. 缩短事务：尽量减少从首次加锁到提交之间的时间
3. 咨询锁：使用 pg_advisory_lock() 实现应用层协调
4. 重试逻辑：捕获死锁错误，并以指数退避方式重试

---

备份与恢复

PostgreSQL

# Full backup
pg_dump -Fc --no-owner dbname > backup.dump
# Restore
pg_restore -d dbname --clean --no-owner backup.dump
# Point-in-time recovery: configure WAL archiving + restore_command

MySQL

# Full backup
mysqldump --single-transaction --routines --triggers dbname > backup.sql
# Restore
mysql dbname < backup.sql
# Binary log for PITR: mysqlbinlog --start-datetime="2025-01-01 00:00:00" binlog.000001

SQLite

# Backup (safe with concurrent reads)
sqlite3 dbname ".backup backup.db"

备份最佳实践

• 自动化：使用 cron 或 systemd timer，不要只依赖手动操作
• 测试恢复：未经测试的备份不算备份
• 异地副本：S3、GCS 或其他区域
• 保留策略：每日备份保留 7 天，每周备份保留 4 周，每月备份保留 12 个月
• 监控备份大小和耗时：突然的变化往往预示着问题

---

反模式

反模式	问题	修复方法
SELECT *	传输不必要的数据，schema 变更时容易出错	显式列出字段
外键列缺少索引	JOIN 变慢，级联删除变慢	为所有外键添加索引
N+1 查询	与数据库之间产生 1 + N 次往返	预加载或批量查询
隐式类型转换	WHERE id = '123' 会导致无法使用索引	在谓词中保持类型一致
未使用连接池	高负载下耗尽连接	使用 PgBouncer、ProxySQL 或 ORM 连接池
无界查询	没有 LIMIT 可能返回数百万行	始终分页
用 FLOAT 存储金额	舍入误差	使用 DECIMAL(19,4) 或以整数分为单位
上帝表	单表 50 个以上字段	规范化或使用垂直分表
到处使用软删除	每个查询都要加 WHERE deleted_at IS NULL，使查询变复杂	使用归档表或事件溯源
原始字符串拼接	SQL 注入	始终使用参数化查询

---

交叉引用

Skill	关系
database-designer	Schema 架构、规范化分析、ERD 生成
database-schema-designer	可视化 ERD 建模、关系映射
migration-architect	复杂的多步骤迁移编排
api-design-reviewer	确保 API 端点与查询模式保持一致
observability-platform	查询性能监控、慢查询告警