diff --git a/misc/python/materialize/zippy/backup_and_restore_actions.py b/misc/python/materialize/zippy/backup_and_restore_actions.py index 130aac4576952..3762ac3813edc 100644 --- a/misc/python/materialize/zippy/backup_and_restore_actions.py +++ b/misc/python/materialize/zippy/backup_and_restore_actions.py @@ -39,6 +39,7 @@ def run(self, c: Composition, state: State) -> None: system_parameter_defaults=state.system_parameter_defaults, sanity_restart=False, restart="on-failure", + additional_system_parameter_defaults=state.additional_system_parameter_defaults, metadata_store="cockroach", default_replication_factor=2, ) diff --git a/misc/python/materialize/zippy/balancerd_actions.py b/misc/python/materialize/zippy/balancerd_actions.py index 9f931bf7d7d55..e74b55aec8761 100644 --- a/misc/python/materialize/zippy/balancerd_actions.py +++ b/misc/python/materialize/zippy/balancerd_actions.py @@ -15,6 +15,20 @@ from materialize.zippy.mz_capabilities import MzIsRunning +def balancerd_service(state: State) -> Balancerd: + """Balancerd configured to route to the current Mz service.""" + return Balancerd( + https_resolver_template=f"{state.mz_service}:6876", + static_resolver_addr=f"{state.mz_service}:6875", + ) + + +def restart_balancerd(c: Composition, state: State) -> None: + with c.override(balancerd_service(state)): + c.kill("balancerd") + c.up("balancerd") + + class BalancerdStart(Action): """Starts balancerd""" @@ -27,12 +41,7 @@ def incompatible_with(cls) -> set[type[Capability]]: return {BalancerdIsRunning} def run(self, c: Composition, state: State) -> None: - with c.override( - Balancerd( - https_resolver_template=f"{state.mz_service}:6876", - static_resolver_addr=f"{state.mz_service}:6875", - ) - ): + with c.override(balancerd_service(state)): c.up("balancerd") def provides(self) -> list[Capability]: @@ -65,11 +74,4 @@ def requires(cls) -> set[type[Capability]]: return {BalancerdIsRunning, MzIsRunning} def run(self, c: Composition, state: State) -> None: - with c.override( - Balancerd( - https_resolver_template=f"{state.mz_service}:6876", - static_resolver_addr=f"{state.mz_service}:6875", - ) - ): - c.kill("balancerd") - c.up("balancerd") + restart_balancerd(c, state) diff --git a/misc/python/materialize/zippy/debezium_actions.py b/misc/python/materialize/zippy/debezium_actions.py index 6711744bfcc07..785ae32675a9e 100644 --- a/misc/python/materialize/zippy/debezium_actions.py +++ b/misc/python/materialize/zippy/debezium_actions.py @@ -35,20 +35,6 @@ def run(self, c: Composition, state: State) -> None: c.up("debezium") -class DebeziumStop(Action): - """Stop the Debezium instance.""" - - @classmethod - def requires(cls) -> set[type[Capability]]: - return {DebeziumRunning} - - def withholds(self) -> set[type[Capability]]: - return {DebeziumRunning} - - def run(self, c: Composition, state: State) -> None: - c.kill("debezium") - - class CreateDebeziumSource(Action): """Creates a Debezium source in Materialized.""" @@ -63,39 +49,47 @@ def requires(cls) -> set[type[Capability]]: } def __init__(self, capabilities: Capabilities) -> None: - # To avoid conflicts, we make sure the postgres table and the debezium source have matching names - postgres_table = random.choice(capabilities.get(PostgresTableExists)) - cluster_name = random.choice(source_capable_clusters(capabilities)) - debezium_source_name = f"debezium_source_{postgres_table.name}" - this_debezium_source = DebeziumSourceExists(name=debezium_source_name) - - existing_debezium_sources = [ - s - for s in capabilities.get(DebeziumSourceExists) - if s.name == this_debezium_source.name - ] - - if len(existing_debezium_sources) == 0: - self.new_debezium_source = True - - self.debezium_source = this_debezium_source - self.postgres_table = postgres_table - self.debezium_source.postgres_table = self.postgres_table - self.cluster_name = cluster_name - elif len(existing_debezium_sources) == 1: - self.new_debezium_source = False - - self.debezium_source = existing_debezium_sources[0] - assert self.debezium_source.postgres_table is not None - self.postgres_table = self.debezium_source.postgres_table - else: - raise RuntimeError("More than one Debezium source exists") + # ENVELOPE DEBEZIUM requires a key, so only Postgres tables with a + # primary key can back a Debezium source. + pk_tables = [t for t in capabilities.get(PostgresTableExists) if t.has_pk] + + self.new_debezium_source = False + self.debezium_source: DebeziumSourceExists | None = None + + if pk_tables: + # To avoid conflicts, we make sure the postgres table and the debezium source have matching names + postgres_table = random.choice(pk_tables) + cluster_name = random.choice(source_capable_clusters(capabilities)) + debezium_source_name = f"debezium_source_{postgres_table.name}" + this_debezium_source = DebeziumSourceExists(name=debezium_source_name) + + existing_debezium_sources = [ + s + for s in capabilities.get(DebeziumSourceExists) + if s.name == this_debezium_source.name + ] + + if len(existing_debezium_sources) == 0: + self.new_debezium_source = True + + self.debezium_source = this_debezium_source + self.postgres_table = postgres_table + self.debezium_source.postgres_table = self.postgres_table + self.cluster_name = cluster_name + elif len(existing_debezium_sources) == 1: + self.debezium_source = existing_debezium_sources[0] + assert self.debezium_source.postgres_table is not None + self.postgres_table = self.debezium_source.postgres_table + else: + raise RuntimeError("More than one Debezium source exists") super().__init__(capabilities) def run(self, c: Composition, state: State) -> None: if self.new_debezium_source: - c.testdrive(dedent(f""" + assert self.debezium_source is not None + c.testdrive( + dedent(f""" $ http-request method=POST url=http://debezium:8083/connectors content-type=application/json {{ "name": "{self.debezium_source.name}", @@ -113,7 +107,7 @@ def run(self, c: Composition, state: State) -> None: "publication.name": "dbz_publication_{self.debezium_source.name}", "publication.autocreate.mode": "filtered", "slot.name" : "slot_{self.postgres_table.name}", - "database.history.kafka.bootstrap.servers": "kafka:9092", + "database.history.kafka.bootstrap.servers": "redpanda:9092", "database.history.kafka.topic": "schema-changes.history", "truncate.handling.mode": "include", "decimal.handling.mode": "precise", @@ -134,7 +128,12 @@ def run(self, c: Composition, state: State) -> None: > CREATE TABLE {self.debezium_source.get_name_for_query()} FROM SOURCE {self.debezium_source.name} (REFERENCE "postgres.public.{self.postgres_table.name}") FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn ENVELOPE DEBEZIUM - """)) + """), + mz_service=state.mz_service, + ) def provides(self) -> list[Capability]: - return [self.debezium_source] if self.new_debezium_source else [] + if self.new_debezium_source: + assert self.debezium_source is not None + return [self.debezium_source] + return [] diff --git a/misc/python/materialize/zippy/framework.py b/misc/python/materialize/zippy/framework.py index 8ab253b4f5a9e..14344a837b115 100644 --- a/misc/python/materialize/zippy/framework.py +++ b/misc/python/materialize/zippy/framework.py @@ -7,7 +7,9 @@ # the Business Source License, use of this software will be governed # by the Apache License, Version 2.0. +import os import random +import threading from collections.abc import Sequence from datetime import datetime, timedelta from typing import TYPE_CHECKING, TypeVar @@ -19,6 +21,20 @@ from materialize.zippy.scenarios import Scenario +def ci_additional_system_parameter_defaults() -> dict[str, str]: + """Parse the CI_MZ_SYSTEM_PARAMETER_DEFAULT environment variable.""" + defaults = {} + system_parameter_default = os.getenv("CI_MZ_SYSTEM_PARAMETER_DEFAULT", "") + if system_parameter_default: + for val in system_parameter_default.split(";"): + x = val.split("=", maxsplit=1) + assert ( + len(x) == 2 + ), f"CI_MZ_SYSTEM_PARAMETER_DEFAULT '{val}' should be the format =" + defaults[x[0]] = x[1] + return defaults + + class State: mz_service: str deploy_generation: int @@ -28,8 +44,15 @@ def __init__(self): self.mz_service = "materialized" self.deploy_generation = 0 self.system_parameter_defaults = get_default_system_parameters() - self.iceberg_username: str | None = None - self.iceberg_key: str | None = None + # Every action that (re)starts Mz must pass these along, otherwise a + # feature flag under test silently reverts to its builtin default. + self.additional_system_parameter_defaults = ( + ci_additional_system_parameter_defaults() + ) + # Threads spawned by actions (e.g. parallel Kafka ingestion). They are + # joined at the end of the test and any errors they collected fail it. + self.background_threads: list[threading.Thread] = [] + self.background_errors: list[Exception] = [] class Capability: @@ -101,8 +124,10 @@ def get_free_capability_name( ] existing_object_names = self.get_capability_names(capability) remaining_object_names = set(all_object_names) - set(existing_object_names) + # sorted() so that the same --seed picks the same names regardless of + # set iteration order (which depends on PYTHONHASHSEED). return ( - random.choice(list(remaining_object_names)) + random.choice(sorted(remaining_object_names)) if len(remaining_object_names) > 0 else None ) @@ -191,6 +216,10 @@ def __init__( self._actions: list[Action] = [] self._final_actions: list[Action] = [] self._capabilities = Capabilities() + # The exact Capability instances each action added at generation time. + # Trimming in run() removes by identity, and provides() may build new + # instances on each call, so the originals must be remembered here. + self._provided_capabilities: dict[Action, list[Capability]] = {} self._actions_with_weight: dict[ActionOrFactory, float] = ( self._scenario.actions_with_weight() ) @@ -215,11 +244,10 @@ def generate_actions(self, action_def: ActionOrFactory) -> list[Action]: ) for action in actions: - print("test:", action) - self._capabilities._extend(action.provides()) - print(" - ", self._capabilities, action.provides()) + provided = action.provides() + self._provided_capabilities[action] = provided + self._capabilities._extend(provided) self._capabilities._remove(action.withholds()) - print(" - ", self._capabilities, action.withholds()) return actions @@ -237,10 +265,19 @@ def run(self, c: Composition) -> None: executed_count = i + 1 break + # Surface failures from background threads instead of dropping them. + for thread in self._state.background_threads: + thread.join() + if self._state.background_errors: + raise self._state.background_errors[0] + # Remove capability effects of actions that were never executed, # so that finalization only validates objects that actually exist. + # Capabilities withheld by unexecuted actions are not restored. That + # is fine because only the *IsRunning capabilities are ever withheld + # and finalization unconditionally starts those services again. for action in self._actions[executed_count:]: - for cap in action.provides(): + for cap in self._provided_capabilities[action]: self._capabilities.remove_capability_instance(cap) # Generate finalization actions now, after the main loop, so that diff --git a/misc/python/materialize/zippy/iceberg_actions.py b/misc/python/materialize/zippy/iceberg_actions.py index 5fd70f96f780d..e8f89aeed34a4 100644 --- a/misc/python/materialize/zippy/iceberg_actions.py +++ b/misc/python/materialize/zippy/iceberg_actions.py @@ -45,8 +45,6 @@ def run(self, c: Composition, state: State) -> None: Service("polaris", idle=True), ) username, key = setup_polaris_for_iceberg(c) - state.iceberg_username = username - state.iceberg_key = key c.testdrive( dedent(f""" diff --git a/misc/python/materialize/zippy/kafka_actions.py b/misc/python/materialize/zippy/kafka_actions.py index e9cafd4bba471..743c22af6a99c 100644 --- a/misc/python/materialize/zippy/kafka_actions.py +++ b/misc/python/materialize/zippy/kafka_actions.py @@ -12,8 +12,6 @@ import threading from textwrap import dedent -import numpy as np - from materialize.mzcompose.composition import Composition from materialize.zippy.framework import ( Action, @@ -55,20 +53,6 @@ def run(self, c: Composition, state: State) -> None: c.up("redpanda") -class KafkaStop(Action): - """Stop the Kafka instance.""" - - @classmethod - def requires(cls) -> set[type[Capability]]: - return {KafkaRunning} - - def withholds(self) -> set[type[Capability]]: - return {KafkaRunning} - - def run(self, c: Composition, state: State) -> None: - c.kill("redpanda") - - class CreateTopicParameterized(ActionFactory): """Creates a Kafka topic and decides on the envelope that will be used.""" @@ -139,8 +123,10 @@ def requires(cls) -> set[type[Capability]]: def __init__(self, capabilities: Capabilities) -> None: self.topic = random.choice(capabilities.get(TopicExists)) self.delta = random.randint(1, 10000) - # This gives 67% pads of up to 10 bytes, 25% of up to 100 bytes and outliers up to 256 bytes - self.pad = min(np.random.zipf(1.6, 1)[0], 256) * random.choice( + # Heavy-tailed pad sizes: mostly up to 10 bytes, some up to 100 bytes + # and outliers up to 256 bytes. int(paretovariate(0.6)) approximates a + # zipf(1.6) sample while staying reproducible under random.seed(). + self.pad = min(int(random.paretovariate(0.6)), 256) * random.choice( string.ascii_letters ) super().__init__(capabilities) @@ -167,7 +153,17 @@ def run(self, c: Composition, state: State) -> None: """) if self.parallel(): - threading.Thread(target=c.testdrive, args=[testdrive_str]).start() + mz_service = state.mz_service + + def ingest() -> None: + try: + c.testdrive(testdrive_str, mz_service=mz_service) + except Exception as e: + state.background_errors.append(e) + + thread = threading.Thread(target=ingest) + thread.start() + state.background_threads.append(thread) else: c.testdrive(testdrive_str, mz_service=state.mz_service) diff --git a/misc/python/materialize/zippy/mysql_actions.py b/misc/python/materialize/zippy/mysql_actions.py index 9d343b4c776a6..706618c78d590 100644 --- a/misc/python/materialize/zippy/mysql_actions.py +++ b/misc/python/materialize/zippy/mysql_actions.py @@ -40,20 +40,6 @@ def run(self, c: Composition, state: State) -> None: ) -class MySqlStop(Action): - """Stop the MySQL instance.""" - - @classmethod - def requires(cls) -> set[type[Capability]]: - return {MySqlRunning} - - def withholds(self) -> set[type[Capability]]: - return {MySqlRunning} - - def run(self, c: Composition, state: State) -> None: - c.kill("mysql") - - class MySqlRestart(Action): """Restart the MySql instance.""" @@ -80,8 +66,7 @@ def __init__(self, capabilities: Capabilities) -> None: if len(existing_mysql_tables) == 0: self.new_mysql_table = True - # A PK is now required for Debezium - this_mysql_table.has_pk = True + this_mysql_table.has_pk = random.choice([True, False]) self.mysql_table = this_mysql_table elif len(existing_mysql_tables) == 1: diff --git a/misc/python/materialize/zippy/mz_actions.py b/misc/python/materialize/zippy/mz_actions.py index 03389996dab7d..4a76858f1a830 100644 --- a/misc/python/materialize/zippy/mz_actions.py +++ b/misc/python/materialize/zippy/mz_actions.py @@ -8,54 +8,27 @@ # by the Apache License, Version 2.0. -import os - from materialize.mzcompose.composition import Composition from materialize.mzcompose.services.materialized import ( LEADER_STATUS_HEALTHCHECK, DeploymentStatus, Materialized, ) +from materialize.zippy.balancerd_actions import restart_balancerd from materialize.zippy.balancerd_capabilities import BalancerdIsRunning from materialize.zippy.blob_store_capabilities import BlobStoreIsRunning from materialize.zippy.crdb_capabilities import CockroachIsRunning from materialize.zippy.framework import ( Action, - ActionFactory, - Capabilities, Capability, Mz0dtDeployBaseAction, State, ) from materialize.zippy.mz_capabilities import MzIsRunning +from materialize.zippy.table_capabilities import TableExists from materialize.zippy.view_capabilities import ViewExists -class MzStartParameterized(ActionFactory): - """Starts a Mz instance with custom paramters.""" - - @classmethod - def requires(cls) -> set[type[Capability]]: - return {CockroachIsRunning, BlobStoreIsRunning} - - @classmethod - def incompatible_with(cls) -> set[type[Capability]]: - return {MzIsRunning} - - def __init__( - self, additional_system_parameter_defaults: dict[str, str] = {} - ) -> None: - self.additional_system_parameter_defaults = additional_system_parameter_defaults - - def new(self, capabilities: Capabilities) -> list[Action]: - return [ - MzStart( - capabilities=capabilities, - additional_system_parameter_defaults=self.additional_system_parameter_defaults, - ) - ] - - class MzStart(Action): """Starts a Mz instance (all components are running in the same container).""" @@ -67,31 +40,9 @@ def requires(cls) -> set[type[Capability]]: def incompatible_with(cls) -> set[type[Capability]]: return {MzIsRunning} - def __init__( - self, - capabilities: Capabilities, - additional_system_parameter_defaults: dict[str, str] = {}, - ) -> None: - if additional_system_parameter_defaults: - self.additional_system_parameter_defaults = ( - additional_system_parameter_defaults - ) - else: - self.additional_system_parameter_defaults = {} - system_parameter_default = os.getenv("CI_MZ_SYSTEM_PARAMETER_DEFAULT", "") - if system_parameter_default: - for val in system_parameter_default.split(";"): - x = val.split("=", maxsplit=1) - assert ( - len(x) == 2 - ), f"CI_MZ_SYSTEM_PARAMETER_DEFAULT '{val}' should be the format =" - self.additional_system_parameter_defaults[x[0]] = x[1] - - super().__init__(capabilities) - def run(self, c: Composition, state: State) -> None: print( - f"Starting Mz with additional_system_parameter_defaults = {self.additional_system_parameter_defaults}" + f"Starting Mz with additional_system_parameter_defaults = {state.additional_system_parameter_defaults}" ) with c.override( @@ -104,7 +55,7 @@ def run(self, c: Composition, state: State) -> None: system_parameter_defaults=state.system_parameter_defaults, sanity_restart=False, restart="on-failure", - additional_system_parameter_defaults=self.additional_system_parameter_defaults, + additional_system_parameter_defaults=state.additional_system_parameter_defaults, metadata_store="cockroach", default_replication_factor=2, ) @@ -183,6 +134,7 @@ def run(self, c: Composition, state: State) -> None: system_parameter_defaults=state.system_parameter_defaults, sanity_restart=False, restart="on-failure", + additional_system_parameter_defaults=state.additional_system_parameter_defaults, metadata_store="cockroach", default_replication_factor=2, ) @@ -218,6 +170,7 @@ def run(self, c: Composition, state: State) -> None: sanity_restart=False, restart="on-failure", healthcheck=LEADER_STATUS_HEALTHCHECK, + additional_system_parameter_defaults=state.additional_system_parameter_defaults, metadata_store="cockroach", default_replication_factor=2, ), @@ -237,13 +190,20 @@ def run(self, c: Composition, state: State) -> None: wait=True, ) + # Balancerd's resolver is fixed at startup and still points at the + # previous generation. Re-point it at the new leader. + if c.is_running("balancerd"): + restart_balancerd(c, state) + class KillClusterd(Action): """Kills the clusterd processes in the environmentd container. The process orchestrator will restart them.""" @classmethod - def requires(cls) -> set[type[Capability]]: - return {MzIsRunning, ViewExists} + def requires(cls) -> list[set[type[Capability]]]: + # Only kill once a dataflow-bearing object exists, so the kill has + # something to disrupt. + return [{MzIsRunning, ViewExists}, {MzIsRunning, TableExists}] def run(self, c: Composition, state: State) -> None: # Depending on the workload, clusterd may not be running, hence the || true diff --git a/misc/python/materialize/zippy/postgres_actions.py b/misc/python/materialize/zippy/postgres_actions.py index 512335cb5969c..c30454cee6fc2 100644 --- a/misc/python/materialize/zippy/postgres_actions.py +++ b/misc/python/materialize/zippy/postgres_actions.py @@ -27,20 +27,6 @@ def run(self, c: Composition, state: State) -> None: c.up("postgres") -class PostgresStop(Action): - """Stop the Postgres instance.""" - - @classmethod - def requires(cls) -> set[type[Capability]]: - return {PostgresRunning} - - def withholds(self) -> set[type[Capability]]: - return {PostgresRunning} - - def run(self, c: Composition, state: State) -> None: - c.kill("postgres") - - class PostgresRestart(Action): """Restart the Postgres instance.""" @@ -69,8 +55,7 @@ def __init__(self, capabilities: Capabilities) -> None: if len(existing_postgres_tables) == 0: self.new_postgres_table = True - # A PK is now required for Debezium - this_postgres_table.has_pk = True + this_postgres_table.has_pk = random.choice([True, False]) self.postgres_table = this_postgres_table elif len(existing_postgres_tables) == 1: diff --git a/misc/python/materialize/zippy/scenarios.py b/misc/python/materialize/zippy/scenarios.py index 2169b06d0d82e..2351b43120035 100644 --- a/misc/python/materialize/zippy/scenarios.py +++ b/misc/python/materialize/zippy/scenarios.py @@ -75,21 +75,36 @@ StoragedStart, ) from materialize.zippy.copy_s3_actions import CopyFromS3, CopyToS3 -from materialize.zippy.table_actions import DML, CreateTableParameterized, ValidateTable -from materialize.zippy.view_actions import CreateViewParameterized, ValidateView +from materialize.zippy.table_actions import ( + DML, + CreateTableParameterized, + DropTable, + ValidateTable, +) +from materialize.zippy.view_actions import ( + CreateViewParameterized, + DropView, + ValidateView, +) class Scenario: def bootstrap(self) -> list[ActionOrFactory]: - return [ + bootstrap: list[ActionOrFactory] = [ KafkaStart, CockroachStart, BlobStoreStart, MzStart, StoragedStart, BalancerdStart, - IcebergStart, ] + # Polaris is only needed if the scenario can create Iceberg sinks. + if any( + isinstance(action, CreateIcebergSinkParameterized) + for action in self.actions_with_weight() + ): + bootstrap.append(IcebergStart) + return bootstrap def actions_with_weight(self) -> dict[ActionOrFactory, float]: raise RuntimeError @@ -120,6 +135,7 @@ def actions_with_weight(self) -> dict[ActionOrFactory, float]: CreateTopicParameterized(): 5, CreateSourceParameterized(): 5, CreateViewParameterized(max_inputs=2): 5, + DropView: 5, CreateSinkParameterized(): 5, CreateIcebergSinkParameterized(): 5, ValidateView: 10, @@ -163,12 +179,16 @@ def actions_with_weight(self) -> dict[ActionOrFactory, float]: Mz0dtDeploy: 10, BalancerdStop: 1, BalancerdRestart: 1, - BackupAndRestore: 1, + # No mid-test BackupAndRestore since the state of the sinks is not + # recorded across a backup&restore cycle, see + # https://github.com/MaterializeInc/database-issues/issues/9589 KillClusterd: 10, # Disabled because a separate clusterd is not supported by Mz0dtDeploy yet # StoragedRestart: 5, CreateTableParameterized(): 10, CreateViewParameterized(): 10, + DropTable: 5, + DropView: 5, CreateSinkParameterized(): 10, CreateIcebergSinkParameterized(): 5, ValidateTable: 20, diff --git a/misc/python/materialize/zippy/sql_server_actions.py b/misc/python/materialize/zippy/sql_server_actions.py index 046a607e38b61..1b0952c854780 100644 --- a/misc/python/materialize/zippy/sql_server_actions.py +++ b/misc/python/materialize/zippy/sql_server_actions.py @@ -45,20 +45,6 @@ def run(self, c: Composition, state: State) -> None: ) -class SqlServerStop(Action): - """Stop the SQL Server instance.""" - - @classmethod - def requires(cls) -> set[type[Capability]]: - return {SqlServerRunning} - - def withholds(self) -> set[type[Capability]]: - return {SqlServerRunning} - - def run(self, c: Composition, state: State) -> None: - c.kill("sql-server") - - class SqlServerRestart(Action): """Restart the SqlServer instance.""" @@ -87,8 +73,7 @@ def __init__(self, capabilities: Capabilities) -> None: if len(existing_sql_server_tables) == 0: self.new_sql_server_table = True - # A PK is now required for Debezium - this_sql_server_table.has_pk = True + this_sql_server_table.has_pk = random.choice([True, False]) self.sql_server_table = this_sql_server_table elif len(existing_sql_server_tables) == 1: diff --git a/misc/python/materialize/zippy/table_actions.py b/misc/python/materialize/zippy/table_actions.py index 5316a3a1901e5..e88efe9d95f28 100644 --- a/misc/python/materialize/zippy/table_actions.py +++ b/misc/python/materialize/zippy/table_actions.py @@ -21,6 +21,7 @@ ) from materialize.zippy.mz_capabilities import MzIsRunning from materialize.zippy.table_capabilities import TableExists +from materialize.zippy.view_capabilities import ViewExists MAX_ROWS_PER_ACTION = 10000 @@ -131,6 +132,39 @@ def run(self, c: Composition, state: State) -> None: ) +class DropTable(Action): + """Drops a table that no view reads from.""" + + @classmethod + def requires(cls) -> set[type[Capability]]: + return {BalancerdIsRunning, MzIsRunning, TableExists} + + def __init__(self, capabilities: Capabilities) -> None: + referenced: set[str] = set() + for view in capabilities.get(ViewExists): + referenced.update(input.name for input in view.inputs) + + candidates = [ + t for t in capabilities.get(TableExists) if t.name not in referenced + ] + self.table: TableExists | None = ( + random.choice(candidates) if candidates else None + ) + if self.table is not None: + capabilities.remove_capability_instance(self.table) + super().__init__(capabilities) + + def __str__(self) -> str: + return f"{Action.__str__(self)} {self.table.name if self.table else ''}" + + def run(self, c: Composition, state: State) -> None: + if self.table is not None: + c.testdrive( + f"> DROP TABLE {self.table.name};", + mz_service=state.mz_service, + ) + + class DML(Action): """Performs an INSERT, DELETE or UPDATE against a table.""" diff --git a/misc/python/materialize/zippy/view_actions.py b/misc/python/materialize/zippy/view_actions.py index 475546c482457..71608044d50b2 100644 --- a/misc/python/materialize/zippy/view_actions.py +++ b/misc/python/materialize/zippy/view_actions.py @@ -20,9 +20,11 @@ Capability, State, ) +from materialize.zippy.iceberg_capabilities import IcebergSinkExists from materialize.zippy.mysql_cdc_capabilities import MySqlCdcTableExists from materialize.zippy.mz_capabilities import MzIsRunning from materialize.zippy.pg_cdc_capabilities import PostgresCdcTableExists +from materialize.zippy.sink_capabilities import SinkExists from materialize.zippy.source_capabilities import SourceExists from materialize.zippy.storaged_capabilities import StoragedRunning from materialize.zippy.table_capabilities import TableExists @@ -137,6 +139,44 @@ def provides(self) -> list[Capability]: return [self.view] +class DropView(Action): + """Drops a view that no sink and no other view depends on.""" + + @classmethod + def requires(cls) -> set[type[Capability]]: + return {BalancerdIsRunning, MzIsRunning, ViewExists} + + def __init__(self, capabilities: Capabilities) -> None: + referenced: set[str] = set() + for sink in capabilities.get(SinkExists): + referenced.add(sink.source_view.name) + referenced.add(sink.dest_view.name) + for iceberg_sink in capabilities.get(IcebergSinkExists): + referenced.add(iceberg_sink.source_view.name) + for view in capabilities.get(ViewExists): + referenced.update( + input.name for input in view.inputs if isinstance(input, ViewExists) + ) + + candidates = [ + v for v in capabilities.get(ViewExists) if v.name not in referenced + ] + self.view: ViewExists | None = random.choice(candidates) if candidates else None + if self.view is not None: + capabilities.remove_capability_instance(self.view) + super().__init__(capabilities) + + def __str__(self) -> str: + return f"{Action.__str__(self)} {self.view.name if self.view else ''}" + + def run(self, c: Composition, state: State) -> None: + if self.view is not None: + c.testdrive( + f"> DROP MATERIALIZED VIEW {self.view.name};", + mz_service=state.mz_service, + ) + + class ValidateView(Action): """Validates a view.""" diff --git a/test/zippy/mzcompose.py b/test/zippy/mzcompose.py index 5dd73d0e7889a..7d07106d10350 100644 --- a/test/zippy/mzcompose.py +++ b/test/zippy/mzcompose.py @@ -13,7 +13,6 @@ expected state it can verify results for correctness. """ -import os import random import re import time @@ -47,7 +46,7 @@ setup_default_ssh_test_connection, ) from materialize.mzcompose.services.testdrive import Testdrive -from materialize.zippy.framework import Test +from materialize.zippy.framework import Test, ci_additional_system_parameter_defaults from materialize.zippy.mz_actions import Mz0dtDeploy from materialize.zippy.scenarios import * # noqa: F401 F403 @@ -120,12 +119,19 @@ def __str__(self) -> str: def parse_timedelta(arg: str) -> timedelta: p = re.compile( - (r"((?P-?\d+)d)?" r"((?P-?\d+)h)?" r"((?P-?\d+)m)?"), + ( + r"((?P-?\d+)d)?" + r"((?P-?\d+)h)?" + r"((?P-?\d+)m)?" + r"((?P-?\d+)s)?" + ), re.IGNORECASE, ) - m = p.match(arg) - assert m is not None + m = p.fullmatch(arg) + assert ( + m is not None + ), f"invalid timedelta '{arg}', expected a format like '1d2h3m4s'" parts = {k: int(v) for k, v in m.groupdict().items() if v} td = timedelta(**parts) @@ -212,15 +218,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: print(f"Using seed {args.seed}") random.seed(args.seed) - additional_system_parameter_defaults = {} - system_parameter_default = os.getenv("CI_MZ_SYSTEM_PARAMETER_DEFAULT", "") - if system_parameter_default: - for val in system_parameter_default.split(";"): - x = val.split("=", maxsplit=1) - assert ( - len(x) == 2 - ), f"CI_MZ_SYSTEM_PARAMETER_DEFAULT '{val}' should be the format =" - additional_system_parameter_defaults[x[0]] = x[1] + additional_system_parameter_defaults = ci_additional_system_parameter_defaults() with c.override( Cockroach(