seekdb 全文检索基准复核

帮助用户判断一次 FTS 改动是否带来可复现的收益。依据实际数据给出结论；缺失的测量或正确性证据标为未验证。

收集最小上下文

获取基线与候选版本的 commit、二进制标识、测试命令、原始 JSON、数据规模、查询集合、预热和采样次数。用户只给一次结果时可以分析瓶颈，但不能据此计算两个版本之间的收益。

记录 CPU、内存、操作系统、文件系统、数据库参数和并发负载。远程租赁实例可能重置；先确认数据库进程、数据目录与挂载点是否还在，再决定是否需要恢复。不要把缺失的数据目录直接解释为数据已永久丢失。

先确定可比性与正确性

固定数据集、查询集合、参数、连接方式和并发数；保留基准脚本原样作为主要证据。
采用基线、候选交替测试，并保存每轮结果。使用预热，报告中位数和范围；把收益小于波动的结果写为证据不足。
既有库查询对比尽量使用同一份未变化的数据及索引；重新建索引后的统计量可能变化，应把这类实验单独报告。
对完整查询结果检查主键、命中数、排序和相关度分数。仅命中数相同不能证明正确。浮点容差若需要，必须在看结果前定义并说明理由；并列项按查询约定比较。
分词行为改动覆盖中英文、混合文本、标点、空串、NULL、停用词及 Unicode 边界；编码、排序规则或 NULL 比较的快路径另做差分验证。
只有性能收益且语义一致的改动才能推荐保留。不能通过减少结果、绕过过滤或改变基准降低耗时。

按测量阶段定位

| 阶段 | 优先观察 | 下一步证据 | | --- | --- | --- | | 建索引 | 总时间、CPU、分配、I/O、索引大小 | 单独建库实验与采样剖析 | | 分词 | 不同解析器的延迟、输出 token | 重复 TOKENIZE 测量与逐项输出对比 | | 查询 | 不同查询形态、LIMIT、长尾、排序 | 同库执行计划、热点采样与完整结果 | | 连接/环境 | SELECT 1、CPU 抢占、远端 RTT | 同位置连接对比、负载及资源记录 |

一次只改变有证据支持的热点。数据库运行目录须验证支持实际文件操作；遇到预分配或同步失败时检查文件系统能力。备份和运行目录分开记录。对运行中的其他服务保留隔离，不因性能测试清空它们的数据。

离线汇总 JSON

需要复用汇总逻辑时，将下方代码保存为 compare_fts.py。只使用 Python 3 标准库，读取两个本地 JSON，不连接数据库或网络。

支持单次 {"metrics": {...}}，或多次 {"runs": [{"metrics": {...}}, ...]}。优先多次测量。已识别的耗时字段必须在所有轮次齐全且为有限正数；命中数字段如出现，也必须逐轮齐全且一致。

运行：python3 compare_fts.py baseline.json candidate.json

import argparse
import json
import math
import statistics
import sys
from pathlib import Path

TIMES = (
    "build_total_sec",
    "tokenize_beng_avg_ms", "tokenize_ik_avg_ms",
    "query_beng_avg_ms", "query_cn_avg_ms",
    "query_limit_avg_ms", "query_mixed_avg_ms",
    "select1_avg_ms",
)

def number(value, label, positive=False):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(label + ": expected a number")
    try:
        finite = math.isfinite(value)
    except OverflowError:
        finite = False
    if not finite or (positive and value <= 0) or (not positive and value < 0):
        raise ValueError(label + ": invalid numeric value")
    return value

def unpack(obj):
    if not isinstance(obj, dict):
        raise ValueError("input must be a JSON object")
    runs = obj.get("runs", [obj])
    if not isinstance(runs, list) or not runs:
        raise ValueError("runs must be a nonempty list")
    result = []
    for run in runs:
        if not isinstance(run, dict) or not isinstance(run.get("metrics"), dict):
            raise ValueError("each run must contain a metrics object")
        result.append(run["metrics"])
    return result

def compare(baseline, candidate):
    old, new = unpack(baseline), unpack(candidate)
    all_runs = old + new
    keys = set().union(*(r.keys() for r in all_runs))
    hit_keys = sorted(k for k in keys if k.endswith("_hits"))
    for key in hit_keys:
        if any(key not in r for r in all_runs):
            raise ValueError(key + ": missing in some runs")
        values = [number(r[key], key) for r in all_runs]
        if any(v != values[0] for v in values[1:]):
            raise ValueError(key + ": hit counts changed")
    rows = []
    for key in TIMES:
        if key not in keys:
            continue
        if any(key not in r for r in all_runs):
            raise ValueError(key + ": missing in some runs")
        a = [number(r[key], key, True) for r in old]
        b = [number(r[key], key, True) for r in new]
        am, bm = statistics.median(a), statistics.median(b)
        rows.append({
            "metric": key,
            "baseline_median": am, "candidate_median": bm,
            "baseline_range": [min(a), max(a)],
            "candidate_range": [min(b), max(b)],
            "time_reduction_fraction": 1 - bm / am,
        })
    if not rows:
        raise ValueError("no recognized timing metrics")
    return {
        "baseline_runs": len(old), "candidate_runs": len(new),
        "single_run_warning": min(len(old), len(new)) < 2,
        "hit_count_status": "consistent" if hit_keys else "not_provided",
        "semantic_correctness": "not_verified_by_timing_report",
        "comparison": rows,
    }

def main():
    parser = argparse.ArgumentParser(description="Compare seekdb FTS timings offline")
    parser.add_argument("baseline", type=Path)
    parser.add_argument("candidate", type=Path)
    args = parser.parse_args()
    try:
        a = json.loads(args.baseline.read_text(encoding="utf-8-sig"))
        b = json.loads(args.candidate.read_text(encoding="utf-8-sig"))
        report = compare(a, b)
    except (OSError, ValueError, RecursionError) as exc:
        print("comparison failed: " + str(exc), file=sys.stderr)
        return 2
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0

if __name__ == "__main__":
    sys.exit(main())

time_reduction_fraction = 1 - candidate / baseline：0.10 表示耗时下降 10%，负数表示变慢；它不是吞吐提升比例。秒和毫秒字段分别比较，不把不同单位的耗时直接相加。

脚本不推断赛事分数或统计显著性。只有已知官方计分公式时才单独计算分数；没有公式时报告各项耗时及收益，明确哪些类别仍有退化。

输出给用户

提供可复跑的命令和版本、各阶段耗时与波动、正确性检查证据、环境差异、建议保留的改动及下一项实验。明确区分本地观测和平台正式结果。

示例请求

“这是两个版本各三轮的 FTS JSON，帮我判断收益是否稳定。”
“分词变快但查询变慢，如何定位并决定是否保留改动？”
“换了租赁服务器后结果变化很大，先检查哪些可比性条件？”