feat(asap-tools): feed the sketchdb arm in the ClickHouse runner

asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md

@@ -500,7 +500,7 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`.
 - **Example**: `"--concurrency 32"`
 
 #### `experiment_params.dataset.paced_feeder` (dict, optional; `experiment_run_clickhouse.py`)
-- **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. `sketchdb` mode is refused: its engine ingests `dataset.local_data_file`, and feeding it live is not supported yet (#874). `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`
+- **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`
 - **Example**: `{workload_table: /path/to/table.json, metrics: [data, data_total], seed: 42, duration_secs: 4000}`
 
 ### Query Group Configuration

asap-tools/experiments/experiment_run_clickhouse.py

@@ -91,14 +91,18 @@
 
 A sketch-bench workload table (sketch_bench_workloads.py configs) is fed in
 real time instead, one paced feeder per metric table, when
-experiment_params.dataset.paced_feeder is set.
+experiment_params.dataset.paced_feeder is set. Each mode then recreates the
+tables and starts its own feed. sketchdb's feed goes only to the engine's
+remote-write ingest, so it forwards no queries to ClickHouse, and it needs
+controller.prebuilt_output_dir.
 """
 
 import atexit
+import functools
 import json
 import os
 import time
-from typing import Callable, List, Optional
+from typing import Callable, Dict, List, Optional
 from urllib.parse import urlparse
 
 import hydra
@@ -116,6 +120,7 @@
 from experiment_utils.services.misc import ControllerService, DiscoveryBackend
 from experiment_utils.services.paced_feeder import (
     SINK_CLICKHOUSE,
+    SINK_QUERY_ENGINE,
     PacedFeederService,
     aligned_interval_lcm,
     check_paced_feeder_feed,
@@ -331,14 +336,18 @@ def _check_clickhouse_paced_feeder(
     experiment_modes: List[str],
     client_config: dict,
     minimum_experiment_running_time: int,
+    controller_prebuilt_output_dir: Optional[str] = None,
 ) -> int:
     """Fail before ClickHouse starts on a feeder config the run can't serve;
     returns the seconds the feeders' start must align to."""
-    if constants.SKETCHDB_EXPERIMENT_NAME in experiment_modes:
+    if (
+        constants.SKETCHDB_EXPERIMENT_NAME in experiment_modes
+        and not controller_prebuilt_output_dir
+    ):
         raise ValueError(
-            "dataset.paced_feeder feeds ClickHouse only, and sketchdb mode's "
-            "engine ingests dataset.local_data_file; sketchdb on a paced feed "
-            "is not supported yet"
+            "sketchdb on dataset.paced_feeder needs controller.prebuilt_output_dir: "
+            "the planner's input is built from dataset.precompute, which a fed "
+            "dataset has none of"
         )
     check_paced_feeder_feed(
         feeder_config,
@@ -349,7 +358,18 @@ def _check_clickhouse_paced_feeder(
     return aligned_interval_lcm(client_config)
 
 
-def _start_clickhouse_feed(
+def _feed_sinks(
+    experiment_mode: str, clickhouse_url: str, query_engine_url: str
+) -> Dict[str, str]:
+    """The sinks a mode's feed writes to. sketchdb feeds only ASAPQuery: it
+    forwards no queries, and ClickHouse ingesting a copy would load the node
+    its CPU and latency are measured on. Every other mode reads ClickHouse."""
+    if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
+        return {SINK_QUERY_ENGINE: query_engine_url}
+    return {SINK_CLICKHOUSE: clickhouse_url}
+
+
+def _start_feed(
     *,
     provider,
     node_offset: int,
@@ -359,14 +379,14 @@ def _start_clickhouse_feed(
     feeder_align_secs: int,
     dataset_name: str,
     init_sql_file: str,
-    clickhouse_http_port: int,
+    sinks: Dict[str, str],
     experiment_mode: str,
     experiment_root_output_dir: str,
     local_experiment_root_dir: str,
 ) -> int:
-    """Recreate the metric tables, start one feeder per metric, and point the
-    mode's aligned query groups at the feed's start; returns that start in
-    epoch ms."""
+    """Recreate the metric tables, start one feeder per (sink, metric), and
+    point the mode's aligned query groups at the feed's start; returns that
+    start in epoch ms; `sinks` from _feed_sinks."""
     # An earlier mode's feeders (left running by no_teardown, or with an
     # INSERT in flight) would write into the recreated tables.
     feeder_service.stop()
@@ -379,11 +399,7 @@ def _start_clickhouse_feed(
     data_loader.run_init_sql(dataset_name, init_sql_file)
     start_ms = feeder_service.start(
         config=feeder_config,
-        sinks={
-            # The feeder runs on the next node (see main()).
-            SINK_CLICKHOUSE: f"http://{provider.get_node_ip(node_offset)}:"
-            f"{clickhouse_http_port}/?database={CLICKHOUSE_DATABASE}"
-        },
+        sinks=sinks,
         experiment_output_dir=os.path.join(experiment_root_output_dir, experiment_mode),
         local_experiment_dir=os.path.join(local_experiment_root_dir, experiment_mode),
         align_secs=feeder_align_secs,
@@ -632,19 +648,25 @@ def main(cfg: DictConfig) -> None:
         paced_feeder_config["workload_table"] = os.path.abspath(
             paced_feeder_config["workload_table"]
         )
-        # Every mode's client config has the same query timing.
+        # Every mode but sketchdb runs every query group (sketchdb may skip
+        # trailing ones), so any other mode's client config has them all.
+        all_groups_mode = next(
+            (m for m in experiment_modes if m != constants.SKETCHDB_EXPERIMENT_NAME),
+            experiment_modes[0],
+        )
         with open(
             os.path.join(
                 local_experiment_root_dir,
                 "controller_client_configs",
-                f"{experiment_modes[0]}.yaml",
+                f"{all_groups_mode}.yaml",
             )
         ) as f:
             paced_feeder_align_secs = _check_clickhouse_paced_feeder(
                 paced_feeder_config,
                 experiment_modes,
                 yaml.safe_load(f),
                 minimum_experiment_running_time,
+                controller_prebuilt_output_dir,
             )
         # On its own node, as in experiment_run_e2e.py, so the feeder neither
         # takes ClickHouse's cores nor shows up as its CPU.
@@ -688,7 +710,31 @@ def main(cfg: DictConfig) -> None:
         clickhouse_http_port=clickhouse_http_port,
     )
     remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset)
-    if paced_feeder_service is None:
+    if paced_feeder_service is not None:
+        start_feed = functools.partial(
+            _start_feed,
+            provider=provider,
+            node_offset=node_offset,
+            data_loader=data_loader,
+            feeder_service=paced_feeder_service,
+            feeder_config=paced_feeder_config,
+            feeder_align_secs=paced_feeder_align_secs,
+            dataset_name=dataset_name,
+            init_sql_file=init_sql_file,
+            experiment_root_output_dir=experiment_root_output_dir,
+            local_experiment_root_dir=local_experiment_root_dir,
+        )
+        # The feeder runs on the next node, so it reaches both by node1's IP.
+        feed_target_ip = provider.get_node_ip(node_offset)
+        clickhouse_feed_url = (
+            f"http://{feed_target_ip}:{clickhouse_http_port}/"
+            f"?database={CLICKHOUSE_DATABASE}"
+        )
+        query_engine_url = (
+            f"http://{feed_target_ip}:{int(cfg.streaming.remote_write.base_port)}"
+            f"{cfg.streaming.remote_write.path}"
+        )
+    else:
         ingest_mode = _ingest_mode(experiment_modes)
         ingest_client_config = os.path.join(
             experiment_root_output_dir,
@@ -742,7 +788,8 @@ def main(cfg: DictConfig) -> None:
         os.makedirs(local_experiment_dir, exist_ok=True)
 
         if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
-            # --- sketchdb mode: precompute engine + JSON ingest + ClickHouse fallback ---
+            # --- sketchdb mode: precompute engine + ClickHouse fallback, ingesting
+            # the paced feed (remote write) or local_data_file (JSON) ---
             asap_http_port = urlparse(mode_server_urls[experiment_mode]).port
             # Kill any leftover query_engine_rust from a previous run (mirrors
             # query_engine_service.stop() at the top of the e2e mode loop).
@@ -798,13 +845,27 @@ def generate_controller_output() -> None:
                 generate_controller_output,
             )
 
-            # Start query engine (precompute + JSON ingest + ClickHouse fallback)
+            # Start query engine (precompute + ClickHouse fallback), ingesting
+            # the paced feed over remote write or else local_data_file as JSON
+            remote_write_port = int(cfg.streaming.remote_write.base_port)
+            if paced_feeder_service is not None:
+                ingest_json_config = None
+            else:
+                dataset_precompute_cfg = dataset_cfg.precompute
+                ingest_json_config = {
+                    "path": remote_data_file,
+                    "metric_name": str(dataset_cfg.metric_name),
+                    "value_col": str(dataset_precompute_cfg.value_col),
+                    "label_cols": list(dataset_precompute_cfg.label_cols),
+                    "timestamp_col": str(dataset_precompute_cfg.timestamp_col),
+                    "timestamp_unit": "seconds",
+                    "batch_size": 1000,
+                }
             query_engine_service = QueryEngineRustService(
                 provider=provider,
                 use_container=False,
                 node_offset=node_offset,
             )
-            dataset_precompute_cfg = dataset_cfg.precompute
             query_engine_service.start(
                 experiment_output_dir=experiment_output_dir,
                 local_experiment_dir=local_experiment_dir,
@@ -819,40 +880,48 @@ def generate_controller_output() -> None:
                     "type": "clickhouse",
                     "url": clickhouse_url,
                     "database": CLICKHOUSE_DATABASE,
-                    "forward_unsupported_queries": True,
+                    # A fed run feeds ClickHouse only in the baseline modes,
+                    # so it would answer a forwarded query from no data.
+                    "forward_unsupported_queries": paced_feeder_service is None,
                 },
                 http_port=asap_http_port,
                 spill=cfg.query_engine.spill,
                 retention_slack_ms=config.get_retention_slack_ms(
                     ep.query_groups, data_ingestion_interval_ms
                 ),
-                ingest_json_config={
-                    "path": remote_data_file,
-                    "metric_name": str(dataset_cfg.metric_name),
-                    "value_col": str(dataset_precompute_cfg.value_col),
-                    "label_cols": list(dataset_precompute_cfg.label_cols),
-                    "timestamp_col": str(dataset_precompute_cfg.timestamp_col),
-                    "timestamp_unit": "seconds",
-                    "batch_size": 1000,
-                },
+                remote_write_port=remote_write_port,
+                ingest_json_config=ingest_json_config,
             )
 
             query_engine_service.wait_until_ready()
 
+            if paced_feeder_service is not None:
+                # The engine accepts writes now; the aligned query groups'
+                # starting delay, not steady_state_wait, gives data time to
+                # arrive.
+                start_feed(
+                    experiment_mode=experiment_mode,
+                    sinks=_feed_sinks(
+                        experiment_mode, clickhouse_feed_url, query_engine_url
+                    ),
+                )
+                pre_query_wait_seconds = 0
+            else:
+                pre_query_wait_seconds = int(cfg.flow.steady_state_wait)
+
             controller_client_config = os.path.join(
                 experiment_root_output_dir,
                 "controller_client_configs",
                 f"{experiment_mode}.yaml",
             )
 
             # Single continuous monitor spanning ingest + query: the monitor
-            # starts now (engine is ingesting) and waits steady_state_wait
+            # starts now (engine is ingesting) and waits pre_query_wait_seconds
             # before the query client runs, so one monitor_output.json captures
             # precompute CPU (pc-worker threads) during ingest and query CPU
             # during the query phase, with per-thread attribution.
-            steady_state_wait = int(cfg.flow.steady_state_wait)
             print(
-                f"Monitoring precompute+query CPU; waiting {steady_state_wait}s "
+                f"Monitoring precompute+query CPU; waiting {pre_query_wait_seconds}s "
                 "for ingest before queries..."
             )
             _run_query_workload(
@@ -868,9 +937,11 @@ def generate_controller_output() -> None:
                 query_engine_service=query_engine_service,
                 profile_query_engine=profile_query_engine,
                 profile_prometheus_time=profile_prometheus_time,
-                pre_query_wait_seconds=steady_state_wait,
+                pre_query_wait_seconds=pre_query_wait_seconds,
                 monitor_interval_seconds=monitor_interval_seconds,
             )
+            if paced_feeder_service is not None:
+                _finish_clickhouse_feed(paced_feeder_service)
             spill_error = query_engine_service.record_spill()
 
             sync.rsync_experiment_data(
@@ -890,19 +961,11 @@ def generate_controller_output() -> None:
         else:
             # --- baseline mode ---
             if paced_feeder_service is not None:
-                _start_clickhouse_feed(
-                    provider=provider,
-                    node_offset=node_offset,
-                    data_loader=data_loader,
-                    feeder_service=paced_feeder_service,
-                    feeder_config=paced_feeder_config,
-                    feeder_align_secs=paced_feeder_align_secs,
-                    dataset_name=dataset_name,
-                    init_sql_file=init_sql_file,
-                    clickhouse_http_port=clickhouse_http_port,
+                start_feed(
                     experiment_mode=experiment_mode,
-                    experiment_root_output_dir=experiment_root_output_dir,
-                    local_experiment_root_dir=local_experiment_root_dir,
+                    sinks=_feed_sinks(
+                        experiment_mode, clickhouse_feed_url, query_engine_url
+                    ),
                 )
             controller_client_config = os.path.join(
                 experiment_root_output_dir,