Skip to content

Commit a32a396

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 ae38969 commit a32a396

14 files changed

Lines changed: 446 additions & 50 deletions

File tree

‎Cargo.lock‎

Lines changed: 2 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
@@ -662,6 +662,9 @@ def get_default_system_parameters(
662662
# all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
663663
# apply.
664664
UNINTERESTING_SYSTEM_PARAMETERS = [
665+
# Registered here rather than varied, because the interactive runtime cannot serve
666+
# index peeks yet. Moves to get_variable_system_parameters once it can.
667+
"enable_compute_interactive_runtime",
665668
"enable_compute_half_join2",
666669
"enable_mz_join_core",
667670
"linear_join_yielding",

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3134,6 +3134,8 @@ def __init__(
31343134
BOOLEAN_FLAG_VALUES
31353135
)
31363136
self.flags_with_values["enable_upsert_v2"] = BOOLEAN_FLAG_VALUES
3137+
# Pinned off: the interactive runtime cannot serve index peeks yet.
3138+
self.flags_with_values["enable_compute_interactive_runtime"] = ["FALSE"]
31373139
self.flags_with_values["enable_coalesce_case_transform"] = BOOLEAN_FLAG_VALUES
31383140
self.flags_with_values["enable_any_all_null_array_semantics"] = (
31393141
BOOLEAN_FLAG_VALUES

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

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ use mz_controller::clusters::{
3939
ManagedReplicaLocation, ReplicaAllocation, ReplicaLocation, UnmanagedReplicaLocation,
4040
};
4141
use mz_controller_types::{ClusterId, ReplicaId};
42+
use mz_dyncfg::{Config, ConfigDefault, ConfigType};
4243
use mz_expr::{CollectionPlan, OptimizedMirRelationExpr};
4344
use mz_license_keys::ValidatedLicenseKey;
4445
use mz_ore::collections::CollectionExt;
@@ -2527,6 +2528,46 @@ impl CatalogState {
25272528
&self.scoped_system_parameters
25282529
}
25292530

2531+
/// Resolves `config`'s replica-local override for `replica_id`, falling back to its environment
2532+
/// value when the replica has no override.
2533+
///
2534+
/// For configs consumed when a replica is provisioned rather than on the replica itself. Those
2535+
/// cannot read the value from their own `worker_config`, because the decision is made in
2536+
/// `environmentd` before the replica exists.
2537+
///
2538+
/// Parses through `ConfigType` rather than the value type's own `FromStr`, because a stored
2539+
/// override is a var-format string: `bool` formats as `on`/`off`, which `str::parse::<bool>()`
2540+
/// rejects. Parsing it the wrong way silently resolved `false` for an override every other
2541+
/// surface reported as on.
2542+
pub fn replica_scoped<D: ConfigDefault>(
2543+
&self,
2544+
replica_id: ReplicaId,
2545+
config: &Config<D>,
2546+
) -> D::ConfigType {
2547+
// The environment value, not `config.default()`: a config flipped in production is served
2548+
// from the `ConfigSet`, and the compile-time default would silently ignore that flip.
2549+
let environment = config.get(self.system_configuration.dyncfgs());
2550+
let name = config.name();
2551+
let Some(value) = self
2552+
.scoped_system_parameters
2553+
.replica
2554+
.get(&replica_id)
2555+
.and_then(|overrides| overrides.get(name))
2556+
else {
2557+
return environment;
2558+
};
2559+
match D::ConfigType::parse(value) {
2560+
Ok(parsed) => parsed,
2561+
Err(error) => {
2562+
tracing::warn!(
2563+
%name, %value, %replica_id, %error,
2564+
"cannot parse replica-scoped override, falling back to the environment value",
2565+
);
2566+
environment
2567+
}
2568+
}
2569+
}
2570+
25302571
/// Return a mutable reference to the current system configuration.
25312572
pub fn system_config_mut(&mut self) -> &mut SystemVars {
25322573
Arc::make_mut(&mut self.system_configuration)

‎src/adapter/src/coord.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,7 @@ use mz_controller::clusters::{
121121
ClusterConfig, ClusterEvent, ClusterStatus, ManagedReplicaLocation, ProcessId, ReplicaLocation,
122122
};
123123
use mz_controller::{ControllerConfig, Readiness};
124+
use mz_controller_types::dyncfgs::ENABLE_COMPUTE_INTERACTIVE_RUNTIME;
124125
use mz_controller_types::{ClusterId, ReplicaId, WatchSetId};
125126
use mz_dyncfg::{ConfigUpdates, ParameterScope};
126127
use mz_expr::{MapFilterProject, MirRelationExpr, OptimizedMirRelationExpr, RowSetFinishing};
@@ -2656,6 +2657,10 @@ impl Coordinator {
26562657
)?;
26572658
for replica in instance.replicas() {
26582659
let role = instance.role();
2660+
let interactive_runtime = self
2661+
.catalog()
2662+
.state()
2663+
.replica_scoped(replica.replica_id, &ENABLE_COMPUTE_INTERACTIVE_RUNTIME);
26592664
self.controller.create_replica(
26602665
instance.id,
26612666
replica.replica_id,
@@ -2665,6 +2670,7 @@ impl Coordinator {
26652670
replica.config.clone(),
26662671
enable_worker_core_affinity,
26672672
enable_storage_introspection_logs,
2673+
interactive_runtime,
26682674
)?;
26692675
}
26702676
}

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

Lines changed: 7 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::{PeekError, 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;
@@ -1727,6 +1728,11 @@ impl Coordinator {
17271728
// configuration replays with them. Render-frozen flags make a later push
17281729
// too late, which is why the push precedes `create_replica`.
17291730

1731+
let interactive_runtime = self
1732+
.catalog()
1733+
.state()
1734+
.replica_scoped(replica_id, &ENABLE_COMPUTE_INTERACTIVE_RUNTIME);
1735+
17301736
self.controller
17311737
.create_replica(
17321738
cluster_id,
@@ -1737,6 +1743,7 @@ impl Coordinator {
17371743
replica_config,
17381744
enable_worker_core_affinity,
17391745
enable_storage_introspection_logs,
1746+
interactive_runtime,
17401747
)
17411748
.expect("creating replicas must not fail");
17421749

‎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)