问题描述：ETL 审计逻辑太繁琐
你在搭建 ETL 管道，每天需要：
- 向表中插入新记录
- 更新已有记录
- 保留变更日志——哪些行被插入、哪些被更新、哪些被删除
传统做法需要多个查询：
-- 第 1 步：查找已存在的 ID
SELECT id FROM customers WHERE id IN (1, 2, 3);
-- 第 2 步：插入新记录
INSERT INTO customers (id, name) VALUES (1, 'Alice'), (2, 'Bob');
-- 第 3 步：更新已有记录
UPDATE customers SET name = 'Alice Updated' WHERE id = 2;
-- 第 4 步：记录变更日志
INSERT INTO audit_log (action, id, old_value, new_value)
VALUES ('INSERT', 1, NULL, 'Alice'), ('UPDATE', 2, 'Bob', 'Alice Updated');
仅仅 2 行数据就需要 4 个独立查询。如果是 10,000 行，这将成为维护噩梦。
一招解决：RETURNING
DuckDB 支持 RETURNING 子句——这个让 Postgres 如此强大的特性。它让你在一次查询中同时执行 INSERT/UPDATE/DELETE 并捕获结果。
-- 一条查询：插入并捕获结果
INSERT INTO customers (id, name) VALUES (1, 'Alice') RETURNING *;
-- 一条查询：更新并捕获结果
UPDATE customers SET name = 'Alice Updated' WHERE id = 1 RETURNING *;
-- 一条查询：删除并捕获结果
DELETE FROM customers WHERE id = 1 RETURNING *;
一条 SQL 替代 4 个查询。 这就是 RETURNING 的威力。
实战示例：ETL 审计日志
建表
-- 目标表
CREATE TABLE customers (
id INT PRIMARY KEY,
name VARCHAR,
email VARCHAR,
updated_at TIMESTAMP
);
-- 审计日志表
CREATE TABLE audit_log (
action VARCHAR,
id INT,
old_name VARCHAR,
new_name VARCHAR,
changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
旧方案：每行 4 个查询
# Python 伪代码
for record in staging_table:
# 检查是否存在
existing = con.execute(
"SELECT name FROM customers WHERE id = ?", [record.id]
).fetchone()
if existing:
# 更新
con.execute(
"UPDATE customers SET name = ? WHERE id = ?",
[record.name, record.id]
)
# 记录日志
con.execute(
"INSERT INTO audit_log (action, id, old_name, new_name) VALUES (?, ?, ?, ?)",
["UPDATE", record.id, existing[0], record.name]
)
else:
# 插入
con.execute(
"INSERT INTO customers (id, name) VALUES (?, ?)",
[record.id, record.name]
)
# 记录日志
con.execute(
"INSERT INTO audit_log (action, id, old_name, new_name) VALUES (?, ?, ?, ?)",
["INSERT", record.id, None, record.name]
)
每行 4 个查询 × 10,000 行 = 40,000 个查询。难怪你的 ETL 这么慢。
RETURNING 方案：每行 1 个查询
-- 第 1 步：插入新记录并捕获结果
CREATE TEMP TABLE inserted AS
INSERT INTO customers (id, name)
SELECT id, name FROM staging_customers
WHERE id NOT IN (SELECT id FROM customers)
RETURNING id, name, 'INSERT' AS action;
-- 第 2 步：更新已有记录并捕获新旧值
CREATE TEMP TABLE updated AS
UPDATE customers c
SET name = s.name, updated_at = CURRENT_TIMESTAMP
FROM staging_customers s
WHERE c.id = s.id AND c.name != s.name
RETURNING c.id, c.name AS old_name, s.name AS new_name, 'UPDATE' AS action;
-- 第 3 步：一次性批量写入审计日志
INSERT INTO audit_log (action, id, old_name, new_name)
SELECT action, id, NULL, name FROM inserted
UNION ALL
SELECT action, id, old_name, new_name FROM updated;
总共：3 个查询处理 10,000 行数据。 比旧方案少了 13,333 倍查询。
效果量化
| 指标 | 旧方案 | RETURNING 方案 |
|---|---|---|
| 10K 行查询次数 | 40,000 | 3 |
| Python 代码行数 | ~30 行 | ~10 行 |
| 网络往返延迟 | 40,000 × RTT | 3 × RTT |
| 内存开销 | 高（批量 Python 处理） | 低（SQL 原生） |
对于 10,000 行的 ETL 任务：
- 旧方案：40,000 查询 × 1ms RTT = 40 秒网络延迟
- RETURNING 方案：3 查询 × 1ms RTT = 3 毫秒
仅减少查询次数就能实现 13,000 倍加速。
进阶技巧：条件式 RETURNING
你可以将 RETURNING 与 CASE WHEN 结合，实现条件审计：
-- 条件审计：只在名称变更时记录
UPDATE customers c
SET name = s.name, updated_at = CURRENT_TIMESTAMP
FROM staging_customers s
WHERE c.id = s.id
RETURNING
c.id,
CASE WHEN c.name != s.name THEN 'UPDATE' ELSE 'NOCHANGE' END AS action,
c.name AS old_name,
s.name AS new_name;
这让你跳过无变更更新的审计条目——对于增量 ETL 非常有用，因为许多行可能没有变化。
使用场景速查
| 场景 | 是否使用 RETURNING |
|---|---|
| 插入审计日志 | ✅ 推荐 |
| 带变更追踪的更新 | ✅ 推荐 |
| 带软删除日志的删除 | ✅ 推荐 |
| 批量 upsert + 审计 | ✅ 推荐 |
| 简单 SELECT 查询 | ❌ 不需要 |
| 复杂业务逻辑 | ❌ 用 Python |
核心要点
- RETURNING 将 4 个查询变为 1 个——在 INSERT/UPDATE/DELETE 后加
RETURNING *即可 - 批量操作是关键——对批量插入/更新使用 RETURNING，不要逐行处理
- 结合临时表——将 RETURNING 结果捕获到临时表，用于批量日志写入
- 条件 RETURNING——使用
CASE WHEN实现智能审计过滤
下次搭建 ETL 管道时记住：一条 RETURNING 子句可以替代整个审计子系统。
订阅 DuckDB Lab，每周获取一条让你节省数小时的实战技巧。