Skip to content

Commit 6c58a8b

Browse files
feat(asap-tools): feed only ASAPQuery in fed sketchdb runs
A ClickHouse copy of the feed during sketchdb only served queries ASAPQuery forwards, and loaded the node sketchdb's CPU and latency are measured on. Fed sketchdb now feeds the engine alone and forwards no unsupported queries, so an unplanned query fails instead of being answered by ClickHouse. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 7866734 commit 6c58a8b

3 files changed

Lines changed: 64 additions & 48 deletions

File tree

‎asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -500,7 +500,7 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`.
500500
- **Example**: `"--concurrency 32"`
501501

502502
#### `experiment_params.dataset.paced_feeder` (dict, optional; `experiment_run_clickhouse.py`)
503-
- **Description**: The same keys as `exporters.exporter_list.paced_feeder`, for the ClickHouse runner. In place of the bulk load of `dataset.local_data_file`, each mode runs `dataset.init_sql_file`, then feeds every listed metric (records metrics too) from the ClickHouse node into the table named after it while its queries run. In `sketchdb` mode the engine ingests over remote write (`streaming.remote_write.base_port`/`path`) instead of `dataset.local_data_file`, and the same feed (one start, one seed) also goes to ClickHouse, which answers the queries ASAPQuery forwards; it waits for no `flow.steady_state_wait` and needs `controller.prebuilt_output_dir` (a fed dataset has no `dataset.precompute` to build the planner's input from). `duration_secs` must cover the startup check, the query client's startup (30 s) and the query groups' running time; after the queries each mode waits for the feeders to finish, so they write their `--stats-out`. Requires `dataset.name: custom` and `dataset.init_sql_file`
503+
- **Description**: The same keys as `exporters.exporter_list.paced_feeder`, for the ClickHouse runner. In place of the bulk load of `dataset.local_data_file`, each mode runs `dataset.init_sql_file`, then feeds every listed metric (records metrics too) from the ClickHouse node into the table named after it while its queries run. In `sketchdb` mode the engine ingests over remote write (`streaming.remote_write.base_port`/`path`) instead of `dataset.local_data_file`, and the feed goes to ASAPQuery only, so it forwards no unsupported queries to ClickHouse (whose copy of the data would load the measured node); it waits for no `flow.steady_state_wait` and needs `controller.prebuilt_output_dir` (a fed dataset has no `dataset.precompute` to build the planner's input from). `duration_secs` must cover the startup check, the query client's startup (30 s) and the query groups' running time; after the queries each mode waits for the feeders to finish, so they write their `--stats-out`. Requires `dataset.name: custom` and `dataset.init_sql_file`
504504
- **Example**: `{workload_table: /path/to/table.json, metrics: [data, data_total], seed: 42, duration_secs: 4000}`
505505

506506
### Query Group Configuration

‎asap-tools/experiments/experiment_run_clickhouse.py‎

Lines changed: 38 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -92,17 +92,17 @@
9292
A sketch-bench workload table (sketch_bench_workloads.py configs) is fed in
9393
real time instead, one paced feeder per metric table, when
9494
experiment_params.dataset.paced_feeder is set. Each mode then recreates the
95-
tables and starts its own feed; sketchdb's feed goes to the engine's
96-
remote-write ingest as well as ClickHouse (which answers forwarded queries),
97-
from one start, and needs controller.prebuilt_output_dir.
95+
tables and starts its own feed. sketchdb's feed goes only to the engine's
96+
remote-write ingest, so it forwards no queries to ClickHouse, and it needs
97+
controller.prebuilt_output_dir.
9898
"""
9999

100100
import atexit
101101
import functools
102102
import json
103103
import os
104104
import time
105-
from typing import Callable, List, Optional
105+
from typing import Callable, Dict, List, Optional
106106
from urllib.parse import urlparse
107107

108108
import hydra
@@ -358,7 +358,21 @@ def _check_clickhouse_paced_feeder(
358358
return aligned_interval_lcm(client_config)
359359

360360

361-
def _start_clickhouse_feed(
361+
def _feed_sinks(
362+
experiment_mode: str, clickhouse_http_port: int, query_engine_url: str
363+
) -> Dict[str, str]:
364+
"""The sinks a mode's feed writes to. sketchdb feeds only ASAPQuery: it
365+
forwards no queries, and a ClickHouse copy would only load the node its
366+
CPU and latency are measured on. Every other mode reads ClickHouse."""
367+
if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
368+
return {SINK_QUERY_ENGINE: query_engine_url}
369+
return {
370+
SINK_CLICKHOUSE: f"http://localhost:{clickhouse_http_port}/"
371+
f"?database={CLICKHOUSE_DATABASE}"
372+
}
373+
374+
375+
def _start_feed(
362376
*,
363377
provider,
364378
node_offset: int,
@@ -368,20 +382,14 @@ def _start_clickhouse_feed(
368382
feeder_align_secs: int,
369383
dataset_name: str,
370384
init_sql_file: str,
371-
clickhouse_http_port: int,
385+
sinks: Dict[str, str],
372386
experiment_mode: str,
373387
experiment_root_output_dir: str,
374388
local_experiment_root_dir: str,
375-
query_engine_url: Optional[str] = None,
376389
) -> int:
377390
"""Recreate the metric tables, start one feeder per (sink, metric), and
378391
point the mode's aligned query groups at the feed's start; returns that
379-
start in epoch ms.
380-
381-
ClickHouse is always fed; with `query_engine_url` (sketchdb) so is
382-
ASAPQuery's remote-write ingest, from the same start and seed, so the
383-
queries it forwards to ClickHouse read the same data.
384-
"""
392+
start in epoch ms; `sinks` from _feed_sinks."""
385393
# An earlier mode's feeders (left running by no_teardown, or with an
386394
# INSERT in flight) would write into the recreated tables.
387395
feeder_service.stop()
@@ -392,12 +400,6 @@ def _start_clickhouse_feed(
392400
)
393401
# Before any data arrives: views only see rows inserted after they exist.
394402
data_loader.run_init_sql(dataset_name, init_sql_file)
395-
sinks = {
396-
SINK_CLICKHOUSE: f"http://localhost:{clickhouse_http_port}/"
397-
f"?database={CLICKHOUSE_DATABASE}"
398-
}
399-
if query_engine_url is not None:
400-
sinks[SINK_QUERY_ENGINE] = query_engine_url
401403
start_ms = feeder_service.start(
402404
config=feeder_config,
403405
sinks=sinks,
@@ -707,7 +709,7 @@ def main(cfg: DictConfig) -> None:
707709
remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset)
708710
if paced_feeder_service is not None:
709711
start_feed = functools.partial(
710-
_start_clickhouse_feed,
712+
_start_feed,
711713
provider=provider,
712714
node_offset=node_offset,
713715
data_loader=data_loader,
@@ -716,10 +718,13 @@ def main(cfg: DictConfig) -> None:
716718
feeder_align_secs=paced_feeder_align_secs,
717719
dataset_name=dataset_name,
718720
init_sql_file=init_sql_file,
719-
clickhouse_http_port=clickhouse_http_port,
720721
experiment_root_output_dir=experiment_root_output_dir,
721722
local_experiment_root_dir=local_experiment_root_dir,
722723
)
724+
query_engine_url = (
725+
f"http://localhost:{int(cfg.streaming.remote_write.base_port)}"
726+
f"{cfg.streaming.remote_write.path}"
727+
)
723728
else:
724729
ingest_mode = _ingest_mode(experiment_modes)
725730
ingest_client_config = os.path.join(
@@ -866,7 +871,9 @@ def generate_controller_output() -> None:
866871
"type": "clickhouse",
867872
"url": clickhouse_url,
868873
"database": CLICKHOUSE_DATABASE,
869-
"forward_unsupported_queries": True,
874+
# A fed run feeds ClickHouse only in the baseline modes,
875+
# so it would answer a forwarded query from no data.
876+
"forward_unsupported_queries": paced_feeder_service is None,
870877
},
871878
http_port=asap_http_port,
872879
spill=cfg.query_engine.spill,
@@ -885,8 +892,9 @@ def generate_controller_output() -> None:
885892
# arrive.
886893
feed_start_ms = start_feed(
887894
experiment_mode=experiment_mode,
888-
query_engine_url=f"http://localhost:{remote_write_port}"
889-
f"{cfg.streaming.remote_write.path}",
895+
sinks=_feed_sinks(
896+
experiment_mode, clickhouse_http_port, query_engine_url
897+
),
890898
)
891899
pre_query_wait_seconds = 0
892900
else:
@@ -948,7 +956,12 @@ def generate_controller_output() -> None:
948956
else:
949957
# --- baseline mode ---
950958
if paced_feeder_service is not None:
951-
feed_start_ms = start_feed(experiment_mode=experiment_mode)
959+
feed_start_ms = start_feed(
960+
experiment_mode=experiment_mode,
961+
sinks=_feed_sinks(
962+
experiment_mode, clickhouse_http_port, query_engine_url
963+
),
964+
)
952965
controller_client_config = os.path.join(
953966
experiment_root_output_dir,
954967
"controller_client_configs",

‎asap-tools/experiments/tests/test_experiment_run_clickhouse.py‎

Lines changed: 25 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@
1818
_prepare_clickhouse_controller_output,
1919
_start_clickhouse_controller,
2020
_finish_clickhouse_feed,
21-
_start_clickhouse_feed,
21+
_feed_sinks,
22+
_start_feed,
2223
)
2324
from experiment_utils.config import validate_experiment_config
2425
from experiment_utils.services.paced_feeder import PACED_FEEDER_STARTUP_CHECK_SECS
@@ -163,8 +164,25 @@ def test_feed_must_outlast_the_query_client_startup(self):
163164
)
164165

165166

166-
class StartClickHouseFeedTest(unittest.TestCase):
167-
def start(self, calls, exited=True, mode="baseline", query_engine_url=None):
167+
class FeedSinksTest(unittest.TestCase):
168+
def test_baseline_modes_feed_clickhouse(self):
169+
for mode in ("baseline", "baseline_mv"):
170+
self.assertEqual(
171+
_feed_sinks(mode, 8123, "http://localhost:8080/api/v1/write"),
172+
{"clickhouse": "http://localhost:8123/?database=default"},
173+
)
174+
175+
def test_sketchdb_feeds_only_the_engine(self):
176+
# It forwards no queries, so a ClickHouse copy would only load the
177+
# node its CPU and latency are measured on.
178+
self.assertEqual(
179+
_feed_sinks("sketchdb", 8123, "http://localhost:8080/api/v1/write"),
180+
{"query_engine": "http://localhost:8080/api/v1/write"},
181+
)
182+
183+
184+
class StartFeedTest(unittest.TestCase):
185+
def start(self, calls, exited=True, mode="baseline"):
168186
calls.feeder.start.return_value = 1700000060000
169187
calls.feeder.wait_for_exit.return_value = exited
170188
with tempfile.TemporaryDirectory() as root, patch(
@@ -176,7 +194,7 @@ def start(self, calls, exited=True, mode="baseline", query_engine_url=None):
176194
with open(config_path, "w") as f:
177195
yaml.safe_dump(aligned_client_config(60), f)
178196

179-
start_ms = _start_clickhouse_feed(
197+
start_ms = _start_feed(
180198
provider=Mock(),
181199
node_offset=0,
182200
data_loader=calls.loader,
@@ -185,11 +203,10 @@ def start(self, calls, exited=True, mode="baseline", query_engine_url=None):
185203
feeder_align_secs=60,
186204
dataset_name="custom",
187205
init_sql_file="init.sql",
188-
clickhouse_http_port=8123,
206+
sinks={"clickhouse": "http://localhost:8123/?database=default"},
189207
experiment_mode=mode,
190208
experiment_root_output_dir="/exp",
191209
local_experiment_root_dir=root,
192-
query_engine_url=query_engine_url,
193210
)
194211
with open(config_path) as f:
195212
self.client_config = yaml.safe_load(f)
@@ -223,24 +240,10 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self):
223240
)
224241
self.rsync.assert_called_once()
225242

226-
def test_sketchdb_feeds_the_engine_and_clickhouse_from_one_start(self):
227-
# ClickHouse answers the queries ASAPQuery forwards, so it gets the
228-
# same data from the same start.
243+
def test_sketchdb_queries_follow_its_own_feed(self):
229244
calls = Mock()
230-
self.start(
231-
calls,
232-
mode="sketchdb",
233-
query_engine_url="http://localhost:8080/api/v1/write",
234-
)
235-
calls.feeder.start.assert_called_once()
245+
self.start(calls, mode="sketchdb")
236246
start_kwargs = calls.feeder.start.call_args.kwargs
237-
self.assertEqual(
238-
start_kwargs["sinks"],
239-
{
240-
"clickhouse": "http://localhost:8123/?database=default",
241-
"query_engine": "http://localhost:8080/api/v1/write",
242-
},
243-
)
244247
self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/sketchdb")
245248
self.assertEqual(
246249
self.client_config["query_groups"][0]["client_options"]["data_start_time"],

0 commit comments

Comments
 (0)