DuckDB Version Matrix Result Analysis Script

analyze_results.py

#!/usr/bin/env python3
"""Analyze DuckDB multi-version BenchBox results for the version-performance draft.

This script expects the 72-run matrix produced by run_version_matrix.sh:
- Versions:   1.0.0, 1.1.3, 1.2.2, 1.3.2, 1.4.4, 1.5.0.dev311
- Benchmarks: tpch(sf10), tpcds(sf10), clickbench(sf1), ssb(sf10)
- Reps:       3 per (version, benchmark)

It computes median metrics per matrix cell and writes structured artifacts
(JSON + CSV) for table population in version-performance.md.
"""

from __future__ import annotations

import argparse
import csv
import json
import math
import statistics
import sys
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

VERSIONS: list[str] = ["1.0.0", "1.1.3", "1.2.2", "1.3.2", "1.4.4", "1.5.0.dev311"]
VERSION_LABELS: dict[str, str] = {
    "1.0.0": "v1.0.0",
    "1.1.3": "v1.1.3",
    "1.2.2": "v1.2.2",
    "1.3.2": "v1.3.2",
    "1.4.4": "v1.4.4",
    "1.5.0.dev311": "v1.5.0-dev",
}

BENCHMARK_SCALES: dict[str, float] = {
    "tpch": 10.0,
    "tpcds": 10.0,
    "clickbench": 10.0,
    "ssb": 10.0,
}
BENCHMARK_ORDER: list[str] = ["tpch", "tpcds", "clickbench", "ssb"]

TPCH_CATEGORIES: dict[str, list[str]] = {
    "full-scan": ["1", "6"],
    "join-heavy": ["9", "21"],
    "aggregation": ["5", "18"],
    "sorting": ["3", "4", "10", "16"],
    "subquery": ["17", "20"],
}

CLICKBENCH_PATTERNS: dict[str, list[str]] = {
    "COUNT(*)": ["Q1", "Q2", "Q21"],
    "GROUP BY (low cardinality)": [
        "Q8",
        "Q9",
        "Q10",
        "Q11",
        "Q12",
        "Q40",
        "Q42",
        "Q43",
    ],
    "GROUP BY (high cardinality)": [
        "Q13",
        "Q14",
        "Q15",
        "Q16",
        "Q17",
        "Q18",
        "Q19",
        "Q28",
        "Q29",
        "Q31",
        "Q32",
        "Q33",
        "Q34",
        "Q35",
        "Q36",
        "Q37",
        "Q38",
        "Q39",
        "Q41",
    ],
    "String matching (LIKE)": ["Q21", "Q22", "Q23", "Q24", "Q25", "Q26", "Q27"],
    "ORDER BY with LIMIT": [
        "Q9",
        "Q10",
        "Q11",
        "Q12",
        "Q13",
        "Q14",
        "Q15",
        "Q16",
        "Q17",
        "Q19",
        "Q22",
        "Q23",
        "Q24",
        "Q25",
        "Q26",
        "Q27",
        "Q28",
        "Q29",
        "Q31",
        "Q32",
        "Q33",
        "Q34",
        "Q35",
        "Q36",
        "Q37",
        "Q38",
        "Q39",
        "Q40",
        "Q41",
        "Q42",
    ],
}

SORTING_DEEP_DIVE_QUERIES: list[str] = ["3", "4", "10", "16"]
LIMIT_DEEP_DIVE_QUERIES: list[str] = CLICKBENCH_PATTERNS["ORDER BY with LIMIT"]


@dataclass
class RunRecord:
    path: Path
    version: str
    benchmark: str
    scale: float
    power_runtime_ms: float | None
    load_phase_ms: float | None
    tpc_metrics: dict[str, float]
    query_counts: dict[str, int]
    per_query_ms: dict[str, float]
    timestamp: str | None


def normalize_version(raw: Any) -> str:
    value = str(raw or "").strip().lstrip("v")
    return value.replace("-rc", "rc")


_BENCHMARK_ALIASES: dict[str, str] = {
    "star_schema": "ssb",
}


def normalize_benchmark(raw: Any) -> str:
    name = str(raw or "").strip().lower()
    return _BENCHMARK_ALIASES.get(name, name)


def is_close(a: float, b: float, *, tol: float = 1e-9) -> bool:
    return math.isclose(a, b, rel_tol=0.0, abs_tol=tol)


def median(values: list[float]) -> float | None:
    if not values:
        return None
    return float(statistics.median(values))


def pct_change(new: float, old: float) -> float | None:
    if old == 0:
        return None
    return ((new - old) / old) * 100.0


def pct_faster(new_runtime_ms: float, old_runtime_ms: float) -> float | None:
    if old_runtime_ms == 0:
        return None
    return ((old_runtime_ms - new_runtime_ms) / old_runtime_ms) * 100.0


def speedup_factor(old_runtime_ms: float, new_runtime_ms: float) -> float | None:
    if new_runtime_ms == 0:
        return None
    return old_runtime_ms / new_runtime_ms


def normalize_query_id(benchmark: str, raw_query_id: Any) -> str:
    text = str(raw_query_id or "").strip()
    if not text:
        return ""

    if benchmark == "clickbench":
        if text.upper().startswith("Q"):
            return text.upper()
        return f"Q{text}"

    if benchmark in {"tpch", "tpcds"}:
        upper = text.upper()
        if upper.startswith("Q"):
            upper = upper[1:]
        return upper

    return text.upper()


def extract_power_runtime_ms(payload: dict[str, Any]) -> float | None:
    summary_timing = payload.get("summary", {}).get("timing", {})
    if isinstance(summary_timing.get("total_ms"), (int, float)):
        return float(summary_timing["total_ms"])

    run = payload.get("run", {})
    if isinstance(run.get("query_time_ms"), (int, float)):
        return float(run["query_time_ms"])

    return None


def extract_load_phase_ms(payload: dict[str, Any]) -> float | None:
    phases = payload.get("phases", {})
    data_loading = phases.get("data_loading", {})
    if isinstance(data_loading.get("duration_ms"), (int, float)):
        return float(data_loading["duration_ms"])

    summary_data = payload.get("summary", {}).get("data", {})
    if isinstance(summary_data.get("load_time_ms"), (int, float)):
        return float(summary_data["load_time_ms"])

    return None


def extract_tpc_metrics(payload: dict[str, Any]) -> dict[str, float]:
    metrics = payload.get("summary", {}).get("tpc_metrics", {})
    output: dict[str, float] = {}
    if not isinstance(metrics, dict):
        return output

    for key, value in metrics.items():
        if isinstance(value, (int, float)):
            output[str(key)] = float(value)
    return output


def extract_query_counts(payload: dict[str, Any]) -> dict[str, int]:
    summary_queries = payload.get("summary", {}).get("queries", {})

    total = summary_queries.get("total")
    passed = summary_queries.get("passed")
    failed = summary_queries.get("failed")

    timeout = summary_queries.get("timeout")
    if timeout is None:
        timeout = summary_queries.get("timed_out")

    statuses: list[str] = []
    for item in payload.get("queries", []):
        status = item.get("status")
        if status is not None:
            statuses.append(str(status).strip().upper())

    if timeout is None:
        timeout = sum(1 for status in statuses if "TIMEOUT" in status)

    if total is None:
        total = len(statuses)
    if passed is None:
        passed = sum(
            1
            for status in statuses
            if status in {"SUCCESS", "PASSED", "COMPLETED", "OK"}
        )
    if failed is None:
        failed = max(int(total) - int(passed) - int(timeout), 0)

    return {
        "total": int(total or 0),
        "passed": int(passed or 0),
        "failed": int(failed or 0),
        "timeout": int(timeout or 0),
    }


def extract_per_query_ms(payload: dict[str, Any], benchmark: str) -> dict[str, float]:
    grouped: dict[str, list[float]] = defaultdict(list)

    for item in payload.get("queries", []):
        query_id = normalize_query_id(benchmark, item.get("id"))
        if not query_id:
            continue

        ms_value = item.get("ms")
        if not isinstance(ms_value, (int, float)):
            continue

        run_type = str(item.get("run_type", "")).strip().lower()
        if run_type and run_type != "measurement":
            continue

        status = str(item.get("status", "")).strip().upper()
        if status and status not in {"SUCCESS", "PASSED", "COMPLETED", "OK"}:
            continue

        grouped[query_id].append(float(ms_value))

    output: dict[str, float] = {}
    for query_id, values in grouped.items():
        med = median(values)
        if med is not None:
            output[query_id] = med
    return output


def pick_primary_tpc_metric(
    benchmark: str, tpc_metrics: dict[str, float]
) -> tuple[str, float] | None:
    if not tpc_metrics:
        return None

    preferred_keys = {
        "tpch": ["qpph", "power_at_size"],
        "tpcds": ["qppds", "power_at_size"],
    }.get(benchmark, [])

    lowercase_lookup = {key.lower(): key for key in tpc_metrics}
    for candidate in preferred_keys:
        if candidate in lowercase_lookup:
            original_key = lowercase_lookup[candidate]
            return original_key, tpc_metrics[original_key]

    first_key = sorted(tpc_metrics.keys())[0]
    return first_key, tpc_metrics[first_key]


def parse_result_file(path: Path) -> tuple[RunRecord | None, str | None]:
    try:
        with path.open("r", encoding="utf-8") as fh:
            payload = json.load(fh)
    except Exception as exc:  # noqa: BLE001 - keep script robust over partial files.
        return None, f"{path.name}: JSON parse error: {exc}"

    benchmark = normalize_benchmark(payload.get("benchmark", {}).get("id"))
    if benchmark not in BENCHMARK_SCALES:
        return None, None

    scale_value = payload.get("benchmark", {}).get("scale_factor")
    if not isinstance(scale_value, (int, float)):
        return None, f"{path.name}: missing numeric benchmark.scale_factor"
    scale_float = float(scale_value)

    expected_scale = BENCHMARK_SCALES[benchmark]
    if not is_close(scale_float, expected_scale):
        return None, None

    version = normalize_version(payload.get("platform", {}).get("client_version"))
    if version not in VERSIONS:
        return None, None

    config = payload.get("config", {})
    phases = [
        str(phase).strip().lower()
        for phase in config.get("phases", [])
        if phase is not None
    ]
    if phases:
        if not ({"load", "power"} & set(phases)):
            return None, None

    query_subset = config.get("query_subset")
    if isinstance(query_subset, list) and query_subset:
        return None, None

    record = RunRecord(
        path=path,
        version=version,
        benchmark=benchmark,
        scale=scale_float,
        power_runtime_ms=extract_power_runtime_ms(payload),
        load_phase_ms=extract_load_phase_ms(payload),
        tpc_metrics=extract_tpc_metrics(payload),
        query_counts=extract_query_counts(payload),
        per_query_ms=extract_per_query_ms(payload, benchmark),
        timestamp=payload.get("run", {}).get("timestamp"),
    )

    return record, None


def aggregate_records(records: list[RunRecord]) -> dict[str, Any]:
    power_values = [
        rec.power_runtime_ms for rec in records if rec.power_runtime_ms is not None
    ]
    load_values = [
        rec.load_phase_ms for rec in records if rec.load_phase_ms is not None
    ]

    query_count_values: dict[str, list[float]] = defaultdict(list)
    for rec in records:
        for key, value in rec.query_counts.items():
            query_count_values[key].append(float(value))

    per_query_values: dict[str, list[float]] = defaultdict(list)
    for rec in records:
        for query_id, value in rec.per_query_ms.items():
            per_query_values[query_id].append(value)

    tpc_values: dict[str, list[float]] = defaultdict(list)
    for rec in records:
        for key, value in rec.tpc_metrics.items():
            tpc_values[key].append(value)

    aggregated_query_counts: dict[str, int] = {}
    for key, values in query_count_values.items():
        med = median(values)
        aggregated_query_counts[key] = int(round(med or 0.0))

    aggregated_per_query: dict[str, float] = {}
    for query_id, values in per_query_values.items():
        med = median(values)
        if med is not None:
            aggregated_per_query[query_id] = med

    aggregated_tpc: dict[str, float] = {}
    for key, values in tpc_values.items():
        med = median(values)
        if med is not None:
            aggregated_tpc[key] = med

    return {
        "repetitions": len(records),
        "power_runtime_ms": median(power_values),
        "load_phase_ms": median(load_values),
        "query_counts": aggregated_query_counts,
        "per_query_ms": aggregated_per_query,
        "tpc_metrics": aggregated_tpc,
        "source_files": [rec.path.name for rec in records],
    }


def build_matrix(records: list[RunRecord]) -> dict[str, dict[str, dict[str, Any]]]:
    grouped: dict[tuple[str, str], list[RunRecord]] = defaultdict(list)
    for rec in records:
        grouped[(rec.version, rec.benchmark)].append(rec)

    matrix: dict[str, dict[str, dict[str, Any]]] = {
        benchmark: {} for benchmark in BENCHMARK_ORDER
    }

    for benchmark in BENCHMARK_ORDER:
        for version in VERSIONS:
            recs = grouped.get((version, benchmark), [])
            if not recs:
                continue
            matrix[benchmark][version] = aggregate_records(recs)

    return matrix


def runtime_trends(
    matrix: dict[str, dict[str, dict[str, Any]]],
) -> dict[str, list[dict[str, Any]]]:
    trends: dict[str, list[dict[str, Any]]] = {}

    for benchmark in BENCHMARK_ORDER:
        rows: list[dict[str, Any]] = []
        benchmark_rows = matrix.get(benchmark, {})

        baseline_runtime: float | None = None
        baseline_tpc_metric: float | None = None
        baseline_tpc_key: str | None = None

        prev_runtime: float | None = None
        prev_tpc_metric: float | None = None

        for version in VERSIONS:
            cell = benchmark_rows.get(version)
            if not cell:
                continue

            runtime = cell.get("power_runtime_ms")
            tpc_metric_info = pick_primary_tpc_metric(
                benchmark, cell.get("tpc_metrics", {})
            )

            row: dict[str, Any] = {
                "version": version,
                "version_label": VERSION_LABELS.get(version, f"v{version}"),
                "repetitions": cell.get("repetitions", 0),
                "power_runtime_ms": runtime,
                "load_phase_ms": cell.get("load_phase_ms"),
                "query_counts": cell.get("query_counts", {}),
                "power_runtime_vs_baseline_pct_faster": None,
                "power_runtime_vs_previous_pct_faster": None,
                "power_runtime_speedup_vs_baseline": None,
                "tpc_metric_key": None,
                "tpc_metric": None,
                "tpc_metric_vs_baseline_pct": None,
                "tpc_metric_vs_previous_pct": None,
            }

            if isinstance(runtime, (int, float)):
                runtime = float(runtime)
                if baseline_runtime is None:
                    baseline_runtime = runtime
                else:
                    row["power_runtime_vs_baseline_pct_faster"] = pct_faster(
                        runtime, baseline_runtime
                    )
                    row["power_runtime_speedup_vs_baseline"] = speedup_factor(
                        baseline_runtime, runtime
                    )

                if prev_runtime is not None:
                    row["power_runtime_vs_previous_pct_faster"] = pct_faster(
                        runtime, prev_runtime
                    )

                prev_runtime = runtime

            if tpc_metric_info is not None:
                metric_key, metric_value = tpc_metric_info
                row["tpc_metric_key"] = metric_key
                row["tpc_metric"] = metric_value

                if baseline_tpc_metric is None:
                    baseline_tpc_metric = metric_value
                    baseline_tpc_key = metric_key
                else:
                    row["tpc_metric_vs_baseline_pct"] = pct_change(
                        metric_value, baseline_tpc_metric
                    )

                if prev_tpc_metric is not None:
                    row["tpc_metric_vs_previous_pct"] = pct_change(
                        metric_value, prev_tpc_metric
                    )

                prev_tpc_metric = metric_value

                if baseline_tpc_key is not None and metric_key != baseline_tpc_key:
                    row["tpc_metric_key_warning"] = (
                        f"metric key changed from {baseline_tpc_key} to {metric_key}; review comparability"
                    )

            rows.append(row)

        trends[benchmark] = rows

    return trends


def query_winners_and_regressions(
    matrix: dict[str, dict[str, dict[str, Any]]],
    *,
    baseline_version: str,
    latest_version: str,
    regression_threshold_pct: float = 5.0,
) -> dict[str, dict[str, Any]]:
    output: dict[str, dict[str, Any]] = {}

    for benchmark in BENCHMARK_ORDER:
        bench_data = matrix.get(benchmark, {})
        baseline = bench_data.get(baseline_version, {})
        latest = bench_data.get(latest_version, {})

        baseline_queries = baseline.get("per_query_ms", {})
        latest_queries = latest.get("per_query_ms", {})

        winners: list[dict[str, Any]] = []
        minimal_change: list[dict[str, Any]] = []

        for query_id in sorted(set(baseline_queries) & set(latest_queries)):
            base = baseline_queries[query_id]
            curr = latest_queries[query_id]
            if base <= 0 or curr <= 0:
                continue

            improvement_pct = pct_faster(curr, base)
            speedup = speedup_factor(base, curr)
            if improvement_pct is None or speedup is None:
                continue

            entry = {
                "query_id": query_id,
                "baseline_ms": base,
                "latest_ms": curr,
                "improvement_pct": improvement_pct,
                "speedup_factor": speedup,
            }

            if improvement_pct >= 0:
                winners.append(entry)
            minimal_change.append(entry)

        winners.sort(key=lambda item: item["speedup_factor"], reverse=True)
        minimal_change.sort(key=lambda item: abs(item["improvement_pct"]))

        regressions: list[dict[str, Any]] = []
        for index in range(1, len(VERSIONS)):
            older_version = VERSIONS[index - 1]
            newer_version = VERSIONS[index]
            older_cell = bench_data.get(older_version, {})
            newer_cell = bench_data.get(newer_version, {})
            older_queries = older_cell.get("per_query_ms", {})
            newer_queries = newer_cell.get("per_query_ms", {})

            for query_id in sorted(set(older_queries) & set(newer_queries)):
                older_ms = older_queries[query_id]
                newer_ms = newer_queries[query_id]
                if older_ms <= 0:
                    continue

                slower_pct = ((newer_ms - older_ms) / older_ms) * 100.0
                if slower_pct <= regression_threshold_pct:
                    continue

                regressions.append(
                    {
                        "query_id": query_id,
                        "from_version": older_version,
                        "to_version": newer_version,
                        "older_ms": older_ms,
                        "newer_ms": newer_ms,
                        "slower_pct": slower_pct,
                    }
                )

        regressions.sort(key=lambda item: item["slower_pct"], reverse=True)

        output[benchmark] = {
            "winners": winners,
            "minimal_change": minimal_change,
            "regressions": regressions,
        }

    return output


def tpch_category_breakdown(
    matrix: dict[str, dict[str, dict[str, Any]]],
) -> list[dict[str, Any]]:
    bench_data = matrix.get("tpch", {})
    baseline_queries = bench_data.get("1.0.0", {}).get("per_query_ms", {})
    latest_queries = bench_data.get("1.5.0.dev311", {}).get("per_query_ms", {})

    rows: list[dict[str, Any]] = []
    for category, queries in TPCH_CATEGORIES.items():
        baseline_values = [
            baseline_queries[q] for q in queries if q in baseline_queries
        ]
        latest_values = [latest_queries[q] for q in queries if q in latest_queries]

        baseline_mean = statistics.mean(baseline_values) if baseline_values else None
        latest_mean = statistics.mean(latest_values) if latest_values else None

        improvement = None
        if baseline_mean is not None and latest_mean is not None:
            improvement = pct_faster(latest_mean, baseline_mean)

        rows.append(
            {
                "category": category,
                "queries": queries,
                "baseline_mean_ms": baseline_mean,
                "latest_mean_ms": latest_mean,
                "improvement_pct": improvement,
            }
        )

    return rows


def clickbench_pattern_analysis(
    matrix: dict[str, dict[str, dict[str, Any]]],
) -> list[dict[str, Any]]:
    bench_data = matrix.get("clickbench", {})
    baseline_queries = bench_data.get("1.0.0", {}).get("per_query_ms", {})
    latest_queries = bench_data.get("1.5.0.dev311", {}).get("per_query_ms", {})

    rows: list[dict[str, Any]] = []
    for pattern_name, query_ids in CLICKBENCH_PATTERNS.items():
        baseline_values = [
            baseline_queries[q] for q in query_ids if q in baseline_queries
        ]
        latest_values = [latest_queries[q] for q in query_ids if q in latest_queries]

        baseline_mean = statistics.mean(baseline_values) if baseline_values else None
        latest_mean = statistics.mean(latest_values) if latest_values else None

        improvement = None
        if baseline_mean is not None and latest_mean is not None:
            improvement = pct_faster(latest_mean, baseline_mean)

        rows.append(
            {
                "pattern": pattern_name,
                "query_ids": query_ids,
                "baseline_mean_ms": baseline_mean,
                "latest_mean_ms": latest_mean,
                "improvement_pct": improvement,
            }
        )

    return rows


def deep_dive_csv_import(
    matrix: dict[str, dict[str, dict[str, Any]]],
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    baseline_load_ms = matrix.get("tpch", {}).get("1.0.0", {}).get("load_phase_ms")
    dataset_size_gb = (
        10.0  # Proxy: TPC-H SF10 dataset size for relative throughput comparison.
    )

    for version in VERSIONS:
        cell = matrix.get("tpch", {}).get(version, {})
        load_ms = cell.get("load_phase_ms")
        if load_ms is None:
            rows.append(
                {
                    "version": version,
                    "version_label": VERSION_LABELS[version],
                    "load_phase_ms": None,
                    "throughput_gb_per_s": None,
                    "vs_baseline_pct_faster": None,
                }
            )
            continue

        throughput = dataset_size_gb / (load_ms / 1000.0) if load_ms > 0 else None
        improvement = None
        if baseline_load_ms is not None:
            improvement = pct_faster(load_ms, baseline_load_ms)

        rows.append(
            {
                "version": version,
                "version_label": VERSION_LABELS[version],
                "load_phase_ms": load_ms,
                "throughput_gb_per_s": throughput,
                "vs_baseline_pct_faster": improvement,
            }
        )

    return rows


def per_query_subset_trend(
    matrix: dict[str, dict[str, dict[str, Any]]],
    *,
    benchmark: str,
    query_ids: list[str],
    baseline_version: str,
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []

    baseline_cell = matrix.get(benchmark, {}).get(baseline_version, {})
    baseline_queries = baseline_cell.get("per_query_ms", {})
    baseline_values = [baseline_queries[q] for q in query_ids if q in baseline_queries]
    baseline_mean = statistics.mean(baseline_values) if baseline_values else None

    for version in VERSIONS:
        cell = matrix.get(benchmark, {}).get(version, {})
        per_query = cell.get("per_query_ms", {})
        values = [per_query[q] for q in query_ids if q in per_query]

        mean_ms = statistics.mean(values) if values else None
        speedup = None
        pct_vs_baseline = None
        if baseline_mean is not None and mean_ms is not None:
            speedup = speedup_factor(baseline_mean, mean_ms)
            pct_vs_baseline = pct_faster(mean_ms, baseline_mean)

        rows.append(
            {
                "version": version,
                "version_label": VERSION_LABELS[version],
                "query_count": len(values),
                "mean_ms": mean_ms,
                "speedup_vs_baseline": speedup,
                "vs_baseline_pct_faster": pct_vs_baseline,
            }
        )

    return rows


def diminishing_returns(
    trends: dict[str, list[dict[str, Any]]],
) -> list[dict[str, Any]]:
    tpch_rows = trends.get("tpch", [])
    results: list[dict[str, Any]] = []

    for index in range(1, len(tpch_rows)):
        older = tpch_rows[index - 1]
        newer = tpch_rows[index]

        entry: dict[str, Any] = {
            "from_version": older["version"],
            "to_version": newer["version"],
            "jump_label": f"{older['version']} -> {newer['version']}",
            "qpph_improvement_pct": None,
            "runtime_improvement_pct_faster": None,
        }

        if older.get("tpc_metric") is not None and newer.get("tpc_metric") is not None:
            entry["qpph_improvement_pct"] = pct_change(
                newer["tpc_metric"], older["tpc_metric"]
            )

        if (
            older.get("power_runtime_ms") is not None
            and newer.get("power_runtime_ms") is not None
        ):
            entry["runtime_improvement_pct_faster"] = pct_faster(
                newer["power_runtime_ms"], older["power_runtime_ms"]
            )

        results.append(entry)

    return results


def write_summary_csv(path: Path, matrix: dict[str, dict[str, dict[str, Any]]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)

    with path.open("w", encoding="utf-8", newline="") as fh:
        writer = csv.writer(fh)
        writer.writerow(
            [
                "benchmark",
                "version",
                "version_label",
                "repetitions",
                "power_runtime_ms",
                "load_phase_ms",
                "tpc_metric_key",
                "tpc_metric",
                "queries_total",
                "queries_passed",
                "queries_failed",
                "queries_timeout",
            ]
        )

        for benchmark in BENCHMARK_ORDER:
            bench_rows = matrix.get(benchmark, {})
            for version in VERSIONS:
                cell = bench_rows.get(version)
                if not cell:
                    continue

                tpc_metric_info = pick_primary_tpc_metric(
                    benchmark, cell.get("tpc_metrics", {})
                )
                tpc_metric_key = tpc_metric_info[0] if tpc_metric_info else ""
                tpc_metric_value = tpc_metric_info[1] if tpc_metric_info else ""

                counts = cell.get("query_counts", {})
                writer.writerow(
                    [
                        benchmark,
                        version,
                        VERSION_LABELS.get(version, f"v{version}"),
                        cell.get("repetitions"),
                        cell.get("power_runtime_ms"),
                        cell.get("load_phase_ms"),
                        tpc_metric_key,
                        tpc_metric_value,
                        counts.get("total", ""),
                        counts.get("passed", ""),
                        counts.get("failed", ""),
                        counts.get("timeout", ""),
                    ]
                )


def coverage_summary(matrix: dict[str, dict[str, dict[str, Any]]]) -> dict[str, Any]:
    expected_cells: list[tuple[str, str]] = [
        (benchmark, version) for benchmark in BENCHMARK_ORDER for version in VERSIONS
    ]

    present_cells: list[tuple[str, str]] = []
    for benchmark in BENCHMARK_ORDER:
        for version in VERSIONS:
            if version in matrix.get(benchmark, {}):
                present_cells.append((benchmark, version))

    missing_cells = [cell for cell in expected_cells if cell not in present_cells]

    per_benchmark_reps: dict[str, dict[str, int]] = {}
    for benchmark in BENCHMARK_ORDER:
        per_benchmark_reps[benchmark] = {}
        for version in VERSIONS:
            cell = matrix.get(benchmark, {}).get(version)
            if cell:
                per_benchmark_reps[benchmark][version] = int(cell.get("repetitions", 0))

    return {
        "expected_cells": len(expected_cells),
        "present_cells": len(present_cells),
        "missing_cells": [
            {
                "benchmark": benchmark,
                "version": version,
                "version_label": VERSION_LABELS.get(version, f"v{version}"),
            }
            for benchmark, version in missing_cells
        ],
        "repetitions": per_benchmark_reps,
    }


def print_human_summary(
    *,
    records: list[RunRecord],
    rejected_records: list[str],
    matrix: dict[str, dict[str, dict[str, Any]]],
    trends: dict[str, list[dict[str, Any]]],
    winners_and_regressions: dict[str, dict[str, Any]],
    coverage: dict[str, Any],
    output_json: Path,
    output_csv: Path,
) -> None:
    print("DuckDB Version Matrix Analysis")
    print("=" * 80)
    print(f"Accepted result files: {len(records)}")
    print(f"Rejected result files (parse/validation): {len(rejected_records)}")
    print(
        f"Matrix coverage: {coverage['present_cells']}/{coverage['expected_cells']} "
        "(benchmark/version cells)"
    )

    if coverage["missing_cells"]:
        print("Missing cells:")
        for missing in coverage["missing_cells"]:
            print(f"  - {missing['benchmark']} {missing['version_label']}")

    print("")
    for benchmark in BENCHMARK_ORDER:
        rows = trends.get(benchmark, [])
        if not rows:
            print(f"{benchmark}: no matrix rows")
            continue

        print(f"{benchmark}:")
        for row in rows:
            runtime = row.get("power_runtime_ms")
            speedup = row.get("power_runtime_speedup_vs_baseline")
            rep_count = row.get("repetitions")

            runtime_text = "n/a" if runtime is None else f"{runtime:,.2f} ms"
            if speedup is None:
                speedup_text = "baseline"
            else:
                speedup_text = f"{speedup:.3f}x vs baseline"

            metric_text = ""
            if row.get("tpc_metric") is not None:
                metric_text = f", {row['tpc_metric_key']}={row['tpc_metric']:,.4f}"

            print(
                f"  - {row['version_label']}: runtime={runtime_text}, reps={rep_count}, "
                f"{speedup_text}{metric_text}"
            )
        print("")

    tpch_winners = winners_and_regressions.get("tpch", {}).get("winners", [])
    tpch_regressions = winners_and_regressions.get("tpch", {}).get("regressions", [])

    print(f"TPC-H Top Winners (v1.0.0 -> {VERSION_LABELS.get(VERSIONS[-1], VERSIONS[-1])}):")
    if not tpch_winners:
        print("  - none")
    else:
        for entry in tpch_winners[:5]:
            print(
                f"  - Q{entry['query_id']}: {entry['baseline_ms']:.2f} -> {entry['latest_ms']:.2f} ms "
                f"({entry['speedup_factor']:.2f}x, {entry['improvement_pct']:.2f}% faster)"
            )

    print("TPC-H Regressions (>5% slower on adjacent version jumps):")
    if not tpch_regressions:
        print("  - none detected")
    else:
        for entry in tpch_regressions[:10]:
            print(
                f"  - Q{entry['query_id']} {entry['from_version']} -> {entry['to_version']}: "
                f"{entry['slower_pct']:.2f}% slower"
            )

    print("")
    print(f"JSON output: {output_json}")
    print(f"CSV output:  {output_csv}")


def parse_args(argv: list[str]) -> argparse.Namespace:
    script_dir = Path(__file__).resolve().parent

    parser = argparse.ArgumentParser(
        description="Analyze DuckDB multi-version benchmark results and compute medians/derived metrics."
    )
    parser.add_argument(
        "--manifest",
        type=Path,
        default=None,
        help="Path to a manifest file (one result JSON path per line) produced by run_version_matrix.sh. "
        "When provided, only those files are analyzed. Mutually exclusive with --results-dir scanning.",
    )
    parser.add_argument(
        "--results-dir",
        type=Path,
        default=Path("benchmark_runs/results"),
        help="Directory containing BenchBox JSON result files (default: benchmark_runs/results). "
        "Ignored when --manifest is provided.",
    )
    parser.add_argument(
        "--output-json",
        type=Path,
        default=script_dir / "version_matrix_metrics.json",
        help="Output path for structured JSON metrics.",
    )
    parser.add_argument(
        "--output-csv",
        type=Path,
        default=script_dir / "version_matrix_summary.csv",
        help="Output path for summary CSV.",
    )
    parser.add_argument(
        "--strict",
        action="store_true",
        help="Exit non-zero if any benchmark/version cell is missing or has fewer than 3 reps.",
    )

    return parser.parse_args(argv)


def main(argv: list[str]) -> int:
    args = parse_args(argv)

    records: list[RunRecord] = []
    rejected_records: list[str] = []

    if args.manifest is not None:
        if not args.manifest.exists():
            print(
                f"ERROR: manifest file does not exist: {args.manifest}", file=sys.stderr
            )
            return 2
        paths: list[Path] = [
            Path(line.strip())
            for line in args.manifest.read_text(encoding="utf-8").splitlines()
            if line.strip()
        ]
    else:
        if not args.results_dir.exists():
            print(
                f"ERROR: results directory does not exist: {args.results_dir}",
                file=sys.stderr,
            )
            return 2
        paths = sorted(args.results_dir.glob("*.json"))

    for path in paths:
        record, error = parse_result_file(path)
        if error:
            rejected_records.append(error)
            continue
        if record is not None:
            records.append(record)

    matrix = build_matrix(records)
    coverage = coverage_summary(matrix)
    trends = runtime_trends(matrix)
    winners_and_regressions = query_winners_and_regressions(
        matrix,
        baseline_version="1.0.0",
        latest_version="1.5.0.dev311",
        regression_threshold_pct=5.0,
    )

    derived = {
        "runtime_trends": trends,
        "tpch_category_breakdown": tpch_category_breakdown(matrix),
        "clickbench_pattern_analysis": clickbench_pattern_analysis(matrix),
        "query_winners_and_regressions": winners_and_regressions,
        "deep_dives": {
            "csv_import_proxy": deep_dive_csv_import(matrix),
            "limit_orderby_clickbench_proxy": per_query_subset_trend(
                matrix,
                benchmark="clickbench",
                query_ids=LIMIT_DEEP_DIVE_QUERIES,
                baseline_version="1.2.2",
            ),
            "sorting_tpch_proxy": per_query_subset_trend(
                matrix,
                benchmark="tpch",
                query_ids=SORTING_DEEP_DIVE_QUERIES,
                baseline_version="1.3.2",
            ),
        },
        "diminishing_returns": diminishing_returns(trends),
    }

    output_payload = {
        "metadata": {
            "generated_at_utc": datetime.now(timezone.utc).isoformat(),
            "script": Path(__file__).name,
            "versions": VERSIONS,
            "version_labels": VERSION_LABELS,
            "benchmarks": BENCHMARK_ORDER,
            "benchmark_scales": BENCHMARK_SCALES,
            "notes": [
                "All aggregate metrics use median across available repetitions.",
                "Load-phase CSV import metrics are a proxy from TPC-H load timings.",
                "Per-query regressions use >5% slower threshold on adjacent versions.",
            ],
        },
        "coverage": coverage,
        "matrix": matrix,
        "derived": derived,
        "rejected_files": rejected_records,
    }

    args.output_json.parent.mkdir(parents=True, exist_ok=True)
    with args.output_json.open("w", encoding="utf-8") as fh:
        json.dump(output_payload, fh, indent=2, sort_keys=True)
        fh.write("\n")

    write_summary_csv(args.output_csv, matrix)

    print_human_summary(
        records=records,
        rejected_records=rejected_records,
        matrix=matrix,
        trends=trends,
        winners_and_regressions=winners_and_regressions,
        coverage=coverage,
        output_json=args.output_json,
        output_csv=args.output_csv,
    )

    if args.strict:
        if coverage["missing_cells"]:
            print(
                "STRICT CHECK FAILED: matrix coverage is incomplete.", file=sys.stderr
            )
            return 1

        for benchmark in BENCHMARK_ORDER:
            for version in VERSIONS:
                reps = matrix.get(benchmark, {}).get(version, {}).get("repetitions")
                if reps is None:
                    continue
                if reps < 3:
                    print(
                        f"STRICT CHECK FAILED: {benchmark} {version} has {reps} rep(s), expected >=3.",
                        file=sys.stderr,
                    )
                    return 1

    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))