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. The `ReadIsolationUnderHydration`
parallel-benchmark scenario A/Bs the flag, measuring peek latency with the
feature on against off while hydration saturates the maintenance workers.

Three 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. `singlereplica_attribution_sources.slt`
asserted a materialized view's timely dataflow id, which counted the peek
dataflows rendered before it. Those now render on the interactive runtime, so the
assertion is reduced to the id-to-collection pairing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk

Author: antiguru

misc/python/materialize/mzcompose/__init__.py

@@ -364,6 +364,11 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_compute_interactive_runtime",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",
@@ -662,9 +667,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_workload/action.py

@@ -3134,8 +3134,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_any_all_null_array_semantics"] = (
             BOOLEAN_FLAG_VALUES

src/clusterd-test-driver/src/script.rs

@@ -194,6 +194,9 @@ pub enum ExplainTarget {
         exports: Vec<ExportSpec>,
         /// The dataflow's `as_of`.
         as_of: u64,
+        /// The dataflow's `until`, if bounded. `as_of + 1` makes the dataflow single-time.
+        #[serde(default)]
+        until: Option<u64>,
         /// Run the MIR optimizer before lowering. Off by default.
         #[serde(default)]
         optimize: bool,
@@ -451,6 +454,11 @@ pub enum Command {
         /// The shard's exclusive write upper (see `PersistSource::upper`).
         upper: u64,
     },
+    /// Submit a dataflow a prior `create-dataflow name=<name> defer` registered.
+    SubmitDataflow {
+        /// The `create-dataflow` name to submit.
+        name: String,
+    },
     /// Schedule a previously-submitted collection so it makes progress.
     Schedule {
         /// The collection's global id.
@@ -524,10 +532,17 @@ pub enum Command {
         exports: Vec<ExportSpec>,
         /// The dataflow's `as_of`.
         as_of: u64,
+        /// The dataflow's `until`, if bounded. `as_of + 1` makes the dataflow single-time.
+        #[serde(default)]
+        until: Option<u64>,
         /// Run the MIR optimizer before lowering (needed for e.g. joins). Off by
         /// default, so the caller's MIR is lowered faithfully.
         #[serde(default)]
         optimize: bool,
+        /// Register the dataflow's exports without submitting it, so a later dataflow
+        /// can import them before `submit-dataflow` renders this one.
+        #[serde(default)]
+        defer: bool,
     },
     /// Render a dataflow's lowered LIR plan as text, the output assertion being the
     /// plan shape itself. It submits nothing and records no index, subscribe, or
@@ -631,6 +646,7 @@ struct DataflowSpec {
     builds: Vec<BuildSpec>,
     exports: Vec<ExportSpec>,
     as_of: u64,
+    until: Option<u64>,
     optimize: bool,
 }
 
@@ -801,6 +817,7 @@ impl ScriptState {
         builds: Vec<BuildSpec>,
         exports: Vec<ExportSpec>,
         as_of: u64,
+        until: Option<u64>,
         optimize: bool,
     ) -> anyhow::Result<(DataflowBuilder, PendingRegistrations)> {
         let mut builder =
@@ -937,9 +954,35 @@ impl ScriptState {
             }
         }
         builder.as_of(Timestamp::from(as_of));
+        if let Some(until) = until {
+            builder.until(Timestamp::from(until));
+        }
         Ok((builder, registrations))
     }
 
+    /// Finishes and submits `builder`, then applies `registrations`.
+    ///
+    /// Registers only after a successful submit, so a rejected dataflow leaves no dangling
+    /// index entry or subscribe buffer.
+    fn submit(
+        &mut self,
+        builder: DataflowBuilder,
+        registrations: PendingRegistrations,
+    ) -> anyhow::Result<()> {
+        let df = builder.finish()?;
+        self.driver.submit_dataflow(df)?;
+        for (index_id, entry) in registrations.indexes {
+            self.indexes.insert(index_id, entry);
+        }
+        for sink_id in registrations.subscribes {
+            self.driver.register_subscribe(sink_id);
+        }
+        for (sink_id, metadata) in registrations.mv_outputs {
+            self.mv_outputs.insert(sink_id, metadata);
+        }
+        Ok(())
+    }
+
     /// Execute a single command, returning its golden output text.
     pub async fn execute(&mut self, cmd: Command) -> anyhow::Result<String> {
         match cmd {
@@ -1085,10 +1128,13 @@ impl ScriptState {
                 builds,
                 exports,
                 as_of,
+                until,
                 optimize,
+                defer,
             } => {
                 // Record the spec under its name so `explain ref=<name>` can render
-                // this dataflow's plan later without repeating the body.
+                // this dataflow's plan later without repeating the body, and so
+                // `submit-dataflow` can submit a deferred one.
                 if let Some(name) = &name {
                     self.dataflows.insert(
                         name.clone(),
@@ -1097,39 +1143,63 @@ impl ScriptState {
                             builds: builds.clone(),
                             exports: exports.clone(),
                             as_of,
+                            until,
                             optimize,
                         },
                     );
                 }
-                let (builder, registrations) =
-                    self.configure_dataflow(name, imports, builds, exports, as_of, optimize)?;
-                let df = builder.finish()?;
-                self.driver.submit_dataflow(df)?;
-                // Register only after a successful submit, so a rejected dataflow
-                // leaves no dangling index entry or subscribe buffer.
-                for (index_id, entry) in registrations.indexes {
-                    self.indexes.insert(index_id, entry);
-                }
-                for sink_id in registrations.subscribes {
-                    self.driver.register_subscribe(sink_id);
-                }
-                for (sink_id, metadata) in registrations.mv_outputs {
-                    self.mv_outputs.insert(sink_id, metadata);
+                if defer {
+                    anyhow::ensure!(
+                        name.is_some(),
+                        "`defer` needs a name for the later `submit-dataflow`"
+                    );
+                    // Only the index registrations, which is what a later import
+                    // resolves against. Everything else registers at submit.
+                    let (_builder, registrations) = self.configure_dataflow(
+                        name, imports, builds, exports, as_of, until, optimize,
+                    )?;
+                    for (index_id, entry) in registrations.indexes {
+                        self.indexes.insert(index_id, entry);
+                    }
+                    return Ok("deferred".to_string());
                 }
+                let (builder, registrations) = self
+                    .configure_dataflow(name, imports, builds, exports, as_of, until, optimize)?;
+                self.submit(builder, registrations)?;
+                Ok("ok".to_string())
+            }
+            Command::SubmitDataflow { name } => {
+                let spec = self.dataflows.get(&name).ok_or_else(|| {
+                    anyhow::anyhow!(
+                        "unknown dataflow {name:?}; declare it with \
+                         create-dataflow name={name} defer first"
+                    )
+                })?;
+                let (builder, registrations) = self.configure_dataflow(
+                    Some(name.clone()),
+                    spec.imports.clone(),
+                    spec.builds.clone(),
+                    spec.exports.clone(),
+                    spec.as_of,
+                    spec.until,
+                    spec.optimize,
+                )?;
+                self.submit(builder, registrations)?;
                 Ok("ok".to_string())
             }
             Command::Explain { target } => {
                 // Resolve the target to a dataflow body: either given inline, or the
                 // spec a prior `create-dataflow name=<name>` recorded.
-                let (name, imports, builds, exports, as_of, optimize) = match target {
+                let (name, imports, builds, exports, as_of, until, optimize) = match target {
                     ExplainTarget::Inline {
                         name,
                         imports,
                         builds,
                         exports,
                         as_of,
+                        until,
                         optimize,
-                    } => (name, imports, builds, exports, as_of, optimize),
+                    } => (name, imports, builds, exports, as_of, until, optimize),
                     ExplainTarget::Reference { name } => {
                         let spec = self.dataflows.get(&name).ok_or_else(|| {
                             anyhow::anyhow!(
@@ -1143,6 +1213,7 @@ impl ScriptState {
                             spec.builds.clone(),
                             spec.exports.clone(),
                             spec.as_of,
+                            spec.until,
                             spec.optimize,
                         )
                     }
@@ -1151,8 +1222,8 @@ impl ScriptState {
                 // LIR plan instead of submitting it. The registrations are discarded:
                 // explain has no side effects, so it neither installs a dataflow nor
                 // records an index / subscribe / materialized-view output.
-                let (builder, _registrations) =
-                    self.configure_dataflow(name, imports, builds, exports, as_of, optimize)?;
+                let (builder, _registrations) = self
+                    .configure_dataflow(name, imports, builds, exports, as_of, until, optimize)?;
                 // The LIR render separates objects with blank lines; the `----` block
                 // preserves them via the doubled-separator form (see `crate::text`).
                 // Trim the trailing newline so the golden matches like every other

src/clusterd-test-driver/src/text.rs

@@ -433,6 +433,7 @@ struct DataflowBody {
     builds: Vec<BuildSpec>,
     exports: Vec<ExportSpec>,
     as_of: u64,
+    until: Option<u64>,
     optimize: bool,
 }
 
@@ -446,6 +447,7 @@ fn parse_dataflow_body(
 ) -> anyhow::Result<DataflowBody> {
     let name = opt_string(args, "name");
     let as_of = req_u64(args, "as-of")?;
+    let until = opt_u64(args, "until")?;
     let optimize = flags.iter().any(|f| f == "optimize");
     let mut imports = Vec::new();
     let mut builds = Vec::new();
@@ -485,6 +487,7 @@ fn parse_dataflow_body(
         builds,
         exports,
         as_of,
+        until,
         optimize,
     })
 }
@@ -568,6 +571,7 @@ fn parse_command(input: &str) -> anyhow::Result<Command> {
                 builds,
                 exports,
                 as_of,
+                until,
                 optimize,
             } = parse_dataflow_body(&args, &flags, body)?;
             Command::CreateDataflow {
@@ -576,9 +580,14 @@ fn parse_command(input: &str) -> anyhow::Result<Command> {
                 builds,
                 exports,
                 as_of,
+                until,
                 optimize,
+                defer: flags.iter().any(|f| f == "defer"),
             }
         }
+        "submit-dataflow" => Command::SubmitDataflow {
+            name: req(&args, "name")?.to_string(),
+        },
         "explain" => {
             // `explain ref=<name>` renders a previously declared dataflow; otherwise
             // the dataflow is given inline with the `create-dataflow` body.
@@ -595,6 +604,7 @@ fn parse_command(input: &str) -> anyhow::Result<Command> {
                     builds,
                     exports,
                     as_of,
+                    until,
                     optimize,
                 } = parse_dataflow_body(&args, &flags, body)?;
                 ExplainTarget::Inline {
@@ -603,6 +613,7 @@ fn parse_command(input: &str) -> anyhow::Result<Command> {
                     builds,
                     exports,
                     as_of,
+                    until,
                     optimize,
                 }
             };
@@ -785,7 +796,9 @@ mod tests {
                     key: vec![0]
                 }],
                 as_of: 0,
+                until: None,
                 optimize: false,
+                defer: false,
             }
         );
 
@@ -835,6 +848,7 @@ mod tests {
                         key: vec![0],
                     }],
                     as_of: 0,
+                    until: None,
                     optimize: true,
                 }
             }
@@ -1019,4 +1033,21 @@ mod tests {
             content
         );
     }
+
+    /// `defer` on the directive line is picked up, and `submit-dataflow` names the deferred
+    /// dataflow.
+    #[mz_ore::test]
+    fn parses_deferred_create_and_submit() {
+        let input = "create-dataflow name=d as-of=0 defer\n  import index=1001\n  build id=2000\n    Get u1000\n  export index=2001 on=2000 key=[0]";
+        assert!(matches!(
+            parse_command(input).unwrap(),
+            Command::CreateDataflow { defer: true, .. }
+        ));
+        assert_eq!(
+            parse_command("submit-dataflow name=d").unwrap(),
+            Command::SubmitDataflow {
+                name: "d".to_string()
+            }
+        );
+    }
 }