more checkers

Author: def-

misc/python/materialize/invariants/checkers.py

@@ -12,6 +12,7 @@
 import random
 from abc import abstractmethod
 from collections import Counter
+from collections.abc import Callable
 
 from materialize.invariants.framework import (
     Checker,
@@ -404,3 +405,80 @@ def check_once(self) -> None:
                 f" {self.last_cluster}: {rows[:5]}"
             )
         self.validations += 1
+
+
+class TernaryPartitionPeek(PeekChecker):
+    """Splitting a relation by a predicate must not create or lose rows.
+
+    For any total predicate p, the rows where p holds, where it does not, and
+    where it is NULL partition the relation exactly, at every timestamp and
+    whatever the data is. So this holds without knowing anything about the
+    workload or which of its operations committed, which is what lets it run
+    during a disruption, and it is a different question from the conserved
+    totals the scenarios otherwise assert: it is about whether predicates are
+    evaluated correctly, including when persist skips parts by their
+    statistics rather than reading them.
+
+    Both the row count and a summed value are partitioned, because a filter
+    that drops one row and duplicates another keeps the count.
+
+    `predicate` must produce **total** expressions. Anything that can raise,
+    a division or a fallible cast, breaks the comparison: the three branches
+    would fail differently rather than disagree.
+    """
+
+    pause = (0.5, 2.0)
+
+    def __init__(
+        self,
+        rng,
+        ctx,
+        name: str,
+        relation: str,
+        predicate: Callable[[random.Random], str],
+        value: str,
+        history: Watermark | None = None,
+    ) -> None:
+        super().__init__(rng, ctx, name, ["quickstart", "compute"])
+        self.relation = relation
+        self.predicate = predicate
+        self.value = value
+        self.history = history
+
+    def check_once(self) -> None:
+        p = self.predicate(self.rng)
+        # One statement, so all four reads share a timestamp. Written as
+        # differences so the check is "both zero" rather than a comparison of
+        # four numbers that a disruption could interleave.
+        query = (
+            f"SELECT (SELECT count(*) FROM {self.relation})"
+            f" - (SELECT count(*) FROM {self.relation} WHERE {p})"
+            f" - (SELECT count(*) FROM {self.relation} WHERE NOT ({p}))"
+            f" - (SELECT count(*) FROM {self.relation} WHERE ({p}) IS NULL),"
+            f" (SELECT coalesce(sum({self.value}), 0) FROM {self.relation})"
+            f" - (SELECT coalesce(sum({self.value}), 0) FROM {self.relation}"
+            f" WHERE {p})"
+            f" - (SELECT coalesce(sum({self.value}), 0) FROM {self.relation}"
+            f" WHERE NOT ({p}))"
+            f" - (SELECT coalesce(sum({self.value}), 0) FROM {self.relation}"
+            f" WHERE ({p}) IS NULL)"
+        )
+        at = "now"
+        recent = self.history.get() if self.history is not None else 0
+        if recent > 0 and self.rng.random() < 0.25:
+            as_of = self.rng.randint(max(recent - 120_000, 1), recent)
+            query, at = f"{query} AS OF {as_of}", str(as_of)
+        try:
+            rows = self.peek(query)
+        except UnexpectedQueryError as e:
+            if "could not find a valid timestamp" in str(e):
+                raise TransientError(f"{at} no longer readable") from None
+            raise
+        count_diff, value_diff = int(rows[0][0]), int(rows[0][1])
+        if count_diff or value_diff:
+            raise InvariantViolation(
+                f"{self.name}: `{p}` does not partition {self.relation} at {at}"
+                f" on {self.last_cluster}: {count_diff} rows and {value_diff}"
+                f" of {self.value} unaccounted for"
+            )
+        self.validations += 1

misc/python/materialize/invariants/scenarios/table_bank.py

@@ -36,6 +36,7 @@
     GroupCompletenessPeek,
     PeekChecker,
     SubscribeChecker,
+    TernaryPartitionPeek,
 )
 from materialize.invariants.framework import (
     CONVERGE_TIMEOUT,
@@ -639,6 +640,131 @@ def check_once(self) -> None:
         self.validations += 1
 
 
+def ledger_predicate(rng: random.Random) -> str:
+    """A random total predicate over the ledger, for the partition oracle.
+
+    Total is the requirement: no division, no fallible cast, nothing that can
+    raise, or the three branches of the partition fail differently instead of
+    disagreeing. Everything here is a comparison, a NULL test, or a boolean
+    combination of them.
+
+    The columns are chosen for the values that make filters and statistics
+    pushdown awkward. `flt` carries NaN, both infinities, negative zero and a
+    denormal, `day` is nullable, so predicates over it are the ones that put
+    rows in the IS NULL branch at all, and `amount` is signed.
+    """
+    atoms = [
+        f"amount > {rng.randint(-100, 100)}",
+        f"amount < {rng.randint(-100, 100)}",
+        f"amount = {rng.randint(-100, 100)}",
+        f"account >= {rng.randint(0, 32)}",
+        f"seq % {rng.randint(2, 9)} = 0",
+        "flt > 0",
+        "flt < 0",
+        "flt = 'NaN'::double precision",
+        "flt = 'Infinity'::double precision",
+        f"amount_dec > {rng.randint(-100, 100)}",
+        "tag LIKE 'w%'",
+        f"tag LIKE '%s{rng.randint(0, 9)}'",
+        "day IS NULL",
+        "day > DATE '2024-01-01'",
+        f"day < DATE '20{rng.randint(20, 30)}-06-01'",
+    ]
+    p = rng.choice(atoms)
+    # Compound predicates, since a filter can be right on each half and wrong
+    # on the combination, and NULL propagation through AND/OR is its own trap.
+    if rng.random() < 0.4:
+        p = f"({p}) {rng.choice(['AND', 'OR'])} ({rng.choice(atoms)})"
+    if rng.random() < 0.2:
+        p = f"NOT ({p})"
+    return p
+
+
+# A DAG whose edges only ever point from a lower node to a higher one, so the
+# graph is acyclic no matter which inserts and deletes committed, and the
+# closure of whatever edges are currently visible is the thing to check. Small
+# enough that a dense closure stays bounded.
+GRAPH_NODES = 40
+
+# Reachability recomputed from scratch, for comparison against the maintained
+# view. UNION rather than UNION ALL so the iteration converges.
+GRAPH_REACH_CTE = (
+    "WITH MUTUALLY RECURSIVE reach (src int, dst int) AS ("
+    " SELECT src, dst FROM edges"
+    " UNION"
+    " SELECT r.src, e.dst FROM reach r JOIN edges e ON r.dst = e.src"
+    ")"
+)
+GRAPH_CLOSURE_SQL = f"{GRAPH_REACH_CTE} SELECT src, dst FROM reach"
+
+# Three properties of the closure of whatever edges are visible right now.
+# None of them depend on which operations succeeded, and each fails for a
+# different reason: a missing transitive pair means the iteration stopped
+# early, a missing edge means the base case was lost, and a self-loop means a
+# retraction left a cycle behind in a graph that cannot contain one.
+GRAPH_NOT_CLOSED_SQL = (
+    "SELECT a.src, a.dst, b.dst FROM closure a"
+    " JOIN closure b ON a.dst = b.src"
+    " LEFT JOIN closure c ON c.src = a.src AND c.dst = b.dst"
+    " WHERE c.src IS NULL LIMIT 5"
+)
+GRAPH_MISSING_EDGE_SQL = (
+    "SELECT e.src, e.dst FROM edges e"
+    " LEFT JOIN closure c ON c.src = e.src AND c.dst = e.dst"
+    " WHERE c.src IS NULL LIMIT 5"
+)
+GRAPH_SELF_LOOP_SQL = "SELECT src, dst FROM closure WHERE src = dst LIMIT 5"
+
+# The maintained view against the same question recomputed in one shot, at one
+# timestamp. This is the one that catches the incremental iteration diverging
+# from the batch answer, which is what retractions through a recursive
+# dataflow are most likely to get wrong.
+# The recursion is defined once and referenced from both sides: a WITH cannot
+# be an operand of a set operation, and this keeps it to one statement, so
+# both sides are answered at one timestamp.
+GRAPH_DIFFERENTIAL_SQL = (
+    f"{GRAPH_REACH_CTE}"
+    " (SELECT src, dst FROM closure EXCEPT ALL SELECT src, dst FROM reach)"
+    " UNION ALL"
+    " (SELECT src, dst FROM reach EXCEPT ALL SELECT src, dst FROM closure)"
+)
+
+
+class GraphChurn(Action):
+    """Insert and delete edges of an acyclic graph.
+
+    Deletions are the point. An insert only ever grows the closure, which an
+    iterative dataflow can get right by accumulating, while a delete has to
+    retract every path that went through the removed edge, including paths
+    discovered several iterations deep. Nothing else in this scenario asks the
+    engine to iterate at all.
+
+    Edges always point from a lower node to a higher one, so the graph stays
+    acyclic whatever subset of the operations applied, which is what lets the
+    self-loop check be an invariant rather than a race.
+    """
+
+    name = "graph-churn"
+
+    def __init__(self, rng: random.Random, client: MzClient) -> None:
+        super().__init__(rng)
+        self.client = client
+
+    def run(self) -> Outcome | None:
+        if self.rng.random() < 0.4:
+            # Delete by predicate rather than by a remembered edge: an
+            # UNKNOWN insert may or may not exist, and this is correct either
+            # way.
+            src = self.rng.randrange(GRAPH_NODES - 1)
+            return self.client.write(f"DELETE FROM edges WHERE src = {src}")
+        src = self.rng.randrange(GRAPH_NODES - 1)
+        dst = self.rng.randrange(src + 1, GRAPH_NODES)
+        return self.client.write(f"INSERT INTO edges VALUES ({src}, {dst})")
+
+    def close(self) -> None:
+        self.client.reset()
+
+
 class RollbackNoop(Action):
     """A transaction that never commits must leave nothing behind.
 
@@ -737,9 +863,22 @@ def __init__(self, rng: random.Random, client: MzClient, scenario) -> None:
 
     def run(self) -> Outcome | None:
         self.client.query(f"SET cluster = {self.rng.choice(['quickstart', 'compute'])}")
-        return self.client.write_txn(
-            [("INSERT INTO cross_probe SELECT total FROM total", None)]
-        )
+        # A read and a write in one transaction, which is only expressible
+        # this way: Materialize rejects both UPDATE and INSERT .. SELECT
+        # inside a transaction block, so the read has to be its own statement
+        # and the write has to carry the value the client saw.
+        try:
+            self.client.query("BEGIN")
+            rows = self.client.query("SELECT total FROM total")
+            self.client.query(f"INSERT INTO cross_probe VALUES ({int(rows[0][0])})")
+            self.client.query("COMMIT")
+        except TransientError:
+            self.client.reset()
+            raise
+        except Exception:
+            self.client.reset()
+            return Outcome.UNKNOWN
+        return Outcome.COMMITTED
 
     def close(self) -> None:
         self.client.reset()
@@ -1769,6 +1908,9 @@ def setup(self) -> None:
             "CREATE TABLE counters (id int, n bigint)",
             "CREATE TABLE rollback_probe (id bigint)",
             "CREATE TABLE cross_probe (total bigint)",
+            "CREATE TABLE edges (src int, dst int)",
+            "CREATE MATERIALIZED VIEW closure IN CLUSTER compute AS"
+            f" {GRAPH_CLOSURE_SQL}",
             f"INSERT INTO counters SELECT generate_series(0, {COUNTER_KEYS - 1}), 0",
             *(
                 f"CREATE TABLE {table} (worker int, seq bigint, idx int)"
@@ -1829,12 +1971,13 @@ def make_worker(self, index: int, rng: random.Random) -> WorkerBundle:
             ReadThenWrite(rng, index, client, self.counter_oplog),
             RollbackNoop(rng, index, self.ctx),
             CrossObjectTxn(rng, MzClient(self.ctx, f"cross-{index}"), self),
+            GraphChurn(rng, MzClient(self.ctx, f"graph-{index}")),
         ]
         # The transfer actions carry the conservation and ledger-identity
         # oracles, which are the strongest ones here, so they keep the bulk of
         # the op budget. The rest only need to happen often enough to produce
         # the states their checkers watch continuously.
-        weights = [10, 10, 3, 3, 6, 1, 1, 2, 2, 1, 1]
+        weights = [10, 10, 3, 3, 6, 1, 1, 2, 2, 1, 1, 3]
         if index == 0:
             # Single-instance churns: concurrent swaps of the same schema
             # pair or replacements of the same MV would only race each other
@@ -1847,7 +1990,7 @@ def make_worker(self, index: int, rng: random.Random) -> WorkerBundle:
         return WorkerBundle(actions=actions, weights=weights)
 
     def checkers(self) -> list[Checker]:
-        rngs = [random.Random(self.ctx.rng.randrange(SEED_RANGE)) for _ in range(19)]
+        rngs = [random.Random(self.ctx.rng.randrange(SEED_RANGE)) for _ in range(24)]
         return [
             BankTotalPeek(rngs[0], self.ctx, self),
             LedgerDirectPeek(rngs[1], self.ctx, self),
@@ -1870,6 +2013,31 @@ def checkers(self) -> list[Checker]:
             ReplicaDivergence(rngs[14], self.ctx),
             CounterSumPeek(rngs[16], self.ctx, self.counter_oplog),
             RollbackProbePeek(rngs[17], self.ctx),
+            GroupCompletenessPeek(
+                rngs[20], self.ctx, "closure-transitive", GRAPH_NOT_CLOSED_SQL
+            ),
+            GroupCompletenessPeek(
+                rngs[21], self.ctx, "closure-has-edges", GRAPH_MISSING_EDGE_SQL
+            ),
+            GroupCompletenessPeek(
+                rngs[22], self.ctx, "closure-acyclic", GRAPH_SELF_LOOP_SQL
+            ),
+            GroupCompletenessPeek(
+                rngs[23],
+                self.ctx,
+                "closure-differential",
+                GRAPH_DIFFERENTIAL_SQL,
+                clusters=["compute"],
+            ),
+            TernaryPartitionPeek(
+                rngs[19],
+                self.ctx,
+                "ledger-partition",
+                "ledger",
+                ledger_predicate,
+                "amount",
+                history=self.recent_ts,
+            ),
             GroupCompletenessPeek(
                 rngs[18],
                 self.ctx,
@@ -1935,6 +2103,15 @@ def final_check(self) -> None:
             raise InvariantViolation(
                 f"a transaction reading the total MV recorded {wrong}, not {self.total}"
             )
+        for name, sql in (
+            ("not transitively closed", GRAPH_NOT_CLOSED_SQL),
+            ("missing an edge", GRAPH_MISSING_EDGE_SQL),
+            ("has a self loop", GRAPH_SELF_LOOP_SQL),
+            ("disagrees with a recomputation", GRAPH_DIFFERENTIAL_SQL),
+        ):
+            bad = client.query(sql)
+            if bad:
+                raise InvariantViolation(f"closure {name}: {bad[:10]}")
         leaked = client.query("SELECT id FROM rollback_probe LIMIT 20")
         if leaked:
             raise InvariantViolation(