sql-expert

SQL 专家技能

针对 PostgreSQL、MySQL、SQLite 和 SQL Server 数据库的编写、优化与管理提供专家指导。

核心能力

此技能使你能够：

• 编写复杂的 SQL 查询，包括 JOIN、子查询、CTE 和窗口函数
• 使用 EXPLAIN 执行计划和索引建议优化慢查询
• 设计数据库架构，并进行适当的规范化（1NF、2NF、3NF、BCNF）
• 创建有效的索引，提升查询性能
• 安全地编写数据库迁移，并支持回滚
• 调试 SQL 错误并理解错误消息
• 使用适当的隔离级别处理事务
• 使用 JSON/JSONB 数据类型
• 生成测试用的样例数据
• 转换不同数据库方言（PostgreSQL ↔ MySQL ↔ SQLite）

---

支持的数据库系统

PostgreSQL

适用场景：复杂查询、JSON 数据、高级功能、ACID 合规性

pip install psycopg2-binary sqlalchemy

MySQL/MariaDB

适用场景：Web 应用、WordPress、高读负载

pip install mysql-connector-python sqlalchemy

SQLite

适用场景：本地开发、嵌入式数据库、测试

pip install sqlite3  # Built into Python

SQL Server

适用场景：企业应用、Windows 环境

pip install pyodbc sqlalchemy

---

查询编写

基本 SELECT 与 JOIN

-- Simple SELECT with filtering
SELECT
    column1,
    column2,
    column3
FROM
    table_name
WHERE
    condition = 'value'
    AND another_condition > 100
ORDER BY
    column1 DESC
LIMIT 10;

-- INNER JOIN
SELECT
    users.name,
    orders.order_date,
    orders.total_amount
FROM
    users
INNER JOIN
    orders ON users.id = orders.user_id
WHERE
    orders.status = 'completed';

-- LEFT JOIN (include all users, even without orders)
SELECT
    users.name,
    COUNT(orders.id) as order_count,
    COALESCE(SUM(orders.total_amount), 0) as total_spent
FROM
    users
LEFT JOIN
    orders ON users.id = orders.user_id
GROUP BY
    users.id, users.name;

子查询和 CTE

-- Subquery in WHERE clause
SELECT name, salary
FROM employees
WHERE salary > (SELECT AVG(salary) FROM employees);

-- Common Table Expression (CTE)
WITH high_value_customers AS (
    SELECT
        user_id,
        SUM(total_amount) as lifetime_value
    FROM orders
    GROUP BY user_id
    HAVING SUM(total_amount) > 1000
)
SELECT
    users.name,
    users.email,
    hvc.lifetime_value
FROM users
INNER JOIN high_value_customers hvc ON users.id = hvc.user_id;

窗口函数

-- Ranking within groups
SELECT
    name,
    department,
    salary,
    ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) as salary_rank
FROM
    employees;

-- Running totals
SELECT
    order_date,
    total_amount,
    SUM(total_amount) OVER (ORDER BY order_date) as running_total
FROM
    orders;

-- Moving averages
SELECT
    order_date,
    total_amount,
    AVG(total_amount) OVER (
        ORDER BY order_date
        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ) as moving_avg_7days
FROM
    daily_sales;

有关更高级的查询模式，请参阅 examples/complex_queries.sql。

---

查询优化

使用 EXPLAIN

-- Analyze query performance
EXPLAIN ANALYZE
SELECT
    users.name,
    COUNT(orders.id) as order_count
FROM users
LEFT JOIN orders ON users.id = orders.user_id
GROUP BY users.id, users.name;

-- Look for:
-- - Seq Scan (bad) vs Index Scan (good)
-- - High cost numbers
-- - Large row counts being processed

快速优化技巧

-- BAD: Function on indexed column
SELECT * FROM users WHERE LOWER(email) = '[email protected]';

-- GOOD: Keep indexed column clean
SELECT * FROM users WHERE email = LOWER('[email protected]');

-- BAD: SELECT *
SELECT * FROM large_table WHERE id = 123;

-- GOOD: Select only needed columns
SELECT id, name, email FROM large_table WHERE id = 123;

有关全面的优化技巧，请参阅 references/query-optimization.md。

---

架构设计

规范化原则

第一范式（1NF）：消除重复组，使用原子值

-- GOOD: Separate table for order items
CREATE TABLE orders (
    order_id INT PRIMARY KEY,
    customer_name VARCHAR(100)
);

CREATE TABLE order_items (
    order_item_id INT PRIMARY KEY,
    order_id INT REFERENCES orders(order_id),
    product_name VARCHAR(100)
);

第二范式（2NF）：所有非键属性都依赖于整个主键

-- GOOD: Separate product information
CREATE TABLE products (
    product_id INT PRIMARY KEY,
    product_name VARCHAR(100),
    product_price DECIMAL(10, 2)
);

CREATE TABLE order_items (
    order_id INT,
    product_id INT,
    quantity INT,
    PRIMARY KEY (order_id, product_id),
    FOREIGN KEY (product_id) REFERENCES products(product_id)
);

第三范式（3NF）：不存在传递依赖

常见架构模式

一对多：

CREATE TABLE authors (
    author_id INT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100) UNIQUE
);

CREATE TABLE books (
    book_id INT PRIMARY KEY,
    title VARCHAR(200),
    author_id INT NOT NULL,
    published_date DATE,
    FOREIGN KEY (author_id) REFERENCES authors(author_id)
);

多对多：

CREATE TABLE students (
    student_id INT PRIMARY KEY,
    name VARCHAR(100)
);

CREATE TABLE courses (
    course_id INT PRIMARY KEY,
    course_name VARCHAR(100)
);

-- Junction table
CREATE TABLE enrollments (
    enrollment_id INT PRIMARY KEY,
    student_id INT NOT NULL,
    course_id INT NOT NULL,
    enrollment_date DATE,
    grade CHAR(2),
    FOREIGN KEY (student_id) REFERENCES students(student_id),
    FOREIGN KEY (course_id) REFERENCES courses(course_id),
    UNIQUE (student_id, course_id)
);

有关更多架构模式，请参阅 examples/schema_examples.sql。

---

索引与性能

创建索引

-- Single column index
CREATE INDEX idx_users_email ON users(email);

-- Composite index (order matters!)
CREATE INDEX idx_orders_user_date ON orders(user_id, order_date);

-- Unique index
CREATE UNIQUE INDEX idx_users_username ON users(username);

-- Partial index (PostgreSQL)
CREATE INDEX idx_active_users ON users(email) WHERE status = 'active';

索引指南

适合创建索引的情况：

• ✅ WHERE 子句中使用的列
• ✅ JOIN 条件中使用的列
• ✅ ORDER BY 中使用的列
• ✅ 外键列

不适合创建索引的情况：

• ❌ 小型表（< 1000 行）
• ❌ 选择性低的列（布尔字段）
• ❌ 频繁更新的列

有关详细的索引策略，请参阅 references/indexes-performance.md。

---

迁移

安全迁移模式

-- Step 1: Add column as nullable
ALTER TABLE users ADD COLUMN status VARCHAR(20);

-- Step 2: Populate existing rows
UPDATE users SET status = 'active' WHERE status IS NULL;

-- Step 3: Make it NOT NULL
ALTER TABLE users ALTER COLUMN status SET NOT NULL;

-- Step 4: Add default for new rows
ALTER TABLE users ALTER COLUMN status SET DEFAULT 'active';

-- Rollback plan
ALTER TABLE users DROP COLUMN status;

零停机迁移

-- GOOD: Add column as nullable first, then backfill
ALTER TABLE large_table ADD COLUMN new_column VARCHAR(100);

-- Backfill in batches
UPDATE large_table SET new_column = 'value' WHERE new_column IS NULL LIMIT 1000;
-- Repeat until complete

-- Then make it NOT NULL
ALTER TABLE large_table ALTER COLUMN new_column SET NOT NULL;

有关更多迁移模式，请参阅 examples/migrations.sql。

---

高级模式

UPSERT（插入或更新）

-- PostgreSQL
INSERT INTO users (user_id, name, email, updated_at)
VALUES (1, 'John Doe', '[email protected]', NOW())
ON CONFLICT (user_id)
DO UPDATE SET
    name = EXCLUDED.name,
    email = EXCLUDED.email,
    updated_at = NOW();

-- MySQL
INSERT INTO users (user_id, name, email, updated_at)
VALUES (1, 'John Doe', '[email protected]', NOW())
ON DUPLICATE KEY UPDATE
    name = VALUES(name),
    email = VALUES(email),
    updated_at = NOW();

递归 CTE

-- Hierarchical data traversal
WITH RECURSIVE employee_hierarchy AS (
    -- Anchor: top-level employees
    SELECT id, name, manager_id, 1 as level
    FROM employees
    WHERE manager_id IS NULL

    UNION ALL

    -- Recursive: employees reporting to previous level
    SELECT e.id, e.name, e.manager_id, eh.level + 1
    FROM employees e
    INNER JOIN employee_hierarchy eh ON e.manager_id = eh.id
)
SELECT * FROM employee_hierarchy ORDER BY level, name;

有关更多高级模式，包括数据透视表、JSON 操作和批量操作，请参阅 references/advanced-patterns.md。

---

最佳实践

关键指南

1. 始终使用参数化查询，防止 SQL 注入
2. 对相关操作使用事务，确保原子性
3. 添加适当的约束（PRIMARY KEY、FOREIGN KEY、NOT NULL、CHECK）
4. 在表中包含时间戳（created_at、updated_at）
5. 为表和列使用有意义的名称
6. **避免使用 SELECT ***——仅指定所需的列
7. 为外键建立索引，提升连接性能
8. 可变长度字符串使用 VARCHAR 而非 CHAR
9. 使用 IS NULL / IS NOT NULL 正确处理 NULL 值
10. 使用适当的数据类型（金额使用 DECIMAL，而非 FLOAT）

多项最佳实践示例：

CREATE TABLE orders (
    order_id INT PRIMARY KEY,
    user_id INT NOT NULL,
    order_date DATE NOT NULL DEFAULT CURRENT_DATE,
    total_amount DECIMAL(10, 2) CHECK (total_amount >= 0),
    status VARCHAR(20) CHECK (status IN ('pending', 'completed', 'cancelled')),
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(user_id)
);

CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_orders_status ON orders(status);

有关全面的最佳实践，请参阅 references/best-practices.md。

---

常见陷阱

注意以下常见问题：

1. N+1 查询问题 - 使用 JOIN，而不是在循环中执行查询
2. 对大型表进行探索性查询时未使用 LIMIT
3. 隐式类型转换导致无法使用索引
4. EXISTS 已足够时仍使用 COUNT(*)
5. 未正确处理 NULL（NULL = NULL 的结果始终是 NULL，而不是 TRUE）
6. 将 SELECT DISTINCT 当作临时补救手段，而不是修正查询
7. 忘记为相关操作使用事务
8. 在带索引的列上使用函数导致无法使用索引

示例 - 避免 N+1：

# BAD: N+1 queries
users = db.query("SELECT * FROM users")
for user in users:
    orders = db.query("SELECT * FROM orders WHERE user_id = ?", user.id)

# GOOD: Single query with JOIN
result = db.query("""
    SELECT users.*, orders.*
    FROM users
    LEFT JOIN orders ON users.id = orders.user_id
""")

如需查看完整的陷阱及解决方案列表，请参阅 references/common-pitfalls.md。

---

辅助脚本和示例

可用资源

辅助脚本（scripts/）：

• sql_helper.py - 用于构建查询、检查架构、分析索引和辅助迁移的实用函数

示例（examples/）：

• complex_queries.sql - 包含 CTE、窗口函数和子查询的高级查询模式
• schema_examples.sql - 适用于各种用例的完整架构设计示例
• migrations.sql - 安全迁移模式和零停机技术

参考资料（references/）：

• query-optimization.md - 全面的查询优化技术和 EXPLAIN 分析
• indexes-performance.md - 详细的索引策略、维护和监控
• advanced-patterns.md - UPSERT、批量操作、数据透视表、JSON 操作、递归查询
• best-practices.md - 完整的 SQL 最佳实践指南
• common-pitfalls.md - 常见错误及其避免方法

快速入门

1. 对于基本查询，请使用上文所示的模式
2. 对于优化，请从 EXPLAIN 入手，并查看 references/query-optimization.md
3. 对于架构设计，请查看规范化模式和 examples/schema_examples.sql
4. 对于复杂场景，请查看 references/advanced-patterns.md
5. 对于实用工具，请使用 scripts/sql_helper.py

---

工作流

使用 SQL 数据库时：

1. 了解需求 - 需要查询或存储哪些数据？
2. 设计架构 - 应用规范化，选择合适的数据类型
3. 创建索引 - 为外键和经常查询的列创建索引
4. 编写查询 - 从简单开始，按需增加复杂度
5. 优化 - 使用 EXPLAIN 查找瓶颈
6. 测试 - 使用示例数据和边界情况进行验证
7. 编写文档 - 为复杂查询添加注释

迁移时：

1. 规划变更 - 确定受影响的表和依赖关系
2. 编写迁移 - 同时创建 up 和 down 迁移
3. 在副本上测试 - 先在开发数据库上测试
4. 备份 - 运行迁移前务必备份
5. 执行 - 在流量较低时运行迁移
6. 验证 - 迁移后检查数据完整性