Implementation:MaterializeInc Materialize Check Initialize Manipulate Validate

Knowledge Sources	misc/python/materialize/checks/checks.py
Domains	Upgrade Testing, State-Transition Testing, Python API
Last Updated	2026-02-08

Overview

Concrete base class for upgrade checks provided by the Materialize platform-checks framework, implementing the initialize/manipulate/validate lifecycle through the Check class in materialize.checks.checks.

Description

The Check class is the abstract base that every individual upgrade check must subclass. It enforces the three-phase lifecycle contract:

• initialize() — returns a Testdrive or PyAction that sets up the check's precondition state. The default implementation is a no-op (Testdrive("$ nop")).
• manipulate() — returns a list of exactly two Testdrive or PyAction objects corresponding to manipulation phase 1 and phase 2. This method must be overridden.
• validate() — returns a Testdrive or PyAction that asserts post-upgrade correctness. This method must be overridden and may be invoked multiple times.

The class also provides convenience helpers (start_initialize, join_initialize, etc.) that wrap the lifecycle methods with guard logic: each phase is skipped if the check is disabled or if _can_run() returns False for the current executor.

Class-level flags control participation in specific scenarios:

• enabled — can be set to False via the @disabled decorator.
• externally_idempotent — indicates whether the check tolerates repeated external events.
• supports_forced_migrations — indicates whether the check works with forced catalog migrations.

Usage

Use the Check class when writing a new upgrade-safety test for a Materialize feature. Subclass it, override manipulate() and validate(), and optionally override initialize(). The framework will automatically include the check in all compatible scenarios.

Code Reference

Source Location

misc/python/materialize/checks/checks.py, lines 25–105.

Signature

class Check:
    enabled: bool = True
    externally_idempotent: bool = True
    supports_forced_migrations: bool = True

    def __init__(self, base_version: MzVersion, rng: Random | None) -> None: ...

    def _can_run(self, e: Executor) -> bool: ...

    def initialize(self) -> Testdrive | PyAction: ...

    def manipulate(self) -> list[Testdrive] | list[PyAction]: ...

    def validate(self) -> Testdrive | PyAction: ...

    def start_initialize(self, e: Executor, a: "Action") -> None: ...
    def join_initialize(self, e: Executor) -> None: ...

    def start_manipulate(self, e: Executor, a: "Action") -> None: ...
    def join_manipulate(self, e: Executor, a: "Action") -> None: ...

    def start_validate(self, e: Executor, a: "Action") -> None: ...
    def join_validate(self, e: Executor) -> None: ...

Import

from materialize.checks.checks import Check

I/O Contract

Inputs

Parameter	Type	Description
base_version	MzVersion	The Materialize version at which the check's state was first created. Used by checks that need to branch behavior based on version.
rng	None	Optional random number generator for checks that want deterministic randomization controlled by the scenario seed.

Outputs

Method	Return Type	Description
initialize()	PyAction	A single action that establishes precondition state.
manipulate()	list[PyAction]	Exactly two actions for phase 1 and phase 2 manipulation.
validate()	PyAction	A single action that asserts correctness; may be called multiple times.

Usage Examples

Minimal custom check:

from materialize.checks.checks import Check
from materialize.checks.actions import Testdrive


class CheckMyFeature(Check):
    def initialize(self) -> Testdrive:
        return Testdrive(
            """
            > CREATE TABLE my_table (a INT);
            > INSERT INTO my_table VALUES (1), (2), (3);
            """
        )

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive("$ nop"),
            Testdrive(
                """
                > INSERT INTO my_table VALUES (4);
                """
            ),
        ]

    def validate(self) -> Testdrive:
        return Testdrive(
            """
            > SELECT count(*) FROM my_table;
            4
            """
        )

Disabling a check:

from materialize.checks.checks import Check, disabled


@disabled("Feature X is temporarily broken; see #12345")
class CheckBrokenFeature(Check):
    ...

Related Pages

Implements Principle

• Principle:MaterializeInc_Materialize_Check_Lifecycle