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..b26b84ca6d80d --- /dev/null +++ b/misc/python/materialize/workload_replay/executor_test.py @@ -0,0 +1,141 @@ +# 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