diff --git a/asap-tools/experiments/HYDRA_CONFIG_USAGE.md b/asap-tools/experiments/HYDRA_CONFIG_USAGE.md index e7cb3eb3..85bd6c66 100644 --- a/asap-tools/experiments/HYDRA_CONFIG_USAGE.md +++ b/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 diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index faf14952..7280945d 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/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) diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index 5f8ce768..d0efd7ce 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/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)] diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index d39f8c25..6bbf4b60 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/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() diff --git a/asap-tools/experiments/tests/test_prebuilt_controller_output.py b/asap-tools/experiments/tests/test_prebuilt_controller_output.py index b140933c..c11d242a 100644 --- a/asap-tools/experiments/tests/test_prebuilt_controller_output.py +++ b/asap-tools/experiments/tests/test_prebuilt_controller_output.py @@ -9,6 +9,7 @@ from experiment_utils.config import copy_prebuilt_controller_output QUERY = "sum by (job) (data)" +SQL_QUERY = "SELECT count() FROM hits" class CopyPrebuiltControllerOutputTest(unittest.TestCase): @@ -41,6 +42,15 @@ def copy(self, src=None, dest=None, scrape_interval_ms=1000): src or self.src, dest or self.dest, self.client, scrape_interval_ms ) + def copy_sql(self, src=None, dest=None, scrape_interval_ms=1000): + copy_prebuilt_controller_output( + src or self.src, + dest or self.dest, + self.client, + scrape_interval_ms, + query_language="sql", + ) + def test_replaces_stale_output(self): os.makedirs(self.dest) with open(os.path.join(self.dest, "controller.log"), "w") as f: @@ -82,6 +92,42 @@ def test_refuses_another_scrape_interval(self): with self.assertRaisesRegex(ValueError, "15000 ms scrape interval"): self.copy(scrape_interval_ms=15000) + def test_sql_output_requires_a_sql_schema(self): + with self.assertRaisesRegex(ValueError, "not an SQL inference config"): + self.copy_sql() + + def test_sql_output_with_a_sql_schema_is_copied(self): + self.write(self.client, {"query_groups": [{"queries": [SQL_QUERY]}]}) + self.write( + os.path.join(self.src, "inference_config.yaml"), + { + "tables": [ + { + "name": "hits", + "time_column": "EventTime", + "value_columns": ["ResolutionWidth"], + "metadata_columns": [], + } + ], + "queries": [{"query": SQL_QUERY}], + }, + ) + + self.copy_sql() + + self.assertTrue( + os.path.isfile(os.path.join(self.dest, "inference_config.yaml")) + ) + + def test_sql_output_refuses_an_invalid_table_schema(self): + self.write( + os.path.join(self.src, "inference_config.yaml"), + {"tables": [{}], "queries": [{"query": QUERY}]}, + ) + + with self.assertRaisesRegex(ValueError, "invalid table schema"): + self.copy_sql() + if __name__ == "__main__": unittest.main()