diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 6486063e..7cd49acc 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -31,6 +31,7 @@ pub struct ClickHousePlannedQuery { pub canonical: QueryExpr, pub canonical_sql: String, pub physical: PhysicalExpr, + pub selection_trace: serde_json::Value, } pub async fn plan_clickhouse_sql( @@ -45,24 +46,29 @@ pub async fn plan_clickhouse_sql( // summary-capable Aggregate. Use ASAPPlanner's recursive selector here; // the PromQL deployment lowering retains its existing conservative rules. let cost_model = ControlPlaneCostModel::new(accuracy.clone()); - let selected = crate::planner_selection::select_workload( - vec![(0, Rc::new(canonical.clone()))], - accuracy, - &cost_model, - )? - .into_iter() - .next() - .map(|(_, node)| node) - .ok_or_else(|| { - crate::planner_selection::SelectionError::Workload( - "SQL workload search returned no root".into(), - ) - })?; + let (selected, selection_trace) = + crate::planner_selection::select_workload_with_accuracy_model_and_trace( + vec![(0, Rc::new(canonical.clone()))], + accuracy, + &cost_model, + &asap_aware_mapping::NoAccuracyEvidence, + &asap_aware_mapping::DefaultAccuracyModel, + )?; + let selected = selected + .into_iter() + .next() + .map(|(_, node)| node) + .ok_or_else(|| { + crate::planner_selection::SelectionError::Workload( + "SQL workload search returned no root".into(), + ) + })?; let physical = PhysicalExpr::committed(selected); Ok(ClickHousePlannedQuery { canonical_sql: canonical_sql_identity(&canonical), canonical, physical, + selection_trace, }) } @@ -103,6 +109,157 @@ pub struct ClickHouseSqlWorkloadEntry { pub cumulative: bool, } +/// Workload input without predeclared summary families or materialization IDs. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ClickHouseSqlAutomaticWorkload { + pub envelope: asap_types::precompute_plan::PlanEnvelope, + pub tables: HashMap, + pub accuracy: AccuracyTarget, + pub queries: Vec, +} + +pub async fn compile_automatic_clickhouse_workload( + request: &ClickHouseSqlAutomaticWorkload, +) -> Result< + ( + crate::physical::publication::PhysicalPlanPublication, + std::collections::BTreeMap, + ), + ClickHousePlanningError, +> { + let catalog = SqlCatalog { + tables: request.tables.clone(), + }; + let mut entries = std::collections::BTreeMap::new(); + let mut installed_dags = std::collections::BTreeMap::new(); + let mut materializations = std::collections::BTreeMap::new(); + let mut selection_traces = std::collections::BTreeMap::new(); + for query in &request.queries { + validate_sql_evaluation(query)?; + // Selection runs once. Compilation installs only SummaryAgg nodes + // actually visited in this selected DAG, never a scripted family. + let mut planned = + plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?; + selection_traces.insert( + planned.canonical_sql.clone(), + std::mem::take(&mut planned.selection_trace), + ); + let (entry, installed) = compile_selected_sql(query, planned, |node, family| { + let config = materialize_selected_sql(node, family, query) + .map_err(crate::query_plan::QueryPlanError::Invalid)?; + let binding = MaterializationBinding { + materialization: config.policy_fingerprint().into(), + output_grouping: PhysicalGrouping::Reduce(config.grouping_labels.labels.clone()), + window_ms: config.slide_interval * 1000, + pane_origin_ms: config.pane_origin_ms, + readout_lookback_ms: Some(query.end_ms - query.start_ms), + item_labels: config.aggregated_labels.labels.clone(), + }; + materializations + .entry(config.policy_fingerprint()) + .or_insert(config); + Ok(binding) + })?; + let key = QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &entry.canonical_query); + if entries.insert(key, entry).is_some() { + return Err(ClickHousePlanningError::Lower( + "duplicate canonical SQL query identity".into(), + )); + } + installed_dags.insert(query.sql.clone(), installed); + } + let configs: Vec<_> = materializations.into_values().collect(); + let sds = SummaryCatalog::from_materializations( + request.envelope.plan_id, + request.envelope.plan_version, + &configs, + ) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + let mut precompute = PrecomputePlan::build_backend_local(request.envelope.clone(), configs) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + precompute.summary_catalog = Some( + sds.reference() + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?, + ); + precompute.executable_dags = installed_dags; + let mut transmission = TransmissionPlan::build( + request.envelope.clone(), + &precompute, + &std::collections::BTreeMap::new(), + ) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + transmission.summary_catalog = precompute.summary_catalog.clone(); + let publication = crate::physical::publication::PhysicalPlanPublication { + summary_catalog: sds, + precompute_plan: precompute, + collector_plans: Vec::new(), + transmission_plan: transmission, + query_plan: QueryPlan { + plan_id: request.envelope.plan_id, + plan_version: request.envelope.plan_version, + clickhouse_context: Some(ClickHousePlanningContext { + tables: request.tables.clone(), + accuracy: request.accuracy.clone(), + }), + entries, + }, + }; + publication + .validate() + .map_err(ClickHousePlanningError::Lower)?; + Ok((publication, selection_traces)) +} + +fn materialize_selected_sql( + node: &planner_types::post_asap::SummaryNode, + family: &planner_types::post_asap::SummaryFamilyType, + query: &ClickHouseSqlWorkloadEntry, +) -> Result { + use crate::physical::colored_dag::emitter::{AggregationInput, BackendAggregation}; + use planner_types::{post_asap::SummaryExpr, pre_asap::Reduction}; + let SummaryExpr::SummaryAgg { + reduction: Reduction::Reduce(keys), + .. + } = &node.expr + else { + return Err("SQL materialization requires a supported reduction".into()); + }; + if keys.is_without() || !keys.keys().is_empty() { + return Err("SQL grouped source projection requires a typed grouping reader".into()); + } + let (table, value, window, population, timestamp) = + clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)?; + let window_secs = window.ok_or("SQL materialization requires a bounded window")?; + let aggregation = BackendAggregation { + aggregation_id: String::new(), + metric_name: format!("{table}.{value}"), + family: crate::physical::compiler::physical_materialization_family(family), + window_secs, + spatial_filter: String::new(), + grouping: Vec::new(), + item_label: None, + heap_update_mode: None, + aggregation_input: AggregationInput::Raw, + }; + let mut config = crate::physical::compiler::aggregation_config_for_materialization( + &aggregation, + asap_types::QueryLanguage::ClickHouseSql, + ) + .map_err(|error| error.to_string())?; + config.table_name = Some(table); + config.value_column = Some(value); + config.table_timestamp_column = Some(timestamp); + config.table_population = Some(population); + config.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped); + config.pane_origin_ms = Some( + i64::try_from(query.start_ms) + .map_err(|_| "SQL evaluation timestamp exceeds runtime range")?, + ); + config.num_aggregates_to_retain = Some(2); + Ok(config) +} + pub async fn compile_clickhouse_workload( request: &ClickHouseSqlWorkload, ) -> Result { @@ -121,70 +278,12 @@ pub async fn compile_clickhouse_workload( let mut installed_dags = std::collections::BTreeMap::new(); for query in &request.queries { let planned = plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?; - let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = - planned.physical - else { - return Err(ClickHousePlanningError::Lower( - "SQL did not produce a summary DAG".into(), - )); - }; - let semantic = planner_types::post_asap::compile_executable_dag_with_node_ids(&root) - .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; - let mut materialization_nodes = std::collections::BTreeMap::new(); - let mut query_nodes = std::collections::BTreeMap::new(); - let executable = QueryPlanEntry::compile_bound_relational_mapped( - query.sql.clone(), - planned.canonical_sql.clone(), - &root, - FixedEvaluationRange { - start_ms: query.start_ms, - end_ms: query.end_ms, - cumulative: query.cumulative, - }, - InstantExecution { - lookback_ms: query.end_ms.saturating_sub(query.start_ms), - full_history: query.start_ms == 0, - cumulative_readout: query.cumulative, - }, - FallbackPolicy::ExactBackend, - |node, family| { - let binding = bind_selected_node(node, family, query, request)?; - let id = semantic.node_ids.node_id(node).ok_or_else(|| { - crate::query_plan::QueryPlanError::Invalid( - "selected SQL node is absent from semantic DAG".into(), - ) - })?; - materialization_nodes.insert(id, binding.materialization); - Ok(binding) - }, - |node, query_node| { - if let Some(id) = semantic.node_ids.node_id(node) { - query_nodes.insert(id, query_node); - } - }, - ) - .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; - let installed = crate::physical::executable_binding::install_selected_dag( - query.sql.clone(), - &semantic.dag, - executable.root, - |id| materialization_nodes.get(&id).copied(), - |id| query_nodes.get(&id).copied(), - ) - .map_err(ClickHousePlanningError::Lower)?; - crate::physical::executable_binding::validate_query_plan(&installed, &executable) - .map_err(ClickHousePlanningError::Lower)?; + let (executable, installed) = compile_selected_sql(query, planned, |node, family| { + bind_selected_node(node, family, query, request) + })?; installed_dags.insert(query.sql.clone(), installed); - if executable - .nodes - .values() - .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. })) - { - return Err(ClickHousePlanningError::Lower( - "compiled SQL contains an unsupported operator; publication refused".into(), - )); - } - let identity = QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &planned.canonical_sql); + let identity = + QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &executable.canonical_query); if entries.insert(identity.clone(), executable).is_some() { return Err(ClickHousePlanningError::Lower(format!( "duplicate canonical SQL query identity `{identity}`" @@ -214,6 +313,100 @@ pub async fn compile_clickhouse_workload( Ok(publication) } +fn compile_selected_sql( + query: &ClickHouseSqlWorkloadEntry, + planned: ClickHousePlannedQuery, + mut bind: F, +) -> Result< + ( + QueryPlanEntry, + crate::physical::executable_binding::InstalledPostAsapDag, + ), + ClickHousePlanningError, +> +where + F: FnMut( + &Rc, + &planner_types::post_asap::SummaryFamilyType, + ) -> Result, +{ + validate_sql_evaluation(query)?; + let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = + planned.physical + else { + return Err(ClickHousePlanningError::Lower( + "SQL did not produce a summary DAG".into(), + )); + }; + let semantic = planner_types::post_asap::compile_executable_dag_with_node_ids(&root) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + let mut materialization_nodes = std::collections::BTreeMap::new(); + let mut query_nodes = std::collections::BTreeMap::new(); + let executable = QueryPlanEntry::compile_bound_relational_mapped( + query.sql.clone(), + planned.canonical_sql.clone(), + &root, + FixedEvaluationRange { + start_ms: query.start_ms, + end_ms: query.end_ms, + cumulative: query.cumulative, + }, + InstantExecution { + lookback_ms: query.end_ms.saturating_sub(query.start_ms), + full_history: query.start_ms == 0, + cumulative_readout: query.cumulative, + }, + FallbackPolicy::ExactBackend, + |node, family| { + let binding = bind(node, family)?; + let id = semantic.node_ids.node_id(node).ok_or_else(|| { + crate::query_plan::QueryPlanError::Invalid( + "selected SQL node is absent from semantic DAG".into(), + ) + })?; + materialization_nodes.insert(id, binding.materialization); + Ok(binding) + }, + |node, query_node| { + if let Some(id) = semantic.node_ids.node_id(node) { + query_nodes.insert(id, query_node); + } + }, + ) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + let installed = crate::physical::executable_binding::install_selected_dag( + query.sql.clone(), + &semantic.dag, + executable.root, + |id| materialization_nodes.get(&id).copied(), + |id| query_nodes.get(&id).copied(), + ) + .map_err(ClickHousePlanningError::Lower)?; + crate::physical::executable_binding::validate_query_plan(&installed, &executable) + .map_err(ClickHousePlanningError::Lower)?; + if executable + .nodes + .values() + .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. })) + { + return Err(ClickHousePlanningError::Lower( + "compiled SQL contains an unsupported operator; publication refused".into(), + )); + } + Ok((executable, installed)) +} + +fn validate_sql_evaluation( + query: &ClickHouseSqlWorkloadEntry, +) -> Result<(), ClickHousePlanningError> { + if query.start_ms >= query.end_ms || query.end_ms > i64::MAX as u64 { + return Err(ClickHousePlanningError::Lower( + "SQL evaluation requires start_ms < end_ms within signed Unix milliseconds".into(), + )); + } + Ok(()) +} + fn bind_selected_node( node: &planner_types::post_asap::SummaryNode, family: &planner_types::post_asap::SummaryFamilyType, @@ -228,7 +421,7 @@ fn bind_selected_node( &request.precompute_plan.materializations, &table_ref, &value_column, - &spatial_filter, + &spatial_filter.canonical(), &expected, source_window.unwrap_or((query.end_ms.saturating_sub(query.start_ms)) / 1000), )?; @@ -251,7 +444,16 @@ fn clickhouse_materialization_leaf_contract( node: &planner_types::post_asap::SummaryNode, evaluation_start_ms: u64, evaluation_end_ms: u64, -) -> Result<(String, String, Option, String, String), String> { +) -> Result< + ( + String, + String, + Option, + asap_types::table_population::TablePopulation, + String, + ), + String, +> { use planner_types::{ post_asap::SummaryExpr, pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source}, @@ -410,7 +612,7 @@ fn clickhouse_materialization_leaf_contract( table_ref.to_owned(), value_column, Some(window_secs), - population.canonical(), + population, schema .time_index .and_then(|index| schema.columns.get(index)) @@ -574,6 +776,19 @@ mod tests { ); } + #[test] + fn sql_evaluation_rejects_empty_reversed_and_unrepresentable_ranges() { + for (start_ms, end_ms) in [(2, 1), (1, 1), (0, u64::MAX)] { + assert!(validate_sql_evaluation(&ClickHouseSqlWorkloadEntry { + sql: "SELECT sum(value) FROM telemetry".into(), + start_ms, + end_ms, + cumulative: true, + }) + .is_err()); + } + } + #[tokio::test] async fn compiles_summary_joined_with_exact_table_into_mixed_dag() { let config = materialization( @@ -638,6 +853,41 @@ mod tests { }], }; let publication = compile_clickhouse_workload(&request).await.unwrap(); + let (automatic, traces) = + compile_automatic_clickhouse_workload(&ClickHouseSqlAutomaticWorkload { + envelope: request.precompute_plan.envelope.clone(), + tables: request.tables.clone(), + accuracy: request.accuracy.clone(), + queries: request + .queries + .iter() + .map(|query| ClickHouseSqlWorkloadEntry { + sql: query.sql.clone(), + start_ms: query.start_ms, + end_ms: query.end_ms, + cumulative: query.cumulative, + }) + .collect(), + }) + .await + .unwrap(); + assert_eq!(automatic.precompute_plan.materializations.len(), 1); + assert_eq!(traces.len(), 1); + assert!(traces + .values() + .any( + |trace| trace["groups"] + .as_array() + .unwrap() + .iter() + .any(|group| group["candidates"] + .as_array() + .unwrap() + .iter() + .any(|candidate| candidate["selected"] == true)) + )); + assert_eq!(automatic.precompute_plan.executable_dags.len(), 1); + automatic.validate().unwrap(); let installed = publication .precompute_plan .executable_dags diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index df8033ea..d3f4861d 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -464,7 +464,12 @@ async fn push_cumulative_entries( let materializations = match cumulative_be .aggregations .iter() - .map(crate::physical::compiler::aggregation_config_for_materialization) + .map(|aggregation| { + crate::physical::compiler::aggregation_config_for_materialization( + aggregation, + asap_types::QueryLanguage::PromQl, + ) + }) .collect::>>() { Ok(materializations) => materializations, diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 5674a178..a7923638 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -567,6 +567,10 @@ async fn main() { "/api/v1/clickhouse-plan/compile-and-publish", post(handle_compile_and_publish_clickhouse_plan), ) + .route( + "/api/v1/clickhouse-plan/automatic/compile-and-publish", + post(handle_compile_and_publish_automatic_clickhouse_plan), + ) .route("/api/v1/plan/auto", post(handle_plan_auto)) .route("/api/v1/plan/pareto", post(handle_pareto)) .route("/api/v1/plan/:metric", get(handle_get_plan)) @@ -818,6 +822,26 @@ async fn handle_compile_and_publish_clickhouse_plan( Ok(publication) => publication, Err(error) => return (StatusCode::BAD_REQUEST, error.to_string()).into_response(), }; + publish_clickhouse_plan(&state, publication, None).await +} + +async fn handle_compile_and_publish_automatic_clickhouse_plan( + State(state): State, + Json(request): Json, +) -> impl IntoResponse { + let (publication, trace) = + match clickhouse::compile_automatic_clickhouse_workload(&request).await { + Ok(publication) => publication, + Err(error) => return (StatusCode::BAD_REQUEST, error.to_string()).into_response(), + }; + publish_clickhouse_plan(&state, publication, Some(serde_json::json!(trace))).await +} + +async fn publish_clickhouse_plan( + state: &AppState, + publication: physical::publication::PhysicalPlanPublication, + selection_trace: Option, +) -> axum::response::Response { let plan_id = publication.summary_catalog.plan_id; let plan_version = publication.summary_catalog.plan_version; let Some(client) = state.backend_client.as_ref() else { @@ -842,7 +866,8 @@ async fn handle_compile_and_publish_clickhouse_plan( Json(serde_json::json!({ "plan_id": plan_id, "plan_version": plan_version, - "status": "active" + "status": "active", + "selection_trace": selection_trace, })) .into_response() } diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 5de2565f..d37872e8 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1871,8 +1871,10 @@ impl PhysicalCompiler { aggregation_id.clone(), environment.target, ); - let precompute_materialization = - aggregation_config_for_materialization(&aggregation)?; + let precompute_materialization = aggregation_config_for_materialization( + &aggregation, + asap_types::QueryLanguage::PromQl, + )?; let materialization = precompute_materialization.policy_fingerprint(); let state_consumers = consumers[&materialization] .iter() @@ -1903,8 +1905,10 @@ impl PhysicalCompiler { // Preserve semantic window and evaluation cadence independently // from the selected storage representation. aggregation.window_secs = window_implementation.window_secs; - let mut runtime_materialization = - aggregation_config_for_materialization(&aggregation)?; + let mut runtime_materialization = aggregation_config_for_materialization( + &aggregation, + asap_types::QueryLanguage::PromQl, + )?; runtime_materialization.window_size = window_implementation.window_secs; runtime_materialization.slide_interval = window_implementation.slide_secs; runtime_materialization.window_type = @@ -3188,12 +3192,22 @@ fn physical_aggregation( /// not create a second registry or wire plan. pub(crate) fn aggregation_config_for_materialization( aggregation: &BackendAggregation, + language: asap_types::QueryLanguage, ) -> anyhow::Result { use anyhow::Context as _; - let json = crate::emit::stage_config::build_backend_aggregation_json(aggregation); - let text = serde_json::to_string(&json).context("serialize synthesized aggregation JSON")?; - let yaml: serde_yaml::Value = - serde_yaml::from_str(&text).context("parse synthesized aggregation JSON as YAML")?; + let mut json = crate::emit::stage_config::build_backend_aggregation_json(aggregation); + // The selected physical duration is authoritative. The legacy edge emitter's + // 5..60 second clamp must not silently change a backend materialization. + json["windowSize"] = serde_json::json!(aggregation.window_secs); + if language == asap_types::QueryLanguage::ClickHouseSql { + // Raw input does not imply PromQL's (start,end] convention. SQL bounds + // are normalized and bound explicitly by the SQL compiler. + json["parameters"] + .as_object_mut() + .expect("emitter parameters are an object") + .remove("promql_right_closed"); + } + let yaml = serde_yaml::to_value(json).context("convert physical aggregation fields")?; asap_types::PrecomputeMaterialization::from_yaml_data( &yaml, None, @@ -3227,12 +3241,10 @@ fn materialization_consumers( { continue; } - let config = aggregation_config_for_materialization(&physical_aggregation( - query, - &state, - query.query_id.clone(), - target, - ))?; + let config = aggregation_config_for_materialization( + &physical_aggregation(query, &state, query.query_id.clone(), target), + asap_types::QueryLanguage::PromQl, + )?; consumers .entry(config.policy_fingerprint()) .or_default() diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index 3dfd3297..c30ca612 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -193,6 +193,38 @@ pub fn select_workload_with_accuracy_model( cost_model: &dyn CostModel, evidence: &dyn AccuracyEvidenceProvider, accuracy_model: &dyn AccuracyModel, +) -> Result)>, SelectionError> { + select_workload_impl(roots, accuracy, cost_model, evidence, accuracy_model, None) +} + +/// Return the candidate ranking and committed choices from the same search +/// that produces the installed roots. Missing numeric costs remain explicit. +pub fn select_workload_with_accuracy_model_and_trace( + roots: Vec<(usize, Rc)>, + accuracy: AccuracyTarget, + cost_model: &dyn CostModel, + evidence: &dyn AccuracyEvidenceProvider, + accuracy_model: &dyn AccuracyModel, +) -> Result<(Vec<(usize, Rc)>, serde_json::Value), SelectionError> { + let mut trace = serde_json::Value::Null; + let selected = select_workload_impl( + roots, + accuracy, + cost_model, + evidence, + accuracy_model, + Some(&mut trace), + )?; + Ok((selected, trace)) +} + +fn select_workload_impl( + roots: Vec<(usize, Rc)>, + accuracy: AccuracyTarget, + cost_model: &dyn CostModel, + evidence: &dyn AccuracyEvidenceProvider, + accuracy_model: &dyn AccuracyModel, + trace: Option<&mut serde_json::Value>, ) -> Result)>, SelectionError> { // Canonical CSE still runs inside search_workload_with_targets. Do not // offer CSE's per-invocation recompute alternative: this runtime currently @@ -218,6 +250,30 @@ pub fn select_workload_with_accuracy_model( accuracy_model, ); let selection = space.global_selection(cost_model); + if let Some(trace) = trace { + let groups = space.cost_sorted(cost_model).iter().enumerate().map(|(index, group)| { + let chosen = selection.groups().find(|selected| Rc::ptr_eq(selected.target, group.target)) + .and_then(|selected| selected.chosen); + let candidates = group.candidates.iter().zip(&group.costs).enumerate() + .map(|(rank, (candidate, cost))| serde_json::json!({ + "rank": rank, + "strategy": candidate.strategy, + "provenance": format!("{:?}", candidate.provenance), + "rationale": candidate.rationale, + "replacement_kind": match &candidate.replacement { + Replacement::Summary(_) => "summary", + Replacement::Rewrite(_) => "rewrite", + Replacement::ExactComposition(_) => "exact_composition", + }, + "estimated_cost": cost.is_finite().then_some(*cost), + "estimated_cost_status": if cost.is_finite() { "available" } else { "not_reported_by_cost_model" }, + "selected": chosen.is_some_and(|chosen| std::ptr::eq(chosen, *candidate)), + })).collect::>(); + serde_json::json!({ "group_id": index, "consumer_count": group.consumer_count, + "candidates": candidates }) + }).collect::>(); + *trace = serde_json::json!({ "schema_version": 1, "group_id_scope": "this_selection", "groups": groups }); + } let roots = space .roots .iter() diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index 84de688e..9089fe47 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -3,7 +3,7 @@ //! Set `CLICKHOUSE_URL` (for example `http://127.0.0.1:8123`) to run it. use std::{ - collections::{BTreeMap, HashMap}, + collections::HashMap, io::Write, net::TcpListener, process::{Child, Command, Stdio}, @@ -51,50 +51,9 @@ async fn wait_http(client: &reqwest::Client, url: &str, child: &mut Child) { panic!("data plane did not become ready at {url}"); } -fn mixed_workload( - sql: &str, -) -> ( - control_plane::clickhouse::ClickHouseSqlWorkload, - asap_types::PrecomputeMaterialization, -) { - use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; - use control_plane::physical::compiler::{ - PlanEnvelope, PrecomputePlan, TransmissionPlan, BACKEND_COMPAT, PLANNER_REVISION, - }; +fn mixed_workload(sql: &str) -> control_plane::clickhouse::ClickHouseSqlAutomaticWorkload { + use control_plane::physical::compiler::{PlanEnvelope, BACKEND_COMPAT, PLANNER_REVISION}; use planner_types::pre_asap::{Column, DataType, Schema}; - - let mut config = PrecomputeMaterialization::new( - AggregationType::Sum, - String::new(), - HashMap::from([("variant".into(), serde_json::json!(1))]), - KeyByLabelNames::empty(), - KeyByLabelNames::empty(), - KeyByLabelNames::empty(), - String::new(), - 2, - 2, - WindowKind::Tumbling, - String::new(), - "telemetry.value".into(), - None, - Some("telemetry".into()), - Some("value".into()), - ); - config.pane_origin_ms = Some(0); - config.table_timestamp_column = Some("timestamp_ms".into()); - config.table_population = Some(asap_types::table_population::TablePopulation { - predicates: vec![asap_types::table_population::TableColumnPredicate { - column: "metric".into(), - operator: planner_types::pre_asap::CompareOpKind::Eq, - value: planner_types::pre_asap::ScalarValue::Utf8("requests".into()), - }], - }); - let sds = asap_types::summary_catalog::SummaryCatalog::from_materializations( - 72, - 1, - &[config.clone()], - ) - .unwrap(); let envelope = PlanEnvelope { plan_id: 72, plan_version: 1, @@ -105,12 +64,6 @@ fn mixed_workload( planner_revision: PLANNER_REVISION.into(), capability_snapshot_id: "clickhouse-mixed-process-e2e".into(), }; - let mut precompute = - PrecomputePlan::build_backend_local(envelope.clone(), vec![config.clone()]).unwrap(); - precompute.summary_catalog = Some(sds.reference().unwrap()); - let mut transmission = - TransmissionPlan::build(envelope, &precompute, &BTreeMap::new()).unwrap(); - transmission.summary_catalog = Some(sds.reference().unwrap()); let schema = |time: &str, value: &str| { Schema::with_time_index( vec![ @@ -122,35 +75,30 @@ fn mixed_workload( vec![], ) }; - ( - control_plane::clickhouse::ClickHouseSqlWorkload { - sds, - precompute_plan: precompute, - transmission_plan: transmission, - tables: HashMap::from([ - ("telemetry".into(), schema("timestamp_ms", "value")), - ( - "divisors".into(), - Schema::with_time_index( - vec![ - Column::new("timestamp", DataType::Int64, false), - Column::new("divisor", DataType::Float64, false), - ], - 0, - vec![], - ), + control_plane::clickhouse::ClickHouseSqlAutomaticWorkload { + envelope, + tables: HashMap::from([ + ("telemetry".into(), schema("timestamp_ms", "value")), + ( + "divisors".into(), + Schema::with_time_index( + vec![ + Column::new("timestamp", DataType::Int64, false), + Column::new("divisor", DataType::Float64, false), + ], + 0, + vec![], ), - ]), - accuracy: planner_types::types::AccuracyTarget::Exact, - queries: vec![control_plane::clickhouse::ClickHouseSqlWorkloadEntry { - sql: sql.into(), - start_ms: 0, - end_ms: 2_000, - cumulative: true, - }], - }, - config, - ) + ), + ]), + accuracy: planner_types::types::AccuracyTarget::Exact, + queries: vec![control_plane::clickhouse::ClickHouseSqlWorkloadEntry { + sql: sql.into(), + start_ms: 0, + end_ms: 2_000, + cumulative: true, + }], + } } #[tokio::test] @@ -235,10 +183,23 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { } let exact = exact_request.send().await.unwrap().bytes().await.unwrap(); - let (workload, config) = mixed_workload(sql); - let publication = control_plane::clickhouse::compile_clickhouse_workload(&workload) - .await + let workload = mixed_workload(sql); + let (publication, selection_trace) = + control_plane::clickhouse::compile_automatic_clickhouse_workload(&workload) + .await + .unwrap(); + if let Ok(path) = std::env::var("CLICKHOUSE_PLANNING_ARTIFACT") { + std::fs::write( + path, + serde_json::to_vec_pretty(&serde_json::json!({ + "publication": &publication, "selection_trace": &selection_trace, + })) + .unwrap(), + ) .unwrap(); + } + assert_eq!(publication.precompute_plan.materializations.len(), 1); + let config = publication.precompute_plan.materializations[0].clone(); let entry = publication.query_plan.entries.values().next().unwrap(); assert!(entry.nodes.values().any(|node| matches!( node, diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index e2d62f53..b90b51e2 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -56,6 +56,37 @@ summary, or maintain a cloned executable catalog. ## HTTP surfaces +`POST /api/v1/clickhouse-plan/automatic/compile-and-publish` accepts a plan +`envelope`, typed `tables`, `accuracy`, and `queries` (SQL plus fixed start/end +milliseconds and cumulative-readout policy). It accepts no preselected family, +materialization, or catalog. The control plane plans each query once and derives +materializations only for supported SummaryAgg nodes in that selected DAG, then +publishes the shared catalog and both execution plans through the normal atomic +install/activate path. + +The initial automatic binder supports bounded, whole-second scalar reductions +over one value column and typed table predicates. It uses the query's fixed +window as the materialization duration; this is not a cost-optimized pane/layout +search. The initial fixed-window policy retains two windows (a completed window +and the next active window); it does not certify arbitrary historical or moving +window coverage. Empty, reversed, and out-of-range evaluation intervals are +rejected before automatic planning. Grouped table projections, count-of-rows literal updates, complex table +types, and arbitrary boundary fragments remain unsupported here. Such inputs +must not be presented as accelerated workload coverage. + +The response includes `selection_trace` from the same Planner search: candidate +strategy, rationale, ranking, selected flag, and estimated cost when available. +The current SQL cost model uses relative ranking and may not report a numeric +cost; those entries carry `null` and `not_reported_by_cost_model`, not zero. +The selected DAG, exact branches, and installed materialization IDs remain in +the publication. The real mixed process regression optionally writes both +publication and trace to `CLICKHOUSE_PLANNING_ARTIFACT`. + +Physical backend conversion preserves the selected window duration instead of +inheriting the legacy edge emitter's 5–60 second clamp. It also takes the query +language explicitly so SQL half-open intervals cannot acquire PromQL's +right-closed range flag merely because both ingest raw samples. + The data-plane listener supports `/`, `/ping`, and standard ClickHouse query parameters. Configure it with: