tests: Extend to run into OCC timestamp failure

Follow-up to #38322,
so currently based on top of it

Author: def-

misc/python/materialize/checks/all_checks/read_then_write_frontier.py

@@ -0,0 +1,123 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+from textwrap import dedent
+
+from materialize.checks.actions import Testdrive
+from materialize.checks.checks import Check
+
+# NOTE: Dedicated schema. Until its near refresh the view below is unreadable,
+# and a transaction's timedomain spans every collection in the queried schemas,
+# so in the default schema that window would block unrelated checks.
+
+
+class ReadThenWriteFarFrontier(Check):
+    """Read-then-writes whose selection reads a far-future write frontier.
+
+    A REFRESH materialized view settles until its next refresh, so its frontier
+    legitimately sits far out while the target table's upper is near the clock.
+    The write timestamp must still come from the timeline's oracle. Taking it
+    from the frontier ratchets the oracle into the future, where it is monotone
+    and durable, so every later write and strict-serializable read blocks until
+    the clock catches up, restarts included.
+
+    NOTE: that failure is environment-wide. A run where this check passes its
+    write and every other check then times out is this check's finding.
+
+    `INSERT ... SELECT` isolates it, since the target is written but not read.
+    The `UPDATE` and `DELETE` read their target too, which pulls the frontier
+    back to the clock: a far-future input must change neither the timestamp nor
+    the answer. `serializable` never consults the oracle, so there the write is
+    invisible rather than slow, which is what the read-back pins.
+    """
+
+    def initialize(self) -> Testdrive:
+        return Testdrive(dedent("""
+            > CREATE SCHEMA rtw_frontier_schema
+
+            > CREATE TABLE rtw_frontier_schema.source (f1 INTEGER)
+            > INSERT INTO rtw_frontier_schema.source VALUES (1), (2), (3)
+
+            > CREATE TABLE rtw_frontier_schema.destination (f1 INTEGER, phase TEXT)
+
+            > CREATE MATERIALIZED VIEW rtw_frontier_schema.frozen_mv
+              WITH (REFRESH AT mz_now()::text::int8 + 2000, REFRESH AT '3000-01-01')
+              AS SELECT f1 FROM rtw_frontier_schema.source
+
+            # Parks until the near refresh, after which the contents are fixed
+            # at these three rows: the only later refresh is in the year 3000.
+            > SELECT count(*) FROM rtw_frontier_schema.frozen_mv
+            3
+
+            > INSERT INTO rtw_frontier_schema.destination SELECT f1, 'initialize' FROM rtw_frontier_schema.frozen_mv
+            """))
+
+    def manipulate(self) -> list[Testdrive]:
+        return [
+            Testdrive(dedent(s))
+            for s in [
+                """
+                > INSERT INTO rtw_frontier_schema.source VALUES (4), (5)
+
+                > INSERT INTO rtw_frontier_schema.destination SELECT f1, 'manipulate1' FROM rtw_frontier_schema.frozen_mv
+
+                # An UPDATE reads its target too, so the table pulls this
+                # selection's frontier back to the clock.
+                > UPDATE rtw_frontier_schema.destination SET f1 = f1 + 10
+                  WHERE phase = 'initialize' AND f1 IN (SELECT f1 FROM rtw_frontier_schema.frozen_mv)
+                """,
+                """
+                > INSERT INTO rtw_frontier_schema.source VALUES (6), (7)
+
+                > INSERT INTO rtw_frontier_schema.destination SELECT f1, 'manipulate2' FROM rtw_frontier_schema.frozen_mv
+
+                > DELETE FROM rtw_frontier_schema.destination
+                  WHERE phase = 'manipulate1' AND f1 IN (SELECT f1 FROM rtw_frontier_schema.frozen_mv)
+
+                # A serializable read picks a timestamp near the clock, so it
+                # sees this write back only if the write landed near it too.
+                > SET transaction_isolation = 'serializable'
+
+                > INSERT INTO rtw_frontier_schema.destination SELECT f1, 'serializable' FROM rtw_frontier_schema.frozen_mv
+
+                > SELECT count(*) FROM rtw_frontier_schema.destination WHERE phase = 'serializable'
+                3
+
+                > RESET transaction_isolation
+                """,
+            ]
+        ]
+
+    def validate(self) -> Testdrive:
+        return Testdrive(dedent("""
+            > SELECT phase, count(*), sum(f1) FROM rtw_frontier_schema.destination GROUP BY phase ORDER BY phase
+            initialize 3 36
+            manipulate2 3 6
+            serializable 3 6
+
+            # A write committed at the view's frontier leaves the oracle in the
+            # year 3000, where this read blocks: a timeout is the same finding.
+            > SELECT mz_now()::text::bigint - (extract(epoch FROM now()) * 1000)::bigint < 60000
+            true
+
+            # TEMPORARY so a second validate() repeats rather than accumulates.
+            > CREATE TEMPORARY TABLE rtw_frontier_probe (f1 INTEGER)
+
+            > INSERT INTO rtw_frontier_probe SELECT f1 FROM rtw_frontier_schema.frozen_mv
+
+            > SELECT count(*), sum(f1) FROM rtw_frontier_probe
+            3 6
+
+            # The timeline still takes a blind write.
+            > INSERT INTO rtw_frontier_probe VALUES (100)
+
+            > SELECT count(*) FROM rtw_frontier_probe
+            4
+
+            > DROP TABLE rtw_frontier_probe
+            """))

misc/python/materialize/parallel_workload/action.py

@@ -262,6 +262,44 @@ def insert_batch_size(self, exe: Executor, table: Table) -> int:
         share = max(1, MAX_ROWS // exe.db.num_threads)
         return self.rng.randint(1, min(available, share))
 
+    def view_predicate(self, exe: Executor, table: Table) -> str | None:
+        """A predicate over a view, adding a second input to a read-then-write's
+        selection.
+
+        The selection's frontier is the minimum over its inputs, and an UPDATE
+        or DELETE reads its target too, so the table holds it near the wall
+        clock while a REFRESH view is free to sit far ahead. Neither may reach
+        the write timestamp, which comes from the timeline's oracle.
+        `InsertSelectAction` covers the view-only case. None when no view offers
+        a comparable column. Views reaching a source are excluded: the adapter
+        refuses such a selection outright, which would make this vacuous."""
+        views = [
+            view
+            for view in exe.db.views
+            if view.read_then_write_input
+            and (not view.temp or view in exe.temp_objects)
+        ]
+        self.rng.shuffle(views)
+        for view in views:
+            pairs = [
+                (table_column, view_column)
+                for table_column in table.columns
+                for view_column in view.columns
+                # A map has no equality operator, so it cannot drive an IN.
+                if table_column.data_type == view_column.data_type
+                and table_column.data_type != TextTextMap
+            ]
+            if not pairs:
+                continue
+            table_column, view_column = self.rng.choice(pairs)
+            # The alias keeps the inner reference off the outer target, the
+            # LIMIT bounds an expensive view body.
+            return (
+                f"{table_column.name(True)} IN (SELECT rtw_src.{view_column.name(True)}"
+                f" FROM {view} AS rtw_src LIMIT 100)"
+            )
+        return None
+
     def create_system_connection(
         self, exe: Executor, num_attempts: int = 10
     ) -> Connection:
@@ -1171,9 +1209,17 @@ def run(self, exe: Executor) -> bool:
         if not tables:
             return False
         table = self.rng.choice(tables)
-        # Reading the insert target itself makes the target a read dependency
-        # too, the most contended shape a read-then-write can have.
-        source = table if self.rng.choice([True, False]) else self.rng.choice(tables)
+        # Reading the insert target itself is the most contended shape a
+        # read-then-write can have. A view is the opposite: the target is
+        # written but not read, so a REFRESH view alone pins the selection's
+        # frontier, see `Action.view_predicate`.
+        sources = tables + [
+            view
+            for view in exe.db.views
+            if view.read_then_write_input
+            and (not view.temp or view in exe.temp_objects)
+        ]
+        source = table if self.rng.choice([True, False]) else self.rng.choice(sources)
 
         column_names = ", ".join(column.name(True) for column in table.columns)
         # The cast is an identity cast: `expression` returns the requested type
@@ -1464,7 +1510,12 @@ def run(self, exe: Executor) -> bool:
             f"{c.name(True)} = {expression(c.data_type, table.columns, self.rng, kind=ExprKind.WRITE)}"
             for c in set_columns
         )
-        query = f"UPDATE {table} SET {set_clause} WHERE {expression(Boolean, table.columns, self.rng, kind=ExprKind.WRITE)}"
+        predicate = expression(Boolean, table.columns, self.rng, kind=ExprKind.WRITE)
+        if self.rng.random() < 0.2:
+            view_predicate = self.view_predicate(exe, table)
+            if view_predicate:
+                predicate = f"({predicate}) AND {view_predicate}"
+        query = f"UPDATE {table} SET {set_clause} WHERE {predicate}"
         if self.rng.choice([True, False]):
             self.stmt_id += 1
             self.exe_prepared(query, f"update{self.stmt_id}", exe)
@@ -1516,7 +1567,8 @@ def errors_to_ignore(self, exe: Executor) -> list[str]:
             "canceling statement due to statement timeout",
             OCC_CONTENTION_EXHAUSTED_ERROR,
         ] + super().errors_to_ignore(exe)
-        if exe.db.scenario == Scenario.Rename:
+        # The predicate can name a view, which DDL drops concurrently.
+        if exe.db.complexity == Complexity.DDL or exe.db.scenario == Scenario.Rename:
             errors += ["does not exist"]
         return errors
 
@@ -1553,7 +1605,14 @@ def run(self, exe: Executor) -> bool:
             query += f" USING {using_table}"
             query += f" WHERE {expression(Boolean, all_columns, self.rng, kind=ExprKind.WRITE)}"
         elif self.rng.random() < 0.95:
-            query += f" WHERE {expression(Boolean, table.columns, self.rng, kind=ExprKind.WRITE)}"
+            predicate = expression(
+                Boolean, table.columns, self.rng, kind=ExprKind.WRITE
+            )
+            if self.rng.random() < 0.2:
+                view_predicate = self.view_predicate(exe, table)
+                if view_predicate:
+                    predicate = f"({predicate}) AND {view_predicate}"
+            query += f" WHERE {predicate}"
         if self.rng.choice([True, False]):
             self.stmt_id += 1
             self.exe_prepared(query, f"delete{self.stmt_id}", exe)

misc/python/materialize/parallel_workload/database.py

@@ -195,6 +195,10 @@ class DBObject:
     # Bounded (UP TO) load generators seal once they finish, and views
     # propagate sealing from their inputs, materialized or not.
     can_seal: bool = False
+    # Whether a read-then-write's selection may (transitively) read this
+    # object. The adapter refuses one that reaches a source or a source-export
+    # table, so only plain tables and views over them qualify.
+    read_then_write_input: bool = False
 
     def __init__(self):
         self.lock = threading.Lock()
@@ -207,6 +211,8 @@ def create(self, exe: Executor) -> None:
 
 
 class Table(DBObject):
+    read_then_write_input = True
+
     table_id: int
     rename: int
     num_rows: int
@@ -309,19 +315,24 @@ def __init__(
 
         self.materialized = not self.temp and rng.choice([True, False])
 
-        self.refresh = (
-            rng.choice(
-                [
-                    "ON COMMIT",
-                    # TODO: Restore minute-scale intervals when CPU-196 is fixed
-                    f"EVERY '{rng.randint(1, 15)} seconds'",
-                    f"EVERY '{rng.randint(1, 15)} seconds' ALIGNED TO (mz_now())",
-                    # Always in the future of all refreshes of previously generated MVs
-                    "AT mz_now()::string::int8 + 1000",
-                ]
-            )
-            if self.materialized
-            else None
+        # (SQL, whether the view's shard seals during a run): a REFRESH AT view
+        # seals once its last refresh has passed.
+        refresh_options = [
+            ("ON COMMIT", False),
+            # TODO: Restore minute-scale intervals when CPU-196 is fixed
+            (f"EVERY '{rng.randint(1, 15)} seconds'", False),
+            (f"EVERY '{rng.randint(1, 15)} seconds' ALIGNED TO (mz_now())", False),
+            # Always in the future of all refreshes of previously generated MVs
+            ("AT mz_now()::string::int8 + 1000", True),
+            # The near refresh makes the view readable, the far one parks its
+            # write frontier a millennium out without ever sealing it. That is
+            # what separates a read-then-write's write timestamp taken from the
+            # oracle from one taken from the selection's frontier, which drags
+            # the monotone, durable oracle into the future with it.
+            ("AT mz_now()::string::int8 + 1000, REFRESH AT '3000-01-01'", False),
+        ]
+        self.refresh, refresh_seals = (
+            rng.choice(refresh_options) if self.materialized else (None, False)
         )
 
         # A materialized view's shard seals (its write frontier advances to the
@@ -333,12 +344,16 @@ def __init__(
         # shards. Unmaterialized views have no shard, but a dataflow reading
         # one inlines its inputs, so sealing must propagate through them too.
         self.can_seal = (
-            (self.refresh or "").startswith("AT")
+            refresh_seals
             or self.repeat_row_const
             or base_object.can_seal
             or (base_object2 is not None and base_object2.can_seal)
         )
 
+        self.read_then_write_input = base_object.read_then_write_input and (
+            base_object2 is None or base_object2.read_then_write_input
+        )
+
         if base_object2:
             # The random boolean alone references an arbitrary subset of the
             # columns, so it usually names only one side or neither, and the