benchmarks: price oversubscription and publication at scale

The first nightly with the interactive runtime showed reads isolated from
hydration and nothing else moving, and left three questions the benchmarks did
not ask. What a saturated interactive runtime does to maintenance, since two
runtimes are twice the worker threads on the same cores. What publishing every
index costs when there are hundreds of them and nobody reads. And whether
`PeekIsolationUnderExpensivePeeks` guards anything, since its two sides came
out identical.

`MaintenanceUnderPeekSaturation` loads an eight-worker replica, half the cores
of the nightly agents, with one join peek in flight per worker and two
hydration churn loops, while a writer advances a small materialized view. The
measured query reads that view strict serializable from a separate one-worker
cluster, so it cannot answer before the view's write frontier passes the write
and nothing else competes with it: its latency is the view's maintenance lag
under the peek load. Read on the loaded replica it would instead queue behind
the joins on the runtime that serves peeks, which is what the join loops' own
latency reports.

`ManyIndexesIdle` and `ManyIndexesRestart` publish two hundred indexed views
over one table, each a distinct arrangement. The first measures one lookup and,
through the clusterd memory column, the resident cost of two hundred
publications. The second restarts the replica and waits for a read that imports
every index, so every publication is rendered and bound again on the way up.
`ManyReexportsIdle` is the other shape: two hundred indexes on one relation and
key share one arrangement and re-export it, each under its own publication.

`PeekIsolationUnderExpensivePeeks` walks five times as many rows half again as
often, about 60% of the worker where the old sizing measured under 10%, at which
the lookups did not notice the walks. Its docstring now says what it guards:
both queries are peeks, so it measures how the serving runtime interleaves
them, not the split between peeks and maintenance.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk

Author: antiguru

misc/python/materialize/feature_benchmark/scenarios/interactive_runtime.py

@@ -17,8 +17,12 @@
 """
 
 from materialize.feature_benchmark.action import Action, TdAction
-from materialize.feature_benchmark.measurement_source import MeasurementSource, Td
-from materialize.feature_benchmark.scenario import Scenario
+from materialize.feature_benchmark.measurement_source import (
+    Lambda,
+    MeasurementSource,
+    Td,
+)
+from materialize.feature_benchmark.scenario import BenchmarkingSequence, Scenario
 
 
 class InteractiveRuntime(Scenario):
@@ -103,3 +107,110 @@ def benchmark(self) -> MeasurementSource:
   /* B */
 1
 """)
+
+
+class ManyIndexes(InteractiveRuntime):
+    """Group parent: `INDEXES` indexed views over one table, plus a view over all of them for a
+    read that answers only once every index has hydrated.
+
+    Each view is a distinct relation, so each index is its own arrangement with its own
+    publication. An index that repeats an existing index's relation and key re-exports that
+    arrangement instead, which `ManyReexportsIdle` prices separately."""
+
+    SCALE = 5
+    INDEXES = 200
+
+    def init(self) -> list[Action]:
+        views = "\n".join(
+            f"> CREATE VIEW v_{i} AS SELECT f1 FROM t WHERE f1 % {self.INDEXES} = {i}\n"
+            f"> CREATE DEFAULT INDEX ON v_{i}\n"
+            for i in range(self.INDEXES)
+        )
+        union = " UNION ALL ".join(f"SELECT f1 FROM v_{i}" for i in range(self.INDEXES))
+        return [
+            self.table_ten(),
+            TdAction(f"""
+> CREATE TABLE t (f1 INTEGER)
+
+> INSERT INTO t SELECT {self.unique_values()} FROM {self.join()}
+
+{views}
+
+> CREATE VIEW all_v AS SELECT count(*) AS c FROM ({union})
+
+> SELECT c FROM all_v
+{self.n()}
+"""),
+        ]
+
+
+class ManyIndexesIdle(ManyIndexes):
+    """One lookup with many published indexes in place. The wallclock is the lookup; the
+    clusterd memory measurement is the resident cost of publishing `INDEXES` arrangements.
+    """
+
+    def benchmark(self) -> MeasurementSource:
+        return Td(f"""
+> SELECT 1
+  /* A */
+1
+
+> SELECT count(*) FROM v_0
+  /* B */
+{self.n() // self.INDEXES}
+""")
+
+
+class ManyIndexesRestart(ManyIndexes):
+    """Restart with many indexes, timed until a read importing all of them answers. Every index
+    is rendered and published again on the way up, and the read binds every publication.
+    """
+
+    def benchmark(self) -> BenchmarkingSequence:
+        return [
+            Lambda(lambda e: e.RestartMzClusterd()),
+            Td(f"""
+> SELECT c FROM all_v
+  /* B */
+{self.n()}
+"""),
+        ]
+
+
+class ManyReexportsIdle(InteractiveRuntime):
+    """One lookup with many re-exported indexes in place: `INDEXES` indexes on one relation and
+    key share one arrangement, and each publishes it under its own id. The clusterd memory
+    measurement is the resident cost of `INDEXES` publications of the same arrangement.
+    """
+
+    SCALE = 5
+    INDEXES = 200
+
+    def init(self) -> list[Action]:
+        indexes = "\n".join(
+            f"> CREATE INDEX t_f1_{i} ON t (f1)\n" for i in range(self.INDEXES)
+        )
+        return [
+            self.table_ten(),
+            TdAction(f"""
+> CREATE TABLE t (f1 INTEGER)
+
+> INSERT INTO t SELECT {self.unique_values()} FROM {self.join()}
+
+{indexes}
+
+> SELECT count(*) FROM t
+{self.n()}
+"""),
+        ]
+
+    def benchmark(self) -> MeasurementSource:
+        return Td("""
+> SELECT 1
+  /* A */
+1
+
+> SELECT f1 FROM t WHERE f1 = 1
+  /* B */
+1
+""")

misc/python/materialize/parallel_benchmark/scenarios.py

@@ -10,6 +10,7 @@
 import queue
 import time
 from copy import deepcopy
+from dataclasses import replace
 
 import psycopg
 
@@ -1528,16 +1529,21 @@ class PeekIsolationUnderExpensivePeeks(Scenario):
     size and the peek response stash out of the measurement. The cheap query is
     a literal lookup on the key, and its p99 is what this reports.
 
+    Both queries are peeks, so they share a runtime wherever peeks are placed.
+    This guards how the runtime that serves peeks interleaves a cheap one with
+    an expensive one, not the split between peeks and maintenance.
+
     Three things the numbers depend on, none of which the output would reveal if
     they stopped holding:
 
     * The expensive loop has to leave the worker idle between walks, or the
       lookups queue without bound and the percentiles report the length of the
       load phase. The measured cluster is one worker, since `Materialized` boots
       at the `bootstrap` replica size and `--size` does not reach it, so the rate
-      is set against one walk: at 100ns to 1us per position, 100,000 positions
-      is 10ms to 100ms, and 2/s of those is 2% to 20% of that worker. Raising
-      the rate or the row count means redoing this.
+      is set against one walk: 100,000 positions measured about 40ms, so 500,000
+      is about 200ms, and 3/s of those is about 60% of that worker. At 100,000
+      positions and 2/s the lookups did not notice the walks at all. Raising the
+      rate or the row count means redoing this.
     * The loops are pooled and the pool is far larger than the ~3 connections
       they hold. Waiting for a connection is timed like waiting for the replica,
       and `ReuseConnQuery` would cap concurrency at one and report a client-side
@@ -1571,7 +1577,7 @@ def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
                     # queued behind it and short enough to leave the worker idle
                     # between walks. The class docstring has the arithmetic.
                     > CREATE TABLE hot (k int, v int)
-                    > INSERT INTO hot SELECT n, n * 2 FROM generate_series(1, 100000) AS n
+                    > INSERT INTO hot SELECT n, n * 2 FROM generate_series(1, 500000) AS n
                     > CREATE INDEX hot_k ON hot (k)
 
                     # Wait for the index to hydrate before measuring.
@@ -1593,7 +1599,7 @@ def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
                         # nothing and every position is examined for no rows.
                         OpenLoop(
                             action=PooledQuery("SELECT v FROM hot WHERE v = -1"),
-                            dist=Periodic(per_second=2),
+                            dist=Periodic(per_second=3),
                             report_regressions=False,
                         ),
                     ],
@@ -1740,10 +1746,9 @@ class FreshnessUnderPeekWalks(Scenario):
     connection; the contention is a fixed rate of full index walks on the same
     replica.
 
-    The walk table is sized as in `PeekIsolationUnderExpensivePeeks`: each walk
-    is long enough to hold the frontier back measurably and short enough to
-    leave the worker idle between walks, so the read does not queue without
-    bound.
+    The walk table is 100,000 rows at 2/s, about 40ms per walk on one worker,
+    so each walk holds the frontier back measurably while leaving the worker
+    idle between walks and the read does not queue without bound.
     """
 
     def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
@@ -1810,3 +1815,129 @@ def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
                 "SELECT count(*) FROM fresh_w WHERE k = 0 (reuse connection)": CONTENDED_THRESHOLDS,
             },
         )
+
+
+class MaintenanceUnderPeekSaturation(Scenario):
+    """Measures how far a saturating peek load holds back maintenance on the same
+    replica.
+
+    The other isolation scenarios saturate maintenance and measure peeks. This
+    is the converse. Cluster `sat` has eight workers, half the cores of the
+    agents the nightly runs on, and carries three loads: eight closed loops of a
+    join peek, one in flight per worker, so every worker that serves peeks is
+    busy; two hydration churn loops, so the maintenance workers are busy too;
+    and a small materialized view over a table a writer keeps advancing. With a
+    second runtime that is sixteen worker threads on sixteen cores, which is the
+    oversubscription this prices. With one runtime the eight workers do all of
+    it in turn.
+
+    The measured query is a strict serializable read of the materialized view
+    from `sat_idle`, a separate one-worker cluster with nothing else to do. It
+    cannot answer until the view's write frontier passes the write, so its
+    latency is the view's maintenance lag on `sat` plus one idle lookup. Read on
+    `sat` itself it would instead measure the lookup queueing behind the joins
+    on the runtime that serves peeks, which the join loops' own latency already
+    reports. That latency is reported too, as the serving runtime's throughput.
+    """
+
+    def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
+        # `connect()` issues `SET cluster` itself, which has to run in autocommit:
+        # `ReuseConnQuery` and `HydrationChurn` turn autocommit on afterwards and
+        # cannot while that statement's transaction is open. The clusters do not
+        # exist yet when these connections open, which is a notice, not an error.
+        sat = replace(conn_infos["materialized"], cluster="sat", autocommit=True)
+        idle = replace(conn_infos["materialized"], cluster="sat_idle", autocommit=True)
+        join = "SELECT count(*) FROM sat_big a JOIN sat_big b USING (k)"
+        churn_view_sql = "SELECT a, count(*) AS c FROM sat_churn GROUP BY a"
+        self.init(
+            [
+                TdPhase(f"""
+                    > DROP TABLE IF EXISTS sat_w CASCADE
+                    > DROP TABLE IF EXISTS sat_big CASCADE
+                    > DROP TABLE IF EXISTS sat_churn CASCADE
+                    > DROP CLUSTER IF EXISTS sat CASCADE
+                    > DROP CLUSTER IF EXISTS sat_idle CASCADE
+
+                    > CREATE CLUSTER sat SIZE 'scale=1,workers=8', REPLICATION FACTOR 1
+                    > CREATE CLUSTER sat_idle SIZE 'scale=1,workers=1', REPLICATION FACTOR 1
+
+                    # The written table and the view whose freshness is measured.
+                    > CREATE TABLE sat_w (k int, v int)
+                    > INSERT INTO sat_w SELECT n, n FROM generate_series(1, 1000) AS n
+                    > CREATE MATERIALIZED VIEW sat_mv IN CLUSTER sat AS SELECT count(*) AS c FROM sat_w
+
+                    # The join input.
+                    > CREATE TABLE sat_big (k int, v int)
+                    > INSERT INTO sat_big SELECT n, n * 2 FROM generate_series(1, 200000) AS n
+                    > CREATE INDEX sat_big_k IN CLUSTER sat ON sat_big (k)
+
+                    # The churn input.
+                    > CREATE TABLE sat_churn (a int, b int)
+                    > INSERT INTO sat_churn SELECT n, n % 1000 FROM generate_series(1, 1000000) AS n
+
+                    > SET cluster = sat_idle
+                    > SELECT c FROM sat_mv
+                    1000
+
+                    > SET cluster = sat
+                    > {join}
+                    200000
+                    """),
+                LoadPhase(
+                    duration=120,
+                    actions=[
+                        # Contention: writes the read has to wait for.
+                        ClosedLoop(
+                            action=ReuseConnQuery(
+                                "INSERT INTO sat_w VALUES (0, 0)",
+                                sat,
+                                strict_serializable=False,
+                            ),
+                            report_regressions=False,
+                        ),
+                        # Measured: a read that waits for the view's write
+                        # frontier to pass the latest write, served off `sat`.
+                        ClosedLoop(
+                            action=ReuseConnQuery(
+                                "SELECT c FROM sat_mv",
+                                idle,
+                                strict_serializable=True,
+                            ),
+                        ),
+                    ]
+                    + [
+                        # Contention and measured: one peek dataflow in flight
+                        # per worker.
+                        ClosedLoop(action=PooledQuery(join))
+                        for _ in range(8)
+                    ]
+                    + [
+                        # Contention: continuous hydration on the maintenance
+                        # workers.
+                        ClosedLoop(
+                            action=HydrationChurn(
+                                sat, f"sat_churn_{i}", churn_view_sql
+                            ),
+                            report_regressions=False,
+                        )
+                        for i in range(2)
+                    ],
+                ),
+                TdPhase("""
+                    > DROP TABLE IF EXISTS sat_w CASCADE
+                    > DROP TABLE IF EXISTS sat_big CASCADE
+                    > DROP TABLE IF EXISTS sat_churn CASCADE
+                    > DROP CLUSTER IF EXISTS sat CASCADE
+                    > DROP CLUSTER IF EXISTS sat_idle CASCADE
+                    """),
+            ],
+            conn_pool_size=100,
+            conn_pool_setup=[
+                "SET TRANSACTION_ISOLATION TO 'SERIALIZABLE'",
+                "SET cluster = sat",
+            ],
+            regression_thresholds={
+                "SELECT c FROM sat_mv (reuse connection)": CONTENDED_THRESHOLDS,
+                f"{join} (pooled)": CONTENDED_THRESHOLDS,
+            },
+        )