workload-replay: Confirm performance regressions with paired reruns

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,
+            )
+        ]
+    )

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()

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