rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-09 23:12:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: port query regression runner to Rust (#8651)

Browse Source 
* refactor: port query regression runner to Rust

Signed-off-by: discord9 <discord9@163.com>

* ci: remove optional OTLP report plotter

Signed-off-by: discord9 <discord9@163.com>

* refactor: split query regression runner into modules

Signed-off-by: discord9 <discord9@163.com>

* style: use crate-qualified imports in query regression runner

Signed-off-by: discord9 <discord9@163.com>

* refactor: simplify query regression runner internals

Signed-off-by: discord9 <discord9@163.com>

* feat: abstract inspect-footer storage access behind object store destination

Add an optional --destination <TOML> to inspect-footer (and
--base-destination/--candidate-destination to finalize-remote) so the
storage inspection reads DB data files through the opendal-backed
object_store abstraction instead of bare std::fs. Local paths keep
working unchanged via the --root shortcut (File backend); remote
backends (S3/GCS/...) are described by a DestinationConfig TOML
reusing the object-store crate's ObjectStoreConfig serde shape.

- inspect_footer: list via ObjectStore::list + ObjectMeta filtering
  (parquet keys, non-zero size, metadata/ segment), read footers
  async via ParquetObjectReader + ParquetMetaDataReader with known
  file size (no extra HEAD); output JSON schema unchanged
- finalize-remote: --base-data-home/--candidate-data-home become
  optional, mutually exclusive with the new --*-destination args
- cmd deps: add object_store_opendal + datafusion_object_store
- tests: fs-backend list+footer integration tests (metadata filtering,
  destination TOML mode, root/destination exclusivity)

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* style: drop needless borrow in inspect footer test

Fix clippy::needless_borrows_for_generic_args in the inspect-footer test
(fs::create_dir_all(table.join("metadata"))). Missed by the earlier
focused clippy run because it only covered --bin targets.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <discord9@163.com>
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-08-06 13:01:32 +00:00
committed by GitHub
parent 1aadfa565c
commit 1693b2727c
25 changed files with 4935 additions and 2294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/scripts/query-regression-run.py
+501 -26
View File
@@ -18,10 +18,17 @@
from __future__ import annotations
import argparse
import json
import os
import re
import signal
import socket
import subprocess
import time
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any
DEFAULT_CASES = [
@@ -100,38 +107,479 @@ def append_step_summary(summary: Path) -> None:
out.write(summary.read_text())
@dataclass(frozen=True)
class RunTarget:
name: str
binary: Path
work_dir: 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
frontend_config: Path | None = None
def allocate_ports(n: int) -> list[int]:
socks = []
try:
for _ in range(n):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 0))
socks.append(sock)
return [sock.getsockname()[1] for sock in socks]
finally:
for sock in socks:
sock.close()
def make_target(
name: str,
binary: Path,
root: Path,
ports: list[int],
frontend_config: Path | None = None,
) -> RunTarget:
work_dir = root / name
return RunTarget(
name=name,
binary=binary,
work_dir=work_dir,
http_port=ports[4],
grpc_port=ports[5],
mysql_port=ports[6],
postgres_port=ports[7],
metasrv_rpc_port=ports[0],
metasrv_http_port=ports[1],
datanode_rpc_port=ports[2],
datanode_http_port=ports[3],
datanode_data_dir=work_dir / "datanode-0" / "data",
frontend_config=frontend_config,
)
def wait_health(port: int, timeout_s: float = 60.0) -> None:
deadline = time.monotonic() + timeout_s
last: Exception | None = None
while time.monotonic() < deadline:
try:
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=2) as response:
if response.status < 500:
return
except Exception as err: # noqa: BLE001 - retain the last health diagnostic
last = err
time.sleep(0.5)
raise TimeoutError(f"health check timed out on port {port}: {last}")
def component_log_dir(target: RunTarget, name: str) -> Path:
return target.work_dir / "logs" / ("datanode-0" if name == "datanode" else name)
def component_command(target: RunTarget, name: str) -> list[str]:
log_dir = component_log_dir(target, name)
if name == "metasrv":
return [
str(target.binary), "metasrv", "start",
"--grpc-bind-addr", f"127.0.0.1:{target.metasrv_rpc_port}",
"--grpc-server-addr", f"127.0.0.1:{target.metasrv_rpc_port}",
"--http-addr", f"127.0.0.1:{target.metasrv_http_port}",
"--backend", "memory-store", "--enable-region-failover", "false",
"--log-dir", str(log_dir),
]
if name == "datanode":
return [
str(target.binary), "datanode", "start",
"--grpc-bind-addr", f"127.0.0.1:{target.datanode_rpc_port}",
"--grpc-server-addr", f"127.0.0.1:{target.datanode_rpc_port}",
"--http-addr", f"127.0.0.1:{target.datanode_http_port}",
"--data-home", str(target.datanode_data_dir), "--log-dir", str(log_dir),
"--node-id", "0", "--metasrv-addrs", f"127.0.0.1:{target.metasrv_rpc_port}",
]
if name == "frontend":
command = [str(target.binary), "frontend", "start"]
if target.frontend_config is not None:
command.extend(["--config-file", str(target.frontend_config)])
command.extend([
"--metasrv-addrs", f"127.0.0.1:{target.metasrv_rpc_port}",
"--http-addr", f"127.0.0.1:{target.http_port}",
"--grpc-bind-addr", f"127.0.0.1:{target.grpc_port}",
"--grpc-server-addr", f"127.0.0.1:{target.grpc_port}",
"--mysql-addr", f"127.0.0.1:{target.mysql_port}",
"--postgres-addr", f"127.0.0.1:{target.postgres_port}",
"--log-dir", str(log_dir),
])
return command
raise ValueError(f"unknown component: {name}")
def start_component(
target: RunTarget,
name: str,
procs: dict[tuple[str, str], subprocess.Popen[bytes]],
) -> None:
if name != "metasrv":
metasrv = procs.get((target.name, "metasrv"))
if metasrv is None or metasrv.poll() is not None:
raise RuntimeError("metasrv exited; memory-store metadata is no longer valid")
if name == "datanode":
target.datanode_data_dir.mkdir(parents=True, exist_ok=True)
logs = component_log_dir(target, name)
logs.mkdir(parents=True, exist_ok=True)
with (logs / "stdout.log").open("ab") as out, (logs / "stderr.log").open("ab") as err:
procs[(target.name, name)] = subprocess.Popen(
component_command(target, name),
stdout=out,
stderr=err,
start_new_session=True,
)
wait_health({
"metasrv": target.metasrv_http_port,
"datanode": target.datanode_http_port,
"frontend": target.http_port,
}[name])
def stop_component(
target: RunTarget,
name: str,
procs: dict[tuple[str, str], subprocess.Popen[bytes]],
) -> None:
proc = procs.pop((target.name, name), None)
if proc is None or proc.poll() is not None:
return
try:
os.killpg(proc.pid, signal.SIGTERM)
except ProcessLookupError:
return
try:
proc.wait(timeout=20)
except subprocess.TimeoutExpired:
try:
os.killpg(proc.pid, signal.SIGKILL)
except ProcessLookupError:
return
proc.wait(timeout=20)
def restart_component(
target: RunTarget,
name: str,
procs: dict[tuple[str, str], subprocess.Popen[bytes]],
) -> None:
stop_component(target, name, procs)
start_component(target, name, procs)
def stop_all(targets: list[RunTarget], procs: dict[tuple[str, str], subprocess.Popen[bytes]]) -> None:
for target in reversed(targets):
for name in ("frontend", "datanode", "metasrv"):
stop_component(target, name, procs)
def load_plan(fixture_generator: Path, case_path: Path) -> dict[str, Any]:
result = subprocess.run(
[str(fixture_generator), "plan", "--case", str(case_path)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
if result.returncode != 0:
raise RuntimeError(f"query_perf_fixture plan failed: {result.stderr[:2000]}")
return json.loads(result.stdout)
def run_direct_case(
args: argparse.Namespace,
case_path: Path,
work_dir: Path,
base_bin: Path,
candidate_bin: Path,
fixture_generator: Path,
runner: Path,
) -> int:
ports = allocate_ports(16)
targets = [
make_target("base", base_bin, work_dir, ports[:8]),
make_target("candidate", candidate_bin, work_dir, ports[8:]),
]
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}")
target.work_dir.mkdir(parents=True, exist_ok=True)
procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
try:
for target in targets:
start_component(target, "metasrv", procs)
start_component(target, "datanode", procs)
start_component(target, "frontend", procs)
prepare = [
str(runner), "prepare-direct", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--base-http-port", str(targets[0].http_port),
"--candidate-http-port", str(targets[1].http_port),
"--fixture-dir", str(work_dir / "fixture"),
"--output", str(work_dir / "prepare-direct.json"),
"--http-timeout", str(args.http_timeout),
]
if parse_bool(args.allow_large_fixture):
prepare.append("--allow-large-fixture")
prepare_status = subprocess.run(prepare, check=False).returncode
if prepare_status != 0:
return prepare_status
for target in targets:
stop_component(target, "datanode", procs)
prepared = json.loads((work_dir / "prepare-direct.json").read_text(encoding="utf-8"))
fixtures = prepared.get("fixtures")
if not isinstance(fixtures, list):
raise RuntimeError("prepare-direct report has no fixtures array")
for target in targets:
destination = target.work_dir / "materialize-destination.toml"
destination.write_text(
f"data_home = {json.dumps(str(target.datanode_data_dir))}\n"
"object_store = { type = \"File\" }\n",
encoding="utf-8",
)
for fixture in fixtures:
fixture_dir = fixture.get("fixture_dir") if isinstance(fixture, dict) else None
if not isinstance(fixture_dir, str):
raise RuntimeError("prepare-direct fixture record has no fixture_dir")
materialize = [
str(runner), "materialize", "--fixture-dir", fixture_dir,
"--destination", str(destination),
]
materialize_status = subprocess.run(materialize, check=False).returncode
if materialize_status != 0:
return materialize_status
for target in targets:
start_component(target, "datanode", procs)
measure = [
str(runner), "measure", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--base-http-port", str(targets[0].http_port),
"--candidate-http-port", str(targets[1].http_port),
"--output", str(work_dir / "query-regression-report.json"),
"--http-timeout", str(args.http_timeout),
]
status = subprocess.run(measure, check=False).returncode
if status != 0:
for target in targets:
restart_component(target, "frontend", procs)
status = subprocess.run(measure, check=False).returncode
return status
finally:
stop_all(targets, procs)
def run_remote_case(
args: argparse.Namespace,
case_path: Path,
work_dir: Path,
base_bin: Path,
candidate_bin: Path,
fixture_generator: Path,
runner: Path,
) -> int:
ports = allocate_ports(16)
targets = [
make_target(
"base",
base_bin,
work_dir,
ports[:8],
work_dir / "base" / "frontend-prom-store.toml",
),
make_target(
"candidate",
candidate_bin,
work_dir,
ports[8:],
work_dir / "candidate" / "frontend-prom-store.toml",
),
]
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}")
target.work_dir.mkdir(parents=True, exist_ok=True)
if target.frontend_config is None:
raise RuntimeError(f"remote target has no frontend config path: {target.name}")
render = [
str(runner), "render-remote-config", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--output", str(target.frontend_config),
]
render_status = subprocess.run(render, check=False).returncode
if render_status != 0:
return render_status
procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
report = work_dir / "query-regression-report.json"
try:
for target in targets:
start_component(target, "metasrv", procs)
start_component(target, "datanode", procs)
start_component(target, "frontend", procs)
prepare = [
str(runner), "prepare-remote", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--base-http-port", str(targets[0].http_port),
"--candidate-http-port", str(targets[1].http_port),
"--output", str(work_dir / "prepare-remote.json"),
"--http-timeout", str(args.http_timeout),
]
prepare_status = subprocess.run(prepare, check=False).returncode
if prepare_status != 0:
return prepare_status
measure = [
str(runner), "measure", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--base-http-port", str(targets[0].http_port),
"--candidate-http-port", str(targets[1].http_port),
"--output", str(report), "--http-timeout", str(args.http_timeout),
]
measure_status = subprocess.run(measure, check=False).returncode
if measure_status != 0 and not report.exists():
return measure_status
for target in targets:
stop_component(target, "datanode", procs)
finalize = [
str(runner), "finalize-remote", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--candidate-bin", str(candidate_bin),
"--base-data-home", str(targets[0].datanode_data_dir),
"--candidate-data-home", str(targets[1].datanode_data_dir),
"--report", str(report),
]
finalize_status = subprocess.run(finalize, check=False).returncode
if finalize_status != 0:
return finalize_status
final_report = json.loads(report.read_text(encoding="utf-8"))
return 1 if measure_status != 0 or final_report.get("status") == "failed" else 0
finally:
stop_all(targets, procs)
def run_otlp_case(
args: argparse.Namespace,
case_path: Path,
work_dir: Path,
base_bin: Path,
candidate_bin: Path,
fixture_generator: Path,
runner: Path,
) -> int:
if args.otelgen_bin is None:
raise ValueError("--otelgen-bin is required for otlp_trace_load")
ports = allocate_ports(16)
targets = [
make_target("base", base_bin, work_dir, ports[:8]),
make_target("candidate", candidate_bin, work_dir, ports[8:]),
]
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}")
target.work_dir.mkdir(parents=True, exist_ok=True)
procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
try:
for target in targets:
try:
start_component(target, "metasrv", procs)
start_component(target, "datanode", procs)
start_component(target, "frontend", procs)
command = [
str(runner), "run-otlp-target", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--otelgen-bin", str(args.otelgen_bin),
"--http-port", str(target.http_port),
"--target-name", target.name,
"--work-dir", str(target.work_dir),
"--output", str(target.work_dir / "report.json"),
"--http-timeout", str(args.http_timeout),
]
status = subprocess.run(command, check=False).returncode
finally:
stop_all([target], procs)
if status != 0:
return status
output = work_dir / "query-regression-report.json"
finalize = [
str(runner), "finalize-otlp", "--case", str(case_path),
"--fixture-generator", str(fixture_generator),
"--base-result", str(targets[0].work_dir / "report.json"),
"--candidate-result", str(targets[1].work_dir / "report.json"),
"--output", str(output),
]
status = subprocess.run(finalize, check=False).returncode
if status != 0:
return status
report = json.loads(output.read_text(encoding="utf-8"))
return 1 if report.get("status") == "failed" else 0
finally:
stop_all(targets, procs)
def run_case(args: argparse.Namespace, case_path: Path, work_dir: Path) -> int:
target_dir = profile_dir(args.cargo_profile)
base_bin = args.base_bin or args.base_src / "target" / target_dir / "greptime"
candidate_bin = args.candidate_bin or args.candidate_src / "target" / target_dir / "greptime"
fixture_generator = args.fixture_generator or args.candidate_src / "target" / target_dir / "query_perf_fixture"
cmd = [
"uv",
"run",
"--no-project",
"python",
str(args.candidate_src / "tests/perf/query_regression_runner.py"),
"--case",
str(case_path),
"--base-bin",
str(base_bin),
"--candidate-bin",
str(candidate_bin),
"--fixture-generator",
str(fixture_generator),
"--work-dir",
str(work_dir),
"--http-timeout",
str(args.http_timeout),
]
if parse_bool(args.allow_large_fixture):
cmd.append("--allow-large-fixture")
if args.otelgen_bin is not None:
cmd.extend(["--otelgen-bin", str(args.otelgen_bin)])
runner = args.runner or args.candidate_src / "target" / target_dir / "query_regression_runner"
plan = load_plan(fixture_generator, case_path)
scenario = plan.get("scenario")
kind = scenario.get("kind") if isinstance(scenario, dict) else None
print(f"::group::Query regression case: {case_path}", flush=True)
try:
return subprocess.run(cmd, check=False).returncode
if kind == "direct_readable_sst":
return run_direct_case(
args,
case_path,
work_dir,
base_bin,
candidate_bin,
fixture_generator,
runner,
)
if kind == "prom_remote_write_then_query":
return run_remote_case(
args,
case_path,
work_dir,
base_bin,
candidate_bin,
fixture_generator,
runner,
)
if kind == "otlp_trace_load":
return run_otlp_case(
args,
case_path,
work_dir,
base_bin,
candidate_bin,
fixture_generator,
runner,
)
raise ValueError(
f"unsupported scenario kind {kind!r}; supported: "
"'direct_readable_sst', 'prom_remote_write_then_query', 'otlp_trace_load'"
)
finally:
print("::endgroup::", flush=True)
@@ -157,6 +605,27 @@ def write_summary(args: argparse.Namespace, reports: list[Path]) -> int:
return subprocess.run(cmd, check=False).returncode
def write_failed_report(work_dir: Path, case_path: Path, error: Exception) -> None:
work_dir.mkdir(parents=True, exist_ok=True)
path = work_dir / "query-regression-report.json"
try:
report = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(report, dict) or not isinstance(report.get("targets"), list) or not isinstance(report.get("thresholds"), list):
raise ValueError("existing report is malformed")
except (OSError, ValueError, json.JSONDecodeError):
report = {
"case_path": str(case_path),
"targets": [],
"thresholds": [],
}
report["status"] = "failed"
report["error"] = repr(error)
path.write_text(
json.dumps(report, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--cases", action="append", help="'all', 'heavy', or comma/space separated case paths")
@@ -170,6 +639,7 @@ def main() -> int:
default=configured_path(os.environ.get("FIXTURE_GENERATOR")),
)
parser.add_argument("--otelgen-bin", type=Path, default=configured_path(os.environ.get("OTELGEN_BIN")))
parser.add_argument("--runner", type=Path, default=configured_path(os.environ.get("QUERY_REGRESSION_RUNNER")))
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"))
@@ -199,7 +669,12 @@ def main() -> int:
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)
try:
case_status = run_case(args, case_path, work_dir)
except Exception as err: # noqa: BLE001 - preserve an aggregate report for the summary step
print(f"error: query regression case {case_path}: {err}", flush=True)
write_failed_report(work_dir, case_path, err)
case_status = 1
if case_status != 0:
status = case_status or 1