feat(asap-tools): run ClickHouse E2E with frozen plans

asap-tools/experiments/HYDRA_CONFIG_USAGE.md

@@ -216,6 +216,32 @@ clickhouse.database=default
 experiment_params.dataset.max_rows=10000
 ```
 
+#### Run a frozen SQL planner output
+
+Generate the SQL plan separately, then point the ClickHouse sketchdb run at
+the directory containing its `streaming_config.yaml` and
+`inference_config.yaml`. The runner validates that the inference config is
+SQL, copies both files, and does not start `asap-planner`.
+
+```bash
+# Generate the plan in a separate SQL MILP-planning step.
+asap-planner --planner milp --query-language sql \
+  --input_config /path/to/planner_input.yaml \
+  --workload-facts /path/to/workload_facts.yaml \
+  --atomic-costs /path/to/rqe_atomic_costs.json \
+  --saturation-dir /path/to/saturation \
+  --streaming_engine precompute \
+  --data-ingestion-interval-ms 1000 \
+  --output-dir /path/to/frozen-sql-plan
+
+# Execute that exact plan during the ClickHouse evaluation.
+python experiment_run_clickhouse.py \
+  experiment_type=clickhouse \
+  experiment.name=frozen_plan_eval \
+  controller.prebuilt_output_dir=/path/to/frozen-sql-plan \
+  'experiment_params.query_groups[0].sql_file=/path/to/queries.sql'
+```
+
 ### experiment_run_e2e.py (Full E2E Pipeline)
 ```bash
 # Full pipeline with query engine

asap-tools/experiments/experiment_run_clickhouse.py

@@ -93,7 +93,7 @@
 import json
 import os
 import time
-from typing import Optional
+from typing import Callable, Optional
 from urllib.parse import urlparse
 
 import hydra
@@ -123,6 +123,87 @@ def _start_clickhouse_controller(controller_service, cfg, **kwargs):
     controller_service.start(punting=cfg.controller.punting, **kwargs)
 
 
+def _deploy_prebuilt_clickhouse_controller_output(
+    provider,
+    prebuilt_output_dir: str,
+    local_controller_dir: str,
+    remote_controller_dir: str,
+    controller_client_config: str,
+    data_ingestion_interval_ms: int,
+    node_offset: int,
+) -> None:
+    """Validate and deploy a frozen SQL planner output without starting a planner."""
+    config.copy_prebuilt_controller_output(
+        prebuilt_output_dir,
+        local_controller_dir,
+        controller_client_config,
+        data_ingestion_interval_ms,
+        query_language=config.QUERY_LANGUAGE_SQL,
+    )
+    sync.rsync_streaming_configs(
+        provider, local_controller_dir, remote_controller_dir, node_offset
+    )
+
+
+def _prepare_clickhouse_controller_output(
+    prebuilt_output_dir: Optional[str],
+    deploy_prebuilt_output: Callable[[str], None],
+    generate_controller_output: Callable[[], None],
+) -> None:
+    """Deploy a frozen output when configured, otherwise generate one."""
+    if prebuilt_output_dir:
+        deploy_prebuilt_output(prebuilt_output_dir)
+    else:
+        generate_controller_output()
+
+
+def _generate_clickhouse_controller_output(
+    provider,
+    cfg,
+    query_groups,
+    dataset_cfg,
+    local_controller_dir: str,
+    remote_controller_dir: str,
+    node_offset: int,
+    clickhouse_url: str,
+    data_ingestion_interval_ms: int,
+) -> None:
+    """Generate and retrieve the SQL planner output for a ClickHouse run."""
+    planner_input_yaml = config.generate_sql_planner_input(
+        query_groups,
+        dataset_cfg,
+        cfg.get("sketch_parameters", None),
+        cfg.get("windowing", None),
+    )
+    local_planner_input = os.path.join(local_controller_dir, "planner_input.yaml")
+    with open(local_planner_input, "w") as planner_input_file:
+        planner_input_file.write(planner_input_yaml)
+    sync.rsync_streaming_configs(
+        provider, local_controller_dir, remote_controller_dir, node_offset
+    )
+
+    controller_service = ControllerService(
+        provider=provider, use_container=False, node_offset=node_offset
+    )
+    _start_clickhouse_controller(
+        controller_service,
+        cfg,
+        controller_input_file=os.path.join(remote_controller_dir, "planner_input.yaml"),
+        streaming_engine="precompute",
+        controller_remote_output_dir=remote_controller_dir,
+        discovery_backend=DiscoveryBackend(
+            type="clickhouse",
+            url=clickhouse_url,
+            database=CLICKHOUSE_DATABASE,
+        ),
+        query_language=config.QUERY_LANGUAGE_SQL,
+        data_ingestion_interval_ms=data_ingestion_interval_ms,
+    )
+    sync.rsync_controller_config_remote_to_local(
+        provider, remote_controller_dir, local_controller_dir, node_offset
+    )
+
+
 def _inline_sql_queries_in_experiment_config(local_experiment_root_dir: str) -> None:
     """Enrich the saved experiment_params.yaml by inlining SQL from sql_file references.
 
@@ -352,6 +433,7 @@ def main(cfg: DictConfig) -> None:
     profile_query_engine = bool(cfg.profiling.query_engine)
     profile_prometheus_time = cfg.profiling.prometheus_time
     monitor_interval_seconds = float(cfg.flow.monitor_interval_seconds)
+    controller_prebuilt_output_dir = cfg.controller.get("prebuilt_output_dir")
     minimum_experiment_running_time = config.get_minimum_experiment_running_time(
         cfg.experiment_params
     )
@@ -524,44 +606,41 @@ def main(cfg: DictConfig) -> None:
             )
             os.makedirs(local_controller_dir, exist_ok=True)
 
-            # Generate and rsync the planner input config to the node
-            planner_input_yaml = config.generate_sql_planner_input(
-                ep.query_groups,
-                dataset_cfg,
-                cfg.get("sketch_parameters", None),
-                cfg.get("windowing", None),
-            )
-            local_planner_input = os.path.join(
-                local_controller_dir, "planner_input.yaml"
-            )
-            with open(local_planner_input, "w") as _f:
-                _f.write(planner_input_yaml)
-            sync.rsync_streaming_configs(
-                provider, local_controller_dir, remote_controller_dir, node_offset
+            controller_client_config = os.path.join(
+                local_experiment_root_dir,
+                "controller_client_configs",
+                f"{experiment_mode}.yaml",
             )
 
-            # Run asap-planner (SQL mode) — writes streaming_config.yaml + inference_config.yaml
-            controller_service = ControllerService(
-                provider=provider, use_container=False, node_offset=node_offset
-            )
-            _start_clickhouse_controller(
-                controller_service,
-                cfg,
-                controller_input_file=os.path.join(
-                    remote_controller_dir, "planner_input.yaml"
-                ),
-                streaming_engine="precompute",
-                controller_remote_output_dir=remote_controller_dir,
-                discovery_backend=DiscoveryBackend(
-                    type="clickhouse",
-                    url=clickhouse_url,
-                    database=CLICKHOUSE_DATABASE,
-                ),
-                query_language="sql",
-                data_ingestion_interval_ms=data_ingestion_interval_ms,
-            )
-            sync.rsync_controller_config_remote_to_local(
-                provider, remote_controller_dir, local_controller_dir, node_offset
+            def deploy_prebuilt_output(prebuilt_output_dir: str) -> None:
+                print(f"Skipping the planner: using configs from {prebuilt_output_dir}")
+                _deploy_prebuilt_clickhouse_controller_output(
+                    provider,
+                    prebuilt_output_dir,
+                    local_controller_dir,
+                    remote_controller_dir,
+                    controller_client_config,
+                    data_ingestion_interval_ms,
+                    node_offset,
+                )
+
+            def generate_controller_output() -> None:
+                _generate_clickhouse_controller_output(
+                    provider,
+                    cfg,
+                    ep.query_groups,
+                    dataset_cfg,
+                    local_controller_dir,
+                    remote_controller_dir,
+                    node_offset,
+                    clickhouse_url,
+                    data_ingestion_interval_ms,
+                )
+
+            _prepare_clickhouse_controller_output(
+                controller_prebuilt_output_dir,
+                deploy_prebuilt_output,
+                generate_controller_output,
             )
 
             # Start query engine (precompute + JSON ingest + ClickHouse fallback)

asap-tools/experiments/experiment_utils/config.py

@@ -539,6 +539,8 @@ def get_prometheus_data_ingestion_interval_ms(prometheus_config):
 
 
 PLANNER_OUTPUT_FILES = ("streaming_config.yaml", "inference_config.yaml")
+QUERY_LANGUAGE_PROMQL = "promql"
+QUERY_LANGUAGE_SQL = "sql"
 
 
 def check_prebuilt_controller_output(src):
@@ -558,8 +560,36 @@ def _is_planned(query, planned):
     return re.fullmatch(r"[\s()+\-*/0-9.]*", query) is not None
 
 
+def _validate_sql_inference_config(inference_config):
+    """Raise a clear error unless `inference_config` has the SQL schema shape."""
+    if not isinstance(inference_config, dict):
+        raise ValueError(
+            "controller.prebuilt_output_dir inference_config.yaml is not an SQL "
+            "inference config"
+        )
+    tables = inference_config.get("tables")
+    if not isinstance(tables, list):
+        raise ValueError(
+            "controller.prebuilt_output_dir inference_config.yaml is not an SQL "
+            "inference config: missing tables"
+        )
+    for table in tables:
+        if not isinstance(table, dict) or not all(
+            isinstance(table.get(key), list if key.endswith("columns") else str)
+            for key in ("name", "time_column", "value_columns", "metadata_columns")
+        ):
+            raise ValueError(
+                "controller.prebuilt_output_dir inference_config.yaml is not an SQL "
+                "inference config: invalid table schema"
+            )
+
+
 def copy_prebuilt_controller_output(
-    src, dest, controller_input_config, scrape_interval_ms
+    src,
+    dest,
+    controller_input_config,
+    scrape_interval_ms,
+    query_language=QUERY_LANGUAGE_PROMQL,
 ):
     """Replaces `dest` with the planner configs in `src`, after checking they
     plan every query of `controller_input_config` at this scrape interval."""
@@ -571,7 +601,16 @@ def copy_prebuilt_controller_output(
             f"dir {dest} must not contain one another"
         )
     with open(os.path.join(src, "inference_config.yaml")) as f:
-        planned = {q["query"] for q in yaml.safe_load(f)["queries"]}
+        inference_config = yaml.safe_load(f)
+    if query_language == QUERY_LANGUAGE_SQL:
+        _validate_sql_inference_config(inference_config)
+    if not isinstance(inference_config, dict) or not isinstance(
+        inference_config.get("queries"), list
+    ):
+        raise ValueError(
+            "controller.prebuilt_output_dir inference_config.yaml has invalid queries"
+        )
+    planned = {q["query"] for q in inference_config["queries"]}
     with open(controller_input_config) as f:
         groups = yaml.safe_load(f)["query_groups"]
     unplanned = [q for g in groups for q in g["queries"] if not _is_planned(q, planned)]

asap-tools/experiments/tests/test_experiment_run_clickhouse.py

@@ -1,11 +1,15 @@
 """Regression tests for ClickHouse experiment controller wiring."""
 
 import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 from omegaconf import OmegaConf
 
-from experiment_run_clickhouse import _start_clickhouse_controller
+from experiment_run_clickhouse import (
+    _deploy_prebuilt_clickhouse_controller_output,
+    _prepare_clickhouse_controller_output,
+    _start_clickhouse_controller,
+)
 
 
 class ClickHouseControllerConfigTest(unittest.TestCase):
@@ -26,6 +30,56 @@ def test_controller_receives_configured_punting_setting(self):
             streaming_engine="precompute",
         )
 
+    @patch("experiment_run_clickhouse.sync.rsync_streaming_configs")
+    @patch("experiment_run_clickhouse.config.copy_prebuilt_controller_output")
+    def test_prebuilt_output_is_validated_and_deployed(
+        self, copy_prebuilt_controller_output, rsync_streaming_configs
+    ):
+        provider = Mock()
+
+        _deploy_prebuilt_clickhouse_controller_output(
+            provider,
+            "/plans/frozen",
+            "/tmp/local-controller",
+            "/remote/controller",
+            "/tmp/client.yaml",
+            1000,
+            2,
+        )
+
+        copy_prebuilt_controller_output.assert_called_once_with(
+            "/plans/frozen",
+            "/tmp/local-controller",
+            "/tmp/client.yaml",
+            1000,
+            query_language="sql",
+        )
+        rsync_streaming_configs.assert_called_once_with(
+            provider, "/tmp/local-controller", "/remote/controller", 2
+        )
+
+    def test_prebuilt_path_does_not_generate_a_controller_output(self):
+        deploy_prebuilt_output = Mock()
+        generate_controller_output = Mock()
+
+        _prepare_clickhouse_controller_output(
+            "/plans/frozen", deploy_prebuilt_output, generate_controller_output
+        )
+
+        deploy_prebuilt_output.assert_called_once_with("/plans/frozen")
+        generate_controller_output.assert_not_called()
+
+    def test_unset_prebuilt_path_generates_a_controller_output(self):
+        deploy_prebuilt_output = Mock()
+        generate_controller_output = Mock()
+
+        _prepare_clickhouse_controller_output(
+            None, deploy_prebuilt_output, generate_controller_output
+        )
+
+        generate_controller_output.assert_called_once_with()
+        deploy_prebuilt_output.assert_not_called()
+
 
 if __name__ == "__main__":
     unittest.main()