From c44f8da6466525acd7c8f340396f8b7f1bc0188f Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 3 Jul 2026 17:09:01 +0800 Subject: [PATCH] feat: add query regression perf harness (#8406) * feat: add query regression perf harness Signed-off-by: discord9 * feat: extend query regression cases Signed-off-by: discord9 * ci: harden query regression workflows Signed-off-by: discord9 * fix: address query regression review comments Signed-off-by: discord9 * ci: limit query regression PR triggers Signed-off-by: discord9 * ci: run full query regression case set Signed-off-by: discord9 * refactor: model query regression scenarios Signed-off-by: discord9 * fix: avoid unenforced query regression thresholds Signed-off-by: discord9 --------- Signed-off-by: discord9 --- .../scripts/query-regression-pr-metadata.py | 44 + .github/scripts/query-regression-run.py | 184 ++++ .github/scripts/query-regression-summary.py | 224 +++++ .../workflows/query-regression-comment.yml | 177 ++++ .github/workflows/query-regression.yml | 152 ++++ .gitignore | 1 + Cargo.lock | 1 + src/cmd/Cargo.toml | 6 + src/cmd/src/bin/query_perf_fixture.rs | 651 ++++++++++++++ tests/perf/AGENTS.md | 24 + tests/perf/README.md | 233 +++++ tests/perf/fixture-format.md | 120 +++ .../promql_pushdown_7913/case.toml | 67 ++ .../query_cases/smoke_direct_sst/case.toml | 61 ++ .../sql_aggregate_order_by/case.toml | 63 ++ .../sql_join_filter_order/case.toml | 95 +++ .../query_cases/sql_topk_order_by/case.toml | 63 ++ tests/perf/query_regression_runner.py | 801 ++++++++++++++++++ 18 files changed, 2967 insertions(+) create mode 100644 .github/scripts/query-regression-pr-metadata.py create mode 100644 .github/scripts/query-regression-run.py create mode 100644 .github/scripts/query-regression-summary.py create mode 100644 .github/workflows/query-regression-comment.yml create mode 100644 .github/workflows/query-regression.yml create mode 100644 src/cmd/src/bin/query_perf_fixture.rs create mode 100644 tests/perf/AGENTS.md create mode 100644 tests/perf/README.md create mode 100644 tests/perf/fixture-format.md create mode 100644 tests/perf/query_cases/promql_pushdown_7913/case.toml create mode 100644 tests/perf/query_cases/smoke_direct_sst/case.toml create mode 100644 tests/perf/query_cases/sql_aggregate_order_by/case.toml create mode 100644 tests/perf/query_cases/sql_join_filter_order/case.toml create mode 100644 tests/perf/query_cases/sql_topk_order_by/case.toml create mode 100644 tests/perf/query_regression_runner.py diff --git a/.github/scripts/query-regression-pr-metadata.py b/.github/scripts/query-regression-pr-metadata.py new file mode 100644 index 0000000000..0ce87e4582 --- /dev/null +++ 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()) diff --git a/.github/scripts/query-regression-run.py b/.github/scripts/query-regression-run.py new file mode 100644 index 0000000000..7af0ccd1fe --- /dev/null +++ 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()) diff --git a/.github/scripts/query-regression-summary.py b/.github/scripts/query-regression-summary.py new file mode 100644 index 0000000000..f07c7d99a1 --- /dev/null +++ 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()) diff --git a/.github/workflows/query-regression-comment.yml b/.github/workflows/query-regression-comment.yml new file mode 100644 index 0000000000..573eda85ca --- /dev/null +++ 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(//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 }} diff --git a/.github/workflows/query-regression.yml b/.github/workflows/query-regression.yml new file mode 100644 index 0000000000..6e8a4618b3 --- /dev/null +++ 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 diff --git a/.gitignore b/.gitignore index 1f46243b3a..90893f8aac 100644 --- a/.gitignore +++ b/.gitignore @@ -73,6 +73,7 @@ greptimedb_data .gemini .opencode .worktrees/ +.slim/deepwork/ # local design docs docs/specs/ diff --git a/Cargo.lock b/Cargo.lock index a58712d54a..c563173084 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2136,6 +2136,7 @@ dependencies = [ "meta-client", "meta-srv", "metric-engine", + "mito-codec", "mito2", "moka", "object-store", diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index 093d9600c1..a11db11962 100644 --- 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 diff --git a/src/cmd/src/bin/query_perf_fixture.rs b/src/cmd/src/bin/query_perf_fixture.rs new file mode 100644 index 0000000000..911bca32ee --- /dev/null +++ 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, + + /// Table name to generate when the case contains multiple [[tables]]. + #[arg(long, value_name = "NAME")] + table: Option, + + /// 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, + tables: Vec, + 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, + #[serde(default)] + sst_format: Option, + primary_key: Vec, + time_index: String, + columns: Vec, +} + +#[derive(Debug, Deserialize)] +struct ColumnConfig { + name: String, + #[serde(rename = "type")] + ty: String, + semantic: String, + distribution: Option, +} + +#[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) -> Vec { + 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::::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::>(), + )) as ArrayRef); + } + ("field", "UINT64") => columns.push(Arc::new(UInt64Array::from( + (0..rows).map(|r| (base_row + r) as u64).collect::>(), + )) 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::>(), + )) 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::>(), + )) as ArrayRef) + } + other => panic!("unsupported column combination {other:?}"), + } + } + + let mut pk_builder = BinaryDictionaryBuilder::::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 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()); +} diff --git a/tests/perf/AGENTS.md b/tests/perf/AGENTS.md new file mode 100644 index 0000000000..eb95544d22 --- /dev/null +++ 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. diff --git a/tests/perf/README.md b/tests/perf/README.md new file mode 100644 index 0000000000..8d82e3a1c3 --- /dev/null +++ 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///...` 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 `. 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 ` 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`. diff --git a/tests/perf/fixture-format.md b/tests/perf/fixture-format.md new file mode 100644 index 0000000000..0086edf627 --- /dev/null +++ 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. diff --git a/tests/perf/query_cases/promql_pushdown_7913/case.toml b/tests/perf/query_cases/promql_pushdown_7913/case.toml new file mode 100644 index 0000000000..de6f73f53c --- /dev/null +++ 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 diff --git a/tests/perf/query_cases/smoke_direct_sst/case.toml b/tests/perf/query_cases/smoke_direct_sst/case.toml new file mode 100644 index 0000000000..1922a134d3 --- /dev/null +++ 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 diff --git a/tests/perf/query_cases/sql_aggregate_order_by/case.toml b/tests/perf/query_cases/sql_aggregate_order_by/case.toml new file mode 100644 index 0000000000..05fc852e8e --- /dev/null +++ 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 diff --git a/tests/perf/query_cases/sql_join_filter_order/case.toml b/tests/perf/query_cases/sql_join_filter_order/case.toml new file mode 100644 index 0000000000..459fe7d698 --- /dev/null +++ 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 diff --git a/tests/perf/query_cases/sql_topk_order_by/case.toml b/tests/perf/query_cases/sql_topk_order_by/case.toml new file mode 100644 index 0000000000..a802fa47f0 --- /dev/null +++ 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 diff --git a/tests/perf/query_regression_runner.py b/tests/perf/query_regression_runner.py new file mode 100644 index 0000000000..176e51f2b8 --- /dev/null +++ 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())