compute: turn the interactive runtime on in test configurations

`enable_compute_interactive_runtime` becomes a variable system parameter
defaulting to on, which is what makes every preceding piece of this work
reachable: sqllogictest, testdrive, and the mzcompose suites now provision
replicas with two runtimes, so peeks and bounded transient dataflows route to the
interactive runtime and maintenance publishes its indexes for it to read.
Production keeps the dyncfg's own default, which is off.

The clusterd mzcompose service grows the second runtime's port and
`--interactive-compute-timely-config`, mirroring what the controller passes in a
real deployment. Two clusterd-test-driver specs cover an index read and a query
dataflow across the runtime boundary, and a parallel-benchmark scenario measures
read isolation with the feature on against off.

Two goldens move. `relations.slt` gains the publisher operators, which are real
operators the maintenance runtime now installs on every published index.
`introspection-sources.td` raises a coarse arrangement-size bound from 16 KiB to
32 KiB, because publication raises the reported size of a one-record index past
the old bound. Whether that overhead is constant per arrangement or scales with
size is not established and wants re-measuring, so the comment records the
measurement without claiming a mechanism.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: antiguru

misc/python/materialize/mzcompose/__init__.py

@@ -297,6 +297,11 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_compute_interactive_runtime",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",
@@ -584,9 +589,6 @@ def get_default_system_parameters(
 # all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
 # apply.
 UNINTERESTING_SYSTEM_PARAMETERS = [
-    # Registered here rather than varied, because the interactive runtime cannot serve
-    # index peeks yet. Moves to get_variable_system_parameters once it can.
-    "enable_compute_interactive_runtime",
     "enable_compute_half_join2",
     "enable_mz_join_core",
     "linear_join_yielding",

misc/python/materialize/mzcompose/services/clusterd.py

@@ -39,6 +39,7 @@ def __init__(
         workers: int = 1,
         process_names: list[str] = [],
         mz_service: str = "materialized",
+        interactive_compute: bool = False,
     ) -> None:
         environment = [
             "CLUSTERD_LOG_FILTER",
@@ -78,6 +79,21 @@ def __init__(
             f"CLUSTERD_STORAGE_TIMELY_CONFIG={storage_timely_config}",
         ]
 
+        # When set, clusterd runs a second, interactive compute runtime alongside the
+        # maintenance one (see `--interactive-compute-timely-config` in
+        # `src/clusterd/src/lib.rs`). It must span the same number of Timely peers as
+        # the maintenance compute config, so it reuses `process_names`/`workers`; its
+        # addresses use a distinct port (2104) so the two runtimes don't collide.
+        ports = [2100, 2101, 6878]
+        if interactive_compute:
+            interactive_compute_timely_config = timely_config(
+                process_names, 2104, workers, DEFAULT_COMPUTE_EXERT_PROPORTIONALITY
+            )
+            environment += [
+                f"CLUSTERD_INTERACTIVE_COMPUTE_TIMELY_CONFIG={interactive_compute_timely_config}"
+            ]
+            ports += [2104]
+
         options = ["clusterd", f"--scratch-directory={scratch_directory}", *options]
 
         config: ServiceConfig = {}
@@ -106,7 +122,7 @@ def __init__(
         config.update(
             {
                 "command": options,
-                "ports": [2100, 2101, 6878],
+                "ports": ports,
                 "environment": environment,
                 "volumes": volumes or DEFAULT_MZ_VOLUMES,
                 "restart": restart,

misc/python/materialize/parallel_benchmark/scenarios.py

@@ -1274,3 +1274,119 @@ def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
             ],
             conn_pool_size=100,
         )
+
+
+class HydrationChurn(Action):
+    """Continuously builds a heavy maintained materialized view, forces it to
+    hydrate, then drops it, keeping the replica's maintenance runtime busy.
+
+    `SELECT count(*)` blocks until the view has hydrated, so each iteration does
+    real hydration work on the maintenance workers before the drop. Runs in a
+    `ClosedLoop`, which is single-threaded, so a fixed view name is safe.
+    """
+
+    def __init__(self, conn_info: PgConnInfo, name: str, view_sql: str):
+        self.conn_info = conn_info
+        self.name = name
+        self.view_sql = view_sql
+        self.conn = conn_info.connect()
+        self.conn.autocommit = True
+        self.cur = self.conn.cursor()
+
+    def _run(self, conns: queue.Queue):
+        self.cur.execute(
+            f"CREATE MATERIALIZED VIEW {self.name} AS {self.view_sql}".encode()
+        )
+        # Force hydration to complete before we drop, so the maintenance runtime
+        # actually does the build work rather than cancelling it immediately.
+        self.cur.execute(f"SELECT count(*) FROM {self.name}".encode())
+        self.cur.fetchall()
+        self.cur.execute(f"DROP MATERIALIZED VIEW {self.name}".encode())
+
+    def __str__(self) -> str:
+        return f"hydration churn {self.name}"
+
+
+class TwoRuntimeReadIsolation(Scenario):
+    """Measures whether peeks stay fast while the replica's maintenance runtime
+    is saturated by continuous dataflow hydration.
+
+    Run the same scenario twice and compare the SELECT p50/p99/qps:
+        bin/mzcompose --find parallel-benchmark run default \
+            --scenario TwoRuntimeReadIsolation \
+            --this-params enable_compute_interactive_runtime=true
+        bin/mzcompose --find parallel-benchmark run default \
+            --scenario TwoRuntimeReadIsolation \
+            --this-params enable_compute_interactive_runtime=false
+
+    With two runtimes ON the interactive runtime serves the peeks off the shared
+    arrangements, isolated from the maintenance workers hydrating the churn MVs.
+    OFF, the peeks contend with hydration on the same workers.
+
+    The peeks run open-loop at a fixed rate, so a serving runtime that cannot
+    keep up under contention accumulates queue-wait latency the reported p50/p99
+    capture. That queue backlog is the signal: it is exactly what a user issuing
+    reads at a steady rate experiences when the maintenance runtime steals the
+    serving capacity.
+    """
+
+    def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
+        mz = conn_infos["materialized"]
+        # Heavy enough that one build takes real CPU, small enough that two
+        # concurrent churn loops do not exhaust memory.
+        churn_view_sql = "SELECT a, count(*) AS c FROM big GROUP BY a"
+        self.init(
+            [
+                TdPhase("""
+                    > DROP TABLE IF EXISTS hot CASCADE
+                    > DROP TABLE IF EXISTS big CASCADE
+
+                    # The peek target: a small indexed table, pre-hydrated.
+                    > CREATE TABLE hot (k int, v int)
+                    > INSERT INTO hot SELECT n, n * 2 FROM generate_series(1, 100000) AS n
+                    > CREATE INDEX hot_k ON hot (k)
+
+                    # The contention source: a large table churned into heavy MVs.
+                    > CREATE TABLE big (a int, b int)
+                    > INSERT INTO big SELECT n, n % 1000 FROM generate_series(1, 1000000) AS n
+
+                    # Wait for the hot index to hydrate before measuring.
+                    > SELECT v FROM hot WHERE k = 42
+                    84
+                    """),
+                LoadPhase(
+                    duration=120,
+                    actions=[
+                        # Measured: fast-path index point lookup.
+                        OpenLoop(
+                            action=ReuseConnQuery(
+                                "SELECT v FROM hot WHERE k = 42",
+                                mz,
+                                strict_serializable=False,
+                            ),
+                            dist=Periodic(per_second=50),
+                            report_regressions=False,
+                        ),
+                        # Measured: slow-path range scan + reduce peek (more
+                        # sensitive to serving-runtime contention).
+                        OpenLoop(
+                            action=ReuseConnQuery(
+                                "SELECT count(*) FROM hot WHERE k < 50000",
+                                mz,
+                                strict_serializable=False,
+                            ),
+                            dist=Periodic(per_second=12),
+                            report_regressions=False,
+                        ),
+                    ]
+                    + [
+                        # Contention: continuous hydration churn on the same replica.
+                        ClosedLoop(
+                            action=HydrationChurn(mz, f"churn_{i}", churn_view_sql),
+                            report_regressions=False,
+                        )
+                        for i in range(2)
+                    ],
+                ),
+            ],
+        )

misc/python/materialize/parallel_workload/action.py

@@ -3020,8 +3020,9 @@ def __init__(
             BOOLEAN_FLAG_VALUES
         )
         self.flags_with_values["enable_upsert_v2"] = BOOLEAN_FLAG_VALUES
-        # Pinned off: the interactive runtime cannot serve index peeks yet.
-        self.flags_with_values["enable_compute_interactive_runtime"] = ["FALSE"]
+        self.flags_with_values["enable_compute_interactive_runtime"] = (
+            BOOLEAN_FLAG_VALUES
+        )
         self.flags_with_values["enable_coalesce_case_transform"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_compute_sync_mv_sink"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_column_paged_batcher"] = BOOLEAN_FLAG_VALUES

test/clusterd-test-driver/mzcompose.py

@@ -98,6 +98,14 @@ def __init__(self, name: str = "headless-driver") -> None:
     "create_time_config.spec",
 ]
 
+# Scenarios run against a clusterd with the interactive compute runtime enabled
+# (see `workflow_two_runtime_compute`). Kept separate from `SCRIPTS` so the
+# single-runtime scenarios above stay byte-for-byte unchanged.
+TWO_RUNTIME_SCRIPTS = [
+    "two_runtime_index.spec",
+    "two_runtime_query_dataflow.spec",
+]
+
 
 def workflow_default(c: Composition) -> None:
     c.up(METADATA_STORE, "minio", ServiceName("headless-driver", idle=True))
@@ -117,3 +125,32 @@ def workflow_default(c: Composition) -> None:
             env_extra={"DRIVER_SCRIPT": f"{SCRIPTS_DIR}/{script}"},
             use_aliases=True,
         )
+    # Also exercise every other workflow (the two-runtime compute path) so the
+    # CI job, which runs this composition's default workflow, covers them too.
+    for name in c.workflows:
+        if name == "default":
+            continue
+        with c.test_case(name):
+            c.workflow(name)
+
+
+def workflow_two_runtime_compute(c: Composition) -> None:
+    """Run the two-runtime scenarios against a clusterd started with a second,
+    interactive compute runtime configured. `Clusterd(interactive_compute=True)`
+    sets `CLUSTERD_INTERACTIVE_COMPUTE_TIMELY_CONFIG`, which turns on the
+    multiplexer that fronts both runtimes on the same `:2101` endpoint this
+    driver already connects to, so no driver changes are needed to reach the
+    interactive runtime.
+    """
+    c.up(METADATA_STORE, "minio", ServiceName("headless-driver", idle=True))
+    with c.override(Clusterd(mz_service="headless-driver", interactive_compute=True)):
+        for i, script in enumerate(TWO_RUNTIME_SCRIPTS):
+            ui.section(f"Running two-runtime scenario {script}")
+            if i > 0:
+                c.kill("clusterd")
+            c.up("clusterd")
+            c.run(
+                "headless-driver",
+                env_extra={"DRIVER_SCRIPT": f"{SCRIPTS_DIR}/{script}"},
+                use_aliases=True,
+            )

test/clusterd-test-driver/scripts/two_runtime_index.spec

@@ -0,0 +1,57 @@
+# 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.
+
+# two_runtime_index scenario: runs against a clusterd started with a second,
+# interactive compute runtime configured (see `Clusterd(interactive_compute=True)`
+# in the `two_runtime_compute` workflow). A multiplexer inside clusterd fronts both
+# runtimes on the single `:2101` endpoint this driver connects to, so the commands
+# below are unchanged from a single-runtime script; only the clusterd underneath
+# differs. The maintenance runtime renders and publishes the index below into the
+# shared arrangement-sharing registry; the multiplexer routes the `peek` to the
+# interactive runtime regardless, so a correct result proves the interactive
+# runtime served the rows from that shared registry rather than rendering its own
+# copy of the index.
+create-instance
+----
+ok
+
+update-configuration
+----
+ok
+
+initialization-complete
+----
+ok
+
+write-rows shard=r ts=0
+  1 alpha
+  2 beta
+  3 gamma
+----
+wrote 3
+
+create-dataflow name=two-runtime-index as-of=0
+  import source=1000 shard=r upper=1
+  build id=2000
+    Project (#0, #1)
+      Get u1000
+  export kind=index index=2001 on=2000 key=[0]
+----
+ok
+
+schedule id=2001
+----
+ok
+
+# Routed to the interactive runtime by the multiplexer.
+peek id=2001 ts=0
+----
+1 "alpha"
+2 "beta"
+3 "gamma"

test/clusterd-test-driver/scripts/two_runtime_query_dataflow.spec

@@ -0,0 +1,98 @@
+# 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.
+
+# two_runtime_query_dataflow scenario: the acceptance test for the interactive
+# slow path (N1-N3). Unlike two_runtime_index.spec, which peeks a maintenance
+# index directly (the fast path), this creates a genuine INTERACTIVE QUERY
+# DATAFLOW: a `count(*)` reduce that imports a maintenance index and is itself
+# exported under a transient id (`t4000`), so the multiplexer routes its
+# `CreateDataflow` to the interactive runtime (see `mz_compute_client::multiplex`).
+#
+# The maintenance index is created but deliberately left unscheduled before the
+# query dataflow is submitted, so nothing has been published to the shared
+# arrangement-sharing registry yet. The interactive runtime builds the query
+# dataflow immediately anyway, binding its import to a real but empty publication
+# point (a placeholder) rather than deferring the build, which would break the
+# deterministic construction order every worker must follow. The query's own
+# `schedule` is then sent immediately, before the maintenance index is scheduled,
+# mirroring what the real compute controller always does for a transient
+# collection (`Instance::maybe_schedule_collection`: scheduled right away, without
+# waiting for its inputs). Only afterward is the maintenance index scheduled, which
+# renders it and adopts the placeholder in place; that publication (never a bare
+# poll) wakes the import, which begins producing. The result peek then returns the
+# correct reduced rows, proving the bind -> fill -> resolve path completed off the
+# maintenance worker.
+create-instance
+----
+ok
+
+update-configuration
+----
+ok
+
+initialization-complete
+----
+ok
+
+write-rows shard=r ts=0
+  1 alpha
+  2 beta
+  3 gamma
+----
+wrote 3
+
+# The maintenance index over shard `r`. Registering it lets `import index=2001`
+# below reference it; its dataflow is intentionally left unscheduled, so no
+# arrangement is published yet.
+create-dataflow name=maint-index as-of=0
+  import source=1000 shard=r upper=1
+  build id=2000
+    Project (#0, #1)
+      Get u1000
+  export kind=index index=2001 on=2000 key=[0]
+----
+ok
+
+# A one-column schema for the count reduce's output (a single bigint).
+define-schema name=count_out
+  count bigint
+----
+ok
+
+# The interactive query dataflow: `count(*)` over the (still unpublished)
+# maintenance index, exported under a transient id (`t4000`) so it is routed to
+# the interactive runtime instead of maintenance.
+create-dataflow name=interactive-count as-of=0
+  import index=2001
+  build id=3000
+    Reduce aggregates=[count(*)]
+      Get u2000
+  export index=t4000 on=3000 key=[0]
+----
+ok
+
+# Scheduled before the maintenance index: the dataflow is built but its import is
+# still an empty placeholder at this point, so this `Schedule` races ahead of the
+# publication, exactly as the real controller's immediate-schedule-of-transient
+# behavior would.
+schedule id=t4000
+----
+ok
+
+# Scheduling the maintenance index renders it and adopts the placeholder, waking
+# the interactive query dataflow's import via the publication notification.
+schedule id=2001
+----
+ok
+
+# The interactive runtime ran the reduce once notified of the publication, off the
+# maintenance worker. The result is the correct count.
+peek id=t4000 schema=count_out ts=0
+----
+3