Function benchmark

misc/python/materialize/workload_replay/executor.py:286–423

Run a benchmark comparing two versions of Materialize.

(
    c: Composition,
    file: pathlib.Path,
    workload: dict,
    compare_against: str,
    factor_initial_data: float,
    factor_ingestions: float,
    factor_queries: float,
    runtime: int,
    verbose: bool,
    seed: str,
    early_initial_data: bool,
    max_concurrent_queries: int,
)

def benchmark(
    c: Composition,
    file: pathlib.Path,
    workload: dict,
    compare_against: str,
    factor_initial_data: float,
    factor_ingestions: float,
    factor_queries: float,
    runtime: int,
    verbose: bool,
    seed: str,
    early_initial_data: bool,
    max_concurrent_queries: int,
) -> None:
    """Run a benchmark comparing two versions of Materialize."""
    import random

    services = [
        "materialized",
        "postgres",
        "mysql",
        "sql-server",
        "kafka",
        "schema-registry",
        "ssh-bastion-host",
        "testdrive",
    ]

    # When scale_data is false, use 100% initial data
    settings = workload.get("settings", {})
    if not settings.get("scale_data", True):
        factor_initial_data = 1.0

    print_workload_stats(file, workload)

    tag = resolve_tag(compare_against)
    print(f"-- Running against materialized:{tag} (reference)")
    random.seed(seed)
    with c.override(
        Materialized(
            image=f"{image_registry()}/materialized:{tag}",
            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={"enable_rbac_checks": "false"},
        )
    ):
        stats_old = test(
            c,
            workload,
            file,
            factor_initial_data,
            factor_ingestions,
            factor_queries,
            runtime,
            verbose,
            True,
            True,