让 Codex 调查订单汇总为什么变慢

从一份变慢的客户订单报表出发,让调查、修复与验收都有可以重新运行的依据。

你能带走什么
  • 先确定报表业务约束
  • 分别观察查询数、计划与耗时
  • 控制索引条件比较查询形状
  • 限定补丁范围并核对完整结果

1. 从一份慢报表开始

后台要展示每位客户的已付款订单数与金额。最初只有几十位客户,后来数据量增大,同一段汇总代码开始拖慢页面。接到这样的任务,可以让 Codex 先读查询、找出重复成本,再提出一个能独立验证的修改。给它一句“优化一下”还不够:哪些客户必须保留、哪些订单不算收入,以及怎样判断结果一致,都要放进任务里。

本文准备了一份 Python 标准库项目,完整源码在后文。它使用合成客户与订单,不连接业务系统。我们会比较无索引逐客户汇总、加上适配索引的逐客户汇总,以及使用同一索引的 LEFT JOIN / GROUP BY 查询。最后交付的应当是一份可以复查的调查结论、一个边界明确的补丁和回归结果。

OpenAI 的性能优化案例提到,用 Codex 分析昂贵查询与重复操作。下面把这类工作落到一个可复现的小场景。文中的代码与候选修复均为人工参考实现;本文没有运行 Codex 模型会话,没有测量模型成功率,也没有把人工修改写成 Codex 的实战成绩。

同一业务结果;索引条件单独控制,集合查询沿用同一索引
订单汇总的三条执行路径。同一业务结果;索引条件单独控制,集合查询沿用同一索引

2. 先写清楚报表不能变的部分

输出按客户 ID 升序排列,每个客户恰好一行。无订单客户和只有取消订单的客户都必须出现,笔数与金额为零。已付款的零金额订单仍计为一笔。金额使用整数分,避免让浮点表示误差混进数据库性能练习。客户姓名可以重复,因此分组依据必须包含客户身份。这里仅有 paid 与 cancelled 两种状态,暂不处理退款、跨币种、日期区间或重复入账。

fixture.py 中的小样本让这些边界都能被看见。测试把四行期望结果直接写出来,再对不同查询逐一核对;大样本另做完整列表比较。这样可以同时防住“所有方案一起算错”与“优化只漏了某一类客户”两种问题。

每一行先按客户身份分组,再计算已付款订单;同名客户不能合并
LEFT JOIN 后,客户仍应完整出现。每一行先按客户身份分组,再计算已付款订单;同名客户不能合并

3. 把 SQLite 的成本判断放回本地进程

逐客户查询具有常见的 N+1 形状:一条 SQL 读客户,后面每位客户再发一条汇总 SQL。SQLite 是嵌入式数据库,查询在应用进程内执行,没有客户端到数据库服务器的一次次网络往返。SQLite 官方的Many Small Queries Are Efficient In SQLite专门解释过这一点。看到 401 条查询,可以继续调查,但不能直接把查询数当成性能结论。

本例的第一个疑点是 customer_id 上缺少适配索引。逐客户汇总每次筛选同一张 orders 表;当执行计划显示全表扫描时,客户数量增长会放大重复读取成本。B 方案保留循环,只增加以 customer_id、status 开头并包含 amount_cents 的索引。C 方案使用与 B 相同的索引,将查询写成集合形式。这一安排让两次比较有清楚的含义,也避免把“加索引”和“改 SQL”产生的收益混在一起。

4. 从空目录运行参考项目

需要带 sqlite3 模块的 Python 3。无需 pip 包、网络、账户或 API 密钥。建立一个空目录,把文末五个源码块分别保存为所标注的文件名;也可以使用随稿提供的 project 目录。以下命令在该目录执行。-B 避免生成字节码文件,数据库在内存中创建并在退出时关闭;重跑不会覆盖现有数据库。

python -B -m unittest -v
python -B benchmark.py
python -B benchmark.py --customers 100 --orders-per-customer 30 --repeats 9

默认生成 400 位客户,每 20 位中有一位没有订单,其余每人生成 60 笔,合计 22,800 笔。状态与金额按固定公式生成,因此每次的业务输入相同。脚本只向终端输出 JSON;如需保存结果,可由你运行 python -B benchmark.py > benchmark-result.json。首次检查输出中的 Python、SQLite 与系统信息,再观察三个方案的样本列表。不要把本文机器的毫秒数当作自己的验收阈值。

5. 先给 Codex 一次只读调查

下面是供读者在已有 Codex CLI 环境中执行的步骤,本文没有执行这些命令或登录账户。先进入仅放置本练习文件的目录,运行 codex --version 与 codex --help,确认安装版本支持以下参数,再启动调查会话。依据当前CLI 命令文档与共享参数源码,--sandbox 选择沙箱,-a 控制审批。这里使用只读沙箱且不允许请求升级;不要增加其他可写目录或绕过沙箱。

codex --version
codex --help
codex --sandbox read-only -a never

会话启动后用 /status 核对工作目录、审批策略和可写范围;若实际策略与预期不符,先退出检查配置及组织限制。只读沙箱约束本地执行,并不意味着模型请求离线。本练习可以把合成源码交给 Codex;真实项目仍需先按所在组织要求处理敏感数据。项目里已经提供人工参考候选,因此这一步练的是证据审阅与方案比较。将下面任务发送给 Codex:

调查 report.py 中订单汇总在数据量增长后变慢的原因,暂时不要修改任何文件。
只读本目录 fixture.py、report.py、benchmark.py、benchmark_candidate.py、test_report.py。
允许运行 python -B -m unittest -v 和 python -B benchmark.py;输出只到终端。
不要联网,不安装依赖,不读真实数据库或凭据,不运行其他项目,不扩大权限。
先重述报表业务约束,再比较三种实现的查询次数、执行计划和完整耗时样本。
SQLite 没有逐查询网络往返,不要因为 N+1 或 SELECT 变少就断定更快。
区分实测、代码推断与尚未验证的结论;不要把计划文本写成跨版本固定断言。
给出证据对应的最小修改建议,并说明新增索引的存储和写入成本。
如命令受限,报告限制及尚未验证部分,不请求升级,不改写配置。

6. 怎样读这次实测,而不夸大结果

benchmark.py 先建库、建索引并执行 ANALYZE,再做查询计数与计划采集。这些工作都在计时区间之外。它使用 Python 的set_trace_callback记录实际执行语句,随后明确传入 None 关闭 trace。触发器与事务管理也可能产生回调,本例只统计诊断阶段的 SELECT;计数不能解释成所有应用都适用的数据库调用指标。

每组先预热一次,随后进行七轮测量,每轮按固定随机序列调整方案次序,减轻总是后运行某组带来的偏差。计时涵盖查询与 Python 结果列表物化;结果一致性断言在停止计时后执行。我们保留每个样本、中位数和范围,没有把一次最快值当作最终结论。它仍是同进程、内存数据库的热态微基准,不能代表磁盘冷启动、并发、网络服务或生产端到端延迟。

400 位客户,22,800 笔订单;每组 7 次,内存数据库,计时关闭 trace
这一次测量留下了什么。400 位客户,22,800 笔订单;每组 7 次,内存数据库,计时关闭 trace

本次环境为 Python 3.12.14、SQLite 3.53.1、Linux x86_64。无索引循环的计划出现 SCAN orders;加入索引后出现使用 orders_customer_status_amount 的覆盖索引查找。集合查询还出现了排序临时 B-tree。SQLite 的执行计划文档说明,这类文本用于交互诊断,格式可能随版本变化。因此测试只比较业务结果,不断言某行计划必须逐字一致。

在本轮样本中,加索引后的逐客户方案中位数为 2.239 ms,集合方案为 2.643 ms,两组范围重叠。可以据此优先调查并修复缺失索引;尚不能据此宣布逐客户查询普遍胜过集合查询,更不能承诺生产加速倍数。换数据分布、查询范围或系统负载后,应重新测量。若报告只显示“401 条降为 1 条”,需要要求 Codex 补上同口径耗时与业务回归。

{
  "environment": {
    "python": "3.12.14 (main, Aug 25 2026, 14:00:49) [Clang 22.1.3 ]",
    "sqlite": "3.53.1",
    "platform": "Linux-6.18.44-x86_64-with-glibc2.41",
    "machine": "x86_64",
    "database": ":memory:",
    "trace_during_timing": false
  },
  "input": {
    "customers": 400,
    "orders_per_customer": 60,
    "repeats": 7
  },
  "orders": 22800,
  "results": {
    "loop_no_index": {
      "select_count": 401,
      "plan": [
        [
          3,
          0,
          161,
          "SCAN orders"
        ]
      ],
      "samples_ms": [
        294.25966,
        275.094234,
        275.701774,
        272.337107,
        283.99456,
        289.019753,
        285.6719
      ],
      "median_ms": 283.99456,
      "min_ms": 272.337107,
      "max_ms": 294.25966
    },
    "loop_indexed": {
      "select_count": 401,
      "plan": [
        [
          3,
          0,
          67,
          "SEARCH orders USING COVERING INDEX orders_customer_status_amount (customer_id=? AND status=?)"
        ]
      ],
      "samples_ms": [
        3.523075,
        2.554719,
        2.461691,
        1.862513,
        1.7449,
        1.802661,
        2.23869
      ],
      "median_ms": 2.23869,
      "min_ms": 1.7449,
      "max_ms": 3.523075
    },
    "set_indexed": {
      "select_count": 1,
      "plan": [
        [
          8,
          0,
          102,
          "SCAN c"
        ],
        [
          10,
          0,
          67,
          "SEARCH o USING COVERING INDEX orders_customer_status_amount (customer_id=? AND status=?) LEFT-JOIN"
        ],
        [
          58,
          0,
          0,
          "USE TEMP B-TREE FOR ORDER BY"
        ]
      ],
      "samples_ms": [
        4.042107,
        5.809229,
        2.791674,
        2.514925,
        2.191416,
        2.223559,
        2.64266
      ],
      "median_ms": 2.64266,
      "min_ms": 2.191416,
      "max_ms": 5.809229
    }
  }
}

7. 批准修改时,把能改的文件说清楚

调查结束后,先检查结论再开启修改阶段。在仅含本练习的副本里启动下面的会话,避免把真实项目其他文件一起置于可写工作区。workspace-write 允许工作区写入,并不会把自然语言中的文件清单变成操作系统级白名单;文件范围仍要靠差异审阅与测试核对。若需要严格逐路径隔离,应另行配置并验证权限,不要认为提示词本身已经实现隔离。

codex --sandbox workspace-write -a on-request

用 /status 再次核对目录和权限。新开的会话不会自动得到上一会话的调查内容,先粘贴你已经审阅的调查结论、业务约束和原始样本,再发送下面的修改任务。这里允许创建候选迁移文件,但不授权对生产数据库执行它。实际索引迁移会消耗存储、增加写入维护成本,建索引期间还需考虑锁与业务流量;这些因素不在内存读取基准中。

根据刚才的调查,在此练习副本做最小修复。
只允许修改 report.py、test_report.py,或新增 migration.sql;不得修改 fixture.py、benchmark.py 和 benchmark_candidate.py。
优先保留业务接口与客户 ID 升序;无订单、仅取消、零金额及同名客户行为不能改变。
若选择索引方案,把 CREATE INDEX 写进 migration.sql 并解释在哪个部署阶段应用;不要在每次 summarize 调用里创建索引。
若改成集合查询,解释 paid 条件、COUNT 与 COALESCE 的位置,并证明完整结果一致。
允许运行现有 Python 标准库测试与基准;候选验收必须运行 benchmark_candidate.py,存在已审阅的 migration.sql 时显式加 --migration migration.sql。不得联网、安装依赖、访问真实库或提交发布。
不要删除边界测试、缩小默认输入或给计时设置任意“通过”阈值。
交付文件差异、回归输出、相同输入的前后样本,以及尚未验证的迁移和写入成本。
如果需要修改范围外文件,先停下说明原因。

8. 人工参考修复:先验证索引,再决定是否合并查询

下面是用于对照的人工参考,不代表 Codex 实际产出了该补丁。本例已有 per_customer 与 set_based 两个函数,summarize 仍指向初始循环。最小索引候选如下;它应在迁移阶段执行一次,不能塞进每次请求。在本项目中 benchmark.py 会对第二份合成库应用同样索引,然后验证循环与集合方案。读者采用该迁移时,应确保测试连接也确实加载迁移,避免“文件里写了 SQL,但应用库没有索引”。

CREATE INDEX orders_customer_status_amount
ON orders(customer_id, status, amount_cents);

customer_id 和 status 对应当前查询的等值条件,amount_cents 放入索引是为了让金额汇总有机会从覆盖索引取得数据。适配范围仅限本例;如果真实查询按日期筛选、订单非常偏斜或写入频繁,索引顺序与宽度需要重新评估。先用真实形状的脱敏副本验证,再设计迁移、备份与回滚。回滚该索引通常要删除对应索引对象,但不应在未知生产环境直接照抄执行。

集合版本的关键在 LEFT JOIN 的 ON 子句里保留 paid 条件。将它移到 WHERE 会筛掉没有匹配已付款订单的客户。空连接会产生一行右侧为空的结果,COUNT(*) 会把这行算进去,COUNT(o.id) 则只数真正订单;零金额订单的 ID 仍存在,所以笔数正确。COALESCE 将没有可汇总值的金额变为零。

SELECT c.id, c.name, COUNT(o.id), COALESCE(SUM(o.amount_cents), 0)
FROM customers AS c LEFT JOIN orders AS o
  ON o.customer_id = c.id AND o.status = 'paid'
GROUP BY c.id, c.name
ORDER BY c.id;

若读者最终选择集合方案,summarize 可以改为 return set_based(db),再运行测试与后面的候选验收脚本。本轮数据并没有给出必须采用这一改动的性能理由;代码可读性、后续字段扩展与实际负载也应进入决定。不要只因为它看起来更像“优化后的 SQL”就接收补丁。

9. 实际验收修改后的入口与迁移

前面的 benchmark.py 固定比较三份人工参考方案,不调用 summarize,也不自动加载迁移。它不能证明你刚收到的补丁已经生效。候选验收使用另一份冻结脚本 benchmark_candidate.py:基线在脚本内保留原始循环,候选调用当前 report.py 的 summarize,并逐行比较完整结果。这样修改候选实现不会顺带改掉基线。

先审阅 migration.sql,再按是否存在获准迁移选择下面一条命令。带 --migration 时,脚本只对一次性的内存候选库和边界样本库执行该文件;不带时不应用任何迁移。输入固定为与正文相同的规模,预热与计时分离,trace 关闭,结果校验位于计时后。没有改 summarize 也没有提供迁移时,两边仍是原始循环,结果可以提醒你修复尚未真正应用。

python -B benchmark_candidate.py
# 已审阅索引迁移时,改用:
python -B benchmark_candidate.py --migration migration.sql

这份脚本也不能证明索引上线过程安全。它的用途是确认候选代码和迁移在合成只读负载中真的生效,并检查这次修改有没有改变报表含义。本文另在一次性副本应用人工索引 SQL 验证了该路径,没有运行 Codex。

10. 完整源码:fixture.py

这是全部输入生成逻辑。small 用于人工可核对的边界结果,large 用于固定规模测量。

"""All input is deterministic synthetic data; no production database needed."""
import sqlite3
SCHEMA = """
CREATE TABLE customers(id INTEGER PRIMARY KEY, name TEXT NOT NULL);
CREATE TABLE orders(id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL
REFERENCES customers(id), status TEXT NOT NULL CHECK(status IN ('paid','cancelled')),
amount_cents INTEGER NOT NULL CHECK(amount_cents >= 0));
"""

def connect():
    db = sqlite3.connect(':memory:')
    db.execute('PRAGMA foreign_keys=ON')
    db.executescript(SCHEMA)
    return db

def small():
    db = connect()
    # Duplicate names deliberately do not identify the same customer.
    db.executemany('INSERT INTO customers VALUES (?,?)',
                   [(4,'same'),(2,'same'),(1,'empty'),(3,'cancelled')])
    db.executemany('INSERT INTO orders VALUES (?,?,?,?)',
                   [(1,2,'paid',125),(2,2,'cancelled',999),
                    (3,3,'cancelled',100),(4,4,'paid',0),(5,2,'paid',75)])
    db.commit()
    return db

def large(customers=400, per_customer=60):
    if customers < 1 or per_customer < 1:
        raise ValueError('positive sizes required')
    db = connect()
    db.executemany('INSERT INTO customers VALUES (?,?)',
                   ((i, 'customer-%04d' % i) for i in range(1, customers+1)))
    rows = []
    order_id = 0
    for cid in range(1, customers+1):
        if cid % 20 == 0:  # No orders for 5% of customers.
            continue
        for j in range(per_customer):
            order_id += 1
            rows.append((order_id,cid,'cancelled' if j % 7 == 0 else 'paid',
                         (cid*31+j*17) % 10000))
    db.executemany('INSERT INTO orders VALUES (?,?,?,?)', rows)
    db.commit()
    return db

11. 完整源码:report.py

summarize 保留待调查入口,另外两个函数是公开的人工对照实现。

"""Synthetic order report. Reference alternatives were written manually."""
CUSTOMERS = 'SELECT id, name FROM customers ORDER BY id'
TOTAL = """SELECT COUNT(*), COALESCE(SUM(amount_cents), 0)
FROM orders WHERE customer_id = ? AND status = 'paid'"""
SET_QUERY = """SELECT c.id, c.name, COUNT(o.id), COALESCE(SUM(o.amount_cents), 0)
FROM customers AS c LEFT JOIN orders AS o
ON o.customer_id = c.id AND o.status = 'paid'
GROUP BY c.id, c.name ORDER BY c.id"""
INDEX = 'CREATE INDEX orders_customer_status_amount ON orders(customer_id, status, amount_cents)'

def per_customer(db):
    result = []
    for customer_id, name in db.execute(CUSTOMERS).fetchall():
        count, amount = db.execute(TOTAL, (customer_id,)).fetchone()
        result.append((customer_id, name, count, amount))
    return result

def set_based(db):
    return db.execute(SET_QUERY).fetchall()

def summarize(db):
    # Starting implementation for the Codex exercise.
    return per_customer(db)

12. 完整源码:test_report.py

五个测试方法覆盖完整业务契约、空数据库、生成数据等价性,并单独展示 COUNT(*) 与 WHERE 条件误放会怎样出错。subTest 会在多种索引状态与实现上重复核对;测试方法数量不等于性能覆盖率。

import unittest
from fixture import small, large
from report import per_customer, set_based, summarize, INDEX
EXPECTED = [(1,'empty',0,0),(2,'same',2,200),
            (3,'cancelled',0,0),(4,'same',1,0)]

class ReportTests(unittest.TestCase):
    def test_business_contract(self):
        for indexed in (False, True):
            with self.subTest(indexed=indexed):
                db = small()
                try:
                    if indexed:
                        db.execute(INDEX)
                    for fn in (per_customer, set_based, summarize):
                        with self.subTest(fn=fn.__name__):
                            self.assertEqual(fn(db), EXPECTED)
                finally:
                    db.close()

    def test_empty_database(self):
        from fixture import connect
        db = connect()
        try:
            self.assertEqual(per_customer(db), [])
            self.assertEqual(set_based(db), [])
            self.assertEqual(summarize(db), [])
        finally:
            db.close()

    def test_generated_equivalence(self):
        db = large(41, 13)
        try:
            baseline = per_customer(db)
            self.assertEqual(baseline, summarize(db))
            self.assertEqual(baseline, set_based(db))
            db.execute(INDEX)
            self.assertEqual(baseline, per_customer(db))
            self.assertEqual(baseline, set_based(db))
            self.assertEqual(len(baseline), 41)
            self.assertEqual([row[0] for row in baseline], list(range(1,42)))
        finally:
            db.close()

    def test_count_star_trap(self):
        db = small()
        try:
            wrong = db.execute('''SELECT c.id, COUNT(*) FROM customers c
                LEFT JOIN orders o ON o.customer_id=c.id AND o.status='paid'
                GROUP BY c.id ORDER BY c.id''').fetchall()
            self.assertEqual(wrong[0], (1,1))
            self.assertEqual(set_based(db)[0][2], 0)
        finally:
            db.close()

    def test_where_trap(self):
        db = small()
        try:
            wrong = db.execute('''SELECT c.id FROM customers c LEFT JOIN orders o
                ON o.customer_id=c.id WHERE o.status='paid'
                GROUP BY c.id ORDER BY c.id''').fetchall()
            self.assertEqual(wrong, [(2,), (4,)])
        finally:
            db.close()

if __name__ == '__main__':
    unittest.main()

13. 完整源码:benchmark.py

本文件负责生成输入、分别观测与计时,并输出环境和所有样本。为保持对比口径,修复阶段不要修改它来迎合候选方案。

"""Separately inspect SQL and time full materialization with tracing disabled."""
import argparse
import json
import platform
import random
import sqlite3
import statistics
import sys
import time
from fixture import large
from report import per_customer, set_based, INDEX, TOTAL, SET_QUERY

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--customers', type=int, default=400)
    parser.add_argument('--orders-per-customer', type=int, default=60)
    parser.add_argument('--repeats', type=int, default=7)
    args = parser.parse_args()
    if args.repeats < 3:
        parser.error('use at least 3 repeats')
    plain = large(args.customers, args.orders_per_customer)
    indexed = large(args.customers, args.orders_per_customer)
    # Fixture creation, index creation and ANALYZE are outside the timed region.
    indexed.execute(INDEX)
    indexed.execute('ANALYZE')
    plain.execute('ANALYZE')
    cases = [('loop_no_index',plain,per_customer),
             ('loop_indexed',indexed,per_customer),
             ('set_indexed',indexed,set_based)]
    try:
        expected = per_customer(plain)
        output = {'environment': {'python':sys.version, 'sqlite':sqlite3.sqlite_version,
                  'platform':platform.platform(), 'machine':platform.machine(),
                  'database':':memory:', 'trace_during_timing':False},
                  'input':vars(args), 'orders':plain.execute('SELECT COUNT(*) FROM orders').fetchone()[0],
                  'results':{}}
        for name, db, fn in cases:
            statements = []
            db.set_trace_callback(statements.append)
            try:
                assert fn(db) == expected
            finally:
                db.set_trace_callback(None)
            sql, params = (SET_QUERY, ()) if fn is set_based else (TOTAL, (2,))
            output['results'][name] = {
                'select_count':sum(s.lstrip().upper().startswith('SELECT') for s in statements),
                'plan':db.execute('EXPLAIN QUERY PLAN '+sql, params).fetchall(),
                'samples_ms':[]}
            fn(db)  # One warm-up per variant, not included.
        rng = random.Random(20261007)
        for _ in range(args.repeats):
            sequence = cases.copy()
            rng.shuffle(sequence)
            for name, db, fn in sequence:
                start = time.perf_counter_ns()
                actual = fn(db)
                elapsed = (time.perf_counter_ns()-start)/1e6
                assert actual == expected  # Validation after stopping timer.
                output['results'][name]['samples_ms'].append(elapsed)
        for result in output['results'].values():
            samples = result['samples_ms']
            result.update(median_ms=statistics.median(samples), min_ms=min(samples), max_ms=max(samples))
        print(json.dumps(output,ensure_ascii=False,indent=2))
    finally:
        plain.close()
        indexed.close()

if __name__ == '__main__':
    main()

14. 完整源码:benchmark_candidate.py

请将它与其余四份源码保存在同一目录。修复任务中保持它不变,让候选入口与迁移接受独立验收。

"""Validate the actual summarize candidate against an unchanged baseline."""
import argparse
import json
import platform
from pathlib import Path
import random
import sqlite3
import statistics
import sys
import time
from fixture import large, small
from report import summarize

def original(db):
    # Deliberately independent of editable report.py.
    result = []
    for cid, name in db.execute('SELECT id,name FROM customers ORDER BY id').fetchall():
        count, amount = db.execute('''SELECT COUNT(*),COALESCE(SUM(amount_cents),0)
            FROM orders WHERE customer_id=? AND status='paid' ''', (cid,)).fetchone()
        result.append((cid,name,count,amount))
    return result

def main():
    p = argparse.ArgumentParser()
    p.add_argument('--migration', type=Path)
    p.add_argument('--repeats', type=int, default=7)
    args = p.parse_args()
    if args.repeats < 3:
        p.error('use at least 3 repeats')
    migration = args.migration.read_text() if args.migration else ''
    edge = small()
    base = large()
    candidate = large()
    try:
        # Only these disposable in-memory connections receive the SQL.
        # Review migration.sql before invoking this optional flag.
        edge.executescript(migration)
        assert summarize(edge) == [(1,'empty',0,0),(2,'same',2,200),
                                   (3,'cancelled',0,0),(4,'same',1,0)]
        candidate.executescript(migration)
        candidate.execute('ANALYZE')
        base.execute('ANALYZE')
        expected = original(base)
        assert summarize(candidate) == expected
        cases = [('original',base,original),('candidate',candidate,summarize)]
        samples = {name:[] for name,_,_ in cases}
        for _, db, fn in cases:
            db.set_trace_callback(None)
            fn(db)
        rng = random.Random(20261007)
        for _ in range(args.repeats):
            sequence = cases.copy()
            rng.shuffle(sequence)
            for name, db, fn in sequence:
                start = time.perf_counter_ns()
                actual = fn(db)
                elapsed = (time.perf_counter_ns()-start)/1e6
                assert actual == expected
                samples[name].append(elapsed)
        print(json.dumps({'python':sys.version,'sqlite':sqlite3.sqlite_version,
              'platform':platform.platform(),'customers':400,'orders':22800,
              'migration_applied':bool(migration),'database':':memory:',
              'samples_ms':samples,
              'median_ms':{k:statistics.median(v) for k,v in samples.items()}},indent=2))
    finally:
        edge.close()
        base.close()
        candidate.close()

if __name__ == '__main__':
    main()

15. 接收调查结果前的最后一次检查

先逐行看四个边界客户,再看生成数据是否完整相等。随后确认耗时采集时 trace 已关闭、输入规模未缩小、索引真的应用到了测试库,最后检查 Codex 是否只改了获准文件。把这些证据与补丁一起保存,之后再讨论线上迁移。若模型没有成功运行某条命令,让报告保留“未验证”,不要将建议措辞改写成已经完成的结果。

这个练习完成后,你应能回答三个具体问题:报表时间主要花在何处,候选修改是否保持业务含义,以及现有测量能支持多大的结论。代码会随业务继续变化,但这三项证据仍能帮助你判断下一次优化是否值得接收。

只发一条 SQL 一定更快吗?

不一定。SQLite 在进程内执行;本例索引循环与集合查询需要按同一输入实际测量。

本文是否记录了一次 Codex 成功修复?

没有。源码和修复是人工参考,本地执行验证了 Python 项目;Codex 会话步骤由读者执行。

能把这条索引直接用于生产吗?

先核对真实查询和数据分布,再验证存储、写入及迁移影响;本文只有内存合成库的读取测量。