generate_metrics.py

#!/usr/bin/env python3
"""
Generate metrics.hpp (struct definitions) from src/common/metrics.json, and
splice the DUCKDB_METRIC list into the existing metrics_manager.cpp between
the 'internal_metrics[]' array opening and the 'FINAL_METRIC' sentinel.
"""

import json
import os

METRICS_JSON = "src/common/metrics.json"
OUTPUT_HEADER = "src/include/duckdb/main/profiler/metrics.hpp"
OUTPUT_SOURCE = "src/main/profiler/metrics_manager.cpp"

CPP_TYPE_MAP = {
    "double": "double",
    "uint64": "uint64_t",
    "string": "string",
    "map": "Value",
}

# Preserve JSON insertion order so the generated files are deterministic.
GROUP_ORDER = [
    "query",
    "system",
    "io",
    "storage",
    "operator",
    "optimizer",
    "parser",
    "planner",
    "physical_planner",
]


ABBREVIATIONS = {"sql", "cpu", "wal", "io"}


def to_pascal_case(s: str) -> str:
    """'physical_planner.total_time' -> 'MetricPhysicalPlannerTotalTime'
    Known abbreviations (SQL, CPU, WAL, IO) are fully uppercased."""
    parts = s.replace(".", "_").split("_")
    result = "Metric"
    for p in parts:
        result += p.upper() if p.lower() in ABBREVIATIONS else p.capitalize()
    return result


def iter_metrics(data: dict):
    """Yield (group, metric_name, metric_info) in a stable order."""
    for group in GROUP_ORDER:
        if group not in data:
            continue
        for metric_name in sorted(data[group].keys()):
            yield group, metric_name, data[group][metric_name]
    # Any groups not explicitly listed — emit them sorted for stability.
    for group in sorted(data.keys()):
        if group in GROUP_ORDER:
            continue
        for metric_name in sorted(data[group].keys()):
            yield group, metric_name, data[group][metric_name]


def generate_header(data: dict) -> str:
    lines = []
    lines += [
        "//===----------------------------------------------------------------------===//",
        "//",
        "//                         DuckDB",
        "//",
        "// duckdb/main/profiler/metrics.hpp",
        "//",
        "// This file is automatically generated by scripts/generate_metrics.py",
        "// Do not edit this file manually, your changes will be overwritten",
        "//===----------------------------------------------------------------------===//",
        "",
        "#pragma once",
        "",
        '#include "duckdb/common/types/value.hpp"',
        "",
        "namespace duckdb {",
        "",
    ]

    current_group = None
    for group, metric_name, info in iter_metrics(data):
        full_name = f"{group}.{metric_name}"
        struct_name = to_pascal_case(full_name)
        cpp_type = CPP_TYPE_MAP[info["metric_type"]]
        description = info["description"].replace('"', '\\"')
        unit = info.get("unit", "").replace('"', '\\"')

        if group != current_group:
            if current_group is not None:
                lines.append("")
            lines.append(f"// {group.replace('_', ' ').title()} metrics")
            current_group = group

        lines += [
            f"struct {struct_name} {{",
            f"\tusing METRIC_TYPE = {cpp_type};",
            f'\tstatic constexpr const char *Name = "{full_name}";',
            f'\tstatic constexpr const char *Description = "{description}";',
            f'\tstatic constexpr const char *Unit = "{unit}";',
            f'\tstatic constexpr const char *TypeStr = "{info["metric_type"]}";',
            "};",
        ]

    lines += [
        "",
        "} // namespace duckdb",
        "",
    ]
    return "\n".join(lines)


def generate_metric_list(data: dict) -> list:
    """Return the DUCKDB_METRIC(...) lines to splice into metrics_manager.cpp."""
    lines = []
    for group, metric_name, _ in iter_metrics(data):
        struct_name = to_pascal_case(f"{group}.{metric_name}")
        lines.append(f"\tDUCKDB_METRIC({struct_name}),")
    return lines


def update_source(data: dict, source_path: str) -> None:
    """Splice the generated metric list into the existing metrics_manager.cpp.

    Looks for the line containing both 'internal_metrics[]' and '= {' as the
    start marker, then the next line containing 'FINAL_METRIC' as the end
    sentinel.  Replaces every line between those two markers with the newly
    generated DUCKDB_METRIC(...) entries.
    """
    with open(source_path, "r") as fh:
        original = fh.read()

    lines = original.splitlines()

    start_idx = None
    end_idx = None
    for i, line in enumerate(lines):
        if start_idx is None and "internal_metrics[]" in line and "= {" in line:
            start_idx = i + 1  # first line inside the array body
        elif start_idx is not None and "FINAL_METRIC" in line:
            end_idx = i
            break

    if start_idx is None or end_idx is None:
        raise RuntimeError(f"Could not find 'internal_metrics[]' / 'FINAL_METRIC' markers in {source_path}")

    metric_lines = generate_metric_list(data)
    new_lines = lines[:start_idx] + metric_lines + lines[end_idx:]
    result = "\n".join(new_lines)
    if not result.endswith("\n"):
        result += "\n"

    with open(source_path, "w") as fh:
        fh.write(result)


def main() -> None:
    script_dir = os.path.dirname(os.path.abspath(__file__))
    repo_root = os.path.dirname(script_dir)

    with open(os.path.join(repo_root, METRICS_JSON), "r") as fh:
        data = json.load(fh)

    header = generate_header(data)
    with open(os.path.join(repo_root, OUTPUT_HEADER), "w") as fh:
        fh.write(header)

    update_source(data, os.path.join(repo_root, OUTPUT_SOURCE))

    total = sum(len(v) for v in data.values())
    print(f"- Generated {total} static metric(s) into {OUTPUT_HEADER}")
    print(f"- Spliced metric list into {OUTPUT_SOURCE}")
    print("  (optimizer.* metrics are emitted as runtime-generated entries)")


if __name__ == "__main__":
    main()