feat: add query regression perf harness (#8406)

* feat: add query regression perf harness Signed-off-by: discord9 <discord9@163.com> * feat: extend query regression cases Signed-off-by: discord9 <discord9@163.com> * ci: harden query regression workflows Signed-off-by: discord9 <discord9@163.com> * fix: address query regression review comments Signed-off-by: discord9 <discord9@163.com> * ci: limit query regression PR triggers Signed-off-by: discord9 <discord9@163.com> * ci: run full query regression case set Signed-off-by: discord9 <discord9@163.com> * refactor: model query regression scenarios Signed-off-by: discord9 <discord9@163.com> * fix: avoid unenforced query regression thresholds Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com>
2026-07-06 14:00:40 +00:00 · 2026-07-03 17:09:01 +08:00
parent 424e400c80
commit c44f8da646
18 changed files with 2967 additions and 0 deletions
--- a/.github/scripts/query-regression-pr-metadata.py
+++ b/.github/scripts/query-regression-pr-metadata.py
@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Write metadata consumed by the trusted query-regression comment workflow."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", default=Path("query-regression-pr.json"), type=Path)
+    args = parser.parse_args()
+
+    metadata = {
+        "pr_number": int(os.environ["PR_NUMBER"]),
+        "head_sha": os.environ["HEAD_SHA"],
+        "head_repo": os.environ["HEAD_REPO"],
+        "base_repo": os.environ["BASE_REPO"],
+        "run_id": int(os.environ["RUN_ID"]),
+        "run_attempt": int(os.environ["RUN_ATTEMPT"]),
+    }
+    args.output.write_text(json.dumps(metadata, sort_keys=True) + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/.github/scripts/query-regression-run.py
+++ b/.github/scripts/query-regression-run.py
@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Run one or more query regression cases after the binaries are built."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import subprocess
+from pathlib import Path
+
+
+DEFAULT_CASES = [
+    "tests/perf/query_cases/smoke_direct_sst/case.toml",
+    "tests/perf/query_cases/promql_pushdown_7913/case.toml",
+    "tests/perf/query_cases/sql_topk_order_by/case.toml",
+    "tests/perf/query_cases/sql_aggregate_order_by/case.toml",
+    "tests/perf/query_cases/sql_join_filter_order/case.toml",
+]
+
+
+def split_cases(values: list[str]) -> list[str]:
+    tokens: list[str] = []
+    for value in values:
+        tokens.extend(part for part in re.split(r"[\s,]+", value.strip()) if part)
+    if not tokens or tokens == ["all"]:
+        return DEFAULT_CASES.copy()
+    if "all" in tokens:
+        raise ValueError("'all' cannot be mixed with explicit case paths")
+    return list(dict.fromkeys(tokens))
+
+
+def parse_bool(value: str) -> bool:
+    return value.lower() in {"1", "true", "yes", "on"}
+
+
+def profile_dir(cargo_profile: str) -> str:
+    if cargo_profile == "dev":
+        return "debug"
+    return cargo_profile
+
+
+def resolve_case_path(candidate_src: Path, case: str) -> Path:
+    path = Path(case)
+    if path.is_absolute() or path.parts[:1] == (candidate_src.name,):
+        return path
+    return candidate_src / path
+
+
+def case_slug(case_path: Path) -> str:
+    raw = case_path.parent.name if case_path.name == "case.toml" else case_path.stem
+    return re.sub(r"[^A-Za-z0-9_.-]+", "-", raw).strip("-") or "case"
+
+
+def append_github_output(path: str | None, status: int) -> None:
+    if not path:
+        return
+    with open(path, "a", encoding="utf-8") as fp:
+        fp.write(f"status={status}\n")
+
+
+def append_step_summary(summary: Path) -> None:
+    step_summary = os.environ.get("GITHUB_STEP_SUMMARY")
+    if not step_summary or not summary.exists():
+        return
+    with open(step_summary, "a", encoding="utf-8") as out:
+        out.write(summary.read_text())
+
+
+def run_case(args: argparse.Namespace, case_path: Path, work_dir: Path) -> int:
+    target_dir = profile_dir(args.cargo_profile)
+    cmd = [
+        "uv",
+        "run",
+        "--no-project",
+        "python",
+        str(args.candidate_src / "tests/perf/query_regression_runner.py"),
+        "--case",
+        str(case_path),
+        "--base-bin",
+        str(args.base_src / "target" / target_dir / "greptime"),
+        "--candidate-bin",
+        str(args.candidate_src / "target" / target_dir / "greptime"),
+        "--fixture-generator",
+        str(args.candidate_src / "target" / target_dir / "query_perf_fixture"),
+        "--work-dir",
+        str(work_dir),
+        "--http-timeout",
+        str(args.http_timeout),
+    ]
+    if parse_bool(args.allow_large_fixture):
+        cmd.append("--allow-large-fixture")
+
+    print(f"::group::Query regression case: {case_path}", flush=True)
+    try:
+        return subprocess.run(cmd, check=False).returncode
+    finally:
+        print("::endgroup::", flush=True)
+
+
+def write_summary(args: argparse.Namespace, reports: list[Path]) -> int:
+    cmd = ["uv", "run", "--no-project", "python", str(args.summary_script)]
+    for report in reports:
+        cmd.extend(["--report", str(report)])
+    cmd.extend(
+        [
+            "--run-url",
+            args.run_url,
+            "--case-name",
+            args.case_name,
+            "--base-ref",
+            args.base_ref,
+            "--candidate-ref",
+            args.candidate_ref,
+            "--output",
+            str(args.summary_output),
+        ]
+    )
+    return subprocess.run(cmd, check=False).returncode
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--cases", action="append", help="'all' or comma/space separated case paths")
+    parser.add_argument("--base-src", type=Path, default=Path("base-src"))
+    parser.add_argument("--candidate-src", type=Path, default=Path("candidate-src"))
+    parser.add_argument("--cargo-profile", default=os.environ.get("CARGO_PROFILE", "nightly"))
+    parser.add_argument("--work-dir", default=Path("query-regression-work"), type=Path)
+    parser.add_argument("--http-timeout", default=os.environ.get("HTTP_TIMEOUT", "300"))
+    parser.add_argument("--allow-large-fixture", default=os.environ.get("ALLOW_LARGE_FIXTURE", "false"))
+    parser.add_argument(
+        "--summary-script",
+        type=Path,
+        default=Path("candidate-src/.github/scripts/query-regression-summary.py"),
+    )
+    parser.add_argument("--summary-output", type=Path, default=Path("query-regression-summary.md"))
+    parser.add_argument("--run-url", default=os.environ.get("RUN_URL", ""))
+    parser.add_argument("--case-name", default=os.environ.get("CASE_NAME", "default case set"))
+    parser.add_argument("--base-ref", default=os.environ.get("BASE_REF", ""))
+    parser.add_argument("--candidate-ref", default=os.environ.get("CANDIDATE_REF", ""))
+    parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT"))
+    args = parser.parse_args()
+
+    try:
+        cases = split_cases(args.cases or [os.environ.get("CASE_PATHS", "all")])
+    except ValueError as err:
+        print(f"error: {err}", flush=True)
+        append_github_output(args.github_output, 1)
+        return 0
+
+    reports: list[Path] = []
+    status = 0
+    for case in cases:
+        case_path = resolve_case_path(args.candidate_src, case)
+        work_dir = args.work_dir / case_slug(case_path)
+        reports.append(work_dir / "query-regression-report.json")
+        case_status = run_case(args, case_path, work_dir)
+        if case_status != 0:
+            status = case_status or 1
+
+    summary_status = write_summary(args, reports)
+    if summary_status != 0 and status == 0:
+        status = summary_status
+    append_step_summary(args.summary_output)
+    append_github_output(args.github_output, status)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/.github/scripts/query-regression-summary.py
+++ b/.github/scripts/query-regression-summary.py
@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Format tests/perf query regression JSON reports as GitHub Markdown."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+
+def fmt_ms(value: Any) -> str:
+    if value is None:
+        return "N/A"
+    try:
+        return f"{float(value):.2f}"
+    except (TypeError, ValueError):
+        return "N/A"
+
+
+def esc(value: Any) -> str:
+    text = "N/A" if value is None else str(value)
+    return text.replace("|", "\\|").replace("\n", " ")
+
+
+def status_emoji(status: str | None) -> str:
+    return {"ok": "✅", "measured": "✅", "failed": "❌", "planned": "📝", "fixture-ready": "🧪"}.get(
+        status or "", "⚠️"
+    )
+
+
+def measurement_map(target: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    return {m.get("name") or f"query-{i}": m for i, m in enumerate(target.get("measurements", []))}
+
+
+def regression_pct(base: Any, candidate: Any) -> str:
+    try:
+        b = float(base)
+        c = float(candidate)
+    except (TypeError, ValueError):
+        return "N/A"
+    if b == 0:
+        return "N/A"
+    return f"{(c - b) / b * 100:+.1f}%"
+
+
+def threshold_status(thresholds: list[dict[str, Any]], query: str) -> str:
+    hits = [t for t in thresholds if t.get("query") == query]
+    if not hits:
+        return "N/A"
+    return ", ".join(f"{t.get('threshold')}: {t.get('status')}" for t in hits)
+
+
+def target_table(targets: list[dict[str, Any]]) -> str:
+    rows = ["| Target | Status | Validation errors | Region | Datanode data home |", "| --- | --- | ---: | --- | --- |"]
+    for target in targets:
+        errors = len(target.get("validation_errors") or [])
+        discovered = target.get("discovered") or {}
+        if isinstance(discovered, list):
+            region = ", ".join(str(item.get("region_id")) for item in discovered)
+        else:
+            region = discovered.get("region_id")
+        rows.append(
+            "| {name} | {status} {raw} | {errors} | {region} | `{data}` |".format(
+                name=esc(target.get("name")),
+                status=status_emoji(target.get("status")),
+                raw=esc(target.get("status")),
+                errors=errors,
+                region=esc(region),
+                data=esc(target.get("datanode_data_home") or target.get("data_dir")),
+            )
+        )
+    return "\n".join(rows)
+
+
+def comparison_table(targets: list[dict[str, Any]], thresholds: list[dict[str, Any]]) -> str:
+    if len(targets) < 2:
+        return "No base/candidate measurements found."
+    base = measurement_map(targets[0])
+    candidate = measurement_map(targets[1])
+    names = sorted(set(base) | set(candidate))
+    rows = [
+        "| Query | Base median ms | Base p95 ms | Candidate median ms | Candidate p95 ms | Regression | Threshold |",
+        "| --- | ---: | ---: | ---: | ---: | ---: | --- |",
+    ]
+    for name in names:
+        bm = base.get(name, {})
+        cm = candidate.get(name, {})
+        rows.append(
+            "| {q} | {bm} | {bp} | {cm} | {cp} | {reg} | {th} |".format(
+                q=esc(name),
+                bm=fmt_ms(bm.get("latency_ms_median")),
+                bp=fmt_ms(bm.get("latency_ms_p95")),
+                cm=fmt_ms(cm.get("latency_ms_median")),
+                cp=fmt_ms(cm.get("latency_ms_p95")),
+                reg=regression_pct(bm.get("latency_ms_median"), cm.get("latency_ms_median")),
+                th=esc(threshold_status(thresholds, name)),
+            )
+        )
+    return "\n".join(rows)
+
+
+def build_markdown(
+    report: dict[str, Any],
+    report_path: Path,
+    run_url: str | None,
+    case_name: str | None,
+    base_ref: str | None,
+    candidate_ref: str | None,
+) -> str:
+    status = report.get("status", "missing")
+    case = report.get("case") or {}
+    title_case = case_name or case.get("name") or report_path.name
+    lines = [f"## {status_emoji(status)} Query regression report: `{esc(title_case)}`", ""]
+    lines.append(f"- **Status:** `{esc(status)}`")
+    lines.append(f"- **Case path:** `{esc(report.get('case_path'))}`")
+    lines.append(f"- **Query mode:** `{esc(report.get('query_mode'))}`")
+    if base_ref:
+        lines.append(f"- **Base ref:** `{esc(base_ref)}`")
+    if candidate_ref:
+        lines.append(f"- **Candidate ref:** `{esc(candidate_ref)}`")
+    if run_url:
+        lines.append(f"- **Workflow run:** {run_url}")
+    if report.get("error"):
+        lines.append(f"- **Error:** `{esc(report.get('error'))}`")
+    lines.append("- **Artifacts:** query-regression-work logs, fixture metadata, and JSON report are uploaded with this run.")
+
+    targets = report.get("targets") or []
+    lines.extend(["", "### Targets", "", target_table(targets)])
+    lines.extend(["", "### Query comparison", "", comparison_table(targets, report.get("thresholds") or [])])
+
+    not_enforced = [t for t in report.get("thresholds") or [] if t.get("status") == "not_enforced"]
+    if not_enforced:
+        lines.extend(["", "### Not enforced thresholds", ""])
+        for item in not_enforced:
+            lines.append(f"- `{esc(item.get('query'))}` / `{esc(item.get('threshold'))}`: {esc(item.get('reason'))}")
+
+    return "\n".join(lines) + "\n"
+
+
+def build_combined_markdown(
+    reports: list[tuple[Path, dict[str, Any]]],
+    run_url: str | None,
+    case_name: str | None,
+    base_ref: str | None,
+    candidate_ref: str | None,
+) -> str:
+    if len(reports) == 1:
+        path, report = reports[0]
+        return build_markdown(report, path, run_url, case_name, base_ref, candidate_ref)
+
+    overall = "failed" if any(report.get("status") != "ok" for _, report in reports) else "ok"
+    lines = [f"## {status_emoji(overall)} Query regression report: `{esc(case_name or 'multiple cases')}`", ""]
+    lines.append(f"- **Status:** `{esc(overall)}`")
+    if base_ref:
+        lines.append(f"- **Base ref:** `{esc(base_ref)}`")
+    if candidate_ref:
+        lines.append(f"- **Candidate ref:** `{esc(candidate_ref)}`")
+    if run_url:
+        lines.append(f"- **Workflow run:** {run_url}")
+    lines.append("- **Artifacts:** query-regression-work logs, fixture metadata, and JSON reports are uploaded with this run.")
+
+    lines.extend(["", "### Cases", "", "| Case | Status | Report |", "| --- | --- | --- |"])
+    for path, report in reports:
+        case = report.get("case") or {}
+        name = case.get("name") or path.parent.name
+        status = report.get("status", "missing")
+        lines.append(f"| `{esc(name)}` | {status_emoji(status)} `{esc(status)}` | `{esc(path)}` |")
+
+    for path, report in reports:
+        case = report.get("case") or {}
+        name = case.get("name") or path.parent.name
+        lines.extend(["", "---", ""])
+        lines.append(build_markdown(report, path, None, name, None, None).rstrip())
+    return "\n".join(lines) + "\n"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--report", required=True, type=Path, action="append")
+    parser.add_argument("--run-url")
+    parser.add_argument("--case-name")
+    parser.add_argument("--base-ref")
+    parser.add_argument("--candidate-ref")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+
+    reports = []
+    for report_path in args.report:
+        if report_path.exists():
+            report = json.loads(report_path.read_text())
+        else:
+            report = {"status": "failed", "error": f"report not found: {report_path}", "targets": []}
+        reports.append((report_path, report))
+    markdown = build_combined_markdown(
+        reports,
+        args.run_url,
+        args.case_name,
+        args.base_ref,
+        args.candidate_ref,
+    )
+    if args.output:
+        args.output.write_text(markdown)
+    else:
+        print(markdown, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/.github/workflows/query-regression-comment.yml
+++ b/.github/workflows/query-regression-comment.yml
@@ -0,0 +1,177 @@
+name: Query Regression Comment
+
+on:
+  workflow_run:
+    workflows: ["Query Regression"]
+    types: [completed]
+
+permissions:
+  contents: read
+  actions: read
+  pull-requests: write
+
+jobs:
+  comment:
+    if: >-
+      ${{ github.event.workflow_run.event == 'pull_request' &&
+          github.event.workflow_run.conclusion != 'cancelled' &&
+          github.event.workflow_run.conclusion != 'skipped' }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Locate query regression comment artifact
+        id: artifact
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const { owner, repo } = context.repo;
+            const run_id = context.payload.workflow_run.id;
+            const { data } = await github.rest.actions.listWorkflowRunArtifacts({
+              owner,
+              repo,
+              run_id,
+              per_page: 100,
+            });
+            const artifact = data.artifacts.find(
+              item => item.name === 'query-regression-comment' && !item.expired
+            );
+            if (!artifact) {
+              core.info('No query-regression-comment artifact found; skipping.');
+              core.setOutput('found', 'false');
+              return;
+            }
+            const maxBytes = 1024 * 1024;
+            if (artifact.size_in_bytes > maxBytes) {
+              core.warning(`Comment artifact too large: ${artifact.size_in_bytes} bytes`);
+              core.setOutput('found', 'false');
+              return;
+            }
+            core.setOutput('found', 'true');
+            core.setOutput('id', String(artifact.id));
+
+      - name: Download query regression comment artifact
+        id: download
+        if: ${{ steps.artifact.outputs.found == 'true' }}
+        uses: actions/download-artifact@v4
+        continue-on-error: true
+        with:
+          name: query-regression-comment
+          path: query-regression-comment
+          repository: ${{ github.repository }}
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          run-id: ${{ github.event.workflow_run.id }}
+
+      - name: Validate PR metadata and prepare comment
+        id: validate
+        if: ${{ steps.download.outcome == 'success' }}
+        uses: actions/github-script@v7
+        env:
+          WORKFLOW_RUN_ID: ${{ github.event.workflow_run.id }}
+          WORKFLOW_RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt }}
+        with:
+          script: |
+            const fs = require('fs');
+            const path = require('path');
+
+            const artifactDir = 'query-regression-comment';
+            const metadataPath = path.join(artifactDir, 'query-regression-pr.json');
+            const summaryPath = path.join(artifactDir, 'query-regression-summary.md');
+
+            function skip(message) {
+              core.info(message);
+              core.setOutput('should_post', 'false');
+            }
+
+            if (!fs.existsSync(metadataPath)) {
+              return skip('Missing query-regression-pr.json; skipping sticky comment.');
+            }
+
+            let metadata;
+            try {
+              metadata = JSON.parse(fs.readFileSync(metadataPath, 'utf8'));
+            } catch (error) {
+              core.warning(`Invalid PR metadata JSON: ${error.message}`);
+              return skip('Invalid PR metadata JSON; skipping sticky comment.');
+            }
+
+            const expectedRunId = Number(process.env.WORKFLOW_RUN_ID);
+            const expectedRunAttempt = Number(process.env.WORKFLOW_RUN_ATTEMPT);
+            if (metadata.run_id !== expectedRunId || metadata.run_attempt !== expectedRunAttempt) {
+              return skip('Artifact metadata does not match this workflow_run; skipping.');
+            }
+
+            if (metadata.base_repo !== context.repo.owner + '/' + context.repo.repo) {
+              return skip(`PR targets ${metadata.base_repo}, not this repository; skipping.`);
+            }
+
+            const prNumber = Number(metadata.pr_number);
+            if (!Number.isInteger(prNumber) || prNumber <= 0) {
+              return skip('Invalid PR number in metadata; skipping.');
+            }
+
+            const run = context.payload.workflow_run;
+            const trustedPrNumbers = new Set(
+              (run.pull_requests || [])
+                .map(pr => Number(pr.number))
+                .filter(Number.isInteger)
+            );
+            try {
+              const { data: associatedPrs } = await github.rest.repos.listPullRequestsAssociatedWithCommit({
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                commit_sha: run.head_sha,
+              });
+              for (const pr of associatedPrs) {
+                trustedPrNumbers.add(Number(pr.number));
+              }
+            } catch (error) {
+              core.warning(`Could not list PRs associated with workflow_run head SHA: ${error.message}`);
+            }
+            if (!trustedPrNumbers.has(prNumber)) {
+              return skip(`PR #${prNumber} is not associated with workflow_run ${run.id}; skipping.`);
+            }
+
+            const { data: pull } = await github.rest.pulls.get({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              pull_number: prNumber,
+            });
+
+            if (pull.state !== 'open') {
+              return skip(`PR #${prNumber} is ${pull.state}; skipping.`);
+            }
+            if (pull.base.repo.full_name !== metadata.base_repo || pull.head.repo.full_name !== metadata.head_repo) {
+              return skip('Current PR repository metadata does not match artifact; skipping.');
+            }
+            if (pull.head.sha !== metadata.head_sha) {
+              return skip('Current PR head SHA differs from artifact; skipping stale run.');
+            }
+
+            let body = 'Query regression completed, but no summary artifact was found.';
+            if (fs.existsSync(summaryPath)) {
+              const maxSummaryBytes = 1024 * 1024;
+              const summarySize = fs.statSync(summaryPath).size;
+              if (summarySize > maxSummaryBytes) {
+                return skip(`Summary file too large: ${summarySize} bytes.`);
+              }
+              body = fs.readFileSync(summaryPath, 'utf8');
+            }
+            body = body
+              .replace(/<!--[\s\S]*?-->/g, '')
+              .replace(/@/g, '@\u200b');
+            if (body.length > 59999) {
+              body = body.slice(0, 59500) + '\n\n…\n\n_Comment truncated because it exceeded 60000 characters._\n';
+            }
+            fs.writeFileSync(summaryPath, body);
+
+            core.setOutput('should_post', 'true');
+            core.setOutput('pr_number', String(prNumber));
+            core.setOutput('summary_path', summaryPath);
+
+      - name: Post sticky PR comment
+        if: ${{ steps.validate.outputs.should_post == 'true' }}
+        uses: marocchino/sticky-pull-request-comment@v2
+        with:
+          header: query-regression-report
+          number: ${{ steps.validate.outputs.pr_number }}
+          path: ${{ steps.validate.outputs.summary_path }}
--- a/.github/workflows/query-regression.yml
+++ b/.github/workflows/query-regression.yml
@@ -0,0 +1,152 @@
+name: Query Regression
+
+on:
+  workflow_dispatch:
+    inputs:
+      case:
+        description: Query perf case path(s) in candidate checkout, or all
+        required: true
+        default: all
+      base_ref:
+        description: Base ref/sha to build
+        required: true
+        default: main
+      candidate_ref:
+        description: Candidate ref/sha to build (empty = current ref)
+        required: false
+        default: ""
+      allow_large_fixture:
+        description: Pass --allow-large-fixture to runner
+        type: boolean
+        default: true
+      http_timeout:
+        description: Runner HTTP timeout seconds
+        required: true
+        default: "300"
+      cargo_profile:
+        description: Cargo profile to build base and candidate binaries with
+        required: true
+        type: choice
+        default: nightly
+        options:
+          - nightly
+          - release
+          - dev
+  pull_request:
+    types: [labeled, ready_for_review, reopened]
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  query-regression:
+    if: >-
+      ${{ github.event_name == 'workflow_dispatch' ||
+          (github.event_name == 'pull_request' &&
+           !github.event.pull_request.draft &&
+           contains(github.event.pull_request.labels.*.name, 'query-regression') &&
+           (github.event.action != 'labeled' || github.event.label.name == 'query-regression')) }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 180
+    env:
+      CARGO_PROFILE: ${{ github.event_name == 'pull_request' && 'nightly' || inputs.cargo_profile }}
+    steps:
+      - name: Checkout base source
+        uses: actions/checkout@v4
+        with:
+          repository: ${{ github.repository }}
+          ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.sha || inputs.base_ref }}
+          path: base-src
+          persist-credentials: false
+
+      - name: Checkout candidate source
+        uses: actions/checkout@v4
+        with:
+          repository: ${{ github.repository }}
+          ref: ${{ github.event_name == 'pull_request' && format('refs/pull/{0}/merge', github.event.pull_request.number) || inputs.candidate_ref || github.ref }}
+          path: candidate-src
+          persist-credentials: false
+
+      - uses: arduino/setup-protoc@v3
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - uses: actions-rust-lang/setup-rust-toolchain@v1
+
+      - uses: astral-sh/setup-uv@v5
+
+      - name: Rust cache (base)
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: base-src
+          shared-key: query-regression-base
+          cache-all-crates: "true"
+          save-if: false
+
+      - name: Rust cache (candidate)
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: candidate-src
+          shared-key: query-regression-candidate
+          cache-all-crates: "true"
+          save-if: false
+
+      - name: Build base greptime
+        working-directory: base-src
+        run: cargo build --profile "${CARGO_PROFILE}" -p cmd --bin greptime
+
+      - name: Build candidate greptime and fixture generator
+        working-directory: candidate-src
+        run: cargo build --profile "${CARGO_PROFILE}" -p cmd --bin greptime --bin query_perf_fixture
+
+      - name: Run query regression
+        id: run
+        env:
+          CASE_PATHS: ${{ github.event_name == 'pull_request' && 'all' || inputs.case }}
+          HTTP_TIMEOUT: ${{ github.event_name == 'pull_request' && '300' || inputs.http_timeout }}
+          ALLOW_LARGE_FIXTURE: ${{ github.event_name == 'pull_request' && 'true' || inputs.allow_large_fixture }}
+          RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          CASE_NAME: ${{ github.event_name == 'pull_request' && 'default case set' || inputs.case }}
+          BASE_REF: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.sha || inputs.base_ref }}
+          CANDIDATE_REF: ${{ github.event_name == 'pull_request' && format('refs/pull/{0}/merge', github.event.pull_request.number) || inputs.candidate_ref || github.ref }}
+        run: uv run --no-project python candidate-src/.github/scripts/query-regression-run.py
+
+      - name: Write PR metadata for trusted comment workflow
+        if: ${{ always() && github.event_name == 'pull_request' }}
+        env:
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+          HEAD_REPO: ${{ github.event.pull_request.head.repo.full_name }}
+          BASE_REPO: ${{ github.event.pull_request.base.repo.full_name }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
+        run: uv run --no-project python candidate-src/.github/scripts/query-regression-pr-metadata.py
+
+      - name: Upload query regression artifacts
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: query-regression-report
+          path: |
+            query-regression-work/**
+            query-regression-summary.md
+          if-no-files-found: warn
+
+      - name: Upload trusted comment artifact
+        if: ${{ always() && github.event_name == 'pull_request' }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: query-regression-comment
+          path: |
+            query-regression-pr.json
+            query-regression-summary.md
+          if-no-files-found: warn
+          retention-days: 7
+
+      - name: Fail on regression failure
+        if: ${{ steps.run.outputs.status != '0' }}
+        run: exit 1
--- a/.gitignore
+++ b/.gitignore
@@ -73,6 +73,7 @@ greptimedb_data
 .gemini
 .opencode
 .worktrees/
+.slim/deepwork/

 # local design docs
 docs/specs/
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2136,6 +2136,7 @@ dependencies = [
 "meta-client",
 "meta-srv",
 "metric-engine",
+ "mito-codec",
 "mito2",
 "moka",
 "object-store",
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -9,6 +9,10 @@ default-run = "greptime"
 name = "greptime"
 path = "src/bin/greptime.rs"

+[[bin]]
+name = "query_perf_fixture"
+path = "src/bin/query_perf_fixture.rs"
+
 [features]
 default = [
    "servers/pprof",
@@ -27,6 +31,7 @@ vector_index = ["mito2/vector_index", "query/vector_index"]
 workspace = true

 [dependencies]
+api.workspace = true
 async-trait.workspace = true
 auth.workspace = true
 base64.workspace = true
@@ -75,6 +80,7 @@ log-store.workspace = true
 meta-client.workspace = true
 meta-srv.workspace = true
 metric-engine.workspace = true
+mito-codec.workspace = true
 mito2.workspace = true
 moka = { workspace = true, features = ["future"] }
 object-store.workspace = true
--- a/src/cmd/src/bin/query_perf_fixture.rs
+++ b/src/cmd/src/bin/query_perf_fixture.rs
@@ -0,0 +1,651 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![allow(clippy::print_stderr, clippy::print_stdout)]
+
+//! Generic direct-readable-SST fixture generator for query performance labs.
+//!
+//! This is a bounded MVP: it reads one table/one region case TOML, builds
+//! synthetic region metadata, writes real Mito readable SST parquet files, and
+//! emits a manifest checkpoint plus validation metadata.  The first case uses
+//! `sst_format = "flat"`; primary-key output is also accepted for early lab use
+//! because the writer can convert the same flat input batch to primary-key SSTs.
+//!
+//! TODO(query-perf): add multi-region generation and seed-manifest/catalog DB
+//! integration so the generated fixture can be mounted directly by a full
+//! GreptimeDB open/query path without manual metadata wiring.
+
+use std::collections::HashMap;
+use std::fs;
+use std::io::{BufWriter, Write};
+use std::num::{NonZeroU64, NonZeroUsize};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use api::v1::{OpType, SemanticType};
+use async_trait::async_trait;
+use clap::Parser;
+use datatypes::arrow::array::{
+    ArrayRef, BinaryDictionaryBuilder, Float64Array, RecordBatch, StringDictionaryBuilder,
+    TimestampMillisecondArray, TimestampNanosecondArray, UInt8Array, UInt64Array,
+};
+use datatypes::arrow::datatypes::UInt32Type;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::schema::{ColumnSchema, SkippingIndexOptions};
+use mito_codec::row_converter::{DensePrimaryKeyCodec, PrimaryKeyCodecExt, SortField};
+use mito2::access_layer::{FilePathProvider, Metrics, WriteType};
+use mito2::config::IndexConfig;
+use mito2::manifest::action::{RegionCheckpoint, RegionManifest, RemovedFilesRecord};
+use mito2::read::FlatSource;
+use mito2::sst::file::{FileMeta, RegionFileId};
+use mito2::sst::index::{Indexer, IndexerBuilder};
+use mito2::sst::parquet::writer::ParquetWriter;
+use mito2::sst::parquet::{SstInfo, WriteOptions};
+use mito2::sst::{
+    DEFAULT_WRITE_BUFFER_SIZE, FlatSchemaOptions, FormatType, to_flat_sst_arrow_schema,
+};
+use object_store::ObjectStore;
+use object_store::services::Fs as FsBuilder;
+use serde::Deserialize;
+use store_api::codec::PrimaryKeyEncoding;
+use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataRef};
+use store_api::path_utils::region_name;
+use store_api::region_request::PathType;
+use store_api::storage::{FileId, RegionId};
+
+#[derive(Parser, Debug)]
+#[command(name = "query_perf_fixture")]
+#[command(about = "Generate direct-readable SST fixtures for query performance cases")]
+struct Args {
+    /// TOML case file, for example tests/perf/query_cases/.../case.toml.
+    #[arg(long, value_name = "PATH")]
+    case: PathBuf,
+
+    /// Output directory (creates object-store/, manifest/, files.jsonl, summary.json).
+    #[arg(long, value_name = "DIR")]
+    out_dir: PathBuf,
+
+    /// Region ID to synthesize. Defaults to table id 1024, region 0.
+    #[arg(long, default_value = "4398046511104")]
+    region_id: u64,
+
+    /// Table directory relative to object-store root. Defaults to data/{database}/{table}/.
+    #[arg(long, value_name = "DIR")]
+    table_dir: Option<String>,
+
+    /// Table name to generate when the case contains multiple [[tables]].
+    #[arg(long, value_name = "NAME")]
+    table: Option<String>,
+
+    /// Manifest/checkpoint version.
+    #[arg(long, default_value = "1000000")]
+    checkpoint_version: u64,
+
+    /// Safety flag when sst_count exceeds 1000.
+    #[arg(long)]
+    allow_large: bool,
+
+    /// Print plan only.
+    #[arg(long)]
+    dry_run: bool,
+}
+
+#[derive(Debug, Deserialize)]
+struct CaseFile {
+    scenario: Scenario,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(tag = "kind")]
+enum Scenario {
+    #[serde(rename = "direct_readable_sst")]
+    DirectReadableSst(DirectReadableSstScenario),
+}
+
+#[derive(Debug, Deserialize)]
+struct DirectReadableSstScenario {
+    #[serde(default)]
+    seed: Option<u64>,
+    tables: Vec<TableConfig>,
+    layout: LayoutConfig,
+}
+
+impl Scenario {
+    fn kind(&self) -> &'static str {
+        match self {
+            Scenario::DirectReadableSst(_) => "direct_readable_sst",
+        }
+    }
+
+    fn direct_readable_sst(&self) -> &DirectReadableSstScenario {
+        match self {
+            Scenario::DirectReadableSst(scenario) => scenario,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+struct TableConfig {
+    database: String,
+    name: String,
+    engine: String,
+    #[serde(default)]
+    append_mode: Option<bool>,
+    #[serde(default)]
+    sst_format: Option<String>,
+    primary_key: Vec<String>,
+    time_index: String,
+    columns: Vec<ColumnConfig>,
+}
+
+#[derive(Debug, Deserialize)]
+struct ColumnConfig {
+    name: String,
+    #[serde(rename = "type")]
+    ty: String,
+    semantic: String,
+    distribution: Option<Distribution>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(tag = "kind")]
+enum Distribution {
+    #[serde(rename = "cardinality")]
+    Cardinality {
+        values: NonZeroUsize,
+        prefix: String,
+    },
+    #[serde(rename = "deterministic_wave")]
+    DeterministicWave { min: f64, max: f64 },
+}
+
+#[derive(Debug, Deserialize)]
+struct LayoutConfig {
+    regions: usize,
+    sst_count: usize,
+    rows_per_sst: usize,
+    row_group_size: usize,
+    series_count: NonZeroUsize,
+    start_unix_nanos: i64,
+    step_nanos: i64,
+    time_range_layout: String,
+    series_layout: String,
+}
+
+struct NoopIndexBuilder;
+
+#[async_trait]
+impl IndexerBuilder for NoopIndexBuilder {
+    async fn build(&self, _file_id: FileId, _index_version: u64) -> Indexer {
+        Indexer::default()
+    }
+}
+
+#[derive(Clone)]
+struct FixedPathProvider {
+    table_dir: String,
+}
+
+impl FilePathProvider for FixedPathProvider {
+    fn build_index_file_path(&self, file_id: RegionFileId) -> String {
+        mito2::sst::location::index_file_path_legacy(&self.table_dir, file_id, PathType::Bare)
+    }
+    fn build_index_file_path_with_version(
+        &self,
+        index_id: mito2::sst::file::RegionIndexId,
+    ) -> String {
+        mito2::sst::location::index_file_path(&self.table_dir, index_id, PathType::Bare)
+    }
+    fn build_sst_file_path(&self, file_id: RegionFileId) -> String {
+        mito2::sst::location::sst_file_path(&self.table_dir, file_id, PathType::Bare)
+    }
+}
+
+fn semantic_type(s: &str) -> SemanticType {
+    match s.to_ascii_lowercase().as_str() {
+        "tag" => SemanticType::Tag,
+        "field" => SemanticType::Field,
+        "timestamp" => SemanticType::Timestamp,
+        other => panic!("unsupported semantic {other}"),
+    }
+}
+
+fn concrete_type(ty: &str) -> ConcreteDataType {
+    match ty.to_ascii_uppercase().as_str() {
+        "STRING" => ConcreteDataType::string_datatype(),
+        "DOUBLE" => ConcreteDataType::float64_datatype(),
+        "UINT64" => ConcreteDataType::uint64_datatype(),
+        "TIMESTAMP(9)" => ConcreteDataType::timestamp_nanosecond_datatype(),
+        "TIMESTAMP(3)" => ConcreteDataType::timestamp_millisecond_datatype(),
+        other => panic!("unsupported column type {other}"),
+    }
+}
+
+fn build_region_metadata(table: &TableConfig, region_id: RegionId) -> RegionMetadata {
+    let mut builder = store_api::metadata::RegionMetadataBuilder::new(region_id);
+    for (idx, col) in table.columns.iter().enumerate() {
+        let mut schema = ColumnSchema::new(
+            &col.name,
+            concrete_type(&col.ty),
+            col.semantic != "timestamp",
+        );
+        if col.semantic == "tag" {
+            schema = schema.with_inverted_index(true);
+        }
+        if col.semantic == "tag" || col.semantic == "field" {
+            schema = schema
+                .with_skipping_options(SkippingIndexOptions {
+                    granularity: 1,
+                    ..Default::default()
+                })
+                .expect("valid skipping index options for query perf fixture");
+        }
+        if col.name == table.time_index {
+            schema = schema.with_time_index(true);
+        }
+        builder.push_column_metadata(ColumnMetadata {
+            column_schema: schema,
+            semantic_type: semantic_type(&col.semantic),
+            column_id: idx as u32,
+        });
+    }
+    let pk_ids = table
+        .primary_key
+        .iter()
+        .map(|name| {
+            table
+                .columns
+                .iter()
+                .position(|c| c.name == *name)
+                .unwrap_or_else(|| panic!("primary key column {name} not found")) as u32
+        })
+        .collect();
+    builder.primary_key(pk_ids);
+    builder.primary_key_encoding(PrimaryKeyEncoding::Dense);
+    builder
+        .build()
+        .expect("region metadata should be valid for fixture table config")
+}
+
+fn encode_dense_primary_key(table: &TableConfig, values: &HashMap<String, String>) -> Vec<u8> {
+    let fields = table
+        .primary_key
+        .iter()
+        .enumerate()
+        .map(|(idx, _)| {
+            (
+                idx as u32,
+                SortField::new(ConcreteDataType::string_datatype()),
+            )
+        })
+        .collect();
+    let converter = DensePrimaryKeyCodec::with_fields(fields);
+    converter
+        .encode(
+            table
+                .primary_key
+                .iter()
+                .map(|name| datatypes::value::ValueRef::String(values[name].as_str())),
+        )
+        .expect("dense primary key encoding should match fixture primary key fields")
+}
+
+fn tag_value(col: &ColumnConfig, series: usize) -> String {
+    match col
+        .distribution
+        .as_ref()
+        .expect("tag distribution is required")
+    {
+        Distribution::Cardinality { values, prefix } => {
+            format!("{}{}", prefix, series % values.get())
+        }
+        _ => panic!("tag column {} requires cardinality distribution", col.name),
+    }
+}
+
+fn wave_value(min: f64, max: f64, row: usize) -> f64 {
+    let span = max - min;
+    let phase = (row % 1024) as f64 / 1023.0;
+    min + span * (0.5 - 0.5 * (std::f64::consts::TAU * phase).cos())
+}
+
+fn generate_record_batch(
+    table: &TableConfig,
+    metadata: &RegionMetadataRef,
+    layout: &LayoutConfig,
+    sst_idx: usize,
+    sequence: u64,
+) -> RecordBatch {
+    let flat_schema = to_flat_sst_arrow_schema(metadata, &FlatSchemaOptions::default());
+    let rows = layout.rows_per_sst;
+    let mut columns = Vec::new();
+    let base_row = sst_idx * rows;
+
+    for col in &table.columns {
+        match (col.semantic.as_str(), col.ty.to_ascii_uppercase().as_str()) {
+            ("tag", "STRING") => {
+                let mut b = StringDictionaryBuilder::<UInt32Type>::new();
+                for row in 0..rows {
+                    let series = if layout.series_layout == "round_robin" {
+                        (base_row + row) % layout.series_count.get()
+                    } else {
+                        sst_idx % layout.series_count.get()
+                    };
+                    let value = tag_value(col, series);
+                    b.append_value(value);
+                }
+                columns.push(Arc::new(b.finish()) as ArrayRef);
+            }
+            ("field", "DOUBLE") => {
+                let (min, max) = match col.distribution.as_ref() {
+                    Some(Distribution::DeterministicWave { min, max }) => (*min, *max),
+                    _ => (0.0, 1.0),
+                };
+                columns.push(Arc::new(Float64Array::from(
+                    (0..rows)
+                        .map(|r| wave_value(min, max, base_row + r))
+                        .collect::<Vec<_>>(),
+                )) as ArrayRef);
+            }
+            ("field", "UINT64") => columns.push(Arc::new(UInt64Array::from(
+                (0..rows).map(|r| (base_row + r) as u64).collect::<Vec<_>>(),
+            )) as ArrayRef),
+            ("timestamp", "TIMESTAMP(9)") => columns.push(Arc::new(TimestampNanosecondArray::from(
+                (0..rows)
+                    .map(|r| layout.start_unix_nanos + ((base_row + r) as i64 * layout.step_nanos))
+                    .collect::<Vec<_>>(),
+            )) as ArrayRef),
+            ("timestamp", "TIMESTAMP(3)") => {
+                columns.push(Arc::new(TimestampMillisecondArray::from(
+                    (0..rows)
+                        .map(|r| {
+                            (layout.start_unix_nanos + ((base_row + r) as i64 * layout.step_nanos))
+                                / 1_000_000
+                        })
+                        .collect::<Vec<_>>(),
+                )) as ArrayRef)
+            }
+            other => panic!("unsupported column combination {other:?}"),
+        }
+    }
+
+    let mut pk_builder = BinaryDictionaryBuilder::<UInt32Type>::new();
+    for row in 0..rows {
+        let series = if layout.series_layout == "round_robin" {
+            (base_row + row) % layout.series_count.get()
+        } else {
+            sst_idx % layout.series_count.get()
+        };
+        let values = table
+            .columns
+            .iter()
+            .filter(|c| c.semantic == "tag")
+            .map(|c| (c.name.clone(), tag_value(c, series)))
+            .collect();
+        pk_builder
+            .append(encode_dense_primary_key(table, &values))
+            .expect("append generated dense primary key to Arrow dictionary builder");
+    }
+    columns.push(Arc::new(pk_builder.finish()));
+    columns.push(Arc::new(UInt64Array::from_value(sequence, rows)));
+    columns.push(Arc::new(UInt8Array::from_value(OpType::Put as u8, rows)));
+    RecordBatch::try_new(flat_schema, columns)
+        .expect("generated fixture columns should match flat SST Arrow schema")
+}
+
+fn file_meta_from_sst_info(
+    info: &SstInfo,
+    region_id: RegionId,
+    file_id: FileId,
+    sequence: u64,
+) -> FileMeta {
+    FileMeta {
+        region_id,
+        file_id,
+        time_range: info.time_range,
+        level: 0,
+        file_size: info.file_size,
+        max_row_group_uncompressed_size: info.max_row_group_uncompressed_size,
+        available_indexes: Default::default(),
+        indexes: Default::default(),
+        index_file_size: 0,
+        index_version: 0,
+        num_rows: info.num_rows as u64,
+        num_row_groups: info.num_row_groups,
+        sequence: NonZeroU64::new(sequence),
+        partition_expr: None,
+        num_series: info.num_series,
+        primary_key_min: None,
+        primary_key_max: None,
+    }
+}
+
+fn deterministic_file_id(seed: u64, index: usize) -> FileId {
+    let seed_hi = ((seed >> 16) & 0xffff) as u16;
+    let seed_lo = (seed & 0xffff) as u16;
+    FileId::parse_str(&format!(
+        "00000000-0000-{seed_hi:04x}-{seed_lo:04x}-{index:012x}"
+    ))
+    .expect("deterministic file id must be a valid UUID")
+}
+
+fn remap_sst_file(
+    object_store_dir: &Path,
+    table_dir: &str,
+    region_id: RegionId,
+    old_file_id: FileId,
+    new_file_id: FileId,
+) {
+    if old_file_id == new_file_id {
+        return;
+    }
+    let old_path = object_store_dir.join(mito2::sst::location::sst_file_path(
+        table_dir,
+        RegionFileId::new(region_id, old_file_id),
+        PathType::Bare,
+    ));
+    let new_path = object_store_dir.join(mito2::sst::location::sst_file_path(
+        table_dir,
+        RegionFileId::new(region_id, new_file_id),
+        PathType::Bare,
+    ));
+    if let Some(parent) = new_path.parent() {
+        fs::create_dir_all(parent).expect("failed to create remapped SST parent directory");
+    }
+    fs::rename(&old_path, &new_path).unwrap_or_else(|err| {
+        panic!(
+            "failed to remap SST file {} -> {}: {err}",
+            old_path.display(),
+            new_path.display()
+        )
+    });
+}
+
+fn case_name_from_path(path: &Path) -> String {
+    path.parent()
+        .and_then(Path::file_name)
+        .and_then(|name| name.to_str())
+        .unwrap_or("query_perf_case")
+        .to_string()
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+    let case_text = fs::read_to_string(&args.case).expect("failed to read query perf case TOML");
+    let case: CaseFile = toml::from_str(&case_text).expect("failed to parse query perf case TOML");
+    let case_name = case_name_from_path(&args.case);
+    let scenario = case.scenario.direct_readable_sst();
+    if scenario.tables.is_empty() || scenario.layout.regions != 1 {
+        panic!("fixture generator supports one or more tables and exactly one region per table");
+    }
+    if scenario.layout.time_range_layout != "non_overlapping_per_sst" {
+        panic!("MVP supports time_range_layout=non_overlapping_per_sst");
+    }
+    if scenario.layout.sst_count > 1000 && !args.allow_large {
+        panic!("sst_count exceeds 1000; pass --allow-large");
+    }
+    let seed = scenario.seed.unwrap_or(0);
+    let table_index = match args.table.as_deref() {
+        Some(name) => scenario
+            .tables
+            .iter()
+            .position(|table| table.name == name)
+            .unwrap_or_else(|| panic!("--table {name} was not found in case {case_name}")),
+        None if scenario.tables.len() == 1 => 0,
+        None => panic!(
+            "case {} contains {} tables; pass --table <name> to choose one",
+            case_name,
+            scenario.tables.len()
+        ),
+    };
+    let table = &scenario.tables[table_index];
+    if table.engine != "mito" {
+        panic!("MVP supports engine=mito");
+    }
+    let region_id = RegionId::from(args.region_id);
+    let table_dir = args
+        .table_dir
+        .unwrap_or_else(|| format!("data/{}/{}/", table.database, table.name));
+    if Path::new(&table_dir).is_absolute() {
+        panic!("--table-dir must be relative");
+    }
+    if table_dir.trim_end_matches('/').rsplit('/').next()
+        == Some(region_name(region_id.table_id(), region_id.region_sequence()).as_str())
+    {
+        panic!("--table-dir must be table-level directory");
+    }
+    let region_dir =
+        mito2::sst::location::region_dir_from_table_dir(&table_dir, region_id, PathType::Bare);
+    let format = match table.sst_format.as_deref().unwrap_or("flat") {
+        "flat" => FormatType::Flat,
+        "primary_key" | "primary-key" => FormatType::PrimaryKey,
+        other => panic!("unsupported sst_format {other}"),
+    };
+    println!(
+        "query_perf_fixture case={} scenario={} table={}.{} ssts={} rows_per_sst={} format={format:?}",
+        case_name,
+        case.scenario.kind(),
+        table.database,
+        table.name,
+        scenario.layout.sst_count,
+        scenario.layout.rows_per_sst
+    );
+    println!(
+        "out_dir={} table_dir={} region_dir={}",
+        args.out_dir.display(),
+        table_dir,
+        region_dir
+    );
+    if args.dry_run {
+        return;
+    }
+
+    let obj_store_dir = args.out_dir.join("object-store");
+    let manifest_dir = args.out_dir.join("manifest");
+    fs::create_dir_all(&obj_store_dir).expect("failed to create fixture object-store directory");
+    fs::create_dir_all(&manifest_dir).expect("failed to create fixture manifest directory");
+    let ostorage = ObjectStore::new(FsBuilder::default().root(&obj_store_dir.to_string_lossy()))
+        .expect("failed to create filesystem object store for fixture output")
+        .finish();
+    let metadata: RegionMetadataRef = Arc::new(build_region_metadata(table, region_id));
+    let mut files = HashMap::with_capacity(scenario.layout.sst_count);
+    let mut next_file_index = 1;
+    for i in 0..scenario.layout.sst_count {
+        let sequence = 1000 + i as u64;
+        let batch = generate_record_batch(table, &metadata, &scenario.layout, i, sequence);
+        let source =
+            FlatSource::new_iter(batch.schema(), Box::new(vec![batch].into_iter().map(Ok)));
+        let mut metrics = Metrics::new(WriteType::Flush);
+        let mut writer = ParquetWriter::new_with_object_store(
+            ostorage.clone(),
+            metadata.clone(),
+            IndexConfig::default(),
+            NoopIndexBuilder,
+            FixedPathProvider {
+                table_dir: table_dir.clone(),
+            },
+            &mut metrics,
+        )
+        .await;
+        let opts = WriteOptions {
+            write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE,
+            row_group_size: scenario.layout.row_group_size,
+            max_file_size: None,
+        };
+        let infos = match format {
+            FormatType::Flat => writer.write_all_flat(source, Some(sequence), &opts).await,
+            FormatType::PrimaryKey => {
+                writer
+                    .write_all_flat_as_primary_key(source, Some(sequence), &opts)
+                    .await
+            }
+        }
+        .expect("failed to write fixture SST through Mito parquet writer");
+        for info in infos {
+            let file_id = deterministic_file_id(
+                seed.wrapping_add((table_index as u64) << 16),
+                next_file_index,
+            );
+            next_file_index += 1;
+            remap_sst_file(&obj_store_dir, &table_dir, region_id, info.file_id, file_id);
+            files.insert(
+                file_id,
+                file_meta_from_sst_info(&info, region_id, file_id, sequence),
+            );
+        }
+    }
+    let overall_seq = files
+        .values()
+        .filter_map(|m| m.sequence)
+        .map(|s| s.get())
+        .max()
+        .unwrap_or(0);
+    let manifest = RegionManifest {
+        metadata,
+        files,
+        removed_files: RemovedFilesRecord::default(),
+        flushed_entry_id: overall_seq,
+        flushed_sequence: overall_seq,
+        committed_sequence: Some(overall_seq),
+        manifest_version: args.checkpoint_version,
+        truncated_entry_id: None,
+        compaction_time_window: None,
+        sst_format: format,
+        append_mode: table.append_mode,
+    };
+    let checkpoint = RegionCheckpoint {
+        last_version: args.checkpoint_version,
+        compacted_actions: manifest.files.len(),
+        checkpoint: Some(manifest.clone()),
+    };
+    let checkpoint_bytes = checkpoint
+        .encode()
+        .expect("failed to encode fixture region checkpoint");
+    let checkpoint_path = manifest_dir.join(format!("{:020}.checkpoint", args.checkpoint_version));
+    fs::write(&checkpoint_path, &checkpoint_bytes).expect("failed to write checkpoint file");
+    fs::write(manifest_dir.join("_last_checkpoint"), serde_json::to_vec_pretty(&serde_json::json!({ "size": checkpoint_bytes.len(), "version": args.checkpoint_version, "checksum": null, "extend_metadata": {} })).expect("failed to serialize _last_checkpoint metadata")).expect("failed to write _last_checkpoint file");
+    let files_jsonl_path = args.out_dir.join("files.jsonl");
+    let mut jsonl = BufWriter::new(
+        fs::File::create(&files_jsonl_path).expect("failed to create fixture files.jsonl"),
+    );
+    let mut entries: Vec<_> = manifest.files.iter().collect();
+    entries.sort_by_key(|(id, _)| id.to_string());
+    for (file_id, meta) in entries {
+        writeln!(jsonl, "{}", serde_json::to_string(&serde_json::json!({ "file_id": file_id.to_string(), "region_id": meta.region_id.as_u64(), "object_path": mito2::sst::location::sst_file_path(&table_dir, RegionFileId::new(meta.region_id, *file_id), PathType::Bare), "time_range_start": meta.time_range.0.value(), "time_range_end": meta.time_range.1.value(), "num_rows": meta.num_rows, "num_row_groups": meta.num_row_groups, "file_size": meta.file_size, "num_series": meta.num_series, "sequence": meta.sequence.map(|s| s.get()) })).expect("failed to serialize fixture SST metadata JSONL entry")).expect("failed to write fixture SST metadata JSONL entry");
+    }
+    jsonl.flush().expect("failed to flush fixture files.jsonl");
+    fs::write(args.out_dir.join("summary.json"), serde_json::to_vec_pretty(&serde_json::json!({ "case": case_name, "seed": seed, "table_index": table_index, "table": table.name, "database": table.database, "region_id": region_id.as_u64(), "table_dir": table_dir, "region_dir": region_dir, "sst_format": format!("{format:?}"), "sst_count": scenario.layout.sst_count, "rows_per_sst": scenario.layout.rows_per_sst, "row_group_size": scenario.layout.row_group_size, "total_rows": scenario.layout.sst_count * scenario.layout.rows_per_sst, "checkpoint_path": checkpoint_path, "files_jsonl_path": files_jsonl_path, "readback_validated": false, "metadata_source": "synthetic" })).expect("failed to serialize fixture summary")).expect("failed to write fixture summary.json");
+    println!("Done. wrote {} SST file entries", manifest.files.len());
+}
--- a/tests/perf/AGENTS.md
+++ b/tests/perf/AGENTS.md
@@ -0,0 +1,24 @@
+# Agent Guidelines for Query Performance Tests
+
+- Keep GitHub Actions YAML thin. Put non-trivial control flow, case expansion,
+  report generation, and metadata writing in scripts under `.github/scripts/`;
+  workflow steps should mostly invoke those scripts.
+- Query regression PR runs should build base/candidate binaries once, then run
+  the default case set. Do not hard-code a single case such as
+  `promql_pushdown_7913` into the workflow path.
+- The case DSL is not required to keep compatibility inside this PR. When the
+  DSL changes, update TOML cases, the Python runner, Rust fixture generator, and
+  docs together.
+- `[case]` is report metadata only. `[scenario]` is the executable regression
+  configuration and must include `kind`, data layout, tables, queries, and
+  thresholds. Keep scenario parsing explicit in both
+  `tests/perf/query_regression_runner.py` and
+  `src/cmd/src/bin/query_perf_fixture.rs`.
+- Keep the direct-SST generator generic. Issue-specific behavior belongs in case
+  files and thresholds, not in Rust generator logic.
+- Before pushing perf harness changes, run at least:
+  - `uv run --no-project python -m py_compile .github/scripts/query-regression-run.py .github/scripts/query-regression-summary.py .github/scripts/query-regression-pr-metadata.py tests/perf/query_regression_runner.py`
+  - `cargo fmt --all -- --check`
+  - `cargo build -p cmd --bin query_perf_fixture`
+  - dry-run the Python runner and Rust fixture generator against all built-in
+    cases when the DSL or workflow case selection changes.
--- a/tests/perf/README.md
+++ b/tests/perf/README.md
@@ -0,0 +1,233 @@
+# Query performance regression harness
+
+This directory is for query performance cases that compare a base build with a
+candidate build. It is not a replacement for sqlness: the goal is to measure the
+effect of optimizer/query-engine changes on realistic scan work.
+
+## Phase 1: direct readable SST fixtures
+
+Phase 1 should generate data by writing readable Mito SST files and matching
+manifest checkpoints directly. This follows the `gc_readable_sst_fixture` lab
+approach from `~/greptimedb-gc-huge-stress`: use Mito's SST writer to create
+queryable files, then write a checkpoint and `_last_checkpoint` that reference
+those files.
+
+The generator itself must be generic. It should not know about a specific issue
+such as #7913 or a specific PromQL query. Cases provide declarative table schema,
+data layout, distributions, and queries; the generator turns those declarations
+into readable SST fixtures.
+
+The intended flow for each case is:
+
+1. Start a GreptimeDB build and create an empty table to seed catalog/table
+   metadata.
+2. Stop the process.
+3. Use the seed region metadata/manifest to generate deterministic readable SSTs
+   and a replacement manifest checkpoint offline.
+4. Start the same build on the generated data directory.
+5. Run warmup and measured queries.
+6. Repeat the same fixture/query process for the candidate build.
+7. Compare base vs candidate metrics and write a regression report.
+
+Direct SST fixtures are the default for phase 1 because they provide stable file
+counts, time ranges, row groups, and label distributions without spending CI time
+on ingestion and flush. Ingestion-path cases can be added later for nightly or
+release-level realism.
+
+## Generator contract
+
+The direct-SST generator should accept a case definition with:
+
+- one or more table definitions: columns, semantic types, primary key, time
+  index, SST format, append mode
+- deterministic distributions: seed, series/tag cardinalities, label/value
+  functions, timestamp layout
+- physical layout: regions, SST count, rows per SST, row group size, time ranges
+  per SST, optional overlap/skew
+- output paths for object-store files, manifest checkpoints, and fixture metadata
+
+This keeps query regression cases reusable: the same generator can produce
+PromQL, SQL, pruning, projection, join, or aggregation fixtures by changing only
+case config.
+
+## What a case owns
+
+Each optimization PR should add or update the query case for the pattern it is
+expected to affect. A case should define:
+
+- schema and seed table SQL
+- deterministic data shape: seed, series count, rows per SST, SST count, time
+  range layout, label distribution, region/partition layout
+- queries to run
+- warmup/measurement repetitions
+- metrics to collect
+- base-vs-candidate thresholds
+
+The `[case]` table is metadata for reports. The executable regression config
+lives under `[scenario]`. A scenario owns data generation, queries, and
+thresholds:
+
+```toml
+[case]
+name = "example"
+description = "what this regression protects"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 12345
+
+[[scenario.tables]]
+# table schema and distributions
+
+[scenario.layout]
+# SST and series layout
+
+[[scenario.queries]]
+# query, warmups, iterations, thresholds
+```
+
+The runner currently supports only `direct_readable_sst`. The enum-style
+`scenario.kind` is reserved for future scenarios such as `write_then_query` and
+`cache_warm_query`.
+
+## Metrics
+
+Primary gates should compare query work rather than plan text:
+
+- scanned files / file ranges
+- scanned rows or row groups
+- bytes read when available
+- pruning ratio
+- query latency median/p95
+- output row count as a sanity check
+
+Plan details such as pushed filters are useful diagnostics, but should not be the
+main pass/fail signal.
+
+## Runner MVP
+
+`query_regression_runner.py` is the base-vs-candidate orchestration layer. The
+current MVP parses a case, creates per-target work directories, and in real query
+mode starts a local distributed cluster for each target: metasrv (memory-store,
+region failover disabled), one datanode (`node_id=0`), and one frontend. It
+creates the configured Mito table(s) through frontend HTTP SQL, discovers the
+real one-region-per-table metadata via `information_schema`, stops only the
+owning datanode, generates one shared direct-SST fixture per table using the
+discovered `--region-id`, `--table-dir`, and `--table`, injects those region
+subtrees into the datanode data home, restarts the datanode, then validates and
+measures through frontend. Reports are written as JSON under the work directory.
+
+The runner intentionally keeps metasrv alive for the whole target run because
+memory-store metadata would otherwise be lost. It replaces only the discovered
+datanode region directory under `data/greptime/<schema>/<table_id>/...` with
+generated SST files and a manifest checkpoint. For multi-table cases this is
+repeated per table, enabling true JOIN fixtures while still requiring exactly one
+region per table. Base and candidate must discover identical per-table
+`table_dir` and `region_id`; otherwise the run fails.
+
+Multi-table direct-SST cases must use unique table names as well as unique
+`(database, name)` pairs because the generator currently selects a table with
+`--table <name>`. Per-table fixture directories are derived from table index,
+database, and table name with path-unsafe characters sanitized.
+
+Currently enforced threshold:
+
+- `max_candidate_latency_regression_pct`, based on client-side median latency.
+
+Server-side scan thresholds such as file ranges and scanned rows are planned for
+a follow-up PR that extracts them from structured `EXPLAIN ANALYZE VERBOSE`
+output. Do not add those threshold keys until the runner enforces them.
+
+Dry-run example:
+
+```bash
+uv run --no-project python tests/perf/query_regression_runner.py \
+  --case tests/perf/query_cases/promql_pushdown_7913/case.toml \
+  --base-bin /path/to/base/greptime \
+  --candidate-bin /path/to/candidate/greptime \
+  --work-dir /tmp/query-perf-work \
+  --dry-run
+```
+
+With a fixture generator:
+
+```bash
+uv run --no-project python tests/perf/query_regression_runner.py \
+  --case tests/perf/query_cases/promql_pushdown_7913/case.toml \
+  --base-bin /path/to/base/greptime \
+  --candidate-bin /path/to/candidate/greptime \
+  --fixture-generator /path/to/query_perf_fixture \
+  --allow-large-fixture \
+  --work-dir /tmp/query-perf-work
+```
+
+This mode launches metasrv, datanode, and frontend for each target with explicit
+localhost HTTP/gRPC/MySQL/Postgres ports and writes component stdout/stderr under
+each target's `logs/` directory.
+
+By default query mode requires fresh base/candidate work directories and fails if
+either target directory already exists with contents. Use `--reuse-work-dir` only
+when intentionally debugging an existing run directory. SQL HTTP requests default
+to a 120 second timeout; override with `--http-timeout <seconds>` for slow lab
+runs.
+
+Fixture generator smoke test:
+
+```bash
+cargo run -p cmd --bin query_perf_fixture -- \
+  --case tests/perf/query_cases/smoke_direct_sst/case.toml \
+  --out-dir /tmp/query-perf-smoke
+```
+
+Runner smoke test with fixture generation only:
+
+```bash
+uv run --no-project python tests/perf/query_regression_runner.py \
+  --case tests/perf/query_cases/smoke_direct_sst/case.toml \
+  --base-bin /path/to/query_perf_fixture \
+  --candidate-bin /path/to/query_perf_fixture \
+  --fixture-generator /path/to/query_perf_fixture \
+  --work-dir /tmp/query-perf-runner-smoke \
+  --fixture-only
+```
+
+`--fixture-only` preserves the earlier smoke behavior: it does not start
+standalone servers, and it materializes the generated fixture into base and
+candidate data directories for plumbing validation.
+
+## GitHub Actions
+
+`.github/workflows/query-regression.yml` provides an opt-in CI entrypoint for
+query regression runs. It builds its own binaries for now:
+
+- base `greptime` from the PR base commit, or `workflow_dispatch` `base_ref`
+- candidate `greptime` and `query_perf_fixture` from the PR merge ref/current
+  candidate checkout
+- runner and summary formatter from the candidate checkout
+
+The workflow runs automatically only for non-draft PRs labeled
+`query-regression` (on label/ready-for-review/reopen events, not every push).
+PR runs build base/candidate once and then run the default case set with
+`--allow-large-fixture`. Manual `workflow_dispatch` runs can pass `all`, one case
+path, or a comma/whitespace-separated list of case paths, and can override refs.
+It always uploads `query-regression-work/**` and `query-regression-summary.md`,
+writes the Markdown summary to the workflow step summary, and updates a sticky PR
+comment through the trusted follow-up workflow.
+
+## Built-in cases
+
+The `promql_pushdown_7913` case is only one case using the generic fixture
+format. It should generate a metric-like table with a nanosecond time index and
+many SSTs with non-overlapping time ranges. A narrow PromQL/TQL query should
+touch only a small time window. The candidate build is expected to scan
+materially fewer files/ranges/rows than the base build when an optimizer PR
+claims to improve this path.
+
+Additional SQL optimizer cases:
+
+- `sql_topk_order_by`: single-table TopK / `ORDER BY` on a DOUBLE field with
+  time and tag predicates.
+- `sql_aggregate_order_by`: grouped aggregate ordered by aggregate value with a
+  `LIMIT`.
+- `sql_join_filter_order`: two direct-SST tables joined on a shared tag with
+  time filters, aggregate ordering, and `LIMIT`.
--- a/tests/perf/fixture-format.md
+++ b/tests/perf/fixture-format.md
@@ -0,0 +1,120 @@
+# Direct-SST fixture format
+
+The phase-1 generator should be a reusable fixture generator, not a collection
+of issue-specific data loaders.
+
+## Inputs
+
+Each case describes:
+
+```toml
+[case]
+name = "example_metric"
+description = "example direct-readable-SST regression"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 12345
+
+[[scenario.tables]]
+database = "public"
+name = "example_metric"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "instance"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 100, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "value"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.0, max = 100.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 1024
+rows_per_sst = 4096
+row_group_size = 512
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "round_robin"
+
+[[scenario.queries]]
+name = "count_all"
+kind = "sql"
+query = "SELECT count(*) FROM example_metric"
+warmup = 0
+iterations = 1
+```
+
+`[scenario]` is required. Other scenario variants are intentionally unsupported
+for now, but `scenario.kind` leaves room for future `write_then_query` and
+`cache_warm_query` configuration.
+
+The generator should use these declarations to produce:
+
+- object-store SST files written through the real Mito SST writer
+- manifest checkpoint and `_last_checkpoint`
+- fixture summary with file IDs, row counts, time ranges, and generated schema
+
+## Why this is generic
+
+The same fixture format should support different query-regression families:
+
+- PromQL/TQL time-index pushdown
+- SQL predicate pruning
+- projection and row-group pruning
+- series scan behavior
+- joins or aggregation over controlled layouts
+
+Issue-specific behavior belongs in case configs and thresholds, not in the
+generator implementation.
+
+## Direct generation vs realism
+
+Direct SST generation is phase-1 because it is fast and reproducible:
+
+- SST count, file time ranges, row groups, and label distributions are fixed by
+  the case config.
+- The same fixture can be used for base and candidate builds.
+- Large pruning-sensitive datasets can be created without spending CI time on
+  ingestion, memtable flush, or compaction.
+
+It is less realistic than ingestion-path data because it bypasses writes,
+memtables, flush scheduling, and compaction. That tradeoff is intentional for
+PR-level query regression. A later nightly/release suite can add ingestion-based
+cases for end-to-end realism.
+
+Multi-table cases are supported by generating one fixture directory per table.
+Each table is still limited to one region, and the runner passes `--table`, the
+discovered `--region-id`, and the discovered `--table-dir` for each table before
+materializing all generated region subtrees into the same datanode data home.
+This supports JOIN regression cases without changing existing single-table case
+files.
+
+Multi-table cases must use unique table names and unique `(database, name)`
+pairs. The runner derives each fixture subdirectory from table index, database,
+and table name, sanitizing path-unsafe characters to avoid collisions and unsafe
+paths.
+
+The preferred compatibility path is:
+
+1. create an empty table with the target build to seed catalog/table metadata;
+2. stop the process;
+3. generate readable SSTs and replacement manifest checkpoints offline using the
+   seeded region metadata;
+4. restart and query the fixture.
+
+Fully synthetic metadata is useful for generator smoke tests, but seeded metadata
+is safer for end-to-end query performance cases.
--- a/tests/perf/query_cases/promql_pushdown_7913/case.toml
+++ b/tests/perf/query_cases/promql_pushdown_7913/case.toml
@@ -0,0 +1,67 @@
+# Query performance regression case for GreptimeDB issue #7913.
+# This is a case config for the generic direct-SST fixture generator. The
+# generator must not hard-code PromQL/#7913 behavior; this file only supplies one
+# schema/layout/query instance.
+
+[case]
+name = "promql_pushdown_7913"
+description = "PromQL/TQL non-ms time-index query should prune SSTs for narrow time ranges"
+issue = "https://github.com/GreptimeTeam/greptimedb/issues/7913"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 7913
+
+
+[[scenario.tables]]
+database = "public"
+name = "promql_pushdown_7913"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "instance"]
+time_index = "greptime_timestamp"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 256, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "instance"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 4096, prefix = "instance" }
+
+[[scenario.tables.columns]]
+name = "download_mbs"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 }
+
+[[scenario.tables.columns]]
+name = "greptime_timestamp"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 1024
+rows_per_sst = 4096
+row_group_size = 512
+series_count = 4096
+start_unix_nanos = 1704067200000000000 # 2024-01-01T00:00:00Z
+step_nanos = 1000000000
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "round_robin"
+
+[[scenario.queries]]
+name = "narrow_window"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704070800000, 1704070860000, '15s') promql_pushdown_7913{host=~'host.*'}"
+warmup = 2
+iterations = 5
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 15
--- a/tests/perf/query_cases/smoke_direct_sst/case.toml
+++ b/tests/perf/query_cases/smoke_direct_sst/case.toml
@@ -0,0 +1,61 @@
+# Small fixture-generator smoke case. This is intentionally tiny so developers
+# can validate direct-SST generation without creating the large #7913 dataset.
+
+[case]
+name = "smoke_direct_sst"
+description = "Tiny generic direct-SST fixture smoke test"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 42
+
+
+[[scenario.tables]]
+database = "public"
+name = "smoke_direct_sst"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "instance"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 2, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "instance"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 4, prefix = "instance" }
+
+[[scenario.tables.columns]]
+name = "value"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.0, max = 10.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 2
+rows_per_sst = 8
+row_group_size = 4
+series_count = 4
+start_unix_nanos = 1704067200000000000
+step_nanos = 1000000000
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "round_robin"
+
+[[scenario.queries]]
+name = "count_all"
+kind = "sql"
+query = "SELECT count(*) FROM smoke_direct_sst"
+warmup = 0
+iterations = 1
--- a/tests/perf/query_cases/sql_aggregate_order_by/case.toml
+++ b/tests/perf/query_cases/sql_aggregate_order_by/case.toml
@@ -0,0 +1,63 @@
+# SQL aggregate ORDER BY case over direct SSTs.
+
+[case]
+name = "sql_aggregate_order_by"
+description = "Grouped aggregate ordered by aggregate value with LIMIT"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 1002
+
+
+[[scenario.tables]]
+database = "public"
+name = "sql_aggregate_order_by"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["service", "host"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "service"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 8, prefix = "service" }
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 64, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "cpu"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.0, max = 100.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 64
+rows_per_sst = 1024
+row_group_size = 256
+series_count = 64
+start_unix_nanos = 1704067200000000000
+step_nanos = 1000000000
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "round_robin"
+
+[[scenario.queries]]
+name = "aggregate_service_top"
+kind = "sql"
+query = "SELECT service, avg(cpu) AS avg_cpu, sum(cpu) AS sum_cpu FROM sql_aggregate_order_by WHERE ts >= '2024-01-01 00:00:00+0000'::TIMESTAMP AND ts < '2024-01-01 08:00:00+0000'::TIMESTAMP GROUP BY service ORDER BY avg_cpu DESC LIMIT 5"
+warmup = 1
+iterations = 3
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 50
--- a/tests/perf/query_cases/sql_join_filter_order/case.toml
+++ b/tests/perf/query_cases/sql_join_filter_order/case.toml
@@ -0,0 +1,95 @@
+# SQL two-table JOIN case over direct SSTs.
+
+[case]
+name = "sql_join_filter_order"
+description = "Join fact and dimension-like tables on host with filters, aggregate ordering, and LIMIT"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 1003
+
+
+[[scenario.tables]]
+database = "public"
+name = "sql_join_fact"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "instance"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 16, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "instance"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 64, prefix = "instance" }
+
+[[scenario.tables.columns]]
+name = "value"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 1.0, max = 500.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[[scenario.tables]]
+database = "public"
+name = "sql_join_dim"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "zone"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 16, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "zone"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 4, prefix = "zone" }
+
+[[scenario.tables.columns]]
+name = "weight"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.1, max = 10.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 32
+rows_per_sst = 512
+row_group_size = 128
+series_count = 64
+start_unix_nanos = 1704067200000000000
+step_nanos = 1000000000
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "round_robin"
+
+[[scenario.queries]]
+name = "join_host_aggregate_top"
+kind = "sql"
+query = "SELECT f.host, d.zone, avg(f.value * d.weight) AS score FROM sql_join_fact f JOIN sql_join_dim d ON f.host = d.host WHERE f.ts >= '2024-01-01 00:00:00+0000'::TIMESTAMP AND f.ts < '2024-01-01 04:00:00+0000'::TIMESTAMP AND d.ts >= '2024-01-01 00:00:00+0000'::TIMESTAMP AND d.ts < '2024-01-01 04:00:00+0000'::TIMESTAMP AND d.zone IN ('zone1', 'zone2') GROUP BY f.host, d.zone ORDER BY score DESC LIMIT 10"
+warmup = 1
+iterations = 3
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 50
--- a/tests/perf/query_cases/sql_topk_order_by/case.toml
+++ b/tests/perf/query_cases/sql_topk_order_by/case.toml
@@ -0,0 +1,63 @@
+# SQL TopK ORDER BY case over direct SSTs.
+
+[case]
+name = "sql_topk_order_by"
+description = "ORDER BY DOUBLE field DESC LIMIT with time and tag filters"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 1001
+
+
+[[scenario.tables]]
+database = "public"
+name = "sql_topk_order_by"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "instance"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 16, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "instance"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 64, prefix = "instance" }
+
+[[scenario.tables.columns]]
+name = "value"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(9)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 64
+rows_per_sst = 1024
+row_group_size = 256
+series_count = 64
+start_unix_nanos = 1704067200000000000
+step_nanos = 1000000000
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "round_robin"
+
+[[scenario.queries]]
+name = "topk_value_desc"
+kind = "sql"
+query = "SELECT host, instance, value, ts FROM sql_topk_order_by WHERE ts >= '2024-01-01 00:10:00+0000'::TIMESTAMP AND ts < '2024-01-01 02:00:00+0000'::TIMESTAMP AND host IN ('host1', 'host3', 'host5') ORDER BY value DESC LIMIT 20"
+warmup = 1
+iterations = 3
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 50
--- a/tests/perf/query_regression_runner.py
+++ b/tests/perf/query_regression_runner.py
@@ -0,0 +1,801 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Base-vs-candidate query performance regression runner."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import shutil
+import socket
+import statistics
+import subprocess
+import sys
+import time
+import tomllib
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+@dataclass(frozen=True)
+class RunTarget:
+    name: str
+    binary: Path
+    work_dir: Path
+    data_dir: Path
+    fixture_dir: Path
+    report_path: Path
+    http_port: int
+    grpc_port: int
+    mysql_port: int
+    postgres_port: int
+    metasrv_rpc_port: int
+    metasrv_http_port: int
+    datanode_rpc_port: int
+    datanode_http_port: int
+    datanode_data_dir: Path
+
+
+def parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description="Run a query perf case against base and candidate binaries.")
+    p.add_argument("--case", required=True, type=Path)
+    p.add_argument("--base-bin", required=True, type=Path)
+    p.add_argument("--candidate-bin", required=True, type=Path)
+    p.add_argument("--fixture-generator", type=Path)
+    p.add_argument("--work-dir", required=True, type=Path)
+    p.add_argument("--reuse-fixture", action="store_true")
+    p.add_argument("--allow-large-fixture", action="store_true")
+    p.add_argument("--dry-run", action="store_true")
+    p.add_argument("--fixture-only", action="store_true", help="old smoke mode: generate/materialize fixture only")
+    p.add_argument("--reuse-work-dir", action="store_true", help="allow non-empty base/candidate work dirs")
+    p.add_argument("--http-timeout", type=float, default=120.0, help="HTTP SQL timeout seconds")
+    return p.parse_args()
+
+
+def load_case(path: Path) -> dict[str, Any]:
+    with path.open("rb") as f:
+        return tomllib.load(f)
+
+
+def require_binary(path: Path, name: str, *, dry_run: bool) -> None:
+    if dry_run:
+        return
+    if not path.exists():
+        raise FileNotFoundError(f"{name} binary does not exist: {path}")
+    if not os.access(path, os.X_OK):
+        raise PermissionError(f"{name} binary is not executable: {path}")
+
+
+def allocate_ports(n: int) -> list[int]:
+    socks = []
+    try:
+        for _ in range(n):
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.bind(("127.0.0.1", 0))
+            socks.append(s)
+        return [s.getsockname()[1] for s in socks]
+    finally:
+        for s in socks:
+            s.close()
+
+
+def make_target(name: str, binary: Path, root: Path, ports: list[int], fixture_dir: Path | None = None) -> RunTarget:
+    work_dir = root / name
+    return RunTarget(
+        name,
+        binary,
+        work_dir,
+        work_dir / "cluster_data",
+        fixture_dir or root / "fixture",
+        work_dir / "report.json",
+        ports[4],
+        ports[5],
+        ports[6],
+        ports[7],
+        ports[0],
+        ports[1],
+        ports[2],
+        ports[3],
+        work_dir / "datanode-0" / "data",
+    )
+
+
+def run_command(cmd: list[str], cwd: Path | None = None) -> dict[str, Any]:
+    started = time.monotonic()
+    proc = subprocess.run(cmd, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
+    return {"cmd": cmd, "cwd": str(cwd) if cwd else None, "returncode": proc.returncode, "elapsed_seconds": time.monotonic() - started, "stdout": proc.stdout, "stderr": proc.stderr}
+
+
+def sql_ident(name: str) -> str:
+    return '"' + name.replace('"', '""') + '"'
+
+
+def column_sql(col: dict[str, Any]) -> str:
+    return f"{sql_ident(col['name'])} {col['type']}"
+
+
+def scenario(case: dict[str, Any]) -> dict[str, Any]:
+    value = case.get("scenario")
+    if not isinstance(value, dict):
+        raise ValueError("case requires [scenario] with kind = 'direct_readable_sst'")
+    kind = value.get("kind")
+    if kind != "direct_readable_sst":
+        raise ValueError(f"unsupported scenario kind {kind!r}; only 'direct_readable_sst' is supported")
+    return value
+
+
+def case_tables(case: dict[str, Any]) -> list[dict[str, Any]]:
+    value = scenario(case)
+    tables = value.get("tables") or []
+    if not tables or value.get("layout", {}).get("regions") != 1:
+        raise ValueError("runner supports one or more tables and exactly one region per table")
+    pairs = [(table.get("database"), table.get("name")) for table in tables]
+    if len(set(pairs)) != len(pairs):
+        raise ValueError("duplicate (database, name) table entries are not supported")
+    names = [table.get("name") for table in tables]
+    if len(set(names)) != len(names):
+        raise ValueError("duplicate table names are not supported because fixture generator --table selects by name")
+    return list(tables)
+
+
+def fixture_subdir(table: dict[str, Any], index: int) -> str:
+    raw = f"{index:02d}_{table['database']}_{table['name']}"
+    safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", raw).strip("._-")
+    return safe or f"table_{index:02d}"
+
+
+def table_fixture_dir(root: Path, tables: list[dict[str, Any]], table: dict[str, Any], index: int) -> Path:
+    return root if len(tables) == 1 else root / fixture_subdir(table, index)
+
+
+def create_table_sql(table: dict[str, Any]) -> str:
+    cols = ",\n  ".join(column_sql(c) for c in table["columns"])
+    pk = ", ".join(sql_ident(c) for c in table.get("primary_key", []))
+    opts = []
+    if "append_mode" in table:
+        opts.append(("append_mode", str(table["append_mode"]).lower()))
+    if table.get("sst_format"):
+        opts.append(("sst_format", table["sst_format"]))
+    with_sql = ""
+    if opts:
+        with_sql = "\nWITH (" + ", ".join(f"'{k}'='{v}'" for k, v in opts) + ")"
+    return f"CREATE TABLE {sql_ident(table['name'])} (\n  {cols},\n  TIME INDEX ({sql_ident(table['time_index'])}),\n  PRIMARY KEY ({pk})\n) ENGINE=mito{with_sql};"
+
+
+def planned_queries(case: dict[str, Any]) -> list[dict[str, Any]]:
+    queries = scenario(case).get("queries", [])
+    if isinstance(queries, dict):
+        return [{"name": name, **value} for name, value in queries.items()]
+    return list(queries)
+
+
+def http_post_sql(port: int, sql: str, db: str, timeout: float) -> dict[str, Any]:
+    data = urllib.parse.urlencode({"sql": sql, "db": db, "format": "json"}).encode()
+    req = urllib.request.Request(f"http://127.0.0.1:{port}/v1/sql", data=data, method="POST")
+    started = time.monotonic()
+    elapsed_ms = (time.monotonic() - started) * 1000.0
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            raw = resp.read().decode()
+            status = resp.status
+        elapsed_ms = (time.monotonic() - started) * 1000.0
+        try:
+            body: Any = json.loads(raw)
+        except json.JSONDecodeError:
+            body = {"raw": raw}
+        ok = status < 400 and not response_has_error(body)
+        return {"ok": ok, "status": status, "latency_ms": elapsed_ms, "response": body, "sql": sql}
+    except urllib.error.HTTPError as e:
+        elapsed_ms = (time.monotonic() - started) * 1000.0
+        raw = e.read().decode(errors="replace")
+        try:
+            body = json.loads(raw)
+        except json.JSONDecodeError:
+            body = {"raw": raw}
+        return {
+            "ok": False,
+            "status": e.code,
+            "latency_ms": elapsed_ms,
+            "response": body,
+            "error": repr(e),
+            "sql": sql,
+        }
+    except Exception as e:  # noqa: BLE001 - report HTTP/query failures as JSON
+        elapsed_ms = (time.monotonic() - started) * 1000.0
+        return {"ok": False, "status": None, "latency_ms": elapsed_ms, "error": repr(e), "sql": sql}
+
+
+def response_has_error(body: Any) -> bool:
+    """Detect top-level GreptimeDB HTTP error envelopes without inspecting rows.
+
+    Query outputs may legitimately contain columns named `code` or `error`, so
+    this intentionally avoids recursive checks through result rows.
+    """
+    if isinstance(body, dict):
+        if body.get("error") or body.get("err_msg") or body.get("error_msg"):
+            return True
+        if "error_code" in body and str(body.get("error_code", "")).lower() not in ("", "0", "success"):
+            return True
+        if "code" in body and "output" not in body and str(body.get("code", "")).lower() not in ("", "0", "success"):
+            return True
+    return False
+
+
+def wait_health(port: int, timeout_s: float = 60.0) -> None:
+    deadline = time.monotonic() + timeout_s
+    last = None
+    while time.monotonic() < deadline:
+        try:
+            with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=2) as r:
+                if r.status < 500:
+                    return
+        except Exception as e:  # noqa: BLE001 - diagnostic loop
+            last = e
+        time.sleep(0.5)
+    raise TimeoutError(f"health check timed out on port {port}: {last}")
+
+
+class DistributedCluster:
+    """Local metasrv + one datanode + frontend cluster for query-mode runs."""
+
+    def __init__(self, target: RunTarget):
+        self.target = target
+        self.procs: dict[str, subprocess.Popen[bytes]] = {}
+
+    def component_report(self) -> dict[str, Any]:
+        return {
+            "metasrv": {
+                "grpc": f"127.0.0.1:{self.target.metasrv_rpc_port}",
+                "http": f"127.0.0.1:{self.target.metasrv_http_port}",
+                "logs": str(self.target.work_dir / "logs" / "metasrv"),
+            },
+            "datanode_0": {
+                "node_id": 0,
+                "grpc": f"127.0.0.1:{self.target.datanode_rpc_port}",
+                "http": f"127.0.0.1:{self.target.datanode_http_port}",
+                "data_home": str(self.target.datanode_data_dir),
+                "logs": str(self.target.work_dir / "logs" / "datanode-0"),
+            },
+            "frontend": {
+                "http": f"127.0.0.1:{self.target.http_port}",
+                "grpc": f"127.0.0.1:{self.target.grpc_port}",
+                "mysql": f"127.0.0.1:{self.target.mysql_port}",
+                "postgres": f"127.0.0.1:{self.target.postgres_port}",
+                "logs": str(self.target.work_dir / "logs" / "frontend"),
+            },
+        }
+
+    def _spawn(self, name: str, args: list[str]) -> None:
+        logs = self.target.work_dir / "logs" / name
+        logs.mkdir(parents=True, exist_ok=True)
+        with (logs / "stdout.log").open("ab") as out, (logs / "stderr.log").open("ab") as err:
+            self.procs[name] = subprocess.Popen(args, stdout=out, stderr=err)
+
+    def _ensure_metasrv_alive(self) -> None:
+        proc = self.procs.get("metasrv")
+        if proc is None or proc.poll() is not None:
+            raise RuntimeError("metasrv exited; memory-store metadata is no longer valid")
+
+    def start_metasrv(self) -> None:
+        log_dir = self.target.work_dir / "logs" / "metasrv"
+        self._spawn(
+            "metasrv",
+            [
+                str(self.target.binary),
+                "metasrv",
+                "start",
+                "--grpc-bind-addr",
+                f"127.0.0.1:{self.target.metasrv_rpc_port}",
+                "--grpc-server-addr",
+                f"127.0.0.1:{self.target.metasrv_rpc_port}",
+                "--http-addr",
+                f"127.0.0.1:{self.target.metasrv_http_port}",
+                "--backend",
+                "memory-store",
+                "--enable-region-failover",
+                "false",
+                "--log-dir",
+                str(log_dir),
+            ],
+        )
+        wait_health(self.target.metasrv_http_port)
+
+    def start_datanode(self) -> None:
+        self._ensure_metasrv_alive()
+        log_dir = self.target.work_dir / "logs" / "datanode-0"
+        self.target.datanode_data_dir.mkdir(parents=True, exist_ok=True)
+        self._spawn(
+            "datanode",
+            [
+                str(self.target.binary),
+                "datanode",
+                "start",
+                "--grpc-bind-addr",
+                f"127.0.0.1:{self.target.datanode_rpc_port}",
+                "--grpc-server-addr",
+                f"127.0.0.1:{self.target.datanode_rpc_port}",
+                "--http-addr",
+                f"127.0.0.1:{self.target.datanode_http_port}",
+                "--data-home",
+                str(self.target.datanode_data_dir),
+                "--log-dir",
+                str(log_dir),
+                "--node-id",
+                "0",
+                "--metasrv-addrs",
+                f"127.0.0.1:{self.target.metasrv_rpc_port}",
+            ],
+        )
+        wait_health(self.target.datanode_http_port)
+
+    def start_frontend(self) -> None:
+        self._ensure_metasrv_alive()
+        log_dir = self.target.work_dir / "logs" / "frontend"
+        self._spawn(
+            "frontend",
+            [
+                str(self.target.binary),
+                "frontend",
+                "start",
+                "--metasrv-addrs",
+                f"127.0.0.1:{self.target.metasrv_rpc_port}",
+                "--http-addr",
+                f"127.0.0.1:{self.target.http_port}",
+                "--grpc-bind-addr",
+                f"127.0.0.1:{self.target.grpc_port}",
+                "--grpc-server-addr",
+                f"127.0.0.1:{self.target.grpc_port}",
+                "--mysql-addr",
+                f"127.0.0.1:{self.target.mysql_port}",
+                "--postgres-addr",
+                f"127.0.0.1:{self.target.postgres_port}",
+                "--log-dir",
+                str(log_dir),
+            ],
+        )
+        wait_health(self.target.http_port)
+
+    def start_all(self) -> None:
+        self.start_metasrv()
+        self.start_datanode()
+        self.start_frontend()
+
+    def stop_component(self, name: str) -> None:
+        proc = self.procs.pop(name, None)
+        if proc is None:
+            return
+        proc.terminate()
+        try:
+            proc.wait(timeout=20)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.wait(timeout=20)
+
+    def stop_all(self) -> None:
+        for name in ("frontend", "datanode", "metasrv"):
+            self.stop_component(name)
+
+    def restart_frontend(self) -> None:
+        self.stop_component("frontend")
+        self.start_frontend()
+
+
+def extract_rows(body: Any) -> list[Any]:
+    """Extract rows from GreptimeDB HTTP JSON in a tolerant way."""
+    rows: list[Any] = []
+    if isinstance(body, dict):
+        for key in ("data", "rows", "records", "output"):
+            if key in body:
+                value = body[key]
+                if key in ("data", "rows") and isinstance(value, list):
+                    rows.extend(value)
+                else:
+                    rows.extend(extract_rows(value))
+    elif isinstance(body, list):
+        if body and all(not isinstance(v, (dict, list)) for v in body):
+            rows.append(body)
+        else:
+            for item in body:
+                rows.extend(extract_rows(item))
+    return rows
+
+
+def row_value(row: Any, idx: int, name: str) -> Any:
+    if isinstance(row, dict):
+        for key in (name, name.upper(), name.lower()):
+            if key in row:
+                return row[key]
+        return None
+    if isinstance(row, list):
+        return row[idx]
+    return row
+
+
+def discover_region_via_frontend(target: RunTarget, table: dict[str, Any], http_timeout: float) -> dict[str, Any]:
+    schema = table["database"]
+    table_name = table["name"].replace("'", "''")
+    schema_name = schema.replace("'", "''")
+    table_sql = f"SELECT table_id FROM information_schema.tables WHERE table_schema = '{schema_name}' AND table_name = '{table_name}'"
+    table_result = http_post_sql(target.http_port, table_sql, schema, http_timeout)
+    if not table_result["ok"]:
+        raise RuntimeError(f"table_id discovery failed: {table_result}")
+    table_rows = extract_rows(table_result.get("response"))
+    if len(table_rows) != 1:
+        raise RuntimeError(f"expected one information_schema.tables row, got {len(table_rows)}: {table_result}")
+    table_id = int(row_value(table_rows[0], 0, "table_id"))
+
+    region_sql = f"SELECT region_id, peer_id, peer_addr, is_leader, status FROM information_schema.region_peers WHERE table_schema = '{schema_name}' AND table_name = '{table_name}'"
+    region_result = http_post_sql(target.http_port, region_sql, schema, http_timeout)
+    if not region_result["ok"]:
+        raise RuntimeError(f"region_peers discovery failed: {region_result}")
+    region_rows = extract_rows(region_result.get("response"))
+    if len(region_rows) != 1:
+        raise RuntimeError(f"expected one information_schema.region_peers row, got {len(region_rows)}: {region_result}")
+    row = region_rows[0]
+    region_id = int(row_value(row, 0, "region_id"))
+    peer_id = int(row_value(row, 1, "peer_id"))
+    peer_addr = row_value(row, 2, "peer_addr")
+    is_leader = row_value(row, 3, "is_leader")
+    status = row_value(row, 4, "status")
+    if str(is_leader).lower() not in ("yes", "true"):
+        raise RuntimeError(f"expected leader region peer, got is_leader={is_leader}: {region_result}")
+    if str(status).upper() != "ALIVE":
+        raise RuntimeError(f"expected ALIVE region peer, got status={status}: {region_result}")
+    if peer_id != 0:
+        raise RuntimeError(f"expected region leader on datanode peer_id=0, got {peer_id}")
+    computed_table_id = region_id >> 32
+    region_seq = region_id & 0xFFFFFFFF
+    if computed_table_id != table_id:
+        raise RuntimeError(f"table_id mismatch: tables={table_id}, region_id-derived={computed_table_id}")
+    table_dir = f"data/greptime/{schema}/{table_id}/"
+    region_dir = f"data/greptime/{schema}/{table_id}/{table_id}_{region_seq:010}"
+    return {
+        "catalog": "greptime",
+        "schema": schema,
+        "table_id": table_id,
+        "region_id": region_id,
+        "region_seq": region_seq,
+        "table_dir": table_dir,
+        "region_dir": region_dir,
+        "peer_id": peer_id,
+        "peer_addr": peer_addr,
+        "is_leader": is_leader,
+        "status": status,
+        "discovery_queries": {"table": table_result, "region": region_result},
+    }
+
+
+def assert_fixture_summary(summary: dict[str, Any], *, region_id: int | None, table_dir: str | None, region_dir: str | None = None, table_name: str | None = None, database: str | None = None) -> None:
+    if table_name is not None and summary.get("table") != table_name:
+        raise RuntimeError(f"fixture table mismatch: {summary.get('table')} != {table_name}")
+    if database is not None and summary.get("database") != database:
+        raise RuntimeError(f"fixture database mismatch: {summary.get('database')} != {database}")
+    if region_id is not None and int(summary["region_id"]) != region_id:
+        raise RuntimeError(f"fixture region_id mismatch: {summary['region_id']} != {region_id}")
+    if table_dir is not None and summary["table_dir"] != table_dir:
+        raise RuntimeError(f"fixture table_dir mismatch: {summary['table_dir']} != {table_dir}")
+    if region_dir is not None and summary["region_dir"].strip("/") != region_dir.strip("/"):
+        raise RuntimeError(f"fixture region_dir mismatch: {summary['region_dir']} != {region_dir}")
+
+
+def generate_fixture(generator: Path | None, case_path: Path, fixture_dir: Path, *, dry_run: bool, reuse_fixture: bool, allow_large_fixture: bool, table_name: str | None = None, database: str | None = None, region_id: int | None = None, table_dir: str | None = None, region_dir: str | None = None) -> dict[str, Any]:
+    if generator is None:
+        return {"status": "skipped", "reason": "no fixture generator provided"}
+    summary_path = fixture_dir / "summary.json"
+    if reuse_fixture and summary_path.exists():
+        summary = json.loads(summary_path.read_text())
+        assert_fixture_summary(summary, region_id=region_id, table_dir=table_dir, region_dir=region_dir, table_name=table_name, database=database)
+        return {"status": "reused", "fixture_dir": str(fixture_dir), "summary": summary}
+    if fixture_dir.exists() and not reuse_fixture:
+        shutil.rmtree(fixture_dir)
+    fixture_dir.mkdir(parents=True, exist_ok=True)
+    cmd = [str(generator), "--case", str(case_path), "--out-dir", str(fixture_dir)]
+    if table_name is not None:
+        cmd += ["--table", table_name]
+    if region_id is not None:
+        cmd += ["--region-id", str(region_id)]
+    if table_dir is not None:
+        cmd += ["--table-dir", table_dir]
+    if allow_large_fixture:
+        cmd.append("--allow-large")
+    if dry_run:
+        return {"status": "dry-run", "cmd": cmd}
+    result = run_command(cmd)
+    result["status"] = "ok" if result["returncode"] == 0 else "failed"
+    if result["returncode"] != 0:
+        raise RuntimeError(f"fixture generator failed: {result['stderr'][:2000]}")
+    summary = json.loads(summary_path.read_text())
+    assert_fixture_summary(summary, region_id=region_id, table_dir=table_dir, region_dir=region_dir, table_name=table_name, database=database)
+    result["summary"] = summary
+    return result
+
+
+def copy_tree_contents(src: Path, dst: Path) -> None:
+    if not src.exists():
+        raise FileNotFoundError(f"fixture path does not exist: {src}")
+    dst.mkdir(parents=True, exist_ok=True)
+    for child in src.iterdir():
+        target = dst / child.name
+        if child.is_dir():
+            shutil.copytree(child, target, dirs_exist_ok=True)
+        else:
+            shutil.copy2(child, target)
+
+
+def materialize_fixture(target: RunTarget, *, dry_run: bool, preserve_state: bool, expected_region_dir: str | None = None, fixture_dir: Path | None = None, reset_data: bool = True) -> dict[str, Any]:
+    fixture_dir = fixture_dir or target.fixture_dir
+    summary_path = fixture_dir / "summary.json"
+    materialize_root = target.datanode_data_dir if preserve_state else target.data_dir
+    if dry_run:
+        return {"status": "dry-run", "summary_path": str(summary_path), "data_dir": str(materialize_root)}
+    summary = json.loads(summary_path.read_text())
+    object_store_dir = fixture_dir / "object-store"
+    manifest_dir = fixture_dir / "manifest"
+    region_dir = summary["region_dir"].strip("/")
+    target_region_dir = materialize_root / region_dir
+    data_root = materialize_root.resolve()
+    resolved_region_dir = target_region_dir.resolve(strict=False)
+    if data_root != resolved_region_dir and data_root not in resolved_region_dir.parents:
+        raise RuntimeError(f"unsafe fixture region_dir escapes data_dir: {region_dir}")
+    if expected_region_dir is not None and region_dir != expected_region_dir.strip("/"):
+        raise RuntimeError(f"fixture region_dir {region_dir} does not equal discovered {expected_region_dir}")
+    target_manifest_dir = target_region_dir / "manifest"
+    if not preserve_state and reset_data and materialize_root.exists():
+        shutil.rmtree(materialize_root)
+    materialize_root.mkdir(parents=True, exist_ok=True)
+    if preserve_state and target_region_dir.exists():
+        shutil.rmtree(target_region_dir)
+    if preserve_state:
+        fixture_region_dir = object_store_dir / region_dir
+        copy_tree_contents(fixture_region_dir, target_region_dir)
+    else:
+        copy_tree_contents(object_store_dir, materialize_root)
+    if target_manifest_dir.exists():
+        shutil.rmtree(target_manifest_dir)
+    copy_tree_contents(manifest_dir, target_manifest_dir)
+    return {"status": "ok", "summary_path": str(summary_path), "data_dir": str(materialize_root), "region_dir": region_dir, "manifest_dir": str(target_manifest_dir), "preserve_state": preserve_state}
+
+
+def response_text(body: Any) -> str:
+    return json.dumps(body, sort_keys=True) if not isinstance(body, str) else body
+
+
+def validate_show_create(result: dict[str, Any], table: dict[str, Any]) -> list[str]:
+    text = response_text(result.get("response", {})).lower()
+    errors = []
+    if table["name"].lower() not in text:
+        errors.append("SHOW CREATE output does not contain table name")
+    if "engine" not in text or "mito" not in text:
+        errors.append("SHOW CREATE output does not mention ENGINE=mito")
+    if "append_mode" in table and "append_mode" not in text:
+        errors.append("SHOW CREATE output does not mention append_mode")
+    if table.get("sst_format") and "sst_format" not in text:
+        errors.append("SHOW CREATE output does not mention sst_format")
+    return errors
+
+
+def run_queries(target: RunTarget, case: dict[str, Any], tables: list[dict[str, Any]], http_timeout: float) -> dict[str, Any]:
+    table = tables[0]
+    db = table["database"]
+    queries = planned_queries(case)
+    if not queries:
+        queries = [{"name": "count_all", "kind": "sql", "query": f"SELECT count(*) FROM {sql_ident(table['name'])}", "warmup": 0, "iterations": 1}]
+    validations = []
+    validation_errors = []
+    for table in tables:
+        for sql in [f"SHOW CREATE TABLE {sql_ident(table['name'])}"]:
+            result = http_post_sql(target.http_port, sql, table["database"], http_timeout)
+            if not result["ok"]:
+                validation_errors.append({"sql": sql, "error": result.get("error"), "response": result.get("response")})
+            else:
+                for error in validate_show_create(result, table):
+                    validation_errors.append({"sql": sql, "error": error, "response": result.get("response")})
+            validations.append(result)
+    if queries:
+        result = http_post_sql(target.http_port, queries[0]["query"], db, http_timeout)
+        if not result["ok"]:
+            validation_errors.append({"sql": queries[0]["query"], "error": result.get("error"), "response": result.get("response")})
+        validations.append(result)
+    measurements = []
+    for q in queries:
+        for _ in range(int(q.get("warmup", 0))):
+            warmup = http_post_sql(target.http_port, q["query"], db, http_timeout)
+            if not warmup["ok"]:
+                validation_errors.append({"sql": q["query"], "phase": "warmup", "error": warmup.get("error"), "response": warmup.get("response")})
+        samples = []
+        for _ in range(int(q.get("iterations", 1))):
+            result = http_post_sql(target.http_port, q["query"], db, http_timeout)
+            result["execution_time_ms"] = extract_execution_time(result.get("response"))
+            samples.append(result)
+        good_lats = [s["latency_ms"] for s in samples if s["ok"]]
+        measurements.append({"name": q.get("name"), "kind": q.get("kind"), "iterations": len(samples), "samples": samples, "latency_ms_median": statistics.median(good_lats) if good_lats else None, "latency_ms_p95": percentile(good_lats, 95) if good_lats else None, "status": "ok" if len(good_lats) == len(samples) else "failed"})
+    return {"validation": validations, "validation_errors": validation_errors, "measurements": measurements, "status": "failed" if validation_errors or any(m["status"] == "failed" for m in measurements) else "ok"}
+
+
+def extract_execution_time(body: Any) -> Any:
+    if isinstance(body, dict):
+        for key in ("execution_time_ms", "execution_time", "elapsed"):
+            if key in body:
+                return body[key]
+        for value in body.values():
+            found = extract_execution_time(value)
+            if found is not None:
+                return found
+    if isinstance(body, list):
+        for value in body:
+            found = extract_execution_time(value)
+            if found is not None:
+                return found
+    return None
+
+
+def percentile(values: list[float], pct: float) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    idx = min(len(ordered) - 1, max(0, round((pct / 100.0) * (len(ordered) - 1))))
+    return ordered[idx]
+
+
+def enforce_thresholds(case: dict[str, Any], base: dict[str, Any], candidate: dict[str, Any]) -> list[dict[str, Any]]:
+    results = []
+    base_by_name = {m["name"]: m for m in base.get("measurements", [])}
+    for cm in candidate.get("measurements", []):
+        bm = base_by_name.get(cm["name"])
+        qcfg = next((q for q in planned_queries(case) if q.get("name") == cm["name"]), {})
+        th = qcfg.get("thresholds") or {}
+        max_reg = th.get("max_candidate_latency_regression_pct")
+        if max_reg is not None and not bm:
+            results.append({"query": cm["name"], "threshold": "max_candidate_latency_regression_pct", "status": "failed", "reason": "missing base measurement"})
+        elif max_reg is not None and bm and bm.get("latency_ms_median") in (None, 0):
+            results.append({"query": cm["name"], "threshold": "max_candidate_latency_regression_pct", "status": "failed", "reason": "base median latency is missing or zero", "base_latency_ms_median": bm.get("latency_ms_median")})
+        elif max_reg is not None and bm and cm.get("latency_ms_median") is None:
+            results.append({"query": cm["name"], "threshold": "max_candidate_latency_regression_pct", "status": "failed", "reason": "missing candidate measurement"})
+        elif bm and max_reg is not None:
+            ratio = (cm["latency_ms_median"] - bm["latency_ms_median"]) / bm["latency_ms_median"] * 100.0
+            results.append({"query": cm["name"], "threshold": "max_candidate_latency_regression_pct", "status": "passed" if ratio <= max_reg else "failed", "actual_pct": ratio, "limit_pct": max_reg})
+        for key in th:
+            if key != "max_candidate_latency_regression_pct":
+                results.append({"query": cm["name"], "threshold": key, "status": "failed", "reason": "unsupported threshold"})
+    return results
+
+
+def write_json(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+
+
+def require_fresh_work_dirs(targets: list[RunTarget], *, reuse_work_dir: bool, dry_run: bool, fixture_only: bool) -> None:
+    if dry_run or reuse_work_dir or fixture_only:
+        return
+    for target in targets:
+        if target.work_dir.exists() and any(target.work_dir.iterdir()):
+            raise RuntimeError(f"target work_dir exists and is non-empty: {target.work_dir}; pass --reuse-work-dir to override")
+
+
+def main() -> int:
+    args = parse_args()
+    case_path = args.case.resolve()
+    case = load_case(case_path)
+    scenario_config = scenario(case)
+    tables = case_tables(case)
+    work_root = args.work_dir.resolve()
+    work_root.mkdir(parents=True, exist_ok=True)
+    require_binary(args.base_bin, "base", dry_run=args.dry_run or args.fixture_only)
+    require_binary(args.candidate_bin, "candidate", dry_run=args.dry_run or args.fixture_only)
+    if not args.dry_run and args.fixture_generator is None:
+        raise ValueError("--fixture-generator is required unless --dry-run is set")
+    if args.fixture_generator is not None:
+        require_binary(args.fixture_generator, "fixture generator", dry_run=args.dry_run)
+
+    ports = allocate_ports(16)
+    targets = [make_target("base", args.base_bin.resolve(), work_root, ports[:8]), make_target("candidate", args.candidate_bin.resolve(), work_root, ports[8:])]
+    require_fresh_work_dirs(targets, reuse_work_dir=args.reuse_work_dir, dry_run=args.dry_run, fixture_only=args.fixture_only)
+    fixture_dir = work_root / "fixture"
+    report: dict[str, Any] = {"case_path": str(case_path), "case": case.get("case", {}), "scenario": scenario_config, "queries": planned_queries(case), "dry_run": args.dry_run, "fixture_only": args.fixture_only, "query_mode": "fixture-only" if args.fixture_only else "distributed", "reuse_work_dir": args.reuse_work_dir, "http_timeout": args.http_timeout, "targets": [], "thresholds": [], "status": "planned" if args.dry_run else "running"}
+
+    if args.fixture_only or args.dry_run:
+        generations = []
+        for table_idx, table in enumerate(tables):
+            per_table_fixture_dir = table_fixture_dir(fixture_dir, tables, table, table_idx)
+            generations.append(generate_fixture(args.fixture_generator, case_path, per_table_fixture_dir, dry_run=args.dry_run, reuse_fixture=args.reuse_fixture, allow_large_fixture=args.allow_large_fixture, table_name=table["name"] if len(tables) > 1 else None, database=table["database"] if len(tables) > 1 else None))
+        report["fixture_generation"] = generations[0] if len(generations) == 1 else generations
+        for target in targets:
+            target.work_dir.mkdir(parents=True, exist_ok=True)
+            mats = []
+            for table_idx, table in enumerate(tables):
+                per_table_fixture_dir = table_fixture_dir(fixture_dir, tables, table, table_idx)
+                mats.append(materialize_fixture(target, dry_run=args.dry_run, preserve_state=False, fixture_dir=per_table_fixture_dir, reset_data=not mats))
+            tr = {"name": target.name, "binary": str(target.binary), "work_dir": str(target.work_dir), "data_dir": str(target.data_dir), "fixture_dir": str(fixture_dir), "fixture_materialization": mats[0] if len(mats) == 1 else mats, "measurements": [], "status": "planned" if args.dry_run else "fixture-ready"}
+            write_json(target.report_path, tr)
+            report["targets"].append(tr)
+        report["status"] = "planned" if args.dry_run else "fixture-ready"
+        write_json(work_root / "query-regression-report.json", report)
+        print(json.dumps(report, indent=2, sort_keys=True))
+        return 0
+
+    clusters: list[DistributedCluster] = []
+    try:
+        discovered = []
+        for target in targets:
+            target.work_dir.mkdir(parents=True, exist_ok=True)
+            cluster = DistributedCluster(target)
+            clusters.append(cluster)
+            cluster.start_all()
+            create_results = []
+            metas = []
+            for table in tables:
+                create_result = http_post_sql(target.http_port, create_table_sql(table), table["database"], args.http_timeout)
+                if not create_result["ok"]:
+                    raise RuntimeError(f"CREATE TABLE {table['name']} failed for {target.name}: {create_result}")
+                create_results.append({"table": table["name"], "result": create_result})
+                metas.append({"table": table["name"], **discover_region_via_frontend(target, table, args.http_timeout)})
+            cluster.stop_component("datanode")
+            cluster._ensure_metasrv_alive()
+            discovered.append(metas)
+            report["targets"].append({"name": target.name, "binary": str(target.binary), "work_dir": str(target.work_dir), "data_dir": str(target.data_dir), "datanode_data_home": str(target.datanode_data_dir), "components": cluster.component_report(), "create_table": create_results[0]["result"] if len(create_results) == 1 else create_results, "discovered": metas[0] if len(metas) == 1 else metas, "discovered_tables": metas})
+
+        if len(discovered[0]) != len(discovered[1]) or any(a["table"] != b["table"] or a["region_id"] != b["region_id"] or a["table_dir"] != b["table_dir"] for a, b in zip(discovered[0], discovered[1])):
+            raise RuntimeError(f"base/candidate metadata mismatch: {discovered}")
+        generations = []
+        for table_idx, meta in enumerate(discovered[0]):
+            per_table_fixture_dir = table_fixture_dir(fixture_dir, tables, tables[table_idx], table_idx)
+            generations.append(generate_fixture(args.fixture_generator, case_path, per_table_fixture_dir, dry_run=False, reuse_fixture=args.reuse_fixture, allow_large_fixture=args.allow_large_fixture, table_name=meta["table"] if len(tables) > 1 else None, database=meta["schema"] if len(tables) > 1 else None, region_id=meta["region_id"], table_dir=meta["table_dir"], region_dir=meta["region_dir"]))
+        report["fixture_generation"] = generations[0] if len(generations) == 1 else generations
+
+        target_results = []
+        for idx, target in enumerate(targets):
+            cluster = clusters[idx]
+            materialize = []
+            for table_idx, meta in enumerate(discovered[idx]):
+                per_table_fixture_dir = table_fixture_dir(fixture_dir, tables, tables[table_idx], table_idx)
+                materialize.append(materialize_fixture(target, dry_run=False, preserve_state=True, expected_region_dir=meta["region_dir"], fixture_dir=per_table_fixture_dir))
+            cluster.start_datanode()
+            query_result = run_queries(target, case, tables, args.http_timeout)
+            if query_result["status"] == "failed":
+                cluster.restart_frontend()
+                retry_result = run_queries(target, case, tables, args.http_timeout)
+                query_result["frontend_restart_retry"] = retry_result
+                if retry_result["status"] == "ok":
+                    query_result = retry_result
+            report["targets"][idx]["fixture_dir"] = str(fixture_dir)
+            report["targets"][idx]["fixture_materialization"] = materialize[0] if len(materialize) == 1 else materialize
+            report["targets"][idx].update(query_result)
+            report["targets"][idx]["status"] = "measured" if query_result["status"] == "ok" else "failed"
+            write_json(target.report_path, report["targets"][idx])
+            target_results.append(query_result)
+
+        report["thresholds"] = enforce_thresholds(case, target_results[0], target_results[1])
+        report["status"] = "failed" if any(t["status"] == "failed" for t in report["thresholds"]) or any(r["status"] == "failed" for r in target_results) else "ok"
+    except Exception as e:  # noqa: BLE001 - write machine-readable failure report
+        report["status"] = "failed"
+        report["error"] = repr(e)
+    finally:
+        for cluster in reversed(clusters):
+            cluster.stop_all()
+    write_json(work_root / "query-regression-report.json", report)
+    print(json.dumps(report, indent=2, sort_keys=True))
+    return 1 if report["status"] == "failed" else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())