Skip to content

Commit e6d27f4

Browse files
antiguruclaude
andcommitted
compute: launch a second, interactive compute runtime
With `--interactive-compute-timely-config`, clusterd runs two compute runtimes in one process: the first takes the `Maintenance` role, the second `Interactive`, and both share the one sharing registry so a reader on worker `i` of either finds the slot a publisher on worker `i` of the other filled. The two must span an equal number of Timely peers, which the config preparation asserts, because the registry pairs workers by ordinal and reads are sound only if both shard keys across the same peer count. Without the flag the process runs a single `Solo` runtime and is byte-unchanged from a deployment that has no second runtime. One controller endpoint still fronts the replica. With two runtimes a `Multiplexer` serves it, routing each command to the owning runtime and merging responses; with one, the maintenance client builder serves it directly. Shared fate is the read-hold mechanism. Both runtimes' worker and reader threads are covered by the process-global panic hook installed at the top of `main`, so a panic on either aborts the whole process. That bounds an interactive import's read hold to the life of the replica without a lease, because there is no way for one runtime to wedge while the other's holds continue. A subprocess test asserts the abort, which cannot be observed from inside the panicking process. `ClusterSpec::cluster_name` lets one process run two clusters of the same kind with distinguishable tracing spans. The interactive runtime takes `compute-interactive`; solo and maintenance keep the bare `compute` so single-runtime logs are unchanged. `enable_compute_interactive_runtime` is replica-scoped but resolved in `environmentd`, because the controller decides `ServiceConfig::ports` before the replica exists and so cannot read the value from the replica's `worker_config`. `CatalogState::replica_scoped_bool` parses the override through the dyncfg rather than `str::parse`, since a stored bool formats as `on`/`off`, which `str::parse::<bool>()` rejects. Off by default in production and in tests. Flipping it changes how a replica is provisioned, so it is not a live toggle: a running replica keeps the layout it was launched with. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent 5c374d3 commit e6d27f4

14 files changed

Lines changed: 708 additions & 53 deletions

File tree

‎Cargo.lock‎

Lines changed: 3 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎misc/python/materialize/mzcompose/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -584,6 +584,9 @@ def get_default_system_parameters(
584584
# all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
585585
# apply.
586586
UNINTERESTING_SYSTEM_PARAMETERS = [
587+
# Registered here rather than varied, because the interactive runtime cannot serve
588+
# index peeks yet. Moves to get_variable_system_parameters once it can.
589+
"enable_compute_interactive_runtime",
587590
"enable_compute_half_join2",
588591
"enable_mz_join_core",
589592
"linear_join_yielding",

‎misc/python/materialize/parallel_workload/action.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3020,6 +3020,8 @@ def __init__(
30203020
BOOLEAN_FLAG_VALUES
30213021
)
30223022
self.flags_with_values["enable_upsert_v2"] = BOOLEAN_FLAG_VALUES
3023+
# Pinned off: the interactive runtime cannot serve index peeks yet.
3024+
self.flags_with_values["enable_compute_interactive_runtime"] = ["FALSE"]
30233025
self.flags_with_values["enable_coalesce_case_transform"] = BOOLEAN_FLAG_VALUES
30243026
self.flags_with_values["enable_compute_sync_mv_sink"] = BOOLEAN_FLAG_VALUES
30253027
self.flags_with_values["enable_column_paged_batcher"] = BOOLEAN_FLAG_VALUES

‎src/adapter/src/catalog/state.rs‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2527,6 +2527,45 @@ impl CatalogState {
25272527
&self.scoped_system_parameters
25282528
}
25292529

2530+
/// Resolves a boolean replica-local override for `replica_id`, falling back to `default` when
2531+
/// the replica has no override for `name`.
2532+
///
2533+
/// For configs consumed when a replica is provisioned rather than on the replica itself. Those
2534+
/// cannot read the value from their own `worker_config`, because the decision is made in
2535+
/// `environmentd` before the replica exists.
2536+
///
2537+
/// Parses through the dyncfg rather than `str::parse`, because a stored override is a var-format
2538+
/// string: `bool` values format as `on`/`off`, which `str::parse::<bool>()` rejects. Parsing it
2539+
/// the wrong way silently resolved `false` for an override every other surface reported as on.
2540+
pub fn replica_scoped_bool(&self, replica_id: ReplicaId, name: &str, default: bool) -> bool {
2541+
let Some(value) = self
2542+
.scoped_system_parameters
2543+
.replica
2544+
.get(&replica_id)
2545+
.and_then(|overrides| overrides.get(name))
2546+
else {
2547+
return default;
2548+
};
2549+
let dyncfgs = self.system_configuration.dyncfgs();
2550+
let parsed = dyncfgs
2551+
.entry(name)
2552+
.and_then(|entry| entry.parse_val(value).ok())
2553+
.and_then(|val| match val {
2554+
mz_dyncfg::ConfigVal::Bool(parsed) => Some(parsed),
2555+
_ => None,
2556+
});
2557+
match parsed {
2558+
Some(parsed) => parsed,
2559+
None => {
2560+
tracing::warn!(
2561+
%name, %value, %replica_id,
2562+
"cannot parse replica-scoped override, falling back to the environment value",
2563+
);
2564+
default
2565+
}
2566+
}
2567+
}
2568+
25302569
/// Return a mutable reference to the current system configuration.
25312570
pub fn system_config_mut(&mut self) -> &mut SystemVars {
25322571
Arc::make_mut(&mut self.system_configuration)

‎src/adapter/src/coord.rs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ use mz_controller::clusters::{
119119
ClusterConfig, ClusterEvent, ClusterStatus, ProcessId, ReplicaLocation,
120120
};
121121
use mz_controller::{ControllerConfig, Readiness};
122+
use mz_controller_types::dyncfgs::ENABLE_COMPUTE_INTERACTIVE_RUNTIME;
122123
use mz_controller_types::{ClusterId, ReplicaId, WatchSetId};
123124
use mz_dyncfg::{ConfigUpdates, ParameterScope};
124125
use mz_expr::{MapFilterProject, MirRelationExpr, OptimizedMirRelationExpr, RowSetFinishing};
@@ -2542,6 +2543,12 @@ impl Coordinator {
25422543
)?;
25432544
for replica in instance.replicas() {
25442545
let role = instance.role();
2546+
let interactive_runtime = self.catalog().state().replica_scoped_bool(
2547+
replica.replica_id,
2548+
ENABLE_COMPUTE_INTERACTIVE_RUNTIME.name(),
2549+
ENABLE_COMPUTE_INTERACTIVE_RUNTIME
2550+
.get(self.catalog().system_config().dyncfgs()),
2551+
);
25452552
self.controller.create_replica(
25462553
instance.id,
25472554
replica.replica_id,
@@ -2551,6 +2558,7 @@ impl Coordinator {
25512558
replica.config.clone(),
25522559
enable_worker_core_affinity,
25532560
enable_storage_introspection_logs,
2561+
interactive_runtime,
25542562
)?;
25552563
}
25562564
}

‎src/adapter/src/coord/catalog_implications.rs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ use mz_cloud_resources::VpcEndpointConfig;
4141
use mz_compute_client::logging::LogVariant;
4242
use mz_compute_client::protocol::response::PeekResponse;
4343
use mz_controller::clusters::{ClusterRole, ReplicaConfig};
44+
use mz_controller_types::dyncfgs::ENABLE_COMPUTE_INTERACTIVE_RUNTIME;
4445
use mz_controller_types::{ClusterId, ReplicaId};
4546
use mz_ore::collections::CollectionExt;
4647
use mz_ore::error::ErrorExt;
@@ -1671,6 +1672,12 @@ impl Coordinator {
16711672
// configuration replays with them. Render-frozen flags make a later push
16721673
// too late, which is why the push precedes `create_replica`.
16731674

1675+
let interactive_runtime = self.catalog().state().replica_scoped_bool(
1676+
replica_id,
1677+
ENABLE_COMPUTE_INTERACTIVE_RUNTIME.name(),
1678+
ENABLE_COMPUTE_INTERACTIVE_RUNTIME.get(self.catalog().system_config().dyncfgs()),
1679+
);
1680+
16741681
self.controller
16751682
.create_replica(
16761683
cluster_id,
@@ -1681,6 +1688,7 @@ impl Coordinator {
16811688
replica_config,
16821689
enable_worker_core_affinity,
16831690
enable_storage_introspection_logs,
1691+
interactive_runtime,
16841692
)
16851693
.expect("creating replicas must not fail");
16861694

‎src/cluster/src/client.rs‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
//! An interactive cluster server.
1111
12+
use std::borrow::Cow;
1213
use std::fmt;
1314
use std::sync::{Arc, Mutex};
1415
use std::thread::Thread;
@@ -167,6 +168,15 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
167168
/// The name of this cluster ("compute" or "storage").
168169
const NAME: &str;
169170

171+
/// The name recorded on the per-worker Timely tracing span.
172+
///
173+
/// Defaults to [`Self::NAME`]. A spec that runs more than one cluster of the same kind in a
174+
/// process (for example the maintenance and interactive compute runtimes) overrides this to
175+
/// keep their spans distinguishable in the logs.
176+
fn cluster_name(&self) -> Cow<'static, str> {
177+
Cow::Borrowed(Self::NAME)
178+
}
179+
170180
/// Run the given Timely worker.
171181
fn run_worker(
172182
&self,
@@ -257,11 +267,12 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
257267
}
258268

259269
let spec = self.clone();
270+
let cluster_name = self.cluster_name();
260271
let worker_guards = execute_from(builders, other, worker_config, move |timely_worker| {
261272
let worker_idx = timely_worker.index();
262273

263274
// Per worker tracing span, lets us identify Timely clusters and workers in the logs.
264-
let span = info_span!("timely", name = Self::NAME, worker_id = worker_idx);
275+
let span = info_span!("timely", name = %cluster_name, worker_id = worker_idx);
265276
let _span_guard = span.enter();
266277

267278
// Every Timely instance in this process names its threads `timely:work-N`, restarting

‎src/clusterd/Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ mz-build-info = { path = "../build-info" }
2323
mz-cloud-resources = { path = "../cloud-resources" }
2424
mz-cluster-client = { path = "../cluster-client" }
2525
mz-compute = { path = "../compute", default-features = false }
26+
mz-compute-client = { path = "../compute-client" }
2627
mz-dyncfgs = { path = "../dyncfgs" }
2728
mz-http-util = { path = "../http-util" }
2829
mz-metrics = { path = "../metrics" }

0 commit comments

Comments
 (0)