Skip to content

Commit 4daa449

Browse files
antiguruclaude
andcommitted
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. 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>
1 parent b288248 commit 4daa449

8 files changed

Lines changed: 271 additions & 10 deletions

File tree

misc/python/materialize/mzcompose/__init__.py

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -351,6 +351,11 @@ def get_variable_system_parameters(
351351
"true",
352352
["true", "false"],
353353
),
354+
VariableSystemParameter(
355+
"enable_compute_interactive_runtime",
356+
"true",
357+
["true", "false"],
358+
),
354359
VariableSystemParameter(
355360
"enable_upsert_v2",
356361
"false",
@@ -649,9 +654,6 @@ def get_default_system_parameters(
649654
# all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
650655
# apply.
651656
UNINTERESTING_SYSTEM_PARAMETERS = [
652-
# Registered here rather than varied, because the interactive runtime cannot serve
653-
# index peeks yet. Moves to get_variable_system_parameters once it can.
654-
"enable_compute_interactive_runtime",
655657
"enable_compute_half_join2",
656658
"enable_mz_join_core",
657659
"linear_join_yielding",

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

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ def __init__(
3939
workers: int = 1,
4040
process_names: list[str] = [],
4141
mz_service: str = "materialized",
42+
interactive_compute: bool = False,
4243
) -> None:
4344
environment = [
4445
"CLUSTERD_LOG_FILTER",
@@ -78,6 +79,21 @@ def __init__(
7879
f"CLUSTERD_STORAGE_TIMELY_CONFIG={storage_timely_config}",
7980
]
8081

82+
# When set, clusterd runs a second, interactive compute runtime alongside the
83+
# maintenance one (see `--interactive-compute-timely-config` in
84+
# `src/clusterd/src/lib.rs`). It must span the same number of Timely peers as
85+
# the maintenance compute config, so it reuses `process_names`/`workers`; its
86+
# addresses use a distinct port (2104) so the two runtimes don't collide.
87+
ports = [2100, 2101, 6878]
88+
if interactive_compute:
89+
interactive_compute_timely_config = timely_config(
90+
process_names, 2104, workers, DEFAULT_COMPUTE_EXERT_PROPORTIONALITY
91+
)
92+
environment += [
93+
f"CLUSTERD_INTERACTIVE_COMPUTE_TIMELY_CONFIG={interactive_compute_timely_config}"
94+
]
95+
ports += [2104]
96+
8197
options = ["clusterd", f"--scratch-directory={scratch_directory}", *options]
8298

8399
config: ServiceConfig = {}
@@ -106,7 +122,7 @@ def __init__(
106122
config.update(
107123
{
108124
"command": options,
109-
"ports": [2100, 2101, 6878],
125+
"ports": ports,
110126
"environment": environment,
111127
"volumes": volumes or DEFAULT_MZ_VOLUMES,
112128
"restart": restart,

misc/python/materialize/parallel_workload/action.py

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3094,8 +3094,9 @@ def __init__(
30943094
BOOLEAN_FLAG_VALUES
30953095
)
30963096
self.flags_with_values["enable_upsert_v2"] = BOOLEAN_FLAG_VALUES
3097-
# Pinned off: the interactive runtime cannot serve index peeks yet.
3098-
self.flags_with_values["enable_compute_interactive_runtime"] = ["FALSE"]
3097+
self.flags_with_values["enable_compute_interactive_runtime"] = (
3098+
BOOLEAN_FLAG_VALUES
3099+
)
30993100
self.flags_with_values["enable_coalesce_case_transform"] = BOOLEAN_FLAG_VALUES
31003101
self.flags_with_values["enable_any_all_null_array_semantics"] = (
31013102
BOOLEAN_FLAG_VALUES

test/clusterd-test-driver/mzcompose.py

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,14 @@ def __init__(self, name: str = "headless-driver") -> None:
9898
"create_time_config.spec",
9999
]
100100

101+
# Scenarios run against a clusterd with the interactive compute runtime enabled
102+
# (see `workflow_two_runtime_compute`). Kept separate from `SCRIPTS` so the
103+
# single-runtime scenarios above stay byte-for-byte unchanged.
104+
TWO_RUNTIME_SCRIPTS = [
105+
"two_runtime_index.spec",
106+
"two_runtime_query_dataflow.spec",
107+
]
108+
101109

102110
def workflow_default(c: Composition) -> None:
103111
c.up(METADATA_STORE, "minio", ServiceName("headless-driver", idle=True))
@@ -117,3 +125,32 @@ def workflow_default(c: Composition) -> None:
117125
env_extra={"DRIVER_SCRIPT": f"{SCRIPTS_DIR}/{script}"},
118126
use_aliases=True,
119127
)
128+
# Also exercise every other workflow (the two-runtime compute path) so the
129+
# CI job, which runs this composition's default workflow, covers them too.
130+
for name in c.workflows:
131+
if name == "default":
132+
continue
133+
with c.test_case(name):
134+
c.workflow(name)
135+
136+
137+
def workflow_two_runtime_compute(c: Composition) -> None:
138+
"""Run the two-runtime scenarios against a clusterd started with a second,
139+
interactive compute runtime configured. `Clusterd(interactive_compute=True)`
140+
sets `CLUSTERD_INTERACTIVE_COMPUTE_TIMELY_CONFIG`, which turns on the
141+
multiplexer that fronts both runtimes on the same `:2101` endpoint this
142+
driver already connects to, so no driver changes are needed to reach the
143+
interactive runtime.
144+
"""
145+
c.up(METADATA_STORE, "minio", ServiceName("headless-driver", idle=True))
146+
with c.override(Clusterd(mz_service="headless-driver", interactive_compute=True)):
147+
for i, script in enumerate(TWO_RUNTIME_SCRIPTS):
148+
ui.section(f"Running two-runtime scenario {script}")
149+
if i > 0:
150+
c.kill("clusterd")
151+
c.up("clusterd")
152+
c.run(
153+
"headless-driver",
154+
env_extra={"DRIVER_SCRIPT": f"{SCRIPTS_DIR}/{script}"},
155+
use_aliases=True,
156+
)
Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
# Copyright Materialize, Inc. and contributors. All rights reserved.
2+
#
3+
# Use of this software is governed by the Business Source License
4+
# included in the LICENSE file at the root of this repository.
5+
#
6+
# As of the Change Date specified in that file, in accordance with
7+
# the Business Source License, use of this software will be governed
8+
# by the Apache License, Version 2.0.
9+
10+
# two_runtime_index scenario: runs against a clusterd started with a second,
11+
# interactive compute runtime configured (see `Clusterd(interactive_compute=True)`
12+
# in the `two_runtime_compute` workflow). A multiplexer inside clusterd fronts both
13+
# runtimes on the single `:2101` endpoint this driver connects to, so the commands
14+
# below are unchanged from a single-runtime script; only the clusterd underneath
15+
# differs. The maintenance runtime renders and publishes the index below into the
16+
# shared arrangement-sharing registry; the multiplexer routes the `peek` to the
17+
# interactive runtime regardless, so a correct result proves the interactive
18+
# runtime served the rows from that shared registry rather than rendering its own
19+
# copy of the index.
20+
create-instance
21+
----
22+
ok
23+
24+
update-configuration
25+
----
26+
ok
27+
28+
initialization-complete
29+
----
30+
ok
31+
32+
write-rows shard=r ts=0
33+
1 alpha
34+
2 beta
35+
3 gamma
36+
----
37+
wrote 3
38+
39+
create-dataflow name=two-runtime-index as-of=0
40+
import source=1000 shard=r upper=1
41+
build id=2000
42+
Project (#0, #1)
43+
Get u1000
44+
export kind=index index=2001 on=2000 key=[0]
45+
----
46+
ok
47+
48+
schedule id=2001
49+
----
50+
ok
51+
52+
# Routed to the interactive runtime by the multiplexer.
53+
peek id=2001 ts=0
54+
----
55+
1 "alpha"
56+
2 "beta"
57+
3 "gamma"
Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
# Copyright Materialize, Inc. and contributors. All rights reserved.
2+
#
3+
# Use of this software is governed by the Business Source License
4+
# included in the LICENSE file at the root of this repository.
5+
#
6+
# As of the Change Date specified in that file, in accordance with
7+
# the Business Source License, use of this software will be governed
8+
# by the Apache License, Version 2.0.
9+
10+
# two_runtime_query_dataflow scenario: the acceptance test for the interactive
11+
# slow path (N1-N3). Unlike two_runtime_index.spec, which peeks a maintenance
12+
# index directly (the fast path), this creates a genuine INTERACTIVE QUERY
13+
# DATAFLOW: a `count(*)` reduce that imports a maintenance index and is itself
14+
# exported under a transient id (`t4000`), so the multiplexer routes its
15+
# `CreateDataflow` to the interactive runtime (see `mz_compute_client::multiplex`).
16+
#
17+
# The maintenance index is created but deliberately left unscheduled before the
18+
# query dataflow is submitted, so nothing has been published to the shared
19+
# arrangement-sharing registry yet. The interactive runtime builds the query
20+
# dataflow immediately anyway, binding its import to a real but empty publication
21+
# point (a placeholder) rather than deferring the build, which would break the
22+
# deterministic construction order every worker must follow. The query's own
23+
# `schedule` is then sent immediately, before the maintenance index is scheduled,
24+
# mirroring what the real compute controller always does for a transient
25+
# collection (`Instance::maybe_schedule_collection`: scheduled right away, without
26+
# waiting for its inputs). Only afterward is the maintenance index scheduled, which
27+
# renders it and adopts the placeholder in place; that publication (never a bare
28+
# poll) wakes the import, which begins producing. The result peek then returns the
29+
# correct reduced rows, proving the bind -> fill -> resolve path completed off the
30+
# maintenance worker.
31+
create-instance
32+
----
33+
ok
34+
35+
update-configuration
36+
----
37+
ok
38+
39+
initialization-complete
40+
----
41+
ok
42+
43+
write-rows shard=r ts=0
44+
1 alpha
45+
2 beta
46+
3 gamma
47+
----
48+
wrote 3
49+
50+
# The maintenance index over shard `r`. Registering it lets `import index=2001`
51+
# below reference it; its dataflow is intentionally left unscheduled, so no
52+
# arrangement is published yet.
53+
create-dataflow name=maint-index as-of=0
54+
import source=1000 shard=r upper=1
55+
build id=2000
56+
Project (#0, #1)
57+
Get u1000
58+
export kind=index index=2001 on=2000 key=[0]
59+
----
60+
ok
61+
62+
# A one-column schema for the count reduce's output (a single bigint).
63+
define-schema name=count_out
64+
count bigint
65+
----
66+
ok
67+
68+
# The interactive query dataflow: `count(*)` over the (still unpublished)
69+
# maintenance index, exported under a transient id (`t4000`) so it is routed to
70+
# the interactive runtime instead of maintenance.
71+
create-dataflow name=interactive-count as-of=0
72+
import index=2001
73+
build id=3000
74+
Reduce aggregates=[count(*)]
75+
Get u2000
76+
export index=t4000 on=3000 key=[0]
77+
----
78+
ok
79+
80+
# Scheduled before the maintenance index: the dataflow is built but its import is
81+
# still an empty placeholder at this point, so this `Schedule` races ahead of the
82+
# publication, exactly as the real controller's immediate-schedule-of-transient
83+
# behavior would.
84+
schedule id=t4000
85+
----
86+
ok
87+
88+
# Scheduling the maintenance index renders it and adopts the placeholder, waking
89+
# the interactive query dataflow's import via the publication notification.
90+
schedule id=2001
91+
----
92+
ok
93+
94+
# The interactive runtime ran the reduce once notified of the publication, off the
95+
# maintenance worker. The result is the correct count.
96+
peek id=t4000 schema=count_out ts=0
97+
----
98+
3

0 commit comments

Comments
 (0)