benchmarks: guard the interactive runtime's read paths

Feature-benchmark scenarios for the four read shapes the interactive runtime
changes, each pinning the flag on and provisioning its own cluster, which the
benchmark's fresh instance per scenario makes safe: 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. They skip on versions before the flag
exists, following the `can_run` precedent.

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,213 @@
+# 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 the interactive compute runtime.
+
+Every scenario here 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 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
+from materialize.mz_version import MzVersion
+
+
+class InteractiveRuntime(Scenario):
+    """Reads whose placement the interactive runtime changes. Group parent."""
+
+    @classmethod
+    def can_run(cls, version: MzVersion) -> bool:
+        # Requires `enable_compute_interactive_runtime`, which lands in 26.41.0. Skips on both
+        # sides while we are still on `26.41.0-dev.0`, because `MzVersion` strips the commit
+        # and cannot distinguish dev builds with the flag from dev builds without it.
+        return version > MzVersion.create(26, 41, 0)
+
+    def cluster(self) -> TdAction:
+        """Provisions cluster `ir` with the interactive runtime on."""
+        return TdAction(f"""
+$ postgres-execute connection=mz_system
+ALTER SYSTEM SET enable_compute_interactive_runtime = true;
+
+> 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 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 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 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
+""")