benchmarks: guard the interactive runtime's read paths

Feature-benchmark scenarios for the four read shapes the interactive runtime
changes, each in an `Interactive` and a `Solo` leaf so the report carries the
cost of the second runtime side by side with each leaf's regression against the
other build: a peek dataflow joining two indexes, a fast-path point lookup,
`CREATE INDEX` plus the first read that uses it, and a per-replica introspection
read. Each leaf pins the flag and provisions its own cluster, which the
benchmark's fresh instance per scenario makes safe.

Parallel-benchmark scenarios for what the second runtime is meant to buy: the
temporary-dataflow floor on a quiet replica, introspection latency under
hydration, and how far expensive peek walks hold back a written index's
frontier, measured as the latency of a strict serializable read after a write.
The two existing isolation scenarios now report regressions on their measured
loops, with looser thresholds for contended tails.

A sqllogictest pins the flag on a two-worker replica and covers same-key
duplicate indexes across drops, error results on the fast path and through a
peek dataflow, strict serializable reads over shared arrangements, and cluster
re-provisioning. The clusterd-test-driver two-runtime workflow also runs with
two workers, so the registry's worker pairing is exercised.

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

@@ -0,0 +1,244 @@
+# 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.
+
+"""Reads on a replica with and without the interactive compute runtime.
+
+Every scenario here comes in two leaves. The `Interactive` leaf provisions its cluster with
+`enable_compute_interactive_runtime` on, so peeks and peek dataflows run on the second runtime and
+read the maintenance runtime's published arrangements. The `Solo` leaf provisions the same cluster
+with the flag off, which is a single-runtime replica. Both leaves run in every benchmark, so the
+report carries the cost of the second runtime for each shape side by side with the regression of
+each leaf against the other build.
+
+The flag is read when a replica is provisioned, and the benchmark boots a fresh instance per
+scenario, so `init` sets it before creating the cluster and nothing else observes the setting.
+"""
+
+from materialize.feature_benchmark.action import Action, TdAction
+from materialize.feature_benchmark.measurement_source import MeasurementSource, Td
+from materialize.feature_benchmark.scenario import Scenario
+
+
+class InteractiveRuntime(Scenario):
+    """Reads whose placement the interactive runtime changes. Group parent."""
+
+    INTERACTIVE: bool
+
+    def cluster(self) -> TdAction:
+        """Provisions cluster `ir` with the interactive runtime on or off."""
+        flag = "true" if self.INTERACTIVE else "false"
+        return TdAction(f"""
+$ postgres-execute connection=mz_system
+ALTER SYSTEM SET enable_compute_interactive_runtime = {flag};
+
+> CREATE CLUSTER ir SIZE 'scale={self._default_size},workers=1', REPLICATION FACTOR 1
+""")
+
+
+class PeekDataflowJoin(InteractiveRuntime):
+    """A join over two indexed views, repeated. Neither input can take the fast path, so each
+    query builds a peek dataflow that imports both indexes, which is the cost of a temporary
+    dataflow on the runtime that serves it."""
+
+    SCALE = 5
+    REPEAT = 10
+
+    def init(self) -> list[Action]:
+        return [
+            self.cluster(),
+            self.table_ten(),
+            TdAction(f"""
+> SET cluster = ir
+
+> CREATE MATERIALIZED VIEW v1 AS SELECT {self.unique_values()} AS f1 FROM {self.join()}
+
+> CREATE MATERIALIZED VIEW v2 AS SELECT {self.unique_values()} AS f1 FROM {self.join()}
+
+> CREATE DEFAULT INDEX ON v1
+
+> CREATE DEFAULT INDEX ON v2
+
+> SELECT count(*) FROM v1 JOIN v2 USING (f1)
+{self.n()}
+"""),
+        ]
+
+    def benchmark(self) -> MeasurementSource:
+        joins = "\n".join(
+            f"> SELECT count(*) FROM v1 JOIN v2 USING (f1)\n{self.n()}\n"
+            for _ in range(self.REPEAT)
+        )
+        return Td(f"""
+> SET cluster = ir
+
+> SELECT 1
+  /* A */
+1
+
+{joins}
+
+> SELECT 1
+  /* B */
+1
+""")
+
+
+class PeekDataflowJoinInteractive(PeekDataflowJoin):
+    INTERACTIVE = True
+
+
+class PeekDataflowJoinSolo(PeekDataflowJoin):
+    INTERACTIVE = False
+
+
+class PointLookup(InteractiveRuntime):
+    """A literal lookup on an indexed view, repeated. On the interactive runtime the walk reads
+    the arrangement the maintenance runtime published rather than a local trace."""
+
+    REPEAT = 1000
+
+    def init(self) -> list[Action]:
+        return [
+            self.cluster(),
+            self.table_ten(),
+            TdAction(f"""
+> SET cluster = ir
+
+> CREATE MATERIALIZED VIEW v1 AS SELECT {self.unique_values()} AS f1 FROM {self.join()}
+
+> CREATE DEFAULT INDEX ON v1
+
+> SELECT count(*) = {self.n()} FROM v1
+true
+"""),
+        ]
+
+    def benchmark(self) -> MeasurementSource:
+        lookups = "\n".join(
+            "> SELECT * FROM v1 WHERE f1 = 1\n1\n" for _ in range(self.REPEAT)
+        )
+        return Td(f"""
+> SET cluster = ir
+
+> SET auto_route_introspection_queries TO false
+
+> BEGIN
+
+> SELECT 1
+  /* A */
+1
+
+{lookups}
+
+> SELECT 1
+  /* B */
+1
+""")
+
+
+class PointLookupInteractive(PointLookup):
+    INTERACTIVE = True
+
+
+class PointLookupSolo(PointLookup):
+    INTERACTIVE = False
+
+
+class CreateIndexPublish(InteractiveRuntime):
+    """CREATE INDEX plus the first read that uses it. A publishing runtime installs a publisher
+    per arrangement, and the first read on the interactive runtime waits for its publication.
+    """
+
+    def init(self) -> list[Action]:
+        return [
+            self.cluster(),
+            self.table_ten(),
+            TdAction(f"""
+> SET cluster = ir
+
+> CREATE TABLE t1 (f1 INTEGER, f2 INTEGER)
+
+> INSERT INTO t1 (f1) SELECT {self.unique_values()} FROM {self.join()}
+
+> SELECT 1 FROM t1 WHERE f1 = 0
+1
+"""),
+        ]
+
+    def benchmark(self) -> MeasurementSource:
+        return Td("""
+> SET cluster = ir
+
+> DROP INDEX IF EXISTS i1
+  /* A */
+
+> CREATE INDEX i1 ON t1(f1)
+
+> SELECT count(*) FROM t1 AS a1, t1 AS a2 WHERE a1.f1 = a2.f1 AND a1.f1 = 0 AND a2.f1 = 0
+  /* B */
+1
+""")
+
+
+class CreateIndexPublishInteractive(CreateIndexPublish):
+    INTERACTIVE = True
+
+
+class CreateIndexPublishSolo(CreateIndexPublish):
+    INTERACTIVE = False
+
+
+class IntrospectionRead(InteractiveRuntime):
+    """A read of a per-replica introspection relation, repeated. The interactive runtime serves it
+    from the maintenance runtime's published logging index."""
+
+    REPEAT = 100
+
+    def init(self) -> list[Action]:
+        return [
+            self.cluster(),
+            self.table_ten(),
+            TdAction(f"""
+> SET cluster = ir
+
+> CREATE MATERIALIZED VIEW v1 AS SELECT {self.unique_values()} AS f1 FROM {self.join()}
+
+> CREATE DEFAULT INDEX ON v1
+
+> SELECT count(*) = {self.n()} FROM v1
+true
+"""),
+        ]
+
+    def benchmark(self) -> MeasurementSource:
+        reads = "\n".join(
+            "> SELECT count(*) > 0 FROM mz_introspection.mz_dataflow_arrangement_sizes\ntrue\n"
+            for _ in range(self.REPEAT)
+        )
+        return Td(f"""
+> SET cluster = ir
+
+> SELECT 1
+  /* A */
+1
+
+{reads}
+
+> SELECT 1
+  /* B */
+1
+""")
+
+
+class IntrospectionReadInteractive(IntrospectionRead):
+    INTERACTIVE = True
+
+
+class IntrospectionReadSolo(IntrospectionRead):
+    INTERACTIVE = False