misc.python.materialize.mzcompose

   1# Copyright Materialize, Inc. and contributors. All rights reserved.
   2#
   3# Use of this software is governed by the Business Source License
   4# included in the LICENSE file at the root of this repository.
   5#
   6# As of the Change Date specified in that file, in accordance with
   7# the Business Source License, use of this software will be governed
   8# by the Apache License, Version 2.0.
   9
  10"""The implementation of the mzcompose system for Docker compositions.
  11
  12For an overview of what mzcompose is and why it exists, see the [user-facing
  13documentation][user-docs].
  14
  15[user-docs]: https://github.com/MaterializeInc/materialize/blob/main/doc/developer/mzbuild.md
  16"""
  17
  18import os
  19import random
  20import subprocess
  21import sys
  22from collections.abc import Iterable
  23from dataclasses import dataclass
  24from typing import Any, Literal, TypeVar
  25
  26import psycopg
  27
  28from materialize import spawn, ui
  29from materialize.mz_version import MzVersion
  30from materialize.rustc_flags import Sanitizer
  31from materialize.ui import UIError
  32
  33T = TypeVar("T")
  34say = ui.speaker("C> ")
  35
  36
  37DEFAULT_CONFLUENT_PLATFORM_VERSION = "8.2.0"
  38
  39DEFAULT_MZ_VOLUMES = [
  40    "mzdata:/mzdata",
  41    "mydata:/var/lib/mysql-files",
  42    "tmp:/share/tmp",
  43    "scratch:/scratch",
  44]
  45
  46
  47# Parameters which disable systems that periodically/unpredictably impact performance
  48# We try to keep this empty, so that we benchmark Materialize as we ship it. If
  49# a new feature causes benchmarks to become flaky, consider that this can also
  50# impact customers' experience and try to find a solution other than disabling
  51# the feature here!
  52ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS = {
  53    # Benchmarks measure the intended production configuration. For hedged
  54    # blob gets that is the planned enablement state (on, at production
  55    # tuning), not the CI-wide coverage tuning below, whose short delay
  56    # would add duplicate fetches to any measured get slower than it.
  57    "persist_blob_hedged_get_enabled": "true",
  58    "persist_blob_hedged_get_delay": "2s",
  59    "persist_blob_hedged_get_budget_ratio": "0.01",
  60    # Curated metric sinks are causing a regression in performance thresholds.
  61    # Currently, the flag defaults off in production, so benchmarking without it
  62    # measures the configuration we ship. Correctness coverage is unaffected:
  63    # the rest of CI # still gets "true" from get_minimal_system_parameters().
  64    # TODO: remove once the regression is fixed
  65    "enable_metric_sink": "false",
  66}
  67
  68
  69def sanitizer_enabled() -> bool:
  70    """Whether the binaries under test were built with a sanitizer.
  71
  72    Sanitizer builds run several times slower and use several times as much
  73    memory as ordinary ones, so tests that assert on timing, or that need
  74    jemalloc (which sanitizer builds drop, as it clashes with the sanitizer
  75    runtimes), have to account for them.
  76    """
  77    return Sanitizer[os.getenv("CI_SANITIZER", "none")] != Sanitizer.none
  78
  79
  80def get_minimal_system_parameters(
  81    version: MzVersion,
  82) -> dict[str, str]:
  83    """Settings we need in order to have tests run at all, but otherwise stay
  84    with the defaults: not changing performance or increasing coverage."""
  85
  86    config = {
  87        # -----
  88        # Unsafe functions
  89        "unsafe_enable_unsafe_functions": "true",
  90        # -----
  91        # Others (ordered by name)
  92        "allow_real_time_recency": "true",
  93        "constraint_based_timestamp_selection": "verify",  # removed from main, keeping it here for old versions
  94        "enable_compute_peek_response_stash": "true",
  95        "enable_0dt_deployment_panic_after_timeout": "true",
  96        "enable_0dt_deployment_sources": (
  97            "true" if version >= MzVersion.parse_mz("v0.132.0-dev") else "false"
  98        ),
  99        "enable_alter_swap": "true",
 100        "enable_arrangement_dictionary_compression_alpha": "false",
 101        "enable_case_literal_transform": "true",
 102        "enable_cast_elimination": "true",
 103        "enable_coalesce_case_transform": "true",
 104        "enable_columnation_lgalloc": "false",
 105        "enable_compute_correction_v2": "true",
 106        "enable_compute_logical_backpressure": "true",
 107        "enable_connection_validation_syntax": "true",
 108        "enable_create_table_from_source": "true",
 109        "enable_eager_delta_joins": "true",
 110        "enable_envelope_debezium_in_subscribe": "true",
 111        "enable_exclude_constraints_option": "true",
 112        "enable_expressions_in_limit_syntax": "true",
 113        "enable_fixed_correlated_cte_lowering": "true",
 114        "enable_introspection_subscribes": "true",
 115        "enable_lgalloc": "false",
 116        "enable_load_generator_counter": "true",
 117        "enable_logical_compaction_window": "true",
 118        "enable_metric_sink": "true",
 119        "enable_multi_worker_storage_persist_sink": "true",
 120        "enable_rbac_checks": "true",
 121        "enable_reduce_mfp_fusion": "true",
 122        "enable_refresh_every_mvs": "true",
 123        "enable_replacement_materialized_views": "true",
 124        "enable_cluster_schedule_refresh": "true",
 125        # Pinned explicitly so runs against older versions (which predate the
 126        # flag or defaulted it off) behave like current ones, where it defaults
 127        # on.
 128        "enable_background_alter_cluster": (
 129            "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false"
 130        ),
 131        "enable_s3_tables_region_check": "false",
 132        "enable_statement_lifecycle_logging": "true",
 133        "enable_storage_introspection_logs": "true",
 134        "enable_compute_error_distinct": "true",
 135        "enable_compute_temporal_bucketing": "true",
 136        "enable_union_cancellation_after_relation_cse": "true",
 137        "enable_variadic_left_join_lowering": "true",
 138        "enable_worker_core_affinity": "true",
 139        "grpc_client_http2_keep_alive_timeout": "5s",
 140        "ore_overflowing_behavior": "panic",
 141        "unsafe_enable_table_keys": "true",
 142        # Keep the 0dt stability soak out of the critical path for tests. The
 143        # production default is much higher. Dedicated workflows override this.
 144        "with_0dt_caught_up_check_stability_period": "0s",
 145        "with_0dt_deployment_max_wait": "1800s",
 146        # End of list (ordered by name)
 147    }
 148
 149    if version >= MzVersion.parse_mz("v26.40.0-dev"):
 150        # Exercise the row-limit check without constraining normal test queries.
 151        config["compute_peek_row_iteration_limit"] = "1000000000"
 152        config["enable_compute_peek_row_iteration_limit"] = "true"
 153
 154        # Exercise the peek offload path in tests while it defaults off in
 155        # production. The budgets stay at their code defaults so tests make the
 156        # same placement decisions production makes.
 157        config["enable_compute_index_peek_offload"] = "true"
 158
 159    if version < MzVersion.parse_mz("v0.163.0-dev"):
 160        config["enable_compute_active_dataflow_cancelation"] = "true"
 161
 162    if version < MzVersion.parse_mz("v26.24.0-dev"):
 163        config["enable_columnar_lgalloc"] = "false"
 164    if version < MzVersion.parse_mz("v26.25.0-dev"):
 165        config["enable_multi_replica_sources"] = "true"
 166
 167    if version >= MzVersion.parse_mz("v26.40.0-dev"):
 168        config["hydration_history_collection_interval"] = "60s"
 169
 170    if sanitizer_enabled():
 171        config["with_0dt_deployment_max_wait"] = "18000s"
 172
 173    # The cluster controller's break-glass gate. Removed in v26.38, where the
 174    # controller runs unconditionally. Older binaries still read it, and
 175    # defaulted it off before v26.29, so pin it on for them to keep mixed-version
 176    # runs exercising the same path as current versions.
 177    if version < MzVersion.parse_mz("v26.38.0-dev"):
 178        config["enable_cluster_controller"] = (
 179            "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false"
 180        )
 181
 182    # The `WITH (WAIT ...)` graceful-reconfiguration surface. Always accepted
 183    # from v26.41 on. Older binaries still gate it behind this feature flag, so
 184    # pin it on for them: the tests that use the surface no longer enable it
 185    # themselves, and in a mixed-version run some of their phases execute
 186    # against the old binary.
 187    if version < MzVersion.parse_mz("v26.41.0-dev"):
 188        config["enable_zero_downtime_cluster_reconfiguration"] = "true"
 189
 190    return config
 191
 192
 193@dataclass
 194class VariableSystemParameter:
 195    key: str
 196    default: str
 197    values: list[str]
 198
 199
 200# TODO: The linter should check this too
 201def get_variable_system_parameters(
 202    version: MzVersion,
 203    force_source_table_syntax: bool,
 204    metadata_store: str,
 205) -> list[VariableSystemParameter]:
 206    """Note: Only the default is tested unless we explicitly select "System Parameters: Random" in trigger-ci.
 207    These defaults are applied _after_ applying the settings from `get_minimal_system_parameters`.
 208    """
 209
 210    # `persist_pg_consensus_read_committed` must stay off on CockroachDB, where
 211    # the lockless CRDB_* consensus queries are only linearizable under
 212    # SERIALIZABLE and persist asserts on the connection's isolation level. On
 213    # Postgres-backed consensus the query family is linearizable under READ
 214    # COMMITTED, so default it on and let it vary.
 215    read_committed_safe = metadata_store in ("postgres-metadata", "alloydb")
 216    persist_pg_consensus_read_committed = VariableSystemParameter(
 217        "persist_pg_consensus_read_committed",
 218        "true" if read_committed_safe else "false",
 219        ["true", "false"] if read_committed_safe else ["false"],
 220    )
 221
 222    params = [
 223        # -----
 224        # To reduce CRDB load as we are struggling with it in CI (values based on load test environment):
 225        VariableSystemParameter(
 226            "persist_next_listen_batch_retryer_clamp",
 227            "16s",
 228            ["100ms", "1s", "10s", "100s"],
 229        ),
 230        VariableSystemParameter(
 231            "persist_next_listen_batch_retryer_initial_backoff",
 232            "100ms",
 233            ["10ms", "100ms", "1s", "10s"],
 234        ),
 235        VariableSystemParameter(
 236            "persist_next_listen_batch_retryer_fixed_sleep",
 237            "1200ms",
 238            ["100ms", "1s", "10s"],
 239        ),
 240        # -----
 241        # Persist internals changes, advance coverage
 242        VariableSystemParameter(
 243            "persist_source_fetch_concurrency", "1", ["1", "2", "8", "16"]
 244        ),
 245        VariableSystemParameter(
 246            "persist_blob_hedged_get_enabled", "true", ["true", "false"]
 247        ),
 248        # 10ms (vs the 2s production default) makes hedges actually fire in
 249        # every CI run; 0s makes every blob get hedge under randomized seeds.
 250        VariableSystemParameter(
 251            "persist_blob_hedged_get_delay", "10ms", ["0s", "10ms", "2s"]
 252        ),
 253        # The production ratio: with the 10ms delay above, a full refill
 254        # would hedge nearly every get and double CI blob traffic. The 1.0
 255        # variant lets randomized runs pair a full budget with delay=0s.
 256        VariableSystemParameter(
 257            "persist_blob_hedged_get_budget_ratio", "0.01", ["1.0", "0.01"]
 258        ),
 259        # -----
 260        # Others (ordered by name),
 261        VariableSystemParameter(
 262            "aws_prefetch_sts_connect_timeout",
 263            "3100ms",
 264            ["3100ms", "30s", "60s"],
 265        ),
 266        VariableSystemParameter(
 267            "compute_correction_v2_chain_proportionality",
 268            "3",
 269            ["2", "3"],
 270        ),
 271        VariableSystemParameter(
 272            "compute_correction_v2_chunk_size",
 273            "8192",
 274            ["8192", "65536", "1048576"],
 275        ),
 276        VariableSystemParameter(
 277            "compute_dataflow_max_inflight_bytes",
 278            "134217728",
 279            ["1048576", "4194304", "16777216", "67108864"],
 280        ),  # 128 MiB
 281        VariableSystemParameter("compute_hydration_concurrency", "2", ["1", "2", "4"]),
 282        VariableSystemParameter(
 283            "compute_replica_expiration_offset", "3d", ["3d", "10d"]
 284        ),
 285        VariableSystemParameter(
 286            "compute_apply_column_demands", "true", ["true", "false"]
 287        ),
 288        # On by default so CI exercises the columnar merge batcher, which is
 289        # off in production while it earns trust.
 290        VariableSystemParameter(
 291            "enable_columnar_merge_batcher", "true", ["true", "false"]
 292        ),
 293        # On by default so CI exercises the columnar accumulable diff layout, which
 294        # is off in production while it earns trust.
 295        VariableSystemParameter(
 296            "enable_columnar_accumulable_diff", "true", ["true", "false"]
 297        ),
 298        VariableSystemParameter(
 299            "compute_peek_response_stash_threshold_bytes",
 300            # 1 MiB, an in-between value
 301            "1048576",
 302            # force-enabled, the in-between, and the production value
 303            ["0", "1048576", "314572800", "67108864"],
 304        ),
 305        VariableSystemParameter(
 306            "compute_subscribe_snapshot_optimization",
 307            "true",
 308            ["true", "false"],
 309        ),
 310        VariableSystemParameter(
 311            "enable_case_literal_transform",
 312            "false",
 313            ["true", "false"],
 314        ),
 315        VariableSystemParameter(
 316            "enable_coalesce_case_transform",
 317            "true",
 318            ["true", "false"],
 319        ),
 320        VariableSystemParameter(
 321            "enable_adapter_frontend_occ_read_then_write",
 322            "true" if version >= MzVersion.parse_mz("v26.36.0-dev") else "false",
 323            ["true", "false"],
 324        ),
 325        VariableSystemParameter(
 326            "enable_cast_elimination",
 327            "true",
 328            ["true", "false"],
 329        ),
 330        VariableSystemParameter(
 331            "enable_fixed_correlated_cte_lowering",
 332            "true",
 333            ["true", "false"],
 334        ),
 335        VariableSystemParameter(
 336            "enable_compute_sync_mv_sink",
 337            "true",
 338            ["true", "false"],
 339        ),
 340        VariableSystemParameter(
 341            "enable_password_auth",
 342            "true",
 343            ["true", "false"],
 344        ),
 345        VariableSystemParameter(
 346            "enable_frontend_peek_sequencing",
 347            "true" if version >= MzVersion.parse_mz("v26.9.0-dev") else "false",
 348            ["true", "false"],
 349        ),
 350        VariableSystemParameter(
 351            "enable_frontend_subscribes",
 352            "true" if version >= MzVersion.parse_mz("v26.18.0-dev") else "false",
 353            ["true", "false"],
 354        ),
 355        VariableSystemParameter(
 356            "enable_simplify_from_less_existence",
 357            "true",
 358            ["true", "false"],
 359        ),
 360        VariableSystemParameter(
 361            "enable_union_cancellation_after_relation_cse",
 362            "true",
 363            ["true", "false"],
 364        ),
 365        VariableSystemParameter(
 366            "enable_upsert_paged_spill",
 367            "true",
 368            ["true", "false"],
 369        ),
 370        # On by default so CI exercises the chunked stash flavor, which is
 371        # off in production while it earns trust. Only meaningful when
 372        # enable_upsert_v2 is true.
 373        VariableSystemParameter(
 374            "enable_upsert_chunked_stash",
 375            "true",
 376            ["true", "false"],
 377        ),
 378        VariableSystemParameter(
 379            "enable_upsert_v2",
 380            "false",
 381            ["true", "false"],
 382        ),
 383        VariableSystemParameter(
 384            "default_timestamp_interval",
 385            "1s",
 386            ["100ms", "1s"],
 387        ),
 388        VariableSystemParameter(
 389            "force_source_table_syntax",
 390            "true" if force_source_table_syntax else "false",
 391            ["true", "false"] if force_source_table_syntax else ["false"],
 392        ),
 393        VariableSystemParameter(
 394            "mysql_source_snapshot_parallelism", "true", ["true", "false"]
 395        ),
 396        # Low default so small tables exercise partitioning.
 397        VariableSystemParameter(
 398            "mysql_source_snapshot_partition_min_rows", "2", ["2", "50000"]
 399        ),
 400        VariableSystemParameter(
 401            "persist_batch_delete_enabled", "true", ["true", "false"]
 402        ),
 403        VariableSystemParameter(
 404            "persist_batch_structured_key_lower_len",
 405            "256",
 406            ["0", "1", "512", "1000"],
 407        ),
 408        VariableSystemParameter(
 409            "persist_batch_max_run_len", "4", ["2", "3", "4", "16"]
 410        ),
 411        VariableSystemParameter(
 412            "persist_catalog_force_compaction_fuel",
 413            "1024",
 414            ["256", "1024", "4096"],
 415        ),
 416        VariableSystemParameter(
 417            "persist_catalog_force_compaction_wait",
 418            "1s",
 419            ["100ms", "1s", "10s"],
 420        ),
 421        VariableSystemParameter(
 422            "persist_stats_audit_percent",
 423            "100",
 424            [
 425                "0",
 426                "1",
 427                "2",
 428                "10",
 429                "100",
 430            ],
 431        ),
 432        VariableSystemParameter("persist_stats_audit_panic", "true", ["true", "false"]),
 433        VariableSystemParameter(
 434            "persist_encoding_enable_dictionary", "true", ["true", "false"]
 435        ),
 436        VariableSystemParameter(
 437            "persist_fast_path_limit",
 438            "1000",
 439            ["100", "1000", "10000"],
 440        ),
 441        VariableSystemParameter("persist_fast_path_order", "true", ["true", "false"]),
 442        VariableSystemParameter(
 443            "persist_gc_use_active_gc",
 444            ("true" if version > MzVersion.parse_mz("v0.143.0-dev") else "false"),
 445            (
 446                ["true", "false"]
 447                if version > MzVersion.parse_mz("v0.127.0-dev")
 448                else ["false"]
 449            ),
 450        ),
 451        VariableSystemParameter(
 452            "persist_gc_min_versions",
 453            "16",
 454            ["16", "256", "1024"],
 455        ),
 456        VariableSystemParameter(
 457            "persist_gc_max_versions",
 458            "128000",
 459            ["256", "128000"],
 460        ),
 461        VariableSystemParameter(
 462            "persist_inline_writes_single_max_bytes",
 463            "4096",
 464            ["256", "1024", "4096", "16384"],
 465        ),
 466        VariableSystemParameter(
 467            "persist_inline_writes_total_max_bytes",
 468            "1048576",
 469            ["65536", "262144", "1048576", "4194304"],
 470        ),
 471        VariableSystemParameter(
 472            "persist_pubsub_client_enabled", "true", ["true", "false"]
 473        ),
 474        VariableSystemParameter(
 475            "persist_pubsub_push_diff_enabled", "true", ["true", "false"]
 476        ),
 477        VariableSystemParameter(
 478            "persist_rollup_use_active_rollup",
 479            ("true" if version > MzVersion.parse_mz("v0.143.0-dev") else "false"),
 480            (
 481                ["true", "false"]
 482                if version > MzVersion.parse_mz("v0.127.0-dev")
 483                else ["false"]
 484            ),
 485        ),
 486        # 16 MiB - large enough to avoid a big perf hit, small enough to get more coverage...
 487        VariableSystemParameter(
 488            "persist_blob_target_size",
 489            "16777216",
 490            ["4096", "1048576", "16777216", "134217728"],
 491        ),
 492        # 5 times the default part size - 4 is the bare minimum.
 493        VariableSystemParameter(
 494            "persist_compaction_memory_bound_bytes",
 495            "83886080",
 496            ["67108864", "134217728", "536870912", "1073741824"],
 497        ),
 498        VariableSystemParameter(
 499            "persist_enable_incremental_compaction",
 500            ("true" if version >= MzVersion.parse_mz("v0.161.0-dev") else "false"),
 501            (
 502                ["true", "false"]
 503                if version >= MzVersion.parse_mz("v0.161.0-dev")
 504                else ["false"]
 505            ),
 506        ),
 507        VariableSystemParameter(
 508            "persist_use_critical_since_catalog", "true", ["true", "false"]
 509        ),
 510        VariableSystemParameter(
 511            "persist_use_critical_since_snapshot",
 512            "false",  # always false, because we always have zero-downtime enabled
 513            ["false"],
 514        ),
 515        VariableSystemParameter(
 516            "persist_use_critical_since_source",
 517            "false",  # always false, because we always have zero-downtime enabled
 518            ["false"],
 519        ),
 520        # 0 disables; otherwise coalesce hydration frontier downgrades until
 521        # this many encoded bytes have been emitted (1 MiB, 16 MiB, 128 MiB).
 522        VariableSystemParameter(
 523            "persist_source_hydration_frontier_coalesce_bytes",
 524            "0",
 525            ["0", "1048576", "16777216", "134217728"],
 526        ),
 527        VariableSystemParameter(
 528            "persist_part_decode_format", "arrow", ["arrow", "row_with_validate"]
 529        ),
 530        VariableSystemParameter(
 531            "persist_blob_cache_scale_with_threads", "true", ["true", "false"]
 532        ),
 533        persist_pg_consensus_read_committed,
 534        VariableSystemParameter(
 535            "persist_state_update_lease_timeout", "1s", ["0s", "1s", "10s"]
 536        ),
 537        VariableSystemParameter(
 538            "arrangement_size_history_collection_interval", "1h", ["1s", "10s", "1h"]
 539        ),
 540        VariableSystemParameter(
 541            "arrangement_size_history_retention_period", "7d", ["1min", "1h", "7d"]
 542        ),
 543        *(
 544            [
 545                VariableSystemParameter(
 546                    "hydration_history_retention_period",
 547                    "30d",
 548                    ["1min", "1h", "30d"],
 549                )
 550            ]
 551            if version >= MzVersion.parse_mz("v26.40.0-dev")
 552            else []
 553        ),
 554        VariableSystemParameter(
 555            "persist_validate_part_bounds_on_read", "false", ["true", "false"]
 556        ),
 557        VariableSystemParameter(
 558            "persist_validate_part_bounds_on_write", "false", ["true", "false"]
 559        ),
 560        VariableSystemParameter(
 561            "statement_logging_default_sample_rate",
 562            "1.0",
 563            ["0", "0.01", "0.5", "0.99", "1.0"],
 564        ),
 565        VariableSystemParameter(
 566            "statement_logging_max_data_credit",
 567            "",
 568            ["", "0", "1024", "1048576", "1073741824"],
 569        ),
 570        VariableSystemParameter(
 571            "statement_logging_max_sample_rate",
 572            "1.0",
 573            ["0", "0.01", "0.5", "0.99", "1.0"],
 574        ),
 575        VariableSystemParameter(
 576            "statement_logging_target_data_rate",
 577            "",
 578            ["", "0", "1", "1000", "2071", "1000000"],
 579        ),
 580        VariableSystemParameter(
 581            "storage_source_decode_fuel",
 582            "100000",
 583            ["10000", "100000", "1000000"],
 584        ),
 585        VariableSystemParameter(
 586            "storage_statistics_collection_interval",
 587            "1000",
 588            ["100", "1000", "10000"],
 589        ),
 590        VariableSystemParameter(
 591            "storage_statistics_interval", "2000", ["100", "1000", "10000"]
 592        ),
 593        VariableSystemParameter(
 594            "storage_use_continual_feedback_upsert", "true", ["true", "false"]
 595        ),
 596        # End of list (ordered by name)
 597    ]
 598
 599    if version < MzVersion.parse_mz("v26.14.0-dev"):
 600        params.append(
 601            VariableSystemParameter(
 602                "storage_reclock_to_latest", "true", ["true", "false"]
 603            )
 604        )
 605    if version < MzVersion.parse_mz("v26.23.0-dev"):
 606        params.append(
 607            VariableSystemParameter(
 608                "persist_enable_arrow_lgalloc_noncc_sizes", "true", ["true", "false"]
 609            )
 610        )
 611        params.append(
 612            VariableSystemParameter(
 613                "persist_enable_s3_lgalloc_noncc_sizes", "true", ["true", "false"]
 614            )
 615        )
 616
 617    return params
 618
 619
 620def get_default_system_parameters(
 621    version: MzVersion | None = None,
 622    force_source_table_syntax: bool = False,
 623    metadata_store: str | None = None,
 624) -> dict[str, str]:
 625    """For upgrade tests we only want parameters set when all environmentd /
 626    clusterd processes have reached a specific version (or higher)
 627
 628    `metadata_store` selects backend-specific defaults. It defaults to the
 629    globally configured metadata store, but callers that target a different
 630    backend than the global (e.g. a local CockroachDB) must pass their own.
 631    """
 632
 633    if not version:
 634        version = MzVersion.parse_cargo()
 635
 636    if metadata_store is None:
 637        from materialize.mzcompose.services.metadata_store import METADATA_STORE
 638
 639        metadata_store = METADATA_STORE
 640
 641    params = get_minimal_system_parameters(version)
 642
 643    system_param_setting = os.getenv("CI_SYSTEM_PARAMETERS", "")
 644    variable_params = get_variable_system_parameters(
 645        version, force_source_table_syntax, metadata_store
 646    )
 647
 648    if system_param_setting == "":
 649        for param in variable_params:
 650            params[param.key] = param.default
 651    elif system_param_setting == "random":
 652        seed = os.getenv("CI_SYSTEM_PARAMETERS_SEED", os.getenv("BUILDKITE_JOB_ID", 1))
 653        rng = random.Random(seed)
 654        for param in variable_params:
 655            params[param.key] = rng.choice(param.values)
 656        print(
 657            f"System parameters with seed CI_SYSTEM_PARAMETERS_SEED={seed}: {params}",
 658            file=sys.stderr,
 659        )
 660    elif system_param_setting == "minimal":
 661        pass
 662    else:
 663        raise ValueError(
 664            f"Unknown value for CI_SYSTEM_PARAMETERS: {system_param_setting}"
 665        )
 666
 667    return params
 668
 669
 670# If you are adding a new config flag in Materialize, consider setting values
 671# for it in get_variable_system_parameters if it can be varied in tests. Set it
 672# in get_minimal_system_parameters if it's required for tests to succeed at
 673# all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
 674# apply.
 675UNINTERESTING_SYSTEM_PARAMETERS = [
 676    "enable_compute_half_join2",
 677    "enable_mz_join_core",
 678    "linear_join_yielding",
 679    "enable_column_paged_batcher",
 680    "enable_column_paged_batcher_spill",
 681    "column_chunk_compress_min_depth",
 682    "column_paged_batcher_budget_fraction",
 683    "column_paged_batcher_lz4",
 684    "column_paged_batcher_swap_pageout",
 685    "column_paged_batcher_spill_worker_count",
 686    "column_paged_batcher_eager_backing",
 687    "column_paged_batcher_pool_rss_target_fraction",
 688    "enable_lgalloc_eager_reclamation",
 689    "lgalloc_background_interval",
 690    "lgalloc_file_growth_dampener",
 691    "lgalloc_local_buffer_bytes",
 692    "lgalloc_slow_clear_bytes",
 693    "memory_limiter_interval",
 694    "memory_limiter_usage_bias",
 695    "memory_limiter_burst_factor",
 696    "catalog_info_metrics_reconcile_interval",
 697    "compute_server_maintenance_interval",
 698    "compute_dataflow_max_inflight_bytes_cc",
 699    "compute_flat_map_fuel",
 700    "compute_temporal_bucketing_summary",
 701    "consolidating_vec_growth_dampener",
 702    "copy_to_s3_parquet_row_group_file_ratio",
 703    "copy_to_s3_arrow_builder_buffer_ratio",
 704    "copy_to_s3_multipart_part_size_bytes",
 705    "enable_replica_targeted_materialized_views",
 706    "compute_mv_sink_advance_persist_frontiers",
 707    "compute_prometheus_introspection_scrape_interval",
 708    "enable_compute_replica_expiration",
 709    "compute_logical_backpressure_max_retained_capabilities",
 710    "compute_logical_backpressure_inflight_slack",
 711    "persist_fetch_semaphore_cost_adjustment",
 712    "persist_fetch_semaphore_permit_adjustment",
 713    "persist_optimize_ignored_data_fetch",
 714    "persist_pubsub_same_process_delegate_enabled",
 715    "persist_pubsub_connect_attempt_timeout",
 716    "persist_pubsub_request_timeout",
 717    "persist_pubsub_connect_max_backoff",
 718    "persist_pubsub_client_sender_channel_size",
 719    "persist_pubsub_client_receiver_channel_size",
 720    "persist_pubsub_server_connection_channel_size",
 721    "persist_pubsub_state_cache_shard_ref_channel_size",
 722    "persist_pubsub_reconnect_backoff",
 723    "persist_encoding_compression_format",
 724    "persist_batch_max_runs",
 725    "persist_write_combine_inline_writes",
 726    "persist_reader_lease_duration",
 727    "persist_consensus_connection_pool_max_size",
 728    "persist_consensus_connection_pool_max_wait",
 729    "persist_consensus_connection_pool_ttl",
 730    "persist_consensus_connection_pool_ttl_stagger",
 731    "persist_use_postgres_tuned_queries",
 732    "crdb_connect_timeout",
 733    "crdb_tcp_user_timeout",
 734    "crdb_keepalives_idle",
 735    "crdb_keepalives_interval",
 736    "crdb_keepalives_retries",
 737    "pg_timestamp_oracle_statement_timeout",
 738    "persist_use_critical_since_txn",
 739    "use_global_txn_cache_source",
 740    "persist_batch_builder_max_outstanding_parts",
 741    "persist_compaction_heuristic_min_inputs",
 742    "persist_compaction_heuristic_min_parts",
 743    "persist_compaction_heuristic_min_updates",
 744    "persist_gc_blob_delete_concurrency_limit",
 745    "persist_state_versions_recent_live_diffs_limit",
 746    "persist_usage_state_fetch_concurrency_limit",
 747    "persist_blob_operation_timeout",
 748    "persist_blob_operation_attempt_timeout",
 749    "persist_blob_connect_timeout",
 750    "persist_blob_read_timeout",
 751    "persist_blob_hedged_get_max_concurrent",
 752    "persist_blob_hedged_get_warm_interval",
 753    "persist_stats_collection_enabled",
 754    "persist_stats_filter_enabled",
 755    "persist_stats_budget_bytes",
 756    "persist_stats_untrimmable_columns_equals",
 757    "persist_stats_untrimmable_columns_prefix",
 758    "persist_stats_untrimmable_columns_suffix",
 759    "persist_expression_cache_force_compaction_fuel",
 760    "persist_expression_cache_force_compaction_wait",
 761    "persist_blob_cache_mem_limit_bytes",
 762    "persist_blob_cache_scale_factor_bytes",
 763    "persist_claim_unclaimed_compactions",
 764    "persist_claim_compaction_percent",
 765    "persist_claim_compaction_min_version",
 766    "persist_next_listen_batch_retryer_multiplier",
 767    "persist_rollup_threshold",
 768    "persist_rollup_fallback_threshold_ms",
 769    "persist_gc_fallback_threshold_ms",
 770    "persist_compaction_minimum_timeout",
 771    "persist_compaction_check_process_flag",
 772    "balancerd_sigterm_connection_wait",
 773    "balancerd_sigterm_listen_wait",
 774    "balancerd_inject_proxy_protocol_header_http",
 775    "balancerd_max_connections",
 776    "balancerd_log_filter",
 777    "balancerd_opentelemetry_filter",
 778    "balancerd_log_filter_defaults",
 779    "balancerd_opentelemetry_filter_defaults",
 780    "balancerd_sentry_filters",
 781    "persist_enable_s3_lgalloc_cc_sizes",
 782    "persist_enable_arrow_lgalloc_cc_sizes",
 783    "controller_past_generation_replica_cleanup_retry_interval",
 784    "wallclock_lag_recording_interval",
 785    "wallclock_lag_histogram_period_interval",
 786    "enable_timely_zero_copy",
 787    "enable_timely_zero_copy_lgalloc",
 788    "timely_zero_copy_limit",
 789    "arrangement_exert_proportionality",
 790    "txn_wal_apply_ensure_schema_match",
 791    "persist_txns_data_shard_retryer_initial_backoff",
 792    "persist_txns_data_shard_retryer_multiplier",
 793    "persist_txns_data_shard_retryer_clamp",
 794    "storage_cluster_shutdown_grace_period",
 795    "storage_dataflow_delay_sources_past_rehydration",
 796    "storage_dataflow_suspendable_sources",
 797    "storage_downgrade_since_during_finalization",
 798    "replica_metrics_history_retention_interval",
 799    "wallclock_lag_history_retention_interval",
 800    "wallclock_global_lag_histogram_retention_interval",
 801    "kafka_client_id_enrichment_rules",
 802    "kafka_poll_max_wait",
 803    "kafka_default_aws_privatelink_endpoint_identification_algorithm",
 804    "kafka_buffered_event_resize_threshold_elements",
 805    "kafka_low_watermark_check",
 806    "mysql_replication_heartbeat_interval",
 807    # Not varied here because statistics tests assert exact
 808    # snapshot_records_known values, which only hold on the exact-count path.
 809    # The estimated path is covered explicitly in mysql-cdc/statistics.td and
 810    # by parallel-workload.
 811    "mysql_source_snapshot_exact_count_max_rows",
 812    "mysql_source_snapshot_partition_probed_prefixes_per_billion_rows",
 813    "postgres_fetch_slot_resume_lsn_interval",
 814    "pg_schema_validation_interval",
 815    "pg_source_validate_timeline",
 816    "sql_server_source_validate_restore_history",
 817    "storage_enforce_external_addresses",
 818    "storage_upsert_prevent_snapshot_buffering",
 819    "storage_rocksdb_use_merge_operator",
 820    "storage_upsert_max_snapshot_batch_buffering",
 821    "storage_rocksdb_cleanup_tries",
 822    "storage_suspend_and_restart_delay",
 823    "storage_server_maintenance_interval",
 824    "storage_sink_progress_search",
 825    "storage_sink_ensure_topic_config",
 826    "sql_server_max_lsn_wait",
 827    "sql_server_snapshot_progress_report_interval",
 828    "sql_server_cdc_cleanup_change_table",
 829    "sql_server_cdc_cleanup_change_table_max_deletes",
 830    "allow_user_sessions",
 831    "group_commit_max_attempts",
 832    "with_0dt_deployment_ddl_check_interval",
 833    "enable_0dt_caught_up_check",
 834    "with_0dt_caught_up_check_allowed_lag",
 835    "with_0dt_caught_up_check_cutoff",
 836    "enable_0dt_caught_up_replica_status_check",
 837    "enable_0dt_caught_up_stability_check",
 838    "enable_0dt_hydrate_migrated_builtin_mvs",
 839    "plan_insights_notice_fast_path_clusters_optimize_duration",
 840    "enable_expression_cache",
 841    "mz_metrics_lgalloc_map_refresh_interval",
 842    "mz_metrics_lgalloc_refresh_interval",
 843    "mz_metrics_rusage_refresh_interval",
 844    "mz_metrics_usage_refresh_interval",
 845    "compute_peek_response_stash_batch_max_runs",
 846    # The offload's budgets, left at their code defaults so tests exercise the
 847    # placement decisions production makes. parallel-workload varies them.
 848    "compute_index_peek_inline_budget",
 849    "compute_index_peek_activation_budget",
 850    "compute_index_peek_yield_granularity",
 851    "compute_index_peek_permit_fraction",
 852    "compute_peek_response_stash_batch_bytes",
 853    "compute_peek_response_stash_read_batch_size_bytes",
 854    "compute_peek_response_stash_read_memory_budget_bytes",
 855    "storage_statistics_retention_duration",
 856    "enable_paused_cluster_readhold_downgrade",
 857    "kafka_retry_backoff",
 858    "kafka_retry_backoff_max",
 859    "kafka_reconnect_backoff",
 860    "kafka_reconnect_backoff_max",
 861    "kafka_sink_message_max_bytes",
 862    "kafka_sink_batch_size",
 863    "kafka_sink_batch_num_messages",
 864    "oidc_issuer",
 865    "oidc_audience",
 866    "oidc_authentication_claim",
 867    "oidc_group_role_sync_enabled",
 868    "oidc_group_claim",
 869    "oidc_group_role_sync_strict",
 870    "console_oidc_client_id",
 871    "console_oidc_scopes",
 872    "enable_public_metrics_endpoint",
 873    "enable_mcp_agent",
 874    "enable_mcp_agent_query_tool",
 875    "enable_mcp_agent_read_data_product_tool",
 876    "enable_mcp_developer",
 877    "enable_mcp_developer_query_tool",
 878    "mcp_max_response_size",
 879    "mcp_request_timeout",
 880    "user_id_pool_batch_size",
 881    "webhook_max_request_size_bytes",
 882    "webhook_validation_memory_budget_bytes",
 883    "subscribe_max_buffered_bytes",
 884    "cluster_controller_tick_interval",
 885    "default_cluster_reconfiguration_timeout",
 886    "read_then_write_max_dependencies",
 887    "enable_hydration_burst",
 888    "default_hydration_burst_linger",
 889]
 890
 891
 892DEFAULT_CRDB_ENVIRONMENT = [
 893    "COCKROACH_ENGINE_MAX_SYNC_DURATION_DEFAULT=120s",
 894    "COCKROACH_LOG_MAX_SYNC_DURATION=120s",
 895]
 896
 897
 898# TODO(benesch): change to `docker-mzcompose` once v0.39 ships.
 899DEFAULT_CLOUD_PROVIDER = "mzcompose"
 900DEFAULT_CLOUD_REGION = "us-east-1"
 901DEFAULT_ORG_ID = "00000000-0000-0000-0000-000000000000"
 902DEFAULT_ORDINAL = "0"
 903DEFAULT_MZ_ENVIRONMENT_ID = f"{DEFAULT_CLOUD_PROVIDER}-{DEFAULT_CLOUD_REGION}-{DEFAULT_ORG_ID}-{DEFAULT_ORDINAL}"
 904
 905
 906# TODO(benesch): replace with Docker health checks.
 907def _check_tcp(
 908    cmd: list[str], host: str, port: int, timeout_secs: int, kind: str = ""
 909) -> list[str]:
 910    cmd.extend(
 911        [
 912            "timeout",
 913            str(timeout_secs),
 914            "bash",
 915            "-c",
 916            f"until [ cat < /dev/null > /dev/tcp/{host}/{port} ] ; do sleep 0.1 ; done",
 917        ]
 918    )
 919    try:
 920        spawn.capture(cmd, stderr=subprocess.STDOUT)
 921    except subprocess.CalledProcessError as e:
 922        ui.log_in_automation(
 923            "wait-for-tcp ({}{}:{}): error running {}: {}, stdout:\n{}\nstderr:\n{}".format(
 924                kind, host, port, ui.shell_quote(cmd), e, e.stdout, e.stderr
 925            )
 926        )
 927        raise
 928    return cmd
 929
 930
 931# TODO(benesch): replace with Docker health checks.
 932def _wait_for_pg(
 933    timeout_secs: int,
 934    query: str,
 935    dbname: str,
 936    port: int,
 937    host: str,
 938    user: str,
 939    password: str | None,
 940    expected: Iterable[Any] | Literal["any"],
 941    print_result: bool = False,
 942    sslmode: str = "disable",
 943) -> None:
 944    """Wait for a pg-compatible database (includes materialized)"""
 945    obfuscated_password = password[0:1] if password is not None else ""
 946    args = f"dbname={dbname} host={host} port={port} user={user} password='{obfuscated_password}...'"
 947    ui.progress(f"waiting for {args} to handle {query!r}", "C")
 948    error = None
 949    for remaining in ui.timeout_loop(timeout_secs, tick=0.5):
 950        try:
 951            conn = psycopg.connect(
 952                dbname=dbname,
 953                host=host,
 954                port=port,
 955                user=user,
 956                password=password,
 957                connect_timeout=1,
 958                sslmode=sslmode,
 959            )
 960            # The default (autocommit = false) wraps everything in a transaction.
 961            conn.autocommit = True
 962            with conn.cursor() as cur:
 963                cur.execute(query.encode())
 964                if expected == "any" and cur.rowcount == -1:
 965                    ui.progress(" success!", finish=True)
 966                    return
 967                result = list(cur.fetchall())
 968                if expected == "any" or result == expected:
 969                    if print_result:
 970                        say(f"query result: {result}")
 971                    else:
 972                        ui.progress(" success!", finish=True)
 973                    return
 974                else:
 975                    say(
 976                        f"host={host} port={port} did not return rows matching {expected} got: {result}"
 977                    )
 978        except Exception as e:
 979            ui.progress(f"{e if print_result else ''} {int(remaining)}")
 980            error = e
 981    ui.progress(finish=True)
 982    raise UIError(f"never got correct result for {args}: {error}")
 983
 984
 985def bootstrap_cluster_replica_size() -> str:
 986    return "bootstrap"
 987
 988
 989def cluster_replica_size_map() -> dict[str, dict[str, Any]]:
 990    """scale=<n>,workers=<n>[,mem=<n>GiB][,legacy]"""
 991
 992    def replica_size(
 993        scale: int,
 994        workers: int,
 995        disabled: bool = False,
 996        is_cc: bool = True,
 997        memory_limit: str = "4 GiB",
 998    ) -> dict[str, Any]:
 999        return {
1000            "cpu_exclusive": False,
1001            "cpu_limit": None,
1002            "credits_per_hour": f"{workers * scale}",
1003            "disabled": disabled,
1004            "disk_limit": None,
1005            "is_cc": is_cc,
1006            "memory_limit": memory_limit,
1007            "scale": scale,
1008            "workers": workers,
1009            # "selectors": {},
1010        }
1011
1012    replica_sizes = {
1013        bootstrap_cluster_replica_size(): replica_size(1, 1),
1014        "scale=2,workers=4": replica_size(2, 4),
1015        "scale=1,workers=1,legacy": replica_size(1, 1, is_cc=False),
1016        "scale=1,workers=2,legacy": replica_size(1, 2, is_cc=False),
1017        # Intentionally not following the naming scheme
1018        "free": replica_size(1, 1, disabled=True),
1019    }
1020
1021    for i in range(0, 6):
1022        workers = 1 << i
1023        replica_sizes[f"scale=1,workers={workers}"] = replica_size(1, workers)
1024        for mem in [4, 8, 16, 32]:
1025            replica_sizes[f"scale=1,workers={workers},mem={mem}GiB"] = replica_size(
1026                1, workers, memory_limit=f"{mem} GiB"
1027            )
1028
1029        replica_sizes[f"scale={workers},workers=1"] = replica_size(workers, 1)
1030        replica_sizes[f"scale={workers},workers={workers}"] = replica_size(
1031            workers, workers
1032        )
1033        replica_sizes[f"scale=1,workers={workers},mem={workers}GiB"] = replica_size(
1034            1, workers, memory_limit=f"{workers} GiB"
1035        )
1036
1037    return replica_sizes