From 24197872f7029617e7eedc6e011a05cbdd8a476b Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Tue, 8 Sep 2026 10:51:49 +0000 Subject: [PATCH 1/2] workload-replay: Confirm performance regressions with paired reruns Require the same metric to exceed its existing threshold in two fresh reference/current pairs. Rerun the baseline as well so an unusually fast reference does not poison confirmation. Preserve both comparison tables and plots, and keep query errors fatal without retrying them. Closes: QAR-153 --- .../materialize/workload_replay/executor.py | 151 +++++++++--------- .../workload_replay/executor_test.py | 137 ++++++++++++++++ .../materialize/workload_replay/stats.py | 25 +-- 3 files changed, 218 insertions(+), 95 deletions(-) create mode 100644 misc/python/materialize/workload_replay/executor_test.py diff --git a/misc/python/materialize/workload_replay/executor.py b/misc/python/materialize/workload_replay/executor.py index c5fe4d8075c07..2ade8efffe4dd 100644 --- a/misc/python/materialize/workload_replay/executor.py +++ b/misc/python/materialize/workload_replay/executor.py @@ -314,9 +314,9 @@ def benchmark( When `compare_against` is set, an older reference version is run first and its stats are compared against the current version. Otherwise only the current version is run, which still exercises that the workload replays - without crashing. + without crashing. Performance failures require the same metric to exceed + its threshold in two fresh reference/current pairs. Query errors are not retried. """ - import random services = [ "materialized", @@ -348,22 +348,24 @@ def benchmark( print_workload_stats(file, workload) - stats_old = None - old_version = None - if compare_against: - tag = resolve_tag(compare_against) - print(f"-- Running against materialized:{tag} (reference)") + filename = posixpath.relpath(file, LOCATION) + tag = resolve_tag(compare_against) if compare_against else None + if compare_against and tag is None: + raise ValueError(f"Could not resolve reference {compare_against}") + + def run(image: str | None) -> tuple[dict[str, Any], str]: + print(f"-- Running against {image or 'current materialized'}") random.seed(seed) with c.override( Materialized( - image=f"{image_registry()}/materialized:{tag}", + image=image, cluster_replica_size=cluster_replica_sizes, ports=[6875, 6874, 6876, 6877, 6878, 6880, 6881, 26257], environment_extra=["MZ_NO_BUILTIN_CONSOLE=0"], additional_system_parameter_defaults=additional_system_parameter_defaults, ) ): - stats_old = test( + stats = test( c, workload, file, @@ -379,82 +381,79 @@ def benchmark( True, max_concurrent_queries, ) - old_version = c.query_mz_version() + version = c.query_mz_version() try: c.kill(*services) except: pass c.rm(*services, destroy_volumes=True) c.rm_volumes("mzdata") - print("-- Running against current materialized") - random.seed(seed) - with c.override( - Materialized( - image=None, - cluster_replica_size=cluster_replica_sizes, - ports=[6875, 6874, 6876, 6877, 6878, 6880, 6881, 26257], - environment_extra=["MZ_NO_BUILTIN_CONSOLE=0"], - additional_system_parameter_defaults=additional_system_parameter_defaults, - ) - ): - stats_new = test( - c, - workload, - file, - factor_initial_data, - factor_ingestions, - factor_queries, - runtime, - verbose, - True, - True, - early_initial_data, - True, - True, - max_concurrent_queries, - ) - new_version = c.query_mz_version() - try: - c.kill(*services) - except: - pass - c.rm(*services, destroy_volumes=True) - c.rm_volumes("mzdata") - filename = posixpath.relpath(file, LOCATION) + return stats, version - if stats_old is None or old_version is None: + if tag is None: + _, new_version = run(None) print(f"-- Ran {new_version} without a reference version to compare against") return - print(f"-- Comparing {old_version} against {new_version}") - plot_docker_stats_compare( - stats_old=stats_old, - stats_new=stats_new, - file=filename, - old_version=old_version, - new_version=new_version, - ) - failures: list[TestFailureDetails] = [] - failures.extend(compare_table(filename, stats_old, stats_new)) - - if "errors" in stats_old["queries"]: - new_errors = [] - for error, occurrences in stats_new["queries"]["errors"].items(): - if error in stats_old["queries"]["errors"]: - continue - # Random data can't satisfy every cast in captured queries. - # E.g. text "bar" cast to bigint, or "005V" cast to uuid. - if "invalid input syntax for type" in error: - continue - new_errors.append(f"{error} in queries: {occurrences}") - if new_errors: - failures.append( - TestFailureDetails( - message=f"Workload {filename} has new errors", - details="\n".join(new_errors), - test_class_name_override=filename, + pending_regressions: set[str] = set() + tables = [] + for attempt in range(1, 3): + print(f"-- Comparison attempt {attempt} for {filename}") + stats_old, old_version = run(f"{image_registry()}/materialized:{tag}") + stats_new, new_version = run(None) + print(f"-- Comparing {old_version} against {new_version}") + plot_docker_stats_compare( + stats_old=stats_old, + stats_new=stats_new, + file=f"{filename}_attempt_{attempt}", + old_version=old_version, + new_version=new_version, + ) + table, regressions = compare_table(stats_old, stats_new) + print(table) + tables.append(f"Attempt {attempt}\n{table}") + + if "errors" in stats_old["queries"]: + new_errors = [] + for error, occurrences in stats_new["queries"]["errors"].items(): + if error in stats_old["queries"]["errors"]: + continue + # Random data can't satisfy every cast in captured queries. + # E.g. text "bar" cast to bigint, or "005V" cast to uuid. + if "invalid input syntax for type" in error: + continue + new_errors.append(f"{error} in queries: {occurrences}") + if new_errors: + raise FailedTestExecutionError( + errors=[ + TestFailureDetails( + message=f"Workload {filename} has new errors", + details="\n".join(new_errors), + test_class_name_override=filename, + ) + ] ) - ) - if failures: - raise FailedTestExecutionError(errors=failures) + if attempt == 1: + pending_regressions = regressions + else: + pending_regressions &= regressions + if not pending_regressions: + if attempt > 1: + print("-- Performance regressions did not reproduce") + return + if attempt == 1: + # An unusually fast reference can cause a false regression too. + # Rerun both versions, not just the current version. + print("-- Confirming performance regressions with a fresh pair of runs") + + raise FailedTestExecutionError( + errors=[ + TestFailureDetails( + message=f"Workload {filename} regressed", + details=f"Confirmed metrics: {', '.join(sorted(pending_regressions))}\n\n" + + "\n\n".join(tables), + test_class_name_override=filename, + ) + ] + ) diff --git a/misc/python/materialize/workload_replay/executor_test.py b/misc/python/materialize/workload_replay/executor_test.py new file mode 100644 index 0000000000000..ef703ef22e603 --- /dev/null +++ b/misc/python/materialize/workload_replay/executor_test.py @@ -0,0 +1,137 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +from contextlib import nullcontext +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from materialize.mzcompose.composition import Composition +from materialize.mzcompose.test_result import FailedTestExecutionError +from materialize.workload_replay import executor + + +def stats(creation: float = 100, cpu: float = 100, error: str | None = None) -> dict[str, Any]: + return { + "object_creation": creation, + "queries": {"errors": {error: ["SELECT 1"]} if error else {}}, + "docker": [(0, {"materialized": {"cpu_percent": cpu, "mem_percent": 10}})], + } + + +@pytest.mark.parametrize( + "runs,compare_against,failure", + [ + pytest.param([stats(), stats(120)], "reference", None, id="threshold-passes"), + pytest.param( + [stats(), stats(150), stats(), stats(120)], + "reference", + None, + id="transient-regression", + ), + pytest.param( + [stats(), stats(150), stats(), stats(130)], + "reference", + "regressed", + id="persistent-same-metric", + ), + pytest.param( + [stats(), stats(150), stats(), stats(cpu=150)], + "reference", + None, + id="different-metric-does-not-confirm", + ), + pytest.param( + [stats(50), stats(), stats(), stats()], + "reference", + None, + id="retry-baseline-too", + ), + pytest.param( + [stats(), stats(150, error="unexpected query failure")], + "reference", + "new errors", + id="query-errors-are-not-retried", + ), + pytest.param( + [stats(), stats(150), stats(), stats(error="unexpected query failure")], + "reference", + "new errors", + id="confirmation-query-error-is-fatal", + ), + pytest.param([stats()], None, None, id="no-reference"), + ], +) +def test_benchmark_paired_confirmation( + runs: list[dict[str, Any]], compare_against: str | None, failure: str | None +) -> None: + c = MagicMock(spec=Composition) + c.query_mz_version.return_value = "test-version" + with ( + patch.object(executor, "test", side_effect=runs) as replay, + patch.object(executor, "resolve_tag", return_value="reference"), + patch.object(executor, "Materialized") as materialized, + patch.object(executor, "print_workload_stats"), + patch.object(executor, "plot_docker_stats_compare") as plots, + ): + with pytest.raises(FailedTestExecutionError) if failure else nullcontext() as exc: + executor.benchmark( + c=c, + file=Path("workload.json"), + workload={}, + compare_against=compare_against, + factor_initial_data=1, + factor_ingestions=1, + factor_queries=1, + runtime=1, + verbose=False, + seed="test-seed", + early_initial_data=False, + max_concurrent_queries=1, + ) + if failure: + assert exc is not None + assert any(failure in error.message for error in exc.value.errors) + if failure == "regressed": + details = exc.value.errors[0].details + assert details is not None + assert "Attempt 1\n" in details and "Attempt 2\n" in details + assert replay.call_count == len(runs) + assert c.rm_volumes.call_count == len(runs) + plot_files = [call.kwargs["file"] for call in plots.call_args_list] + assert len(set(plot_files)) == len(runs) // 2 + assert [call.kwargs["image"] is None for call in materialized.call_args_list] == ( + [False, True] * (len(runs) // 2) if compare_against else [True] + ) + + +def test_benchmark_requires_requested_reference() -> None: + with ( + patch.object(executor, "resolve_tag", return_value=None), + patch.object(executor, "print_workload_stats"), + patch.object(executor, "test") as replay, + ): + with pytest.raises(ValueError, match="Could not resolve reference"): + executor.benchmark( + c=MagicMock(spec=Composition), + file=Path("workload.json"), + workload={}, + compare_against="common-ancestor", + factor_initial_data=1, + factor_ingestions=1, + factor_queries=1, + runtime=1, + verbose=False, + seed="test-seed", + early_initial_data=False, + max_concurrent_queries=1, + ) + replay.assert_not_called() diff --git a/misc/python/materialize/workload_replay/stats.py b/misc/python/materialize/workload_replay/stats.py index 5aabdd377be68..fea7522717329 100644 --- a/misc/python/materialize/workload_replay/stats.py +++ b/misc/python/materialize/workload_replay/stats.py @@ -26,7 +26,6 @@ import numpy from materialize import MZ_ROOT, buildkite -from materialize.mzcompose.test_result import TestFailureDetails # Byte parsing utilities @@ -391,9 +390,9 @@ def fmt_pct(delta: float) -> str: def compare_table( - filename: str, stats_old: dict[str, Any], stats_new: dict[str, Any] -) -> list[TestFailureDetails]: - """Generate a comparison table and check for regressions.""" + stats_old: dict[str, Any], stats_new: dict[str, Any] +) -> tuple[str, set[str]]: + """Return a comparison table and the names of metrics exceeding their thresholds.""" rows = [] if "object_creation" in stats_old: rows.append( @@ -480,21 +479,19 @@ def compare_table( ] ) - failures: list[TestFailureDetails] = [] - output_lines = [ f"{'METRIC':<24} | {'OLD':^12} | {'NEW':^12} | {'CHANGE':^9} | {'THRESHOLD':^9} | {'REGRESSION?':^12}", "-" * 92, ] - regressed = False + regressions = set() for name, old, new, threshold in rows: delta = pct_change(old, new) if threshold is None: flag = "" elif new > old * threshold: - regressed = True + regressions.add(name) flag = "!!YES!!" else: flag = "no" @@ -509,14 +506,4 @@ def compare_table( f"{threshold_field:>9} | " f"{flag:^12}" ) - if regressed: - failures.append( - TestFailureDetails( - message=f"Workload {filename} regressed", - details="\n".join(output_lines), - test_class_name_override=filename, - ) - ) - - print("\n".join(output_lines)) - return failures + return "\n".join(output_lines), regressions From 07ec7c4ff8fe9d5489913eeb0e13b42dd9c229ba Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Tue, 8 Sep 2026 10:56:28 +0000 Subject: [PATCH 2/2] workload-replay: Format confirmation tests --- .../materialize/workload_replay/executor_test.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/misc/python/materialize/workload_replay/executor_test.py b/misc/python/materialize/workload_replay/executor_test.py index ef703ef22e603..b26b84ca6d80d 100644 --- a/misc/python/materialize/workload_replay/executor_test.py +++ b/misc/python/materialize/workload_replay/executor_test.py @@ -19,7 +19,9 @@ from materialize.workload_replay import executor -def stats(creation: float = 100, cpu: float = 100, error: str | None = None) -> dict[str, Any]: +def stats( + creation: float = 100, cpu: float = 100, error: str | None = None +) -> dict[str, Any]: return { "object_creation": creation, "queries": {"errors": {error: ["SELECT 1"]} if error else {}}, @@ -82,7 +84,9 @@ def test_benchmark_paired_confirmation( patch.object(executor, "print_workload_stats"), patch.object(executor, "plot_docker_stats_compare") as plots, ): - with pytest.raises(FailedTestExecutionError) if failure else nullcontext() as exc: + with ( + pytest.raises(FailedTestExecutionError) if failure else nullcontext() + ) as exc: executor.benchmark( c=c, file=Path("workload.json"), @@ -108,9 +112,9 @@ def test_benchmark_paired_confirmation( assert c.rm_volumes.call_count == len(runs) plot_files = [call.kwargs["file"] for call in plots.call_args_list] assert len(set(plot_files)) == len(runs) // 2 - assert [call.kwargs["image"] is None for call in materialized.call_args_list] == ( - [False, True] * (len(runs) // 2) if compare_against else [True] - ) + assert [ + call.kwargs["image"] is None for call in materialized.call_args_list + ] == ([False, True] * (len(runs) // 2) if compare_against else [True]) def test_benchmark_requires_requested_reference() -> None: