#!/usr/bin/env python3 """Repeat the archived JSON_SCHEMA_VALID L20 comparison without project imports. Requires Python 3.9+ and PyMySQL. See README.md for connection examples. Passwords come from MYSQL_OLD_PASSWORD / MYSQL_NEW_PASSWORD or hidden prompts. No server, table, schema, or global configuration is created or modified. """ import argparse import getpass import importlib.metadata import json import os import re import statistics import sys import time from datetime import datetime, timezone from pathlib import Path class BenchmarkError(Exception): """An intentionally credential-free diagnostic.""" def sql_statements(source): """Split SQL outside quoted strings/comments; preserve statement text.""" statements = [] start = 0 position = 0 quote = None comment = None while position < len(source): char = source[position] next_char = source[position + 1:position + 2] if comment == "line": if char in "\r\n": comment = None elif comment == "block": if char == "*" and next_char == "/": comment = None position += 1 elif quote: if char == "\\": position += 1 elif char == quote: if next_char == quote: position += 1 else: quote = None elif char in "'\"`": quote = char elif char == "#": comment = "line" elif char == "-" and next_char == "-" and ( position + 2 == len(source) or source[position + 2].isspace()): comment = "line" position += 1 elif char == "/" and next_char == "*": comment = "block" position += 1 elif char == ";": statements.append(source[start:position + 1].strip()) start = position + 1 position += 1 if quote or comment == "block": raise BenchmarkError("SQL file contains an unterminated quote or comment.") tail = source[start:].strip() if tail: statements.append(tail) return statements def strip_leading_comments(statement): """Remove only leading comments so the executed SELECT itself stays exact.""" while True: statement = statement.lstrip() if statement.startswith("--") or statement.startswith("#"): newline = statement.find("\n") if newline < 0: return "" statement = statement[newline + 1:] elif statement.startswith("/*"): end = statement.find("*/", 2) if end < 0: raise BenchmarkError("SQL file contains an unterminated comment.") statement = statement[end + 2:] else: return statement def load_queries(sql_file): """Accept the supplied session SETs and exactly two scalar SELECTs.""" try: source = sql_file.read_text(encoding="utf-8-sig") except (OSError, UnicodeError): raise BenchmarkError("Cannot read the SQL file as UTF-8.") from None queries = [] allowed_settings = { "SET NAMES UTF8MB4", "SET SESSION TIME_ZONE = '+00:00'", "SET SESSION MAX_EXECUTION_TIME = 10000", } for statement in sql_statements(source): statement = strip_leading_comments(statement) if not statement: continue normalized = " ".join(statement.rstrip(";").split()).upper() if normalized in allowed_settings: continue if not re.match(r"SELECT\s+JSON_SCHEMA_VALID\s*\(", statement, re.I): raise BenchmarkError("SQL file must contain only the documented session SETs and two JSON_SCHEMA_VALID SELECTs.") # The archived PyMySQL statements omit the SQL-script terminator. # Remove that delimiter only; preserve all SQL literal contents. queries.append(statement[:-1] if statement.endswith(";") else statement) if len(queries) != 2: raise BenchmarkError("SQL file must contain exactly two JSON_SCHEMA_VALID SELECT statements, Query A then Query B.") return list(zip(("A_shared_L20", "B_chain_control_L20"), queries)) def arguments(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--sql-file", type=Path, default=Path(__file__).with_name("repro.sql")) parser.add_argument("--output", type=Path, default=Path("benchmark_results.json"), help="New JSON file; an existing file is never overwritten.") for label, port, version in (("old", 3306, "8.0.45"), ("new", 3307, "9.7.2")): prefix = "MYSQL_" + label.upper() + "_" parser.add_argument("--" + label + "-host", default=os.environ.get(prefix + "HOST", "127.0.0.1")) parser.add_argument("--" + label + "-port", type=int, default=int(os.environ.get(prefix + "PORT", port))) parser.add_argument("--" + label + "-user", default=os.environ.get(prefix + "USER", "root")) parser.add_argument("--" + label + "-database", default=os.environ.get(prefix + "DATABASE") or None, help="Optional; these SELECTs need no default database.") parser.add_argument("--" + label + "-expected-version", default=version, help="Exact numeric server version expected; mismatch stops the run.") parser.add_argument("--rounds", type=int, default=3) parser.add_argument("--warmups", type=int, default=3) parser.add_argument("--samples", type=int, default=20) args = parser.parse_args(argv) if args.rounds < 1 or args.warmups < 0 or args.samples < 1: parser.error("rounds and samples must be positive; warmups must be nonnegative") if any(not 1 <= getattr(args, label + "_port") <= 65535 for label in ("old", "new")): parser.error("ports must be between 1 and 65535") return args def query_once(connection, sql): """Match the archived timer: execute through fetchall, excluding SET/warnings.""" with connection.cursor() as cursor: cursor.execute("SET SESSION max_execution_time = 10000") cursor.fetchall() started = time.perf_counter_ns() cursor.execute(sql) rows = cursor.fetchall() elapsed_ms = (time.perf_counter_ns() - started) / 1_000_000.0 cursor.execute("SHOW WARNINGS") warnings = cursor.fetchall() if rows != ((1,),): raise BenchmarkError("A query returned an unexpected result; expected exactly one row containing integer 1.") if warnings: raise BenchmarkError("A query produced server warnings; timing comparison was stopped.") return elapsed_ms def connect_pair(pymysql, args, passwords, identities): connections = {} try: for label in ("old", "new"): connection = pymysql.connect( host=getattr(args, label + "_host"), port=getattr(args, label + "_port"), user=getattr(args, label + "_user"), password=passwords[label], database=getattr(args, label + "_database"), charset="utf8mb4", autocommit=True, connect_timeout=10, read_timeout=20, write_timeout=20, ) connections[label] = connection with connection.cursor() as cursor: cursor.execute("SET NAMES utf8mb4") cursor.execute("SET SESSION time_zone = '+00:00'") cursor.execute("SELECT VERSION(), @@version_comment") version, comment = cursor.fetchone() numeric_version = version.split("-", 1)[0] if numeric_version != getattr(args, label + "_expected_version"): raise BenchmarkError("The " + label + " server does not match the expected numeric version.") identity = {"version": version, "version_comment": comment} if label in identities and identities[label] != identity: raise BenchmarkError("Server identity changed between rounds.") identities[label] = identity return connections except BaseException: for connection in connections.values(): connection.close() raise def benchmark(pymysql, args, queries, passwords): identities = {} result = { "created_utc": datetime.now(timezone.utc).isoformat(), "python_version": sys.version.split()[0], "pymysql_distribution_version": importlib.metadata.version("PyMySQL"), "protocol": { "rounds": args.rounds, "warmups_per_server_per_round": args.warmups, "samples_per_server_per_round": args.samples, "timer": "perf_counter_ns around cursor.execute through fetchall", "excluded_from_timer": ["SET SESSION max_execution_time", "SHOW WARNINGS"], "measured_endpoint_order": "old,new if (zero_based_sample + zero_based_round) is even; otherwise new,old", "warmup_endpoint_order": "all old warmups, then all new warmups", "connections": "one persistent connection per endpoint for each case/round; separate preflight connections", "charset": "utf8mb4", "time_zone": "+00:00", "autocommit": True, "max_execution_time_ms": 10000, "expected_result": [[1]], "warnings": "any warning fails the run", }, "servers": identities, "cases": [], } for name, sql in queries: print("Preflight: " + name, flush=True) connections = connect_pair(pymysql, args, passwords, identities) try: preflight = {label: query_once(connections[label], sql) for label in ("old", "new")} finally: for connection in connections.values(): connection.close() case = {"case": name, "sql": sql, "preflight_ms": preflight, "rounds": []} for round_index in range(args.rounds): connections = connect_pair(pymysql, args, passwords, identities) round_result = {"round": round_index + 1, "warmups_ms": {"old": [], "new": []}, "samples": []} try: for label in ("old", "new"): for _ in range(args.warmups): round_result["warmups_ms"][label].append(query_once(connections[label], sql)) for sample_index in range(args.samples): order = ("old", "new") if (sample_index + round_index) % 2 == 0 else ("new", "old") sample = {"sample": sample_index + 1, "order": list(order)} for label in order: sample[label + "_ms"] = query_once(connections[label], sql) round_result["samples"].append(sample) finally: for connection in connections.values(): connection.close() for label in ("old", "new"): round_result[label + "_median_ms"] = statistics.median(sample[label + "_ms"] for sample in round_result["samples"]) round_result["new_over_old"] = round_result["new_median_ms"] / round_result["old_median_ms"] case["rounds"].append(round_result) print("{} round {}: old={:.6f} ms, new={:.6f} ms, new/old={:.3f}".format( name, round_index + 1, round_result["old_median_ms"], round_result["new_median_ms"], round_result["new_over_old"]), flush=True) case["summary"] = { "old_median_of_round_medians_ms": statistics.median(item["old_median_ms"] for item in case["rounds"]), "new_median_of_round_medians_ms": statistics.median(item["new_median_ms"] for item in case["rounds"]), "median_paired_round_ratio_new_over_old": statistics.median(item["new_over_old"] for item in case["rounds"]), } result["cases"].append(case) return result def main(argv=None): args = arguments(argv) if args.output.exists(): raise BenchmarkError("Output file already exists; choose a new --output filename.") if not args.output.parent.is_dir(): raise BenchmarkError("Output directory does not exist.") queries = load_queries(args.sql_file) try: import pymysql except ImportError: raise BenchmarkError("PyMySQL is required. Install it in a separate environment: python -m pip install PyMySQL") from None passwords = {} for label in ("old", "new"): environment_key = "MYSQL_" + label.upper() + "_PASSWORD" passwords[label] = os.environ[environment_key] if environment_key in os.environ else getpass.getpass(label + " server password: ") result = benchmark(pymysql, args, queries, passwords) with args.output.open("x", encoding="utf-8", newline="\n") as output: json.dump(result, output, indent=2, ensure_ascii=True) output.write("\n") print("Completed; results written to the requested JSON output file.") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: print("Interrupted; no completed result file was written.", file=sys.stderr) raise SystemExit(130) except BenchmarkError as error: print("Error: " + str(error), file=sys.stderr) raise SystemExit(1) except Exception as error: # Driver/network errors may contain usernames, addresses or credentials. # Emit their type and numeric server code, never their raw message. code = error.args[0] if error.args and isinstance(error.args[0], int) else None suffix = " (server code {})".format(code) if code is not None else "" print("Error: {}{}; no raw exception text was printed to avoid exposing connection details.".format(type(error).__name__, suffix), file=sys.stderr) raise SystemExit(1)