MySQL 历史数据清理：让删除和主键断点在同一事务提交
DELETE FROM event_log WHERE created_at < ... 很容易写。真正需要设计的是：删到一半进程退出了，下一次从哪里继续；任务记录已经前进而删除回滚了，会不会永久漏数据；刚好遇到一笔迟提交事务，又该怎样解释“已完成”。
下面把历史事件清理做成一轮有上界的扫描。每批取有限个主键，删除其中已到期的行，删除结果与游标在同一个 InnoDB 事务里提交。示例适用 MySQL 8.4，使用 Python 3 和 PyMySQL，重点是断点协议，不给出脱离业务负载的固定最优批量值。
游标推进的是扫描范围，不是已删除行数
有些实现只拿“本批删了多少条”判断是否结束。第一页含有大量未过期记录，删除为零，它就停了；后面的旧记录根本没有扫描。
这里先按主键选择一段候选，再按时间条件删除候选中的过期记录。游标推进到本批最后一个候选 id，因此零删除仍可前进。任务同时保存固定 cutoff 和 upper_id，下一次恢复不能临时换一个截止时间或混入无限增长的新尾部。
完整建表和样本如下，只放在专用实验库。事件创建时间在这套模型中不可修改，编号也不重用。
CREATE DATABASE IF NOT EXISTS purge36_lab
CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
USE purge36_lab;
CREATE TABLE event_log (
id BIGINT UNSIGNED NOT NULL,
created_at DATETIME(6) NOT NULL,
payload VARCHAR(200) NOT NULL,
PRIMARY KEY (id),
KEY idx_created_id (created_at, id)
) ENGINE=InnoDB;
CREATE TABLE purge_job (
job_name VARCHAR(64) NOT NULL,
cutoff DATETIME(6) NOT NULL,
upper_id BIGINT UNSIGNED NOT NULL,
last_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
deleted_total BIGINT UNSIGNED NOT NULL DEFAULT 0,
status ENUM('RUNNING','DONE') NOT NULL DEFAULT 'RUNNING',
PRIMARY KEY (job_name)
) ENGINE=InnoDB;
INSERT INTO event_log VALUES
(1,'2024-12-01 00:00:00','old-1'),
(2,'2025-02-01 00:00:00','new-2'),
(3,'2024-12-02 00:00:00','old-3'),
(4,'2024-12-03 00:00:00','old-4'),
(5,'2025-02-02 00:00:00','new-5'),
(6,'2024-12-04 00:00:00','old-6'),
(7,'2024-12-05 00:00:00','old-7'),
(8,'2025-02-03 00:00:00','new-8');
INSERT INTO purge_job (job_name,cutoff,upper_id)
SELECT 'demo-round-1','2025-01-01 00:00:00',COALESCE(MAX(id),0)
FROM event_log;
主键从正整数开始，零是初始游标哨兵。idx_created_id 便于按过期时间核对残留；执行器选候选时则按主键顺序走，不做 OFFSET 分页。删除会让前面的记录消失，OFFSET 继续增加容易跳过后面的行。
可执行的清理器
在本机实验环境安装依赖，将代码保存为 purge36.py。密码只从终端读取，代码不保存凭据。目标库名固定为实验库，避免示例误连默认业务库。
python3 -m venv .venv-purge36
. .venv-purge36/bin/activate
python -m pip install 'PyMySQL>=1.1,<2'
import argparse
import getpass
import json
import pymysql
def step(conn, job_name, batch_size, fail_after_delete):
conn.begin()
try:
with conn.cursor(pymysql.cursors.DictCursor) as cur:
cur.execute('SELECT * FROM purge_job WHERE job_name=%s FOR UPDATE',
(job_name,))
job = cur.fetchone()
if job is None:
raise ValueError('job not found')
if job['status'] == 'DONE':
conn.commit()
return job
cur.execute('SELECT id FROM event_log WHERE id>%s AND id<=%s '
'ORDER BY id LIMIT %s',
(job['last_id'], job['upper_id'], batch_size))
ids = [row['id'] for row in cur.fetchall()]
next_id = ids[-1] if ids else job['upper_id']
deleted = 0
if ids:
placeholders = ','.join(['%s'] * len(ids))
cur.execute('DELETE FROM event_log WHERE id IN (' + placeholders
+ ') AND created_at<%s', tuple(ids) + (job['cutoff'],))
deleted = cur.rowcount
if fail_after_delete:
raise RuntimeError('injected failure before checkpoint')
status = 'DONE' if next_id >= job['upper_id'] else 'RUNNING'
cur.execute('UPDATE purge_job SET last_id=%s, '
'deleted_total=deleted_total+%s, status=%s WHERE job_name=%s',
(next_id, deleted, status, job_name))
cur.execute('SELECT * FROM purge_job WHERE job_name=%s', (job_name,))
result = cur.fetchone()
conn.commit()
return result
except BaseException:
try:
conn.rollback()
except pymysql.MySQLError:
pass # Preserve the original failure if the connection is already lost.
raise
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--host', default='127.0.0.1')
parser.add_argument('--port', type=int, default=3306)
parser.add_argument('--user', required=True)
parser.add_argument('--job', default='demo-round-1')
parser.add_argument('--batch', type=int, default=3)
parser.add_argument('--fail-after-delete', action='store_true')
args = parser.parse_args()
if not 1 <= args.batch <= 1000:
parser.error('--batch must be 1..1000')
conn = pymysql.connect(host=args.host, port=args.port, user=args.user,
password=getpass.getpass('Database password: '),
database='purge36_lab', charset='utf8mb4',
autocommit=False, connect_timeout=5)
try:
while True:
result = step(conn, args.job, args.batch, args.fail_after_delete)
print(json.dumps(result, default=str, ensure_ascii=False))
if result['status'] == 'DONE':
break
finally:
conn.close()
if __name__ == '__main__':
main()
动态拼接的只有由候选数量生成的 %s 占位符，主键、时间和任务名仍通过驱动参数绑定。没有把输入值拼进 SQL。批量上限也限制了单条 IN 列表规模，但这不是对锁等待时间的保证。
每批首先锁定任务行，两个相同 job_name 的执行器会依次进入，而不会各自拿旧游标提交。删除采用本次选出的明确主键集合，避免把同一数值区间里后来插入、并未选中的记录顺手删除。业务事务仍可能与 DELETE 争用这些主键，应按实际锁等待配置让任务可失败、可恢复。
先让它在最危险的位置失败
先执行带故障注入的命令。它在 DELETE 之后、写游标之前抛出异常；这次运行应以错误结束。
python purge36.py --user=lab_user --fail-after-delete
回到数据库检查：
SELECT job_name,last_id,deleted_total,status FROM purge_job;
SELECT COUNT(*) AS rows_after_failure FROM event_log;
预期游标仍为 0，累计删除数仍为 0，事件仍有八行。若这里出现“行删了，断点没记”或相反的组合，就不能继续讨论重启安全。这也是两张表都用 InnoDB、使用同一个连接事务的原因。
去掉故障参数再运行：
python purge36.py --user=lab_user
批量为三时，三个成功批次的游标应依次到 3、6、8，累计删除数到 2、4、5。最后状态为 DONE。再查内容：
SELECT id,created_at,payload FROM event_log ORDER BY id;
SELECT * FROM purge_job;
应只剩 2、5、8。再次运行同一任务时，会读取 DONE 并返回相同任务状态，不重新删除。进程若在成功提交后、打印结果前退出，也不需要靠日志猜测是否提交；数据库任务行是恢复依据。
真正断网时，客户端收到异常不总能说明 COMMIT 是否到达服务器。这个执行器不会在一次失败后原地盲目重发 SQL，而是退出。重新连接、重新调用任务，会读取服务器已经提交的游标；若上一笔确实回滚，则从原游标重做。清理动作和断点同事务，使这两种情况都可以根据数据库状态继续。
DONE 不是“从此绝无旧数据”
upper_id 固定了一轮扫描的上界，却没有把所有业务事务冻结。例如另一个事务先申请了较小编号，迟迟未提交；清理器扫描过那个位置以后，它才把一条旧事件提交出来。这条记录会落在游标后方，即使最终 id 小于 upper_id，也可能未被本轮看到。
在示例执行完后，可以直接模拟这类晚到数据：
INSERT INTO event_log VALUES
(4,'2024-12-03 00:00:00','late-arrival-demo');
SELECT id FROM event_log
WHERE created_at < '2025-01-01 00:00:00'
ORDER BY created_at,id;
会重新看到 4。重跑 DONE 的任务不会处理它，必须创建新的一轮，从零扫描当前有界范围：
INSERT INTO purge_job (job_name,cutoff,upper_id)
SELECT 'demo-round-2','2025-01-01 00:00:00',COALESCE(MAX(id),0)
FROM event_log;
python purge36.py --user=lab_user --job=demo-round-2
显式回填编号在真实业务中可能不允许，这个实验只是把迟到可见性摆到桌面上。自增编号同样不自动代表提交顺序。若需求是一次完成后绝不允许出现截止时间之前的数据，就需要上游写入屏障、明确的业务封账状态或其他可证明的边界，不能只凭 MAX(id) 宣称已经获得一致快照。
事件创建时间若允许修改，也有同样的问题：一行被扫过时尚未过期，之后又改成更早时间，本轮不会倒回去。按轮次重新检查残留是这套方案的一部分；它以短事务和可恢复为目标，接受需要后续补扫的写入模型。
主键扫描也有明确代价
这条路线会读过范围内的所有可见主键，包括尚未过期的记录。如果历史数据很少、活跃数据很多，反复从零扫描就不一定划算。时间索引可以迅速定位到期数据，但那时断点通常应同时包含时间与主键，处理相同时间值的稳定排序；不能直接把本例 last_id 套到另一种排序上。选择扫描顺序之前，先看过期记录的分布以及时间字段是否允许回填。
选候选的 SELECT 没有锁住事件行。行在选择后被其他事务删掉，当前批次实际删除数可能小于候选数，这是允许的；游标仍按已检查范围前进。如果应用允许删除后复用同一个主键，或者把同一编号替换成另一条业务事件，本例便失去“编号对应同一事件”的前提。应该修正数据模型或增加版本条件，而不是用累计删除数推断删对了谁。
固定先锁任务行、再删除事件的顺序，只约束清理器自己的代码。如果业务事务反过来先锁事件，再操作任务行，就可能形成死锁。应避免让普通事件写入流程更新这张清理任务表。确需统一操作时，要让各入口遵守共同的锁顺序，并保留死锁受害事务回滚后恢复的能力。
故障注入验证的是应用在未提交位置主动抛错，不能代表所有故障都已演练。进程被终止、数据库重启、提交响应丢失分别涉及连接清理和持久化时机；这里依靠数据库断点恢复，而非承诺客户端每次都知道最后一笔事务的结局。需要进一步验收时，应该在隔离环境分别切断这些阶段，再检查行内容和任务状态是否仍组成一份合法结果。
批量大小应跟着实际约束调整
三条只为让演练容易手算。生产批次要观察单批耗时、锁等待、复制延迟、日志增长以及业务请求延迟，再调整节奏。不能把批次切成一千条，就假设每个事务都同样便宜：级联外键、删除触发器和大字段都会改变一次删除的工作量。
多个不同 job_name 仍可能同时删除同一批业务行。任务行锁只串行化同一个任务，不是全表清理互斥；调度方若允许并发任务，需要额外限制它们的范围，或承受冲突后恢复的成本。这里也没有捕获死锁后无限重试，异常会回滚当前批次并交给外部运行方决定何时重启。
清理成功以后，表文件不一定立即变小，InnoDB 可复用的空间与操作系统看到的文件大小属于另一个层次。验收本轮清理，先查剩余记录、任务断点与实际提交状态；不要用磁盘文件没有缩小来推断 DELETE 没有生效。
