diff --git a/misc/python/materialize/feature_benchmark/scenarios/interactive_runtime.py b/misc/python/materialize/feature_benchmark/scenarios/interactive_runtime.py index 7bda7808e5d57..37d43c43012d9 100644 --- a/misc/python/materialize/feature_benchmark/scenarios/interactive_runtime.py +++ b/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 +""") diff --git a/misc/python/materialize/parallel_benchmark/scenarios.py b/misc/python/materialize/parallel_benchmark/scenarios.py index c0a9576e892c8..5626b5d2745be 100644 --- a/misc/python/materialize/parallel_benchmark/scenarios.py +++ b/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,6 +1529,10 @@ 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: @@ -1535,9 +1540,10 @@ class PeekIsolationUnderExpensivePeeks(Scenario): 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, + }, + )