diff --git a/misc/python/materialize/checks/all_checks/read_then_write_frontier.py b/misc/python/materialize/checks/all_checks/read_then_write_frontier.py new file mode 100644 index 0000000000000..24fec57f5125c --- /dev/null +++ b/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 + """)) diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 58de840777d63..3957ad65a5a34 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/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) diff --git a/misc/python/materialize/parallel_workload/database.py b/misc/python/materialize/parallel_workload/database.py index 59166cd896d1a..96c8db907529c 100644 --- a/misc/python/materialize/parallel_workload/database.py +++ b/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