From bf3b4d7bf0dc79c8eaf6915ad968c554bfd90dcf Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 09:25:12 -0600 Subject: [PATCH 1/3] fix(control_plane): carry the declared query cadence through the YAML registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `WorkloadEntry` had no `repeat_every`, so the startup pre-population loop pinned `QuerySpec::repeat_every` to `None` while the same declaration sent through `POST /api/v1/plan` carried its cadence into `QueryWorkload::repeat_every`. The deployment cost model reads that field as the batch-mode flush-period proxy, so one declaration was costed two ways depending on which entry point registered it. Add the field and collapse the three hand-copied entry -> QuerySpec literals (startup loop, emit test helper, MVP46 test loop) into `workload::query_spec_for_entry` — the duplication is why the field was missed in the first place. Also stop losing registry input silently: `WorkloadEntry` now denies unknown fields, and `WorkloadRegistry::try_load` fails startup on a file that exists but does not parse. A registry that degrades to empty is indistinguishable, at every later step, from a deployment that declared no workloads. A missing file stays non-fatal, which the process e2e tests rely on. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 17 +- control_plane/src/emit/mod.rs | 31 +--- control_plane/src/main.rs | 90 +++-------- control_plane/src/store/workload.rs | 3 + control_plane/src/workload.rs | 237 +++++++++++++++++++++++++--- 5 files changed, 248 insertions(+), 130 deletions(-) diff --git a/README.md b/README.md index 38196adb5..283a0232c 100644 --- a/README.md +++ b/README.md @@ -139,17 +139,20 @@ text and deployment hints: accuracy_sla: 0.99 assign_to_role: agent grouping_labels: [region] + repeat_every: 30s ``` `query_string` is the preferred source for metric, aggregation, filters, grouping and range-window semantics. Optional registry fields include -`sketch_family_override`, `sample_p`, `distinct_keys_per_window`, `item_label` -and `monitor`. `accuracy_sla` is the legacy success fraction: `1.0` requests -exact results and `0.99` permits epsilon `0.01`. - -The startup registry does not currently carry dashboard recurrence. The legacy -planning API accepts an evaluation cadence at `POST /api/v1/plan` -(`CONTROLLER_ADDR`, default port `8080`): +`sketch_family_override`, `sample_p`, `distinct_keys_per_window`, `item_label`, +`monitor` and `repeat_every`. `accuracy_sla` is the legacy success fraction: +`1.0` requests exact results and `0.99` permits epsilon `0.01`. An unknown key +is rejected rather than ignored, and a registry file that exists but does not +parse fails startup instead of degrading to an empty registry. + +The startup registry and the legacy planning API carry the same evaluation +cadence, so a declaration is costed identically through either entry point. +`POST /api/v1/plan` (`CONTROLLER_ADDR`, default port `8080`): ```bash curl -X POST http://localhost:8080/api/v1/plan \ diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 064eaa7b0..e930a7206 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -909,38 +909,11 @@ mod runtime_tests { /// Mimics the pre-population loop in `main()` — turns each /// `WorkloadEntry` into a `QueryWorkload` via the shared `Analyzer`. fn populate_store_from_registry(registry: &WorkloadRegistry, store: &WorkloadStore) { - use crate::pipeline::{Analyzer, QuerySpec}; + use crate::pipeline::Analyzer; use crate::types; - use crate::types_v2; let analyzer = Analyzer::new(); for entry in registry.entries() { - // Mirrors main.rs's QuerySpec construction post-B3/B4: - // thread grouping_labels into group_by_labels; let the - // parser drive time_window when query_string is present. - let spec = QuerySpec { - query_string: entry.query_string.clone(), - metric_name: entry.metric_name.clone(), - label_filters: Default::default(), - group_by_labels: entry.grouping_labels.clone(), - aggregations: vec!["quantile".into()], - time_window: if entry.query_string.is_some() { - String::new() - } else { - "5m".into() - }, - repeat_every: None, - accuracy_sla: entry.accuracy_sla, - latency_sla: None, - sketch_type: entry.sketch_family_override.clone(), - workload: types::WorkloadCharacteristics::default(), - id: None, - language: None, - accuracy: None, - dollars: None, - deployment_model: None, - shape: types_v2::QueryShape::default(), - data: types_v2::DataShape::default(), - }; + let spec = crate::workload::query_spec_for_entry(entry); if let Ok(wl) = analyzer.analyze(spec) { let role = crate::workload::derive_agg_role(entry); store.set( diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 0beeb2ed5..885709567 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -284,7 +284,18 @@ async fn main() { // ── Declarative workload registry ──────────────────────────────────────── let workloads_path = std::env::var("CONTROLLER_WORKLOADS").unwrap_or_else(|_| "workloads.yaml".into()); - let workload_registry = Arc::new(WorkloadRegistry::load(&workloads_path)); + // A registry file that exists but does not parse is a startup failure, not + // an empty registry: every later step (plan pre-population, agent config + // push, cost accounting) would otherwise look exactly like a deployment + // that declared no workloads at all. + let workload_registry = match WorkloadRegistry::try_load(&workloads_path) { + Ok(registry) => Arc::new(registry), + Err(error) => { + tracing::error!(path = %workloads_path, %error, "unusable workload registry"); + eprintln!("unusable workload registry at {workloads_path}: {error}"); + std::process::exit(1); + } + }; // Pre-populate PlanStore from the registry so agents get a config immediately. // @@ -302,56 +313,10 @@ async fn main() { { let analyzer = Analyzer::new(); for entry in workload_registry.entries() { - // MVP blocker B4 — let the analyzer parse `time_window` from - // `query_string` (matrix-selector `[range]`) instead of - // forcing a hardcoded "5m" default that overrides whatever - // the user wrote. The analyzer falls back to its own 5m - // default when the PromQL has no matrix selector (e.g. - // `count(unique_users_per_min)`), so this is strictly an - // improvement for queries that DO carry an explicit range. - // Empty string here means "no override; trust the parsed - // value or the analyzer's fallback". - // - // MVP blocker B3 — thread the WorkloadEntry's declarative - // `grouping_labels` into `QuerySpec.group_by_labels`. The - // analyzer merges these with any `by (...)` keys the - // PromQL parser surfaces, populating `QueryWorkload. - // group_by_labels`, which `collect_metric_to_grouping_labels` - // then drops into `EdgeStageConfig.metric_to_grouping_labels` - // so the agent's `keep_keys(datapoint.attributes, [...])` - // OTTL processor strips wire attrs down to this list - // BEFORE sketching. - let spec = pipeline::QuerySpec { - query_string: entry.query_string.clone(), - metric_name: entry.metric_name.clone(), - label_filters: Default::default(), - group_by_labels: entry.grouping_labels.clone(), - aggregations: vec!["quantile".into()], - // Empty when the entry HAS a `query_string` (the parser - // surfaces the matrix-selector range or its own 5m - // fallback). For entries without a query_string we - // can't trust the parser, so fall back to the - // historical 5m default so the analyzer doesn't error - // out at Step 4. - time_window: if entry.query_string.is_some() { - String::new() - } else { - "5m".into() - }, - repeat_every: None, - accuracy_sla: entry.accuracy_sla, - latency_sla: None, - sketch_type: entry.sketch_family_override.clone(), - workload: types::WorkloadCharacteristics::default(), - // design.md alignment: defaults preserve legacy behaviour. - id: None, - language: None, - accuracy: None, - dollars: None, - deployment_model: None, - shape: types_v2::QueryShape::default(), - data: types_v2::DataShape::default(), - }; + // One conversion, shared with every other caller: see + // `control_plane::workload::query_spec_for_entry` for the B3/B4 + // field notes and for why the cadence must come from the entry. + let spec = control_plane::workload::query_spec_for_entry(entry); // B2 full restructure — derive the AggRole for this entry // BEFORE store insertion so collisions on metric name don't // overwrite a prior role's entry. The pre-B2 loop wrote @@ -1222,6 +1187,7 @@ async fn handle_plan(State(st): State, Json(spec): Json) -> item_label: None, // Role derivation does not depend on monitoring. monitor: None, + repeat_every: None, }; control_plane::workload::derive_agg_role(&entry) }; @@ -1561,6 +1527,7 @@ async fn handle_plan_auto( distinct_keys_per_window: None, item_label: None, monitor, + repeat_every: None, }; let role = derive_agg_role(&entry); // (1) Inject the monitor unconditionally — only needs the metric name; @@ -3719,26 +3686,7 @@ mod api_tests { let analyzer = Analyzer::new(); for entry in registry.entries() { - let spec = pipeline::QuerySpec { - query_string: entry.query_string.clone(), - metric_name: entry.metric_name.clone(), - label_filters: Default::default(), - group_by_labels: vec![], - aggregations: vec!["quantile".into()], - time_window: "5m".into(), - repeat_every: None, - accuracy_sla: entry.accuracy_sla, - latency_sla: None, - sketch_type: entry.sketch_family_override.clone(), - workload: types::WorkloadCharacteristics::default(), - id: None, - language: None, - accuracy: None, - dollars: None, - deployment_model: None, - shape: types_v2::QueryShape::default(), - data: types_v2::DataShape::default(), - }; + let spec = control_plane::workload::query_spec_for_entry(entry); if let Ok(wl) = analyzer.analyze(spec) { let wc = types::WorkloadCharacteristics::default(); let plan = state.planner.plan(&wl, Some(&wc)); diff --git a/control_plane/src/store/workload.rs b/control_plane/src/store/workload.rs index 346eec71f..0799e864a 100644 --- a/control_plane/src/store/workload.rs +++ b/control_plane/src/store/workload.rs @@ -270,6 +270,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, }, WorkloadEntry { metric_name: "http_requests_total".into(), @@ -283,6 +284,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, }, WorkloadEntry { metric_name: "http_requests_total".into(), @@ -296,6 +298,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, }, ]; diff --git a/control_plane/src/workload.rs b/control_plane/src/workload.rs index 29af10722..16f9808ff 100644 --- a/control_plane/src/workload.rs +++ b/control_plane/src/workload.rs @@ -214,7 +214,14 @@ fn collect_agg_intents(expr: &planner_types::pre_asap::QueryExpr, out: &mut Vec< } /// A single workload entry from the workloads YAML file. +/// +/// `deny_unknown_fields`: a misspelled or unsupported key is a planning input +/// the controller cannot honour. Accepting it silently would let the operator +/// believe a declared cadence / hint reached the planner when it never left the +/// YAML, so the registry rejects the file instead (see +/// [`WorkloadRegistry::try_load`]). #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct WorkloadEntry { /// Metric name this workload targets (e.g. `http_request_duration_seconds`). pub metric_name: String, @@ -344,6 +351,80 @@ pub struct WorkloadEntry { /// future, the edge `threshold:` block. `None` / missing ⇒ no monitor. #[serde(default)] pub monitor: Option, + + /// Optional evaluation cadence for this declared query class, in the same + /// `1h5m30s` spelling [`crate::pipeline::parse_duration`] accepts. + /// + /// Why this exists: the cadence is a **cost input**, not a scheduler knob. + /// [`crate::physical::deployment_cost::delta`] uses it as the batch-mode + /// flush-period proxy, so the same declaration costed through the HTTP + /// `POST /api/v1/plan` path (whose [`crate::pipeline::QuerySpec`] has + /// carried `repeat_every` all along) and through this YAML registry used + /// to reach the planner with *different* flush periods — the startup path + /// hardcoded `None`. Threaded into `QuerySpec::repeat_every` by the + /// registry pre-pop loop in `main`, which the analyzer parses into + /// [`crate::types::QueryWorkload::repeat_every`]. + /// + /// `None` / missing ⇒ unchanged behaviour (the cost model falls back to + /// its window-derived flush rate). + #[serde(default)] + pub repeat_every: Option, +} + +/// Build the planning [`crate::pipeline::QuerySpec`] a declarative registry +/// entry describes. +/// +/// **Why this exists**: the YAML → planner conversion used to be an inline +/// struct literal, hand-copied into the controller's startup pre-population +/// loop and into the test helpers that claim to mimic it. Every field added to +/// [`WorkloadEntry`] then had to be re-threaded in each copy, and +/// `repeat_every` is the field that proves the cost: the HTTP +/// `POST /api/v1/plan` path carried the declared cadence into +/// [`crate::types::QueryWorkload::repeat_every`] while the startup path pinned +/// `None`, so the same declaration was costed with two different batch-mode +/// flush periods depending on which entry point registered it. +/// +/// Field notes preserved from the original loop: +/// * MVP blocker B4 — `time_window` is left EMPTY when the entry carries a +/// `query_string` so the analyzer parses the matrix-selector `[range]` +/// instead of a hardcoded `5m` overriding what the operator wrote. Entries +/// without a query string keep the historical `5m` default, because the +/// parser has nothing to read and the analyzer errors out without one. +/// * MVP blocker B3 — `grouping_labels` is threaded into `group_by_labels`. +/// The analyzer merges it with any `by (...)` keys the PromQL parser +/// surfaces, which `collect_metric_to_grouping_labels` drops into +/// `EdgeStageConfig::metric_to_grouping_labels` so the agent's +/// `keep_keys(datapoint.attributes, [...])` OTTL processor strips wire attrs +/// down to this list BEFORE sketching. +/// * `sketch_family_override` is threaded into `QuerySpec::sketch_type`, which +/// populates `QueryWorkload::sketch_type_override` — what `bind_workload_typed` +/// reads to honour the MVP §46 HLL / CountSketch / CountMinSketch pins. +pub fn query_spec_for_entry(entry: &WorkloadEntry) -> crate::pipeline::QuerySpec { + crate::pipeline::QuerySpec { + query_string: entry.query_string.clone(), + metric_name: entry.metric_name.clone(), + label_filters: Default::default(), + group_by_labels: entry.grouping_labels.clone(), + aggregations: vec!["quantile".into()], + time_window: if entry.query_string.is_some() { + String::new() + } else { + "5m".into() + }, + repeat_every: entry.repeat_every.clone(), + accuracy_sla: entry.accuracy_sla, + latency_sla: None, + sketch_type: entry.sketch_family_override.clone(), + workload: crate::types::WorkloadCharacteristics::default(), + // design.md alignment: defaults preserve legacy behaviour. + id: None, + language: None, + accuracy: None, + dollars: None, + deployment_model: None, + shape: crate::types_v2::QueryShape::default(), + data: crate::types_v2::DataShape::default(), + } } /// User-facing continuous-monitoring declaration on a [`WorkloadEntry`]. τ/ε and @@ -471,34 +552,47 @@ impl WorkloadRegistry { } /// Load from a YAML file. Returns an empty registry on any error. + /// + /// Prefer [`Self::try_load`] on the startup path: a registry that parses + /// into nothing is indistinguishable, at every later step, from an + /// operator who declared no workloads at all. pub fn load(path: &str) -> Self { - match std::fs::read_to_string(path) { - Ok(contents) => match serde_yaml::from_str::>(&contents) { - Ok(entries) => { - info!(path, count = entries.len(), "loaded workload registry"); - Self { - entries, - runtime: Default::default(), - } - } - Err(e) => { - warn!(path, error = %e, "invalid workloads YAML; using empty registry"); - Self { - entries: vec![], - runtime: Default::default(), - } - } - }, - Err(_) => { - info!(path, "workloads file not found; using empty registry"); - Self { - entries: vec![], - runtime: Default::default(), - } + match Self::try_load(path) { + Ok(registry) => registry, + Err(error) => { + warn!(path, error = %error, "invalid workloads YAML; using empty registry"); + Self::empty() } } } + /// Load from a YAML file, reporting an unusable file instead of degrading + /// to an empty registry. + /// + /// A **missing** file stays non-fatal — the controller is expected to run + /// without a declarative registry (the process e2e tests boot it with + /// `CONTROLLER_WORKLOADS` pointing at a path that does not exist). A file + /// that exists but does not parse is fatal: it carries planning input the + /// operator wrote down, and silently planning *nothing* from it has the + /// same observable shape as a controller that planned everything. + pub fn try_load(path: &str) -> Result { + let contents = match std::fs::read_to_string(path) { + Ok(contents) => contents, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + info!(path, "workloads file not found; using empty registry"); + return Ok(Self::empty()); + } + Err(error) => return Err(format!("cannot read workload registry: {error}")), + }; + let entries = serde_yaml::from_str::>(&contents) + .map_err(|error| format!("cannot parse workload registry: {error}"))?; + info!(path, count = entries.len(), "loaded workload registry"); + Ok(Self { + entries, + runtime: Default::default(), + }) + } + /// Create an empty registry (no file). pub fn empty() -> Self { Self { @@ -543,6 +637,97 @@ impl WorkloadRegistry { mod tests { use super::*; + /// The declared cadence is a planning cost input, so it has to survive the + /// YAML entry point exactly as it survives `POST /api/v1/plan`. + #[test] + fn yaml_cadence_reaches_the_planning_workload() { + let yaml = r#" +- metric_name: http_requests_total + query_string: "sum by (zone) (rate(http_requests_total[5m]))" + accuracy_sla: 0.99 + repeat_every: 30s +- metric_name: http_errors_total + query_string: "sum by (zone) (rate(http_errors_total[5m]))" + accuracy_sla: 0.99 +"#; + let entries: Vec = serde_yaml::from_str(yaml).unwrap(); + assert_eq!(entries[0].repeat_every.as_deref(), Some("30s")); + assert_eq!(entries[1].repeat_every, None); + + let analyzer = crate::pipeline::Analyzer::new(); + let declared = analyzer.analyze(query_spec_for_entry(&entries[0])).unwrap(); + assert_eq!( + declared.repeat_every, + Some(std::time::Duration::from_secs(30)) + ); + // An entry that declares no cadence keeps the historical `None`, so the + // cost model falls back to its window-derived flush rate. + let undeclared = analyzer.analyze(query_spec_for_entry(&entries[1])).unwrap(); + assert_eq!(undeclared.repeat_every, None); + } + + /// A cadence the duration parser cannot read is a declaration error, not a + /// silently dropped field. + #[test] + fn unparsable_cadence_fails_the_entry() { + let entry = WorkloadEntry { + metric_name: "http_requests_total".into(), + query_string: Some("sum(http_requests_total)".into()), + accuracy_sla: 0.99, + assign_to_role: "agent".into(), + sketch_family_override: None, + target_path: None, + grouping_labels: vec![], + sample_p: 1.0, + distinct_keys_per_window: None, + item_label: None, + monitor: None, + repeat_every: Some("every 30 seconds".into()), + }; + let error = crate::pipeline::Analyzer::new() + .analyze(query_spec_for_entry(&entry)) + .expect_err("unparsable cadence must not plan"); + assert!( + format!("{error:#}").contains("repeat_every"), + "unexpected error: {error:#}" + ); + } + + /// An unsupported key is planning input the controller cannot honour; + /// accepting the file would report a cadence / hint that never left the YAML. + #[test] + fn unknown_registry_key_is_rejected() { + let path = std::env::temp_dir().join(format!( + "asap_workload_unknown_key_{}.yaml", + std::process::id() + )); + std::fs::write( + &path, + "- metric_name: http_requests_total + accuracy_sla: 0.99 + repeat_evry: 30s +", + ) + .unwrap(); + let error = WorkloadRegistry::try_load(path.to_str().unwrap()) + .expect_err("unknown key must not load"); + assert!(error.contains("repeat_evry"), "unexpected error: {error}"); + // The lenient wrapper still degrades, which is why startup uses `try_load`. + assert!(WorkloadRegistry::load(path.to_str().unwrap()) + .entries() + .is_empty()); + std::fs::remove_file(&path).unwrap(); + } + + /// Running without a declarative registry stays supported: the process e2e + /// tests boot the controller with `CONTROLLER_WORKLOADS` pointing nowhere. + #[test] + fn missing_registry_file_is_not_a_startup_failure() { + let registry = WorkloadRegistry::try_load("/definitely/missing/workloads.yaml") + .expect("a missing registry is not an error"); + assert!(registry.entries().is_empty()); + } + #[test] fn load_empty_on_missing_file() { let reg = WorkloadRegistry::load("/nonexistent/workloads.yaml"); @@ -617,6 +802,7 @@ mod tests { epsilon: 0.05, window_secs: 30, }), + repeat_every: None, }); let intents = reg.monitor_intents("dp:4319"); assert_eq!(intents.len(), 1); @@ -644,6 +830,7 @@ mod tests { epsilon: 0.05, window_secs: 30, }), + repeat_every: None, }); let intents = reg.monitor_intents("dp:4319"); assert_eq!(intents.len(), 1, "replaced, not duplicated"); @@ -685,6 +872,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, }, WorkloadEntry { metric_name: "b".into(), @@ -698,6 +886,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, }, WorkloadEntry { metric_name: "c".into(), @@ -711,6 +900,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, }, ]); assert_eq!(reg.for_role("agent").len(), 2); @@ -821,6 +1011,7 @@ mod tests { distinct_keys_per_window: None, item_label: None, monitor: None, + repeat_every: None, } } From 3c6d4d3ed501068484823fd17c7f5ad4952050e0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 09:25:23 -0600 Subject: [PATCH 2/3] feat(clickhouse): admit numeric and nullable SQL value columns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Automatic SQL materialization took a non-null Float64 value column only, so an ordinary nullable or integer ClickHouse column could not be accelerated at all. The refusal was protecting the result — the ingest path had no way to know how to read the column — rather than expressing a semantic limit. Give it that way: `PrecomputeMaterialization::value_source_column` carries the producer column's declared type and nullability alongside the value projection. Typing is a read concern, not an identity one, so it stays out of the policy fingerprint — two materializations over the same column are the same policy however it is declared. The ClickHouse reader then reads the column through its type: * nullable -> `IS NOT NULL`, so the summary skips NULL inputs exactly as the SQL aggregate it stands in for does (and a NULL row no longer fails the row decode). * `Int64` -> explicit `toFloat64`, guarded by `throwIf(abs(col) > 9007199254740992)`. Summary state is f64; beyond 2^53 an integer no longer round-trips, so the read fails loudly instead of summarising a rounded value. Non-numeric columns stay refused: a string, boolean or timestamp column has no value semantics to summarise. The existing test that asserted an Int64 source must be rejected now asserts it is admitted with its producer typing, which is what makes the guard above possible. Co-Authored-By: Claude Opus 5 (1M context) --- control_plane/src/clickhouse.rs | 228 +++++++++++++++--- crates/asap_types/src/aggregation_config.rs | 22 ++ .../drivers/ingest/prometheus_remote_write.rs | 2 + data_plane/src/drivers/query/servers/http.rs | 1 + .../src/precompute_engine/output_sink.rs | 1 + .../sketch_db/backfill/clickhouse_reader.rs | 129 +++++++++- .../sketch_db/lifecycle/eviction.rs | 1 + .../tests/test_utilities/engine_factories.rs | 8 + .../query-engine/clickhouse-sql-support.md | 17 +- 9 files changed, 372 insertions(+), 37 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index c7f9be465..12937c013 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -358,8 +358,15 @@ fn materialize_selected_sql( } let grouping = asap_types::GroupingProjection::new(columns); grouping.validate_table_group_codec()?; - let (table, value, window, population, timestamp) = - clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)?; + let leaf = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)?; + let ClickHouseMaterializationLeaf { + table, + value, + value_source_column, + window_secs: window, + population, + timestamp_column: timestamp, + } = leaf; let window_secs = window.ok_or("SQL materialization requires a bounded window")?; let aggregation = BackendAggregation { aggregation_id: String::new(), @@ -382,6 +389,7 @@ fn materialize_selected_sql( config.value_projection = Some(value); config.table_timestamp_column = Some(timestamp); config.table_population = Some(population); + config.value_source_column = value_source_column; config.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped); config.pane_origin_ms = Some( i64::try_from(query.start_ms) @@ -567,9 +575,15 @@ fn bind_selected_node( query: &ClickHouseSqlWorkloadEntry, request: &ClickHouseSqlWorkload, ) -> Result { - let (table_ref, value_column, source_window, spatial_filter, timestamp_column) = - clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms) - .map_err(crate::query_plan::QueryPlanError::Invalid)?; + let ClickHouseMaterializationLeaf { + table: table_ref, + value: value_column, + window_secs: source_window, + population: spatial_filter, + timestamp_column, + .. + } = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms) + .map_err(crate::query_plan::QueryPlanError::Invalid)?; let expected = crate::physical::compiler::physical_materialization_family(family); let selected = select_materialization( &request.precompute_plan.materializations, @@ -614,20 +628,25 @@ fn constant_int64(expr: &QueryExpr) -> Option { } } +/// The table leaf a SQL summary materialization is admitted against. +#[derive(Debug)] +struct ClickHouseMaterializationLeaf { + table: String, + value: asap_types::sds::ValueProjectionIdentity, + /// Producer typing for a column projection: what the ingest path must know + /// to read the column safely (integer exactness, NULL skipping). `None` + /// for constant projections, which carry their own literal. + value_source_column: Option, + window_secs: Option, + population: asap_types::table_population::TablePopulation, + timestamp_column: String, +} + fn clickhouse_materialization_leaf_contract( node: &planner_types::post_asap::SummaryNode, evaluation_start_ms: u64, evaluation_end_ms: u64, -) -> Result< - ( - String, - asap_types::sds::ValueProjectionIdentity, - Option, - asap_types::table_population::TablePopulation, - String, - ), - String, -> { +) -> Result { use planner_types::{ post_asap::SummaryExpr, pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source}, @@ -676,6 +695,7 @@ fn clickhouse_materialization_leaf_contract( }; use asap_types::sds::ValueProjectionIdentity; use planner_types::post_asap::SummaryInputExpr; + let mut value_source_column = None; let value_projection = match &input.weight { SummaryInputExpr::Column( planner_types::pre_asap::ColumnRef::Wildcard @@ -701,9 +721,24 @@ fn clickhouse_materialization_leaf_contract( .iter() .find(|column| column.name == *name) .ok_or("SQL summary value projection is not a source column")?; - if column.nullable || column.dtype != planner_types::pre_asap::DataType::Float64 { - return Err("SQL value readout requires a non-null Float64 source until typed/null-aware ingest is available".into()); + // Numeric source columns are admitted with their declared type and + // nullability, which the ingest reader honours: integers get an + // exactness guard on the way into f64 summary state, and NULL rows + // are skipped the way a SQL aggregate skips them. Non-numeric + // columns have no value semantics to summarise and stay refused. + if !matches!( + column.dtype, + planner_types::pre_asap::DataType::Float64 + | planner_types::pre_asap::DataType::Int64 + ) { + return Err(format!( + "SQL value readout requires a numeric source column; `{name}` is {:?}", + column.dtype + )); } + let mut source_column = column.clone(); + source_column.table = None; + value_source_column = Some(source_column); ValueProjectionIdentity::Column { name: name.clone() } } SummaryInputExpr::Constant(value) if value.is_finite() => { @@ -827,18 +862,19 @@ fn clickhouse_materialization_leaf_contract( let window_secs = explicit_window.or(inferred_window).ok_or_else(|| { "SQL table summary requires a positive whole-second timestamp range".to_string() })?; - Ok(( - table_ref.to_owned(), - value_projection, - Some(window_secs), + Ok(ClickHouseMaterializationLeaf { + table: table_ref.to_owned(), + value: value_projection, + value_source_column, + window_secs: Some(window_secs), population, - schema + timestamp_column: schema .time_index .and_then(|index| schema.columns.get(index)) .ok_or("SQL summary source has no timestamp projection")? .name .clone(), - )) + }) } fn select_materialization<'a>( @@ -993,6 +1029,133 @@ mod tests { value } + /// A table leaf whose value column has the given producer typing. + fn typed_value_leaf( + dtype: planner_types::pre_asap::DataType, + nullable: bool, + ) -> planner_types::post_asap::SummaryNode { + use planner_types::post_asap::{ + SummaryExpr, SummaryInputExpr, SummaryNode, SummarySchema, SummaryUpdate, + }; + use planner_types::pre_asap::{ + Column, ColumnRef, CompareOpKind, DataType, Predicate, QueryExpr, Reduction, + ScalarValue, Schema, Source, + }; + let schema = Schema::with_time_index( + vec![ + Column::new("timestamp_ms", DataType::Timestamp, false), + Column::new("value", dtype, nullable), + ], + 0, + Vec::new(), + ); + let bound = |op: CompareOpKind, at: i64| { + Predicate(std::rc::Rc::new(QueryExpr::Compare { + left: std::rc::Rc::new(QueryExpr::Column(0)), + op, + right: std::rc::Rc::new(QueryExpr::Literal(ScalarValue::Int64(at))), + })) + }; + let scan = QueryExpr::Scan { + source: Source::Table { + table_ref: "telemetry".into(), + }, + predicates: vec![ + bound(CompareOpKind::Ge, 0), + bound(CompareOpKind::Lt, 60_000), + ], + schema, + }; + let family = materialization( + AggregationType::Sum, + "value", + 60, + 60, + ("variant", serde_json::json!(1)), + ) + .accumulator_spec() + .unwrap() + .family; + let summary_schema = SummarySchema { + fields: vec![], + time_index: None, + }; + SummaryNode { + expr: SummaryExpr::SummaryAgg { + child: std::rc::Rc::new(SummaryNode { + expr: SummaryExpr::KeepPreAsap(std::rc::Rc::new(scan)), + schema: summary_schema.clone(), + guarantee: Default::default(), + }), + family, + input: SummaryUpdate { + item: None, + weight: SummaryInputExpr::Column(ColumnRef::Named("value".into())), + weight_domain: Default::default(), + }, + reduction: Reduction::Reduce(vec![].into()), + grouping: Default::default(), + }, + schema: summary_schema, + guarantee: Default::default(), + } + } + + /// Numeric source columns are admitted with their producer typing, which + /// the ingest reader needs to read them safely. Before this the contract + /// took non-null `Float64` only, so an ordinary nullable or integer + /// ClickHouse column could not be automatically materialized at all. + #[test] + fn numeric_value_columns_are_admitted_with_their_producer_typing() { + use planner_types::pre_asap::DataType; + for (dtype, nullable) in [ + (DataType::Float64, false), + (DataType::Float64, true), + (DataType::Int64, false), + (DataType::Int64, true), + ] { + let leaf = clickhouse_materialization_leaf_contract( + &typed_value_leaf(dtype.clone(), nullable), + 0, + 60_000, + ) + .unwrap_or_else(|error| panic!("{dtype:?}/{nullable}: {error}")); + assert_eq!( + leaf.value, + asap_types::sds::ValueProjectionIdentity::Column { + name: "value".into() + } + ); + let column = leaf + .value_source_column + .expect("a column projection carries its producer typing"); + assert_eq!(column.dtype, dtype); + assert_eq!(column.nullable, nullable); + // Typing is a read concern, not an identity one: the same column + // is the same policy however it is declared. + assert_eq!(column.table, None); + } + } + + /// Non-numeric columns have no value semantics to summarise. Refusing them + /// protects the result; it is not a gap to be widened. + #[test] + fn non_numeric_value_columns_stay_refused() { + use planner_types::pre_asap::DataType; + for dtype in [DataType::Utf8, DataType::Bool, DataType::Timestamp] { + let error = clickhouse_materialization_leaf_contract( + &typed_value_leaf(dtype.clone(), false), + 0, + 60_000, + ) + .unwrap_err(); + assert!( + error.contains("numeric source column"), + "{dtype:?}: {error}" + ); + } + } + #[test] fn keyed_summary_input_is_not_replaced_by_its_unit_weight() { use planner_types::post_asap::{ @@ -1427,12 +1590,21 @@ mod tests { }) .collect(), }; + // An Int64 source is admitted and carries its declared type into the + // materialization, which is what lets the ingest reader widen the + // column explicitly and fail loudly on a value beyond the exact + // Float64 range instead of silently summarising a rounded one. integer_source.tables.get_mut("telemetry").unwrap().columns[1].dtype = DataType::Int64; - assert!( - compile_automatic_clickhouse_workload(&integer_source) - .await - .is_err(), - "an Int64 source may contain values beyond exact Float64 ingest range" + let (integer_plan, _) = compile_automatic_clickhouse_workload(&integer_source) + .await + .expect("an Int64 source is admitted with its producer typing"); + assert_eq!( + integer_plan.precompute_plan.materializations[0] + .value_source_column + .as_ref() + .expect("column projections carry their producer typing") + .dtype, + DataType::Int64 ); integer_source.tables.get_mut("telemetry").unwrap().columns[1].dtype = DataType::Float64; integer_source.queries[0].sql = integer_source.queries[0].sql.replace( diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index b8a5f4e62..0aab0521c 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -156,6 +156,27 @@ pub struct PrecomputeMaterialization { skip_serializing_if = "Option::is_none" )] pub table_population: Option, + /// Producer typing for a SQL table value projection: the source column's + /// declared type and nullability. + /// + /// **Why this is separate from [`Self::value_projection`]**: the + /// projection is the materialization's *identity* — which column or + /// constant is summarised, and part of the policy fingerprint. This is how + /// the ingest path must *read* that column, which the identity does not + /// determine: an integer column needs an exactness guard on its way into + /// f64 summary state, and a nullable column needs SQL's "aggregates skip + /// NULL" rule reproduced at the reader rather than a decode failure on the + /// first NULL row. Two materializations over the same column are the same + /// policy either way, so this deliberately stays out of the fingerprint. + /// + /// `None` ⇒ PromQL-mode materializations and legacy SQL definitions, which + /// keep the pre-typed behaviour (read the column as it comes). + #[serde( + default, + alias = "valueSourceColumn", + skip_serializing_if = "Option::is_none" + )] + pub value_source_column: Option, } /// Policy-match handles for both the key and value dimensions of a @@ -317,6 +338,7 @@ impl PrecomputeMaterialization { .map(|name| crate::sds::ValueProjectionIdentity::Column { name }), table_population: None, table_timestamp_column: None, + value_source_column: None, } } diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 8a7ba8657..b37d4c798 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -1051,6 +1051,7 @@ mod tests { derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let policy_fp = aggregation.policy_fp_u64(); let streaming = StreamingConfig::new(HashMap::from([(policy_fp, aggregation)])); @@ -1187,6 +1188,7 @@ mod tests { derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let cms = config( AggregationType::CountMinSketchWithHeap, diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index ce1bbcee1..a6d63925d 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3727,6 +3727,7 @@ aggregations: derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; // PR 5: streaming-config is keyed on the policy // fingerprint. Build a marker→fingerprint map so the test diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 538940d86..c2e4ffc4c 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -430,6 +430,7 @@ mod tests { derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, } } diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs index fb169e898..d19aad0e0 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs @@ -50,6 +50,9 @@ pub struct ClickHouseReader { http: reqwest::Client, population: Option, value_projection: Option, + /// Producer typing for the projected column, when the materialization + /// carries it. Drives [`ClickHouseReader::value_column_guards`]. + value_source_column: Option, output_metric: Option, grouping_projection: Option, } @@ -77,6 +80,7 @@ impl ClickHouseReader { http: reqwest::Client::new(), population: None, value_projection: None, + value_source_column: None, output_metric: None, grouping_projection: None, }) @@ -139,24 +143,65 @@ impl ClickHouseReader { let value = match &self.value_projection { Some(asap_types::sds::ValueProjectionIdentity::Constant { value: planner_types::pre_asap::ScalarValue::Int64(_), - }) => "{projected_value:Int64}", + }) => "{projected_value:Int64}".to_string(), Some(asap_types::sds::ValueProjectionIdentity::Constant { value: planner_types::pre_asap::ScalarValue::Float64(_), - }) => "{projected_value:Float64}", - _ => c.value_column.as_str(), + }) => "{projected_value:Float64}".to_string(), + // A typed column projection is read through its declared type: + // integers via an explicit widening, everything else as-is. + _ => match self + .value_source_column + .as_ref() + .map(|column| &column.dtype) + { + Some(planner_types::pre_asap::DataType::Int64) => { + format!("toFloat64({})", c.value_column) + } + _ => c.value_column.clone(), + }, }; format!( "SELECT {labels} AS labels, {timestamp} AS timestamp_ms, {value} AS value \ - FROM {database}.{table} WHERE {population} \ + FROM {database}.{table} WHERE {population}{value_guards} \ AND {timestamp} >= {{start_ms:Int64}} AND {timestamp} < {{end_ms:Int64}} \ ORDER BY labels, timestamp_ms FORMAT JSONEachRow", labels = labels, timestamp = c.timestamp_ms_column, value = value, + value_guards = self.value_column_guards(), database = c.database, table = c.table, ) } + + /// Predicates the typed value column needs before its rows may enter + /// summary state. + /// + /// * **NULL**: a SQL aggregate skips NULL inputs, so the summary that + /// stands in for `sum(col)` / `count(col)` / `quantile(col)` has to skip + /// them too. (Without this the JSON decode fails on the first NULL row, + /// which is why nullable columns used to be refused at planning time.) + /// * **Integer exactness**: summary state is f64. Beyond 2^53 an integer + /// no longer round-trips, so instead of silently summarising a rounded + /// value the read fails loudly on the offending row. The guard evaluates + /// to NULL for NULL rows, which the NULL predicate has already excluded. + fn value_column_guards(&self) -> String { + let Some(column) = self.value_source_column.as_ref() else { + return String::new(); + }; + let name = &self.config.value_column; + let mut guards = String::new(); + if column.nullable { + guards.push_str(&format!(" AND {name} IS NOT NULL")); + } + if column.dtype == planner_types::pre_asap::DataType::Int64 { + guards.push_str(&format!( + " AND throwIf(abs({name}) > 9007199254740992, \ + 'ASAP ClickHouse ingest: integer value exceeds the exact Float64 range') = 0" + )); + } + guards + } } /// Resolve a typed table source using deployment-local connection settings. @@ -199,6 +244,7 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor let mut reader = ClickHouseReader::new(source_config)?; reader.population = Some(materialization.table_population.clone().unwrap_or_default()); reader.value_projection = Some(materialization.effective_value_projection().clone()); + reader.value_source_column = materialization.value_source_column.clone(); reader.output_metric = Some(materialization.metric.clone()); reader.grouping_projection = Some(materialization.grouping_labels.clone()); Ok(Arc::new(reader) as Arc) @@ -410,6 +456,81 @@ mod tests { assert!(!sql.contains("{metric:String}")); } + /// SQL aggregates skip NULL inputs, so the summary standing in for one has + /// to skip them too — and a NULL row would otherwise fail the row decode, + /// which is why nullable columns used to be refused at planning time. + #[test] + fn nullable_value_columns_skip_null_rows_the_way_sql_aggregates_do() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column { + name: "value".into(), + }); + reader.value_source_column = Some(planner_types::pre_asap::Column::new( + "value", + planner_types::pre_asap::DataType::Float64, + true, + )); + let sql = reader.sql(); + assert!(sql.contains("AND value IS NOT NULL"), "{sql}"); + // A float column is read as it comes; only integers are widened. + assert!(sql.contains(" value AS value"), "{sql}"); + assert!(!sql.contains("toFloat64"), "{sql}"); + } + + /// A non-null column needs no NULL predicate: the guard tracks the declared + /// nullability, it is not applied blindly. + #[test] + fn non_null_value_columns_are_read_without_extra_predicates() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column { + name: "value".into(), + }); + reader.value_source_column = Some(planner_types::pre_asap::Column::new( + "value", + planner_types::pre_asap::DataType::Float64, + false, + )); + let sql = reader.sql(); + assert!(!sql.contains("IS NOT NULL"), "{sql}"); + assert!(!sql.contains("throwIf"), "{sql}"); + } + + /// Summary state is f64. An integer column is widened explicitly, and a + /// value that no longer round-trips fails the read instead of entering the + /// summary rounded. + #[test] + fn integer_value_columns_are_widened_under_an_exactness_guard() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column { + name: "value".into(), + }); + reader.value_source_column = Some(planner_types::pre_asap::Column::new( + "value", + planner_types::pre_asap::DataType::Int64, + false, + )); + let sql = reader.sql(); + assert!(sql.contains("toFloat64(value) AS value"), "{sql}"); + assert!( + sql.contains("throwIf(abs(value) > 9007199254740992"), + "{sql}" + ); + assert!(!sql.contains("IS NOT NULL"), "{sql}"); + } + + /// A legacy definition without producer typing keeps its previous read. + #[test] + fn untyped_column_projection_reads_exactly_as_before() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column { + name: "value".into(), + }); + let sql = reader.sql(); + assert!(sql.contains(" value AS value"), "{sql}"); + assert!(!sql.contains("IS NOT NULL"), "{sql}"); + assert!(!sql.contains("throwIf"), "{sql}"); + } + #[test] fn constant_projection_uses_a_typed_parameter_without_a_fake_column() { let mut reader = ClickHouseReader::new(config("samples")).unwrap(); diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs index 469db04ae..dbe81fbb2 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -280,6 +280,7 @@ mod tests { derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, } } diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index dd88cab66..b5daec83c 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -113,6 +113,7 @@ pub fn create_engine_single_pop_with_aggregated( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); @@ -212,6 +213,7 @@ pub fn create_engine_dual_input( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let value_id = value_agg_config.policy_fp_u64(); aggregation_configs.insert(value_id, value_agg_config); @@ -241,6 +243,7 @@ pub fn create_engine_dual_input( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let keys_id = keys_agg_config.policy_fp_u64(); aggregation_configs.insert(keys_id, keys_agg_config); @@ -335,6 +338,7 @@ pub fn create_engine_two_metrics( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let id_a = agg_config_a.policy_fp_u64(); aggregation_configs.insert(id_a, agg_config_a); @@ -363,6 +367,7 @@ pub fn create_engine_two_metrics( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let id_b = agg_config_b.policy_fp_u64(); aggregation_configs.insert(id_b, agg_config_b); @@ -467,6 +472,7 @@ pub fn create_engine_three_metrics( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let id = cfg.policy_fp_u64(); ids.push(id); @@ -548,6 +554,7 @@ pub fn create_engine_multi_timestamp( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); @@ -621,6 +628,7 @@ pub fn create_engine_multi_timestamp_with_window( derived_input: None, table_timestamp_column: None, partitioning: None, + value_source_column: None, }; let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 7873028a0..919bf94ab 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -65,8 +65,15 @@ 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 a non-null Float64 value column or a finite numeric literal, plus typed -table predicates. Row counts use the shared typed constant `1` projection. It uses the query's fixed +over a numeric value column — `Float64` or `Int64`, nullable or not — or a +finite numeric literal, plus typed table predicates. The column's declared type +and nullability travel with the materialization (`value_source_column`) because +the ingest reader needs them: a nullable column is read with `IS NOT NULL`, so +the summary skips NULL inputs exactly as the SQL aggregate it stands in for +does, and an `Int64` column is widened explicitly with a guard that fails the +read on a value beyond the exact `Float64` range rather than summarising a +rounded one. Typing is a read concern, not an identity one, so it stays out of +the policy fingerprint. Row counts use the shared typed constant `1` projection. 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 @@ -79,9 +86,9 @@ SQL `count(*)` now installs an exact row-count materialization through the same catalog and readout DAG as other summaries. The selected Count intent uses the existing physical SUM accumulator over typed constant `1`. It includes rows with a NULL value column. Planner rejects nullable `count(value)` until per-aggregate null exclusion is -represented; it must not silently become row count. Non-Float64 named sources -are rejected because the current Float64 ingest path cannot preserve arbitrary -Int64 values exactly. Producer `Project` subtrees are also rejected until their +represented; it must not silently become row count. Non-numeric named sources +are rejected: a string, boolean or timestamp column has no value semantics to +summarise. Producer `Project` subtrees are also rejected until their computation is executed, rather than skipped while binding the original table. The real process test covers SUM and row count mixed with a ClickHouse exact branch, and deletes a source row after materialization to prove summary readout. From 7ea79f1ea647489e84e24eba709f113c94c62234 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 09:25:34 -0600 Subject: [PATCH 3/3] test(data_plane): fix the population maintenance contract and gate data plane tests in CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things kept the data plane out of the default gate, and they compound: 1. A stable failure. `one_sid_with_two_populations_cannot_publish_a_partial_global_summary` asserted `execute_finite_maintenance` refuses a global reduce over one sid carrying two populations. It does not, and it should not: the `CanonicalLabelsV1` path reads the cohort through `read_complete_raw_maintenance_cohort`, which enumerates the whole durable population set for the definition and fails on the first population missing the window. The reduce sees both populations or it sees nothing, so the partial publication the test guarded against is unreachable there. Rewritten as the acceptance contract instead of dropping the assertion: the two populations carry distinct values, and the test proves one output series with one (empty) group and one window, that the reduce consumed both populations' state, that a re-run republishes nothing and writes no new part, and that the output is durably committed. The second half — a later lifetime repeating a logical population — is still refused, and now also proves it published nothing new. 2. Fifteen parallel flakes. Every one waited on a background flusher / sealer thread under a fixed 5s budget. At one test per core the waiter and the thread it waits on compete for the same CPUs and the budget expires while the work is merely queued — which is why they passed on serial re-runs. `tests::test_utilities::timing` scales the budget with test concurrency (`ASAP_TEST_TIMEOUT_SCALE` overrides); a passing wait still returns as soon as its condition holds, and a stuck one still fails. 3. No gate. `cargo test -p asap_types -p data_plane --lib` now runs in MVP CI. Compiling the data plane was never evidence that it runs: plan installation, maintenance execution, persistence recovery and query routing all live in these tests. `--lib` deliberately excludes `data_plane/tests/`, which needs live ClickHouse / Prometheus / VictoriaMetrics endpoints. Data plane library tests: 1194 passing under the default parallel run. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/mvp-ci.yml | 17 +- .../precompute_engine/maintenance_runtime.rs | 6 +- .../sketch_db/backfill/service.rs | 4 +- .../sketch_db/index/maintenance.rs | 152 ++++++++++++++---- .../storage_engines/sketch_db/index/mod.rs | 5 +- .../sketch_db/persistence/flusher.rs | 8 +- data_plane/src/tests/test_utilities/mod.rs | 1 + data_plane/src/tests/test_utilities/timing.rs | 64 ++++++++ 8 files changed, 219 insertions(+), 38 deletions(-) create mode 100644 data_plane/src/tests/test_utilities/timing.rs diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml index a2b9d4cb4..a7b96003f 100644 --- a/.github/workflows/mvp-ci.yml +++ b/.github/workflows/mvp-ci.yml @@ -16,7 +16,7 @@ jobs: backend: name: MVP backend runs-on: ubuntu-latest - timeout-minutes: 45 + timeout-minutes: 60 steps: - name: Checkout ASAPQuery-backend uses: actions/checkout@v4 @@ -78,3 +78,18 @@ jobs: CARGO_NET_GIT_FETCH_WITH_CLI: "true" working-directory: ASAPQuery-backend run: cargo test -p control_plane + + # Compiling the data plane is not evidence that it runs: plan + # installation, maintenance execution, persistence recovery and query + # routing all live in these library tests, and none of them were gated + # before. `--lib` deliberately excludes `data_plane/tests/`, which needs + # live ClickHouse / Prometheus / VictoriaMetrics endpoints. + - name: Test shared types and data plane + env: + CARGO_NET_GIT_FETCH_WITH_CLI: "true" + # Several storage tests wait on background flusher / sealer threads. + # Hosted runners are small and noisy, so give those waits headroom; + # a passing wait still returns as soon as its condition holds. + ASAP_TEST_TIMEOUT_SCALE: "4" + working-directory: ASAPQuery-backend + run: cargo test -p asap_types -p data_plane --lib diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 47fb0428d..3c8d7ab4e 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -2587,7 +2587,8 @@ mod tests { &BTreeMap::new() ) .is_err()); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let deadline = + crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5)); while !store.seal_finite_summary_input(&generation).unwrap() { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); @@ -3023,7 +3024,8 @@ mod tests { assert!(store .series_ids_for_policy(configs[2].policy_fingerprint()) .is_empty()); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let deadline = + crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5)); while !store.seal_finite_summary_input(&generation).unwrap() { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index 3fd4347ac..3b10f79b8 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -389,7 +389,9 @@ mod tests { target: BackfillStatus, timeout_ms: u64, ) -> BackfillStatus { - let deadline = std::time::Instant::now() + std::time::Duration::from_millis(timeout_ms); + let deadline = crate::tests::test_utilities::timing::deadline( + std::time::Duration::from_millis(timeout_ms), + ); loop { if let Some(j) = registry.get(job_id) { if j.status == target || j.status.is_terminal() { diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index a26140f17..7e7ac4d65 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -800,7 +800,8 @@ mod tests { ) .unwrap(); } - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let deadline = + crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5)); while !store.seal_finite_summary_input(&generation).unwrap() { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); @@ -953,7 +954,8 @@ mod tests { }, ) .unwrap(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let deadline = + crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5)); while !store.seal_finite_summary_input(&generation).unwrap() { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); @@ -1067,7 +1069,16 @@ mod tests { } #[test] - fn one_sid_with_two_populations_cannot_publish_a_partial_global_summary() { + fn one_sid_with_two_populations_publishes_one_complete_global_summary() { + // Two logical populations share a physical sid. A global reduce over + // them is publishable exactly when EVERY population contributes the + // window — `read_complete_raw_maintenance_cohort` enumerates the whole + // durable population set and fails on the first one that is missing, + // so the reduce sees both or it sees nothing. The partial publication + // this test used to assert against is therefore unreachable here; what + // still has to hold is that the complete reduce publishes ONCE, with + // both populations' state in it, and that a later lifetime which + // repeats a logical population is still refused (second half). let mut fixture: serde_json::Value = serde_json::from_str(include_str!( "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) @@ -1123,7 +1134,9 @@ mod tests { output.population_labels = Some(population); output.catalog_generation = Some(Arc::clone(&generation)); let mut sum = SumAccumulator::new(); - sum.update(5.0); + // Distinct per-population values: a summary built from only one of + // them reads back as 5 or 7, never as the pair. + sum.update(if instance == "a" { 5.0 } else { 7.0 }); store .publish_admitted_summary_update( &generation, @@ -1138,7 +1151,8 @@ mod tests { assert!(store .complete_raw_maintenance_population(source.policy_fingerprint().into(), &generation) .is_err()); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let deadline = + crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5)); while !store.seal_finite_summary_input(&generation).unwrap() { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); @@ -1168,7 +1182,7 @@ mod tests { .unwrap(); assert!(!frozen.singleton_population_complete); // The read set is deterministic and all-or-nothing, independently of - // the later routing decision (which still rejects this global reduce). + // the routing decision that consumes it. let request = |name: &str| { ( 700, @@ -1208,22 +1222,86 @@ mod tests { &[request("a"), request("absent")] ) .is_err()); - let parts_before = persistence.manifest.live_parts().len(); let resolver = crate::drivers::ingest::series_resolver::SeriesIdResolver::new(); - for _ in 0..2 { - assert!( - crate::precompute_engine::maintenance_runtime::execute_finite_maintenance( - &store, - &resolver, - &plan.precompute_plan + crate::precompute_engine::maintenance_runtime::execute_finite_maintenance( + &store, + &resolver, + &plan.precompute_plan, + ) + .unwrap(); + + // One target series, one output population (the global reduce erases + // the source grouping), one window. + let target_id = target.policy_fingerprint().into(); + let target_sids = store.series_ids_for_policy(target.policy_fingerprint()); + assert_eq!(target_sids.len(), 1, "one global output series"); + let published = store + .completed_maintenance_coordinates(target_id, &generation) + .unwrap(); + assert_eq!(published.len(), 1); + let groups = &published[&target_sids[0]]; + assert_eq!(groups.len(), 1); + assert_eq!(groups[&BTreeMap::new()], BTreeSet::from([(0, 60_000)])); + + // The reduce consumed BOTH populations: the cohort the publication + // path reads is the whole durable population set for this window, and + // it carries the two distinct per-population sums. (Published summary + // state is sketch-encoded, and the immutable read-back path decodes + // exact accumulators only — the reduced VALUE is asserted against this + // same cohort contract by + // `precompute_engine::maintenance_runtime::tests`.) + let cohort = store + .read_complete_raw_maintenance_cohort( + &generation, + &BTreeSet::from([source_id]), + (0, 60_000), + ) + .unwrap(); + assert_eq!(cohort.inputs().len(), 2); + let contributions: Vec<(String, f64)> = cohort + .inputs() + .iter() + .map(|input| { + ( + input.group["instance"].clone(), + input.windows[&(0, 60_000)] + .query_statistic( + asap_types::Statistic::Sum, + &None, + &std::collections::HashMap::new(), + ) + .unwrap(), ) - .is_err() - ); - } - assert!(store - .series_ids_for_policy(target.policy_fingerprint()) - .is_empty()); - assert_eq!(persistence.manifest.live_parts().len(), parts_before); + }) + .collect(); + assert_eq!( + contributions, + vec![("a".to_string(), 5.0), ("b".to_string(), 7.0)] + ); + + // Idempotent: a re-run republishes nothing and writes no new part. + let parts_after_publication = persistence.manifest.live_parts().len(); + crate::precompute_engine::maintenance_runtime::execute_finite_maintenance( + &store, + &resolver, + &plan.precompute_plan, + ) + .unwrap(); + assert_eq!( + persistence.manifest.live_parts().len(), + parts_after_publication + ); + assert_eq!( + store + .completed_maintenance_coordinates(target_id, &generation) + .unwrap(), + published + ); + assert_eq!( + store.series_ids_for_policy(target.policy_fingerprint()), + target_sids + ); + // The published output is durable and committed, not left pending. let records = store .persistence_metadata .read() @@ -1232,9 +1310,15 @@ mod tests { .unwrap() .load_strict() .unwrap(); - assert!(records.iter().all(|record| record.summary_definition_id - != Some(target.policy_fingerprint().into()) - && record.pending_immutable.is_none())); + let published_target_sids: BTreeSet = records + .iter() + .filter(|record| !record.removed && record.summary_definition_id == Some(target_id)) + .map(|record| record.sid) + .collect(); + assert_eq!(published_target_sids.len(), 1); + assert!(records + .iter() + .all(|record| record.pending_immutable.is_none())); persistence.shutdown(); // A catalog transition deliberately leaves old-generation payload @@ -1281,7 +1365,8 @@ mod tests { |writer| writer.ingest_precompute_with_series_id(702, source, &output, &sum), ) .unwrap(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let deadline = + crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5)); while !restarted .seal_finite_summary_input(&next_generation) .unwrap() @@ -1320,18 +1405,27 @@ mod tests { .series_ids_for_policy(target.policy_fingerprint()) .is_empty()); assert_eq!(restored.manifest.live_parts().len(), parts_before); - assert!(restarted + let after_restart = restarted .persistence_metadata .read() .unwrap() .as_ref() .unwrap() .load_strict() - .unwrap() + .unwrap(); + // The refused lifetime publishes nothing NEW: the durable target state + // is exactly what the earlier complete reduce committed. + assert_eq!( + after_restart + .iter() + .filter(|record| !record.removed && record.summary_definition_id == Some(target_id)) + .map(|record| record.sid) + .collect::>(), + published_target_sids + ); + assert!(after_restart .iter() - .all(|record| record.summary_definition_id - != Some(target.policy_fingerprint().into()) - && record.pending_immutable.is_none())); + .all(|record| record.pending_immutable.is_none())); restored.shutdown(); } diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 88ae71d67..d77030e22 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -4791,7 +4791,10 @@ mod tests { } fn wait_until bool>(f: F, timeout: std::time::Duration) -> bool { - let deadline = std::time::Instant::now() + timeout; + // Scaled: these waits are on background flusher / sealer threads that + // compete with the test harness for cores. See + // `crate::tests::test_utilities::timing`. + let deadline = crate::tests::test_utilities::timing::deadline(timeout); while std::time::Instant::now() < deadline { if f() { return true; diff --git a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs index 4c8307db2..ee66370ce 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs @@ -721,7 +721,7 @@ mod tests { let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap(); // Wait until memory is below the low-water mark. - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(2)); while source.approx_memory_bytes() > 600 && Instant::now() < deadline { thread::sleep(Duration::from_millis(20)); } @@ -755,7 +755,7 @@ mod tests { cfg.hot_window_ms = Some(1_000); // 1 second hot window let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap(); - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(2)); while source.approx_memory_bytes() > 0 && Instant::now() < deadline { thread::sleep(Duration::from_millis(20)); } @@ -788,7 +788,7 @@ mod tests { cfg.hot_window_ms = Some(120_000); // 120s hot window, like live let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap(); - let deadline = Instant::now() + Duration::from_secs(3); + let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(3)); while manifest.live_parts().is_empty() && Instant::now() < deadline { thread::sleep(Duration::from_millis(20)); } @@ -809,7 +809,7 @@ mod tests { cfg.delete_older_than_ms = Some(0); // then immediately expire it let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap(); - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(2)); while !manifest.live_parts().is_empty() && Instant::now() < deadline { thread::sleep(Duration::from_millis(20)); } diff --git a/data_plane/src/tests/test_utilities/mod.rs b/data_plane/src/tests/test_utilities/mod.rs index 8515aa5fa..2b419d484 100644 --- a/data_plane/src/tests/test_utilities/mod.rs +++ b/data_plane/src/tests/test_utilities/mod.rs @@ -7,5 +7,6 @@ //! `StoreQueryPlan` legacy types. pub mod engine_factories; +pub mod timing; pub use engine_factories::*; diff --git a/data_plane/src/tests/test_utilities/timing.rs b/data_plane/src/tests/test_utilities/timing.rs new file mode 100644 index 000000000..3b579ff4e --- /dev/null +++ b/data_plane/src/tests/test_utilities/timing.rs @@ -0,0 +1,64 @@ +//! Wall-clock budgets for tests that wait on background progress. +//! +//! Several storage / maintenance tests wait for work that a *background* +//! thread performs — the persistence flusher sealing and evicting epochs, the +//! finite-input sealer draining an admitted revision. A fixed five-second +//! budget is generous when the test owns the machine and far too tight when +//! `cargo test` runs one test per core: on a 64-way runner the waiter and the +//! background thread it is waiting on compete for the same CPUs, and the +//! budget expires while the work is merely queued. That is what made the +//! suite's flush / evict / restart tests fail under the default parallel run +//! and pass on a serial re-run, which in turn kept the data-plane library +//! tests out of the default CI gate. +//! +//! Scaling the budget with the test concurrency keeps both properties: a +//! passing test still returns as soon as its condition holds (these are +//! polling waits, not sleeps), and a genuinely stuck condition still fails — +//! later, but deterministically. + +use std::time::Duration; + +/// Scale a wait budget by the concurrency the test binary is running at. +/// +/// Override with `ASAP_TEST_TIMEOUT_SCALE=` to pin a multiplier (useful on +/// a loaded shared machine, or to re-tighten the budget while debugging a +/// hang). Otherwise the multiplier is derived from `RUST_TEST_THREADS` when +/// the harness was told a thread count, and from the machine's parallelism +/// otherwise, capped so the worst case stays bounded. +pub fn scaled(budget: Duration) -> Duration { + budget * scale() +} + +fn scale() -> u32 { + if let Some(explicit) = std::env::var("ASAP_TEST_TIMEOUT_SCALE") + .ok() + .and_then(|value| value.trim().parse::().ok()) + { + return explicit.max(1); + } + let threads = std::env::var("RUST_TEST_THREADS") + .ok() + .and_then(|value| value.trim().parse::().ok()) + .or_else(|| std::thread::available_parallelism().ok().map(|n| n.get())) + .unwrap_or(1); + ((threads / 8) as u32).clamp(1, 8) +} + +/// A deadline `budget` from now, scaled by [`scaled`]. +pub fn deadline(budget: Duration) -> std::time::Instant { + std::time::Instant::now() + scaled(budget) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn explicit_scale_overrides_the_derived_one() { + // The env var is process-global; assert the parsing contract on the + // pure helper instead of mutating it under a parallel harness. + assert!(scale() >= 1); + assert!(scaled(Duration::from_secs(5)) >= Duration::from_secs(5)); + assert!(scaled(Duration::from_secs(5)) <= Duration::from_secs(40)); + } +}