diff --git a/Cargo.lock b/Cargo.lock index aca9e44af..35a689289 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -393,18 +393,6 @@ dependencies = [ "serde_json", ] -[[package]] -name = "asap-precompute-rs" -version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPCollector?branch=main#1d8efd07e40fc151cbd4678a5c6aa9774b1aed34" -dependencies = [ - "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?branch=main)", - "prost", - "serde", - "serde_json", - "thiserror 1.0.69", -] - [[package]] name = "asap-sql-function-catalog" version = "0.1.0" @@ -431,6 +419,14 @@ dependencies = [ "tonic-build", ] +[[package]] +name = "asap_sketch_codec" +version = "0.1.0" +dependencies = [ + "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?branch=main)", + "prost", +] + [[package]] name = "asap_sketchlib" version = "0.3.0" @@ -1159,9 +1155,9 @@ dependencies = [ "arrow", "asap-aware-mapping", "asap-frontend-promql", - "asap-precompute-rs", "asap-types", "asap_otel_proto", + "asap_sketch_codec", "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?branch=main)", "asap_types", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index dc0e70f09..3079a97fe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ resolver = "2" members = [ "crates/asap_otel_proto", "crates/asap_types", + "crates/asap_sketch_codec", "data_plane", "control_plane", ] @@ -11,12 +12,6 @@ members = [ edition = "2021" version = "0.1.0" -# ASAPCollector's `asap-precompute-rs` currently declares Sketchlib as a -# relative path. When Collector is consumed from Git, resolve that dependency -# to the same Git-sourced Sketchlib package as the backend. -[patch."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/ProjectASAP/ASAPCollector"] -asap_sketchlib = { git = "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/ProjectASAP/asap_sketchlib", branch = "main" } - [workspace.dependencies] # Keep Planner frontends, selection, and IR on the same immutable revision. # Alias upstream asap-types because this workspace also defines asap_types. diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index f1e1b1df0..c01d614f3 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -247,6 +247,7 @@ pub async fn compile_automatic_clickhouse_workload( let mut entries = std::collections::BTreeMap::new(); let mut window_templates = std::collections::BTreeMap::>::new(); let mut installed_dags = std::collections::BTreeMap::new(); + let mut selected_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 { @@ -286,7 +287,13 @@ pub async fn compile_automatic_clickhouse_workload( "duplicate canonical SQL query identity".into(), )); } - installed_dags.insert(query.sql.clone(), installed); + selected_dags.insert(query.sql.clone(), installed.document.clone()); + installed_dags.insert( + query.sql.clone(), + installed + .maintenance_projection() + .map_err(ClickHousePlanningError::Lower)?, + ); } let configs: Vec<_> = materializations.into_values().collect(); let sds = SummaryCatalog::from_materializations( @@ -322,6 +329,7 @@ pub async fn compile_automatic_clickhouse_workload( tables: request.tables.clone(), accuracy: request.accuracy.clone(), }), + selected_dags, entries, }, }; @@ -423,6 +431,7 @@ pub async fn compile_clickhouse_workload( let mut entries = std::collections::BTreeMap::new(); let mut window_templates = std::collections::BTreeMap::>::new(); let mut installed_dags = std::collections::BTreeMap::new(); + let mut selected_dags = std::collections::BTreeMap::new(); for query in &request.queries { let planned = plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?; let template = planned.canonical_sql.clone(); @@ -430,7 +439,13 @@ pub async fn compile_clickhouse_workload( bind_selected_node(node, family, query, request) })?; index_sql_template(&mut window_templates, template, &executable); - installed_dags.insert(query.sql.clone(), installed); + selected_dags.insert(query.sql.clone(), installed.document.clone()); + installed_dags.insert( + query.sql.clone(), + installed + .maintenance_projection() + .map_err(ClickHousePlanningError::Lower)?, + ); let identity = QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &executable.canonical_query); if entries.insert(identity.clone(), executable).is_some() { @@ -454,6 +469,7 @@ pub async fn compile_clickhouse_workload( tables: request.tables.clone(), accuracy: request.accuracy.clone(), }), + selected_dags, entries, }, }; @@ -1641,10 +1657,16 @@ mod tests { installed.binding.nodes.len(), installed.document.nodes.len() ); - assert!(installed.binding.nodes.values().any(|binding| matches!( + assert!(!installed.binding.nodes.values().any(|binding| matches!( binding, crate::physical::executable_binding::BackendNodeBinding::Query { .. } ))); + assert!( + publication.query_plan.selected_dags[&request.queries[0].sql] + .nodes + .len() + > installed.document.nodes.len() + ); let entry = publication.query_plan.entries.values().next().unwrap(); // External SQL retains its literal time range until it can be bound. assert!(!entry.canonical_query.starts_with("moving-window-v1:")); diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 4b9dd1d5b..b17c45e56 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1852,10 +1852,11 @@ impl PhysicalPlanCompiler { }); } } - let query_plan = QueryPlan { + let mut query_plan = QueryPlan { plan_id, plan_version: envelope.plan_version, clickhouse_context: None, + selected_dags: BTreeMap::new(), entries: query_entries, }; let mut installed_dags = BTreeMap::new(); @@ -1884,6 +1885,9 @@ impl PhysicalPlanCompiler { query_id: query_id.clone(), reason, })?; + query_plan + .selected_dags + .insert(query_id.clone(), installed.document.clone()); installed_dags.insert(query_id, installed); } for materialization in &mut materializations { @@ -1935,6 +1939,18 @@ impl PhysicalPlanCompiler { } })?; } + // QueryPlan already owns executable readout nodes. Persist only + // maintenance ancestors under PrecomputePlan. + installed_dags = installed_dags + .into_iter() + .filter(|(_, installed)| !installed.binding.precompute_sinks.is_empty()) + .map(|(query_id, installed)| { + installed + .maintenance_projection() + .map(|projected| (query_id.clone(), projected)) + .map_err(|reason| CompileError::Query { query_id, reason }) + }) + .collect::>()?; let mut precompute_plan = match environment.target { PhysicalDeploymentTarget::DistributedCollectors => { PrecomputePlan::build(envelope.clone(), materializations, &producer_ids).and_then( @@ -3900,8 +3916,8 @@ pub(crate) mod tests { assert_eq!(populations.len(), 1); let installed = serde_json::to_string(&plan.precompute_plan.executable_dags).unwrap(); assert!( - installed.contains("MaintainPopulation"), - "shared state must originate in the installed Planner DAG" + !installed.contains("MaintainPopulation"), + "query-only population readout must not be executable maintenance" ); } @@ -4601,11 +4617,27 @@ pub(crate) mod tests { .precompute_plan .executable_dags .get(&entry.query_id) - .expect("compiled query retains its Planner DAG and backend placement"); + .expect("compiled query retains its maintenance projection"); installed.validate().expect("typed DAG document"); - crate::physical::executable_binding::validate_query_plan(installed, entry) - .expect("query node bindings"); + assert_eq!( + installed.document.schema_version, + asap_types::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION + ); + assert!(installed + .document + .nodes + .iter() + .all(|node| node.output_state.timing + == planner_types::post_asap::ExecutionTiming::MaintenanceTime)); assert_eq!(installed.binding.query_plan_sink, entry.root); + let mut mismatched = plan.to_publication_artifact().unwrap(); + let projected = mismatched + .precompute_plan + .executable_dags + .get_mut(&entry.query_id) + .unwrap(); + projected.binding.query_plan_sink = asap_types::executable_plan::QueryNodeId(u64::MAX); + assert!(mismatched.validate().is_err()); assert!(installed.binding.nodes.values().any(|placement| matches!( placement, crate::physical::executable_binding::BackendNodeBinding::Materialization { .. } diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 15ce225a9..e453f24b9 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -1050,6 +1050,7 @@ mod catalog_binding_tests { plan_id: 7, plan_version: 2, clickhouse_context: None, + selected_dags: Default::default(), entries: BTreeMap::from([(entry.canonical_query.clone(), entry)]), }, catalog, @@ -1328,6 +1329,7 @@ mod tests { tables: Default::default(), accuracy: planner_types::types::AccuracyTarget::Exact, }), + selected_dags: Default::default(), entries: [base, metricsql, clickhouse] .into_iter() .map(|entry| (QueryPlan::catalog_key(entry.language, "shared"), entry)) diff --git a/crates/asap_sketch_codec/Cargo.toml b/crates/asap_sketch_codec/Cargo.toml new file mode 100644 index 000000000..c2d728895 --- /dev/null +++ b/crates/asap_sketch_codec/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "asap_sketch_codec" +version.workspace = true +edition.workspace = true + +[dependencies] +asap_sketchlib = { git = "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/ProjectASAP/asap_sketchlib", branch = "main" } +prost = "0.13" diff --git a/crates/asap_sketch_codec/src/lib.rs b/crates/asap_sketch_codec/src/lib.rs new file mode 100644 index 000000000..e7414e323 --- /dev/null +++ b/crates/asap_sketch_codec/src/lib.rs @@ -0,0 +1,93 @@ +//! Runtime-independent decoding of the sketchlib protobuf envelope. + +use asap_sketchlib::proto::sketchlib::{ + sketch_envelope::SketchState, DdSketchState, KllState, SketchEnvelope, +}; +use asap_sketchlib::DdSketch; +use prost::Message; + +pub fn envelope_state(bytes: &[u8]) -> Result, String> { + SketchEnvelope::decode(bytes) + .map(|envelope| envelope.sketch_state) + .map_err(|error| format!("decode SketchEnvelope: {error}")) +} + +/// Accept the current full envelope and the supported legacy bare state. +pub fn ddsketch_state(bytes: &[u8]) -> Result<(DdSketchState, f64), String> { + match SketchEnvelope::decode(bytes) { + Ok(envelope) => match envelope.sketch_state { + Some(SketchState::Ddsketch(state)) => Ok((state, envelope.sample_p)), + Some(_) => Err("SketchEnvelope contains a non-DDSketch state".into()), + None => DdSketchState::decode(bytes) + .map(|state| (state, 1.0)) + .map_err(|error| format!("decode DdSketchState: {error}")), + }, + Err(_) => DdSketchState::decode(bytes) + .map(|state| (state, 1.0)) + .map_err(|error| format!("decode DdSketchState: {error}")), + } +} + +pub fn reconstruct_ddsketch(bytes: &[u8]) -> Result<(DdSketch, f64), String> { + let (state, sample_p) = ddsketch_state(bytes)?; + if !state.alpha.is_finite() || !(0.0..1.0).contains(&state.alpha) || state.alpha == 0.0 { + return Err("DDSketch alpha must be finite and between zero and one".into()); + } + Ok(( + DdSketch::from_raw(state.alpha, state.store_counts, state.store_offset), + sample_p, + )) +} + +pub fn kll_state(bytes: &[u8]) -> Result { + match SketchEnvelope::decode(bytes) { + Ok(envelope) => match envelope.sketch_state { + Some(SketchState::Kll(state)) => Ok(state), + Some(_) => Err("SketchEnvelope contains a non-KLL state".into()), + None => KllState::decode(bytes).map_err(|error| format!("decode KllState: {error}")), + }, + Err(_) => KllState::decode(bytes).map_err(|error| format!("decode KllState: {error}")), + } +} + +pub fn encode_ddsketch(sketch: &DdSketch) -> Vec { + let envelope = SketchEnvelope { + format_version: 1, + producer: None, + hash_spec: None, + sample_p: 0.0, + sketch_state: Some(SketchState::Ddsketch(DdSketchState { + alpha: sketch.wire_alpha(), + store_counts: sketch.store_counts.clone(), + store_offset: sketch.store_offset, + })), + }; + envelope.encode_to_vec() +} + +pub fn encode_kll(sketch: &asap_sketchlib::sketches::kll::KLL) -> Vec { + use asap_sketchlib::proto::sketchlib::CoinState; + let (state, bit_cache, remaining_bits) = sketch.wire_coin(); + SketchEnvelope { + format_version: 1, + producer: None, + hash_spec: None, + sample_p: 0.0, + sketch_state: Some(SketchState::Kll(KllState { + k: sketch.wire_k(), + m: sketch.wire_m(), + num_levels: sketch.wire_num_levels(), + levels: sketch.wire_levels(), + items: sketch.wire_items(), + coin: Some(CoinState { + state, + bit_cache, + remaining_bits, + }), + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), + })), + } + .encode_to_vec() +} diff --git a/crates/asap_types/src/derived_input.rs b/crates/asap_types/src/derived_input.rs index 964ee8d3f..0c10ff4bf 100644 --- a/crates/asap_types/src/derived_input.rs +++ b/crates/asap_types/src/derived_input.rs @@ -37,7 +37,11 @@ impl DerivedInputIdentity { root: PostAsapNodeId, frontiers: &BTreeMap, ) -> Result { - if document.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION { + if !matches!( + document.schema_version, + crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION + | crate::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION + ) { return Err("unsupported derived program document version".into()); } let decoded = document.decode()?; diff --git a/crates/asap_types/src/executable_plan.rs b/crates/asap_types/src/executable_plan.rs index 04647cd44..0b63b6e83 100644 --- a/crates/asap_types/src/executable_plan.rs +++ b/crates/asap_types/src/executable_plan.rs @@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize}; pub struct QueryNodeId(pub u64); pub const OWNED_POST_ASAP_DAG_SCHEMA_VERSION: u32 = 1; +pub const MAINTENANCE_DAG_SCHEMA_VERSION: u32 = 2; /// Versioned, language-neutral Planner DAG persisted with an installed plan. /// Plan lifecycle belongs to the enclosing `PrecomputePlan`; this document @@ -193,12 +194,58 @@ pub struct InstalledPostAsapDag { impl InstalledPostAsapDag { pub fn validate(&self) -> Result<(), String> { - if self.document.schema_version != OWNED_POST_ASAP_DAG_SCHEMA_VERSION - || self.document.query_id.trim().is_empty() - { + if self.document.query_id.trim().is_empty() { return Err("invalid post-ASAP DAG document identity/version".into()); } - self.binding.validate(&self.document.decode()?) + match self.document.schema_version { + OWNED_POST_ASAP_DAG_SCHEMA_VERSION => self + .binding + .validate(&self.document.decode()?) + .map_err(|error| format!("legacy DAG `{}`: {error}", self.document.query_id)), + MAINTENANCE_DAG_SCHEMA_VERSION => { + self.binding.validate_maintenance(&self.document.decode()?) + } + _ => Err("unsupported post-ASAP DAG document version".into()), + } + } + + /// Project the selected semantic DAG onto the maintenance ancestors of its + /// stored outputs. Version 1 remains readable for installed legacy plans. + pub fn maintenance_projection(mut self) -> Result { + self.validate()?; + if self.binding.precompute_sinks.is_empty() { + return Err("cannot project a DAG without maintenance sinks".into()); + } + let mut included = self + .binding + .precompute_sinks + .iter() + .copied() + .collect::>(); + let mut frontier = self.binding.precompute_sinks.clone(); + while let Some(consumer) = frontier.pop() { + for edge in self + .document + .edges + .iter() + .filter(|edge| edge.consumer == consumer) + { + if included.insert(edge.producer) { + frontier.push(edge.producer); + } + } + } + self.document + .nodes + .retain(|node| included.contains(&node.id)); + self.document + .edges + .retain(|edge| included.contains(&edge.producer) && included.contains(&edge.consumer)); + self.binding.nodes.retain(|id, _| included.contains(id)); + self.document.root = *self.binding.precompute_sinks.first().unwrap(); + self.document.schema_version = MAINTENANCE_DAG_SCHEMA_VERSION; + self.validate()?; + Ok(self) } } @@ -234,6 +281,65 @@ impl BackendExecutableBinding { self.nodes.get(&id) } + /// Scheduler validation for both the legacy complete DAG and a projected + /// maintenance DAG. The installed document version is checked at staging. + pub fn validate_precompute_execution(&self, dag: &ExecutableDag) -> Result<(), String> { + if dag.nodes.iter().any(|node| node.id == self.query_sink) { + self.validate(dag) + } else { + self.validate_maintenance(dag) + } + } + + fn validate_maintenance(&self, dag: &ExecutableDag) -> Result<(), String> { + let ids = dag + .nodes + .iter() + .map(|node| node.id) + .collect::>(); + if ids.is_empty() || self.nodes.keys().copied().collect::>() != ids { + return Err("maintenance binding does not cover its projected DAG".into()); + } + if self.precompute_sinks.is_empty() + || self.precompute_sinks.iter().any(|id| !ids.contains(id)) + { + return Err("maintenance projection has missing sinks".into()); + } + if self.precompute_sinks.iter().any(|id| { + !matches!( + self.node(*id), + Some(BackendNodeBinding::Materialization { .. }) + ) + }) { + return Err("maintenance sink lacks a stored-output binding".into()); + } + for node in &dag.nodes { + match (node.output_state.timing, self.node(node.id)) { + ( + ExecutionTiming::MaintenanceTime, + Some( + BackendNodeBinding::MaintenanceInput + | BackendNodeBinding::Materialization { .. }, + ), + ) => {} + _ => { + return Err(format!( + "query-owned node {} in maintenance projection", + node.id.0 + )) + } + } + } + if dag + .edges + .iter() + .any(|edge| !ids.contains(&edge.producer) || !ids.contains(&edge.consumer)) + { + return Err("maintenance projection has dangling edges".into()); + } + Ok(()) + } + pub fn validate(&self, dag: &ExecutableDag) -> Result<(), String> { let semantic = dag .nodes diff --git a/crates/asap_types/src/plan_publication.rs b/crates/asap_types/src/plan_publication.rs index 3631b6cef..e10fda7be 100644 --- a/crates/asap_types/src/plan_publication.rs +++ b/crates/asap_types/src/plan_publication.rs @@ -33,6 +33,96 @@ pub struct PhysicalPlanInstallRequest { pub adaptation_evidence: Vec, } +impl PhysicalPlanInstallRequest { + /// Convert supported complete-DAG artifacts into the split runtime form + /// before staging. The selected document remains query provenance. + pub fn normalize_legacy_dags(&mut self) -> Result<(), String> { + let mut normalized = std::collections::BTreeMap::new(); + for (query_id, installed) in &self.precompute_plan.executable_dags { + if installed.document.schema_version + != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION + { + normalized.insert(query_id.clone(), installed.clone()); + continue; + } + installed.validate()?; + let selected = installed.document.clone(); + if selected.query_id != *query_id { + return Err("legacy DAG key differs from its query identity".into()); + } + if let Some(existing) = self.query_plan.selected_dags.get(query_id) { + if existing != &selected { + return Err("legacy DAG conflicts with selected query provenance".into()); + } + } else { + self.query_plan + .selected_dags + .insert(query_id.clone(), selected); + } + if !installed.binding.precompute_sinks.is_empty() { + normalized.insert( + query_id.clone(), + installed.clone().maintenance_projection()?, + ); + } + } + self.precompute_plan.executable_dags = normalized; + Ok(()) + } +} + +/// A projected writer must refer to the query entry installed in the same +/// generation. Legacy complete DAGs retain their existing validation path. +pub fn validate_maintenance_query_bindings( + precompute: &PrecomputePlan, + query: &QueryPlan, +) -> Result<(), String> { + for (query_id, installed) in &precompute.executable_dags { + if installed.document.schema_version + != crate::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION + { + continue; + } + let mut entries = query + .entries + .values() + .filter(|entry| &entry.query_id == query_id); + let entry = entries + .next() + .ok_or("maintenance projection has no query entry")?; + if entries.next().is_some() { + return Err("maintenance projection has ambiguous query entries".into()); + } + if entry.root != installed.binding.query_plan_sink { + return Err("maintenance projection and query entry have different roots".into()); + } + let selected = query + .selected_dags + .get(query_id) + .ok_or("maintenance projection has no selected semantic provenance")?; + if selected.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION + || selected.query_id != *query_id + { + return Err("selected semantic provenance has invalid identity/version".into()); + } + selected.decode()?; + if installed + .document + .nodes + .iter() + .any(|node| !selected.nodes.contains(node)) + || installed + .document + .edges + .iter() + .any(|edge| !selected.edges.contains(edge)) + { + return Err("maintenance projection differs from its selected DAG".into()); + } + } + Ok(()) +} + impl PhysicalPlanPublication { /// Validate every plan against the shared catalog snapshot. pub fn validate(&self) -> Result<(), String> { @@ -49,6 +139,7 @@ impl PhysicalPlanPublication { self.query_plan .validate_against_catalog(catalog) .map_err(|e| e.to_string())?; + validate_maintenance_query_bindings(&self.precompute_plan, &self.query_plan)?; let materializations = self .precompute_plan .materializations diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index a219ff03b..ad0c7b8c6 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -115,8 +115,8 @@ pub struct PrecomputePlan { pub schemas: Vec, pub producers: Vec, pub materializations: Vec, - /// Planner semantic DAGs and backend-owned placement for this generation. - /// Empty only for legacy/config-only construction paths. + /// Version 2 holds maintenance projections ending at stored outputs. + /// Version 1 complete DAGs remain readable for installed legacy plans. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub executable_dags: BTreeMap, } diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs index 5bdd2605a..274626bf1 100644 --- a/crates/asap_types/src/query_plan.rs +++ b/crates/asap_types/src/query_plan.rs @@ -27,6 +27,10 @@ pub struct QueryPlan { pub plan_version: u64, #[serde(default, skip_serializing_if = "Option::is_none")] pub clickhouse_context: Option, + /// Selected semantic roots retained for provenance; serving executes + /// `entries` and never reconstructs a plan from these documents. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub selected_dags: BTreeMap, pub entries: BTreeMap, } @@ -55,6 +59,7 @@ impl QueryPlan { plan_id: 0, plan_version: 0, clickhouse_context: None, + selected_dags: BTreeMap::new(), entries: BTreeMap::new(), } } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 60d4549a0..7482c5210 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -6,6 +6,7 @@ edition.workspace = true [dependencies] # Internal crates (workspace) asap_types.workspace = true +asap_sketch_codec = { path = "../crates/asap_sketch_codec" } # Phase 9: the control plane is now an in-process library inside the # backend binary. Wiring up the in-process OpAMP server + capability-map # exposure is a follow-up after Phase 4 (centralized series_id @@ -72,9 +73,6 @@ asap_sketchlib = { git = "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/ProjectASAP/asap_sketchlib", branch # existing PUBLIC `from_legacy_matrix`/`sketch_matrix`/`topk_heap_items` API. # Already in the lock as a transitive dep of `asap_sketchlib`. rmp-serde = "1.3" -# Shared collector ingest implementation. This follows ASAPCollector's -# current main branch alongside the direct Sketchlib dependency above. -asap-precompute-rs = { git = "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/ProjectASAP/ASAPCollector", branch = "main" } # Persistence layer (SketchStore parts / manifest / Tier-2 cache) moka = { version = "0.12", features = ["sync"] } memmap2 = "0.9" diff --git a/data_plane/examples/audit_clickhouse_fallback.rs b/data_plane/examples/audit_clickhouse_fallback.rs index 81781851b..eb057edc9 100644 --- a/data_plane/examples/audit_clickhouse_fallback.rs +++ b/data_plane/examples/audit_clickhouse_fallback.rs @@ -89,6 +89,7 @@ async fn main() { tables: HashMap::from([("raw_samples".into(), schema)]), accuracy: AccuracyTarget::Epsilon(0.01), }), + selected_dags: Default::default(), entries: BTreeMap::new(), }; let active = validate_and_build_runtime_plan( diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 8de15bfd4..f1e698426 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -2565,71 +2565,20 @@ fn decode_modified_otlp_sketch_bytes( match encoding { ENCODING_PROTO => match algorithm { - // Phase 3 step 3: DDSketch and KLL envelope-parsing / - // sketch reconstruction route through the shared - // `edge_runtime_adapter`, which delegates to - // `asap-precompute-rs`'s `Sketch` trait. Backend's - // accumulator wraps the result. Byte parity with Go is - // covered by `asap_sketchlib` PRs #40 (DDSketch) and #41 - // (KLL). - // - // HLL / CountSketch / CountMinSketch byte parity is - // tracked under ProjectASAP/ASAPCollector#243 — until it - // lands those three sketches keep using the backend's - // existing per-accumulator decoder. + // The neutral codec accepts both full envelopes and supported bare + // states. Query accumulators retain their family-specific readouts. SketchAlgorithm::DDSketch => { - use crate::precompute_engine::operators::edge_runtime_adapter::{ - reconstruct_via_runtime, ReconstructedSketch, SketchType as RtSketchType, - }; - // Prefer the asap-precompute-rs runtime path (envelope- - // wrapped bytes, the canonical edge-framework wire format). - // If the input is a bare `DdSketchState` (as some unit-test - // / pre-envelope agent payloads still emit, mirrored by the - // PR #14 contract on `from_sketchlib_proto_bytes`), the - // adapter returns an error decoding the envelope — fall - // back to the backend's native decoder which already - // accepts both shapes. - match reconstruct_via_runtime(RtSketchType::DDSketch, bytes) { - Ok(ReconstructedSketch::DdSketch(inner)) => { - // The runtime reconstruction discards the envelope's - // sample_p; re-read it from the same full-frame bytes - // so a sampled series rescales its Count by 1/p. - let sample_p = DDSketchAccumulator::sample_p_from_envelope_bytes(bytes); - Ok(Box::new(DDSketchAccumulator { inner, sample_p })) - } - Ok(_) => { - Err("edge_runtime_adapter returned non-DDSketch reconstruction".into()) - } - Err(_) => Ok(Box::new(DDSketchAccumulator::from_sketchlib_proto_bytes( - bytes, - )?)), - } - } - SketchAlgorithm::Kll => { - use crate::precompute_engine::operators::edge_runtime_adapter::{ - reconstruct_via_runtime, ReconstructedSketch, SketchType as RtSketchType, + let (inner, sample_p) = asap_sketch_codec::reconstruct_ddsketch(bytes)?; + let sample_p = if sample_p.is_finite() && sample_p > 0.0 && sample_p < 1.0 { + sample_p + } else { + 1.0 }; - // Same envelope-vs-bare-state handling as DDSketch above. - // Backend's KLL accumulator owns the wire-format-aligned - // `KllSketch` rather than the high-throughput `KLL` - // that asap-precompute-rs's `KLLWrapper` wraps internally - // — when the adapter succeeds, bridge by re-feeding the - // wrapper's snapshot bytes through backend's existing - // decoder. The envelope work (decode + state extraction - // + reconstruction) has already happened in the runtime - // adapter; this final step just reshapes into backend's - // accumulator type. On envelope-decode failure (bare - // state bytes) fall through to the native decoder. - match reconstruct_via_runtime(RtSketchType::KLLSketch, bytes) { - Ok(ReconstructedSketch::Kll { snapshot_bytes }) => Ok(Box::new( - DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(&snapshot_bytes)?, - )), - Ok(_) => Err("edge_runtime_adapter returned non-KLL reconstruction".into()), - Err(_) => Ok(Box::new( - DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(bytes)?, - )), - } + Ok(Box::new(DDSketchAccumulator { inner, sample_p })) } + SketchAlgorithm::Kll => Ok(Box::new( + DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(bytes)?, + )), SketchAlgorithm::Cms => Ok(Box::new( CountMinSketchAccumulator::from_sketchlib_proto_bytes(bytes)?, )), diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index a7ec99101..47c15f465 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -2764,6 +2764,7 @@ mod tests { plan_id: 7, plan_version: 1, clickhouse_context: None, + selected_dags: Default::default(), entries: Default::default(), }), storage_routing: Arc::new( @@ -6068,10 +6069,11 @@ pub use asap_types::plan_publication::PhysicalPlanInstallRequest; /// Decode and cross-validate every backend view before it can become visible. /// Used by both startup artifact loading and the staged HTTP install path. pub fn validate_and_build_runtime_plan( - request: PhysicalPlanInstallRequest, + mut request: PhysicalPlanInstallRequest, default_routing: Arc, ) -> Result { use std::collections::BTreeSet; + request.normalize_legacy_dags()?; request .precompute_plan .validate_against_catalog(&request.summary_catalog) @@ -6153,6 +6155,10 @@ pub fn validate_and_build_runtime_plan( .query_plan .validate(&typed_fps) .map_err(|error| format!("QueryPlan validation error: {error}"))?; + asap_types::plan_publication::validate_maintenance_query_bindings( + &request.precompute_plan, + &request.query_plan, + )?; let storage_routing = match request.storage_routing.as_ref() { Some(value) => Arc::new( crate::storage_engines::types::BackendStorageRouting::from_json_payload(value) @@ -7195,6 +7201,53 @@ mod catalog_install_tests { ) } + #[test] + fn legacy_complete_dag_is_normalized_before_install() { + use asap_types::executable_plan::{BackendNodeBinding, InstalledPostAsapDag}; + + let mut request = request(); + let (query_id, projected) = request + .precompute_plan + .executable_dags + .iter() + .next() + .map(|(id, dag)| (id.clone(), dag.clone())) + .expect("fixture has a maintained summary"); + let selected = request.query_plan.selected_dags.remove(&query_id).unwrap(); + let semantic = selected.decode().unwrap(); + let mut binding = projected.binding; + binding.nodes = semantic + .nodes + .iter() + .map(|node| { + ( + node.id, + binding + .nodes + .get(&node.id) + .cloned() + .unwrap_or(BackendNodeBinding::QueryInput), + ) + }) + .collect(); + request.precompute_plan.executable_dags.insert( + query_id.clone(), + InstalledPostAsapDag { + document: selected, + binding, + }, + ); + + let installed = install(request).unwrap(); + assert_eq!( + installed.precompute_plan.executable_dags[&query_id] + .document + .schema_version, + asap_types::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION + ); + assert!(installed.query_plan.selected_dags.contains_key(&query_id)); + } + #[test] fn invalid_clickhouse_entry_cannot_change_active_generation() { let active = install(request()).expect("baseline plan installs"); diff --git a/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs b/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs index 86b95a5ba..20421ccb0 100644 --- a/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs @@ -56,45 +56,11 @@ impl DatasketchesKLLAccumulator { /// DataCollector's `kllprocessor` emits when /// `encoding = KLL_SKETCH_ENCODING_PROTO`. /// - /// ⚠ This is a **lossy statistical reconstruction**, not a - /// bit-identical round-trip: the `KllState` proto carries the - /// retained items in level order plus an explicit `levels[]` - /// boundary array, but sketch-core's `KllSketch` backend types - /// keep their level structure private. Rather than touch - /// upstream `asap_sketchlib` to add a typed-state constructor, - /// we build a fresh `DatasketchesKLLAccumulator` with the same - /// `k` and replay every retained item through `update()`. - /// Quantile estimates on the reconstructed sketch are - /// approximately equivalent to the source's — within KLL's - /// own rank-error bound, which is the same bound the source - /// already inherited — so Phase 1 hot-path queries that hit - /// the reconstructed sketch return answers the user would - /// already have accepted from the source. Bit-identical - /// reconstruction is tracked as a sketchlib upstream follow-up. + /// The neutral codec decodes the full envelope or legacy bare state. + /// The level-aware constructor below preserves the supplied retained + /// sample layout without replaying updates. pub fn from_sketchlib_proto_bytes(buffer: &[u8]) -> Result> { - use asap_sketchlib::proto::sketchlib::{sketch_envelope, KllState, SketchEnvelope}; - use prost::Message; - - // DataCollector's kllprocessor wraps the state in a - // `SketchEnvelope{kll: KllState}` via sketchlib-go's - // `SerializePortableFO` + `proto.Marshal`. Try envelope first, - // fall back to bare `KllState` for callers (e.g. unit tests) - // that encode the state directly. Mirrors the PR #14 fix on - // `CountMinSketchAccumulator::from_sketchlib_proto_bytes`. - let state = match SketchEnvelope::decode(buffer) { - Ok(env) => match env.sketch_state { - Some(sketch_envelope::SketchState::Kll(st)) => st, - Some(other) => { - return Err(format!( - "SketchEnvelope contains non-KLL sketch: {:?}", - std::mem::discriminant(&other) - ) - .into()); - } - None => KllState::decode(buffer).map_err(|e| format!("decode KllState: {e}"))?, - }, - Err(_) => KllState::decode(buffer).map_err(|e| format!("decode KllState: {e}"))?, - }; + let state = asap_sketch_codec::kll_state(buffer)?; if state.k < 8 { return Err(format!("KllState.k must be >= 8 (got {})", state.k).into()); } diff --git a/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs b/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs index 926a1e883..8fad2209a 100644 --- a/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs @@ -91,42 +91,7 @@ impl DDSketchAccumulator { /// DataCollector's `ddsketchprocessor` emits when /// `encoding = DD_SKETCH_ENCODING_PROTO`. pub fn from_sketchlib_proto_bytes(buffer: &[u8]) -> Result> { - use asap_sketchlib::proto::sketchlib::{sketch_envelope, DdSketchState, SketchEnvelope}; - use prost::Message; - - // DataCollector's ddsketchprocessor wraps the state in a - // `SketchEnvelope{ddsketch: DdSketchState}` via sketchlib-go's - // `SerializePortableFO` + `proto.Marshal`. Try envelope first, - // fall back to bare `DdSketchState` for callers (e.g. unit - // tests) that encode the state directly. Mirrors the PR #14 - // fix on `CountMinSketchAccumulator::from_sketchlib_proto_bytes`. - // Capture the envelope's `sample_p` alongside the state so a `Count` - // query can rescale by `1/p`. Bare `DdSketchState` bytes (no envelope) - // carry no sampling info → `sample_p` 1.0 (no rescale). - let (state, sample_p) = match SketchEnvelope::decode(buffer) { - Ok(env) => { - let sp = env.sample_p; - match env.sketch_state { - Some(sketch_envelope::SketchState::Ddsketch(st)) => (st, sp), - Some(other) => { - return Err(format!( - "SketchEnvelope contains non-DDSketch sketch: {:?}", - std::mem::discriminant(&other) - ) - .into()); - } - None => ( - DdSketchState::decode(buffer) - .map_err(|e| format!("decode DDSketchState: {e}"))?, - 1.0, - ), - } - } - Err(_) => ( - DdSketchState::decode(buffer).map_err(|e| format!("decode DDSketchState: {e}"))?, - 1.0, - ), - }; + let (state, sample_p) = asap_sketch_codec::ddsketch_state(buffer)?; if !(state.alpha > 0.0 && state.alpha < 1.0) { return Err(format!( "DDSketchState alpha {} out of range (expected 0 < alpha < 1)", diff --git a/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs b/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs deleted file mode 100644 index fba53304a..000000000 --- a/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs +++ /dev/null @@ -1,391 +0,0 @@ -//! Edge-runtime adapter — Phase 3 step 3 of the ASAP edge-framework -//! migration (`docs/design-asap-edge-framework.md`, ADR-0002). -//! -//! Routes the **shared** ingest work (envelope wire-format parsing, -//! per-sketch state extraction, sketch reconstruction, sketch merge) -//! through the [`asap_precompute_rs`] crate so the same code runs in -//! agents (Rust edge runtime) and the backend (this repo). What stays -//! in this repo: the QUERY-side engine — PromQL aggregation, storage, -//! and query planning. See README §"Ingest path consumes -//! asap-precompute-rs" for the contract. -//! -//! # What's shared -//! -//! - **Envelope wire-format parsing.** The runtime view -//! [`SketchEnvelope`] (renamed from a backend-internal struct to -//! asap-precompute-rs's canonical type) decodes the on-the-wire -//! `asap_sketchlib::proto::sketchlib::SketchEnvelope` proto plus the -//! surrounding metadata (window bounds, `agg_id`, encoding tag, -//! sketch type, host-neutral labels, metric name, count, -//! temporality). -//! - **Per-sketch state extraction.** [`unwrap_envelope_state`] dispatches -//! the [`asap_sketchlib::proto::sketchlib::sketch_envelope::SketchState`] -//! oneof into a typed `*State` proto for the requested sketch type, -//! producing the same diagnostic shape every accumulator's -//! `from_sketchlib_proto_bytes` used to repeat by hand. -//! - **Sketch reconstruction (DDSketch + KLL today).** -//! [`reconstruct_via_runtime`] constructs an asap-precompute-rs -//! `*Wrapper`, runs `Sketch::apply_delta(envelope_bytes)` (which is -//! the Layer-3 runtime's reconstruct-from-bytes path), then extracts -//! the underlying `asap_sketchlib::sketches::*` state via -//! `wrapper.inner().clone()`. DDSketch and KLL byte parity holds in -//! `asap_sketchlib::main` (PRs #40, #41); HLL / CountSketch / -//! CountMinSketch are tracked under -//! ProjectASAP/ASAPCollector#243 — until those land we keep the -//! backend's per-accumulator decoders for the three sketches and -//! only delegate envelope-parsing. -//! - **Cross-runtime sketch merge.** [`merge_via_runtime`] uses -//! asap-precompute-rs's `Sketch::merge` + `Sketch::snapshot` round- -//! trip so the merge logic lives in one place. Identical results to -//! `asap_sketchlib::sketches::*::merge_refs` because both call into -//! the same underlying merge implementation. -//! -//! # What stays put -//! -//! - The backend's **per-accumulator query-side surface** (`AggregateCore`, -//! `query_statistic`, `MergeableAccumulator`, ...) — query-side and -//! not what asap-precompute-rs is for. -//! - **Sparse delta application** (`apply_proto_delta_bytes` on the -//! backend's accumulators) — `asap_sketchlib` doesn't yet expose the -//! `compute_delta` family (Go's `sketchlib-go` has it; tracked -//! upstream), so asap-precompute-rs's wrappers fall back to "always -//! full" delta encoding. Backend's typed-delta apply is independent -//! and stays. - -use asap_precompute_rs::{ - envelope::ProtoSketchEnvelope, sketches::DDSketchWrapper, sketches::KLLWrapper, Sketch, -}; -use asap_sketchlib::proto::sketchlib::sketch_envelope::SketchState; -use prost::Message; - -/// Re-export of asap-precompute-rs's runtime view of the -/// `SketchEnvelope` proto (the prost-generated wire-format type plus -/// surrounding host-neutral metadata: window bounds, `agg_id`, -/// encoding tag, sketch-type tag, labels, metric name, count, -/// temporality). -/// -/// Kept as a re-export so all backend code that touches the runtime -/// envelope reaches for the canonical type from the edge-framework -/// runtime — preventing a "backend-flavored" runtime view from -/// drifting alongside the agent-flavored one. -pub use asap_precompute_rs::envelope::{Encoding, SketchEnvelope, SketchType}; - -/// Re-export the [`asap_precompute_rs::Sketch`] trait family so backend -/// code that wants to operate on envelope bytes via the host-neutral -/// runtime trait (`snapshot`, `apply_delta`, `merge`, `reset`) imports -/// from one place. -pub use asap_precompute_rs::{CardinalitySketch, FrequencySketch, QuantileSketch}; - -/// Decode the wire-format `asap_sketchlib` `SketchEnvelope` proto from -/// `bytes` and return the inner [`SketchState`] oneof variant. -/// -/// Mirrors the per-accumulator `match SketchEnvelope::decode(buffer) -/// → Some(SketchState::X(st)) → st` ladder that every -/// `from_sketchlib_proto_bytes` used to inline. The fall-back to -/// decoding `bytes` as the bare typed-state proto is preserved at the -/// caller so the existing PR #14 contract continues to work for unit -/// tests that encode the state directly (without an envelope wrapper). -/// -/// This is the **single shared envelope-unwrap path** between -/// asap-precompute-rs and backend ingest. asap-precompute-rs's -/// per-wrapper `decode_envelope` does the same prost-decode + -/// oneof-extraction work; the difference is that asap-precompute-rs -/// then constructs a typed `asap_sketchlib::sketches::*` from the -/// state, which the backend may or may not want depending on the -/// downstream caller. -pub fn unwrap_envelope_state( - bytes: &[u8], -) -> Result, Box> { - let env = - ProtoSketchEnvelope::decode(bytes).map_err(|e| format!("decode SketchEnvelope: {e}"))?; - Ok(env.sketch_state) -} - -/// Result of [`reconstruct_via_runtime`] — backend uses the inner -/// `asap_sketchlib::sketches::*` to construct its own accumulator. -pub enum ReconstructedSketch { - /// Reconstructed [`asap_sketchlib::DdSketch`] state. - DdSketch(asap_sketchlib::DdSketch), - /// Reconstructed KLL — asap-precompute-rs's [`KLLWrapper`] owns - /// the high-throughput `asap_sketchlib::sketches::kll::KLL` - /// internally; backend's KLL accumulator wraps the wire-format- - /// aligned `KllSketch` instead. We surface the wrapper's - /// re-snapshot bytes so the caller can route them through - /// backend's existing `KllSketch::deserialize_msgpack` / - /// proto-state path or replay items via `KllSketch::update()`. - Kll { - /// Bytes of the reconstructed sketch's snapshot — same shape - /// as the input envelope, validated round-trip. - snapshot_bytes: Vec, - }, -} - -/// Reconstruct an `asap_sketchlib` sketch from envelope bytes by -/// delegating envelope parsing + sketch construction to -/// asap-precompute-rs's `Sketch` trait family. -/// -/// The function is sketch-type-aware because the wrappers' constructor -/// parameters (alpha for DDSketch, k+seed for KLL, ...) live partly in -/// the envelope state proto. We peek the state, construct a wrapper -/// with matching parameters, then call [`Sketch::apply_delta`] which -/// runs asap-precompute-rs's canonical decode + reconstruct pathway. -/// -/// Today wires DDSketch (byte parity per asap_sketchlib#40) and KLL -/// (byte parity per asap_sketchlib#41). HLL / CountSketch / -/// CountMinSketch are tracked under ProjectASAP/ASAPCollector#243 — -/// callers fall back to backend's per-accumulator decoder for those. -pub fn reconstruct_via_runtime( - sketch_type: SketchType, - envelope_bytes: &[u8], -) -> Result> { - match sketch_type { - SketchType::DDSketch => { - // Peek the state to learn alpha, then construct the - // wrapper with matching alpha so `apply_delta`'s merge - // step doesn't trip on `DdSketch::merge`'s alpha-equality - // guard. - let state = unwrap_envelope_state(envelope_bytes)?; - let alpha = match &state { - Some(SketchState::Ddsketch(s)) => s.alpha, - Some(_) => { - return Err("envelope is not a DDSketch".into()); - } - None => return Err("envelope has no sketch_state".into()), - }; - if !(alpha > 0.0 && alpha < 1.0) { - return Err(format!("DDSketch alpha {alpha} out of (0,1)").into()); - } - let mut wrapper = DDSketchWrapper::new(alpha); - wrapper - .apply_delta(envelope_bytes) - .map_err(|e| format!("DDSketchWrapper apply_delta: {e}"))?; - Ok(ReconstructedSketch::DdSketch(wrapper.inner().clone())) - } - SketchType::KLLSketch => { - // KLL: peek `k`, construct an empty wrapper, apply. - let state = unwrap_envelope_state(envelope_bytes)?; - let k = match state { - Some(SketchState::Kll(s)) => { - if s.k > i32::MAX as u32 { - return Err(format!("KllState.k too large: {}", s.k).into()); - } - s.k as i32 - } - Some(_) => return Err("envelope is not a KLL".into()), - None => return Err("envelope has no sketch_state".into()), - }; - let mut wrapper = KLLWrapper::new(k, None); - wrapper - .apply_delta(envelope_bytes) - .map_err(|e| format!("KLLWrapper apply_delta: {e}"))?; - // Re-snapshot via the wrapper's `Sketch::snapshot` — - // canonical asap-precompute-rs encode of the reconstructed - // state. Backend's KLL accumulator can then re-decode via - // its existing `from_sketchlib_proto_bytes` path; the - // edge-runtime adapter has done the envelope + state - // unwrap, the wire-format reshape, and the reconstruction - // round-trip. - let snapshot_bytes = wrapper - .snapshot() - .map_err(|e| format!("KLLWrapper snapshot: {e}"))?; - Ok(ReconstructedSketch::Kll { snapshot_bytes }) - } - SketchType::HLLSketch | SketchType::CountSketch | SketchType::CountMinSketch => { - Err(format!( - "reconstruct_via_runtime({sketch_type:?}): byte parity for \ - HLL / CountSketch / CountMinSketch not yet in upstream \ - asap_sketchlib — tracked at ProjectASAP/ASAPCollector#243. \ - Caller must fall back to backend's per-accumulator decoder." - ) - .into()) - } - SketchType::Unspecified => Err("reconstruct_via_runtime: SketchType::Unspecified".into()), - } -} - -/// Snapshot a backend-side `asap_sketchlib::DdSketch` through -/// asap-precompute-rs's `Sketch` trait — the canonical encode path -/// shared with the agent runtime. Used by the round-trip test -/// (`tests/edge_runtime_adapter.rs`). -pub fn snapshot_ddsketch_via_runtime( - sk: &asap_sketchlib::DdSketch, -) -> Result, Box> { - let mut wrapper = DDSketchWrapper::new(sk.alpha); - // Bridge into the wrapper by merging in the existing sketch. - // We can't move-construct the wrapper from a non-empty `DdSketch`, - // but `Sketch::apply_delta` against the existing snapshot bytes - // is equivalent. - if sk.total_count() > 0 { - // Re-encode the source's state into the canonical envelope - // shape that asap-precompute-rs's wrapper recognizes, then - // round-trip through `apply_delta`. Mirrors the agent runtime's - // own merge path. - let bridge_bytes = encode_ddsketch_envelope(sk); - wrapper - .apply_delta(&bridge_bytes) - .map_err(|e| format!("DDSketchWrapper apply_delta (bridge): {e}"))?; - } - wrapper - .snapshot() - .map_err(|e| format!("DDSketchWrapper snapshot: {e}").into()) -} - -/// Encode a backend-side `DdSketch` as the same `SketchEnvelope` proto -/// shape that asap-precompute-rs's `DDSketchWrapper::snapshot` emits. -/// -/// Matches asap-precompute-rs's `DDSketchWrapper::build_state` + -/// `encode_envelope` byte-for-byte — they call into the same -/// `asap_sketchlib::proto::sketchlib::*` types. Lives here so the -/// backend's existing accumulators don't need to import the wrapper -/// internals. -pub fn encode_ddsketch_envelope(sk: &asap_sketchlib::DdSketch) -> Vec { - use asap_sketchlib::proto::sketchlib::{ - sketch_envelope, DdSketchState, SketchEnvelope as ProtoEnvelope, - }; - let state = DdSketchState { - // Use `wire_alpha` so the bytes match Go's - // `sketchlib-go::DDSketch.SerializePortable` (PR - // asap_sketchlib#40 closes this). - alpha: sk.wire_alpha(), - store_counts: sk.store_counts.clone(), - store_offset: sk.store_offset, - // The DataPoint-level scalars (count/sum/min/max) were dropped from - // `DDSketchState` (ProjectASAP/sketchlib-go#243 / asap_sketchlib#57); - // the bucket counts carry all reconstructable state. - }; - let env = ProtoEnvelope { - format_version: 1, - producer: None, - hash_spec: None, - // No edge sampling on this path: 0.0 is the proto3 default (dual-read as - // 1.0) so the encoded envelope stays byte-identical. - sample_p: 0.0, - sketch_state: Some(sketch_envelope::SketchState::Ddsketch(state)), - }; - let mut buf = Vec::with_capacity(env.encoded_len()); - env.encode(&mut buf).expect("prost encode"); - buf -} - -#[cfg(test)] -/// Merge two `DdSketch` instances by routing through asap-precompute-rs's -/// runtime `Sketch::merge`. The result is byte-identical to -/// `asap_sketchlib::DdSketch::merge_refs(&[a, b])` because -/// both paths call the same underlying merge logic. -/// -/// Used by the cross-runtime parity test -/// (`tests/edge_runtime_adapter.rs::ddsketch_merge_via_runtime_matches_native`). -pub fn merge_ddsketches_via_runtime( - a: &asap_sketchlib::DdSketch, - b: &asap_sketchlib::DdSketch, -) -> Result> { - if (a.alpha - b.alpha).abs() > f64::EPSILON { - return Err(format!( - "merge_ddsketches_via_runtime: alpha mismatch ({} vs {})", - a.alpha, b.alpha - ) - .into()); - } - let mut wrapper_a = DDSketchWrapper::new(a.alpha); - if a.total_count() > 0 { - let bridge = encode_ddsketch_envelope(a); - wrapper_a - .apply_delta(&bridge) - .map_err(|e| format!("merge_ddsketches_via_runtime/a: {e}"))?; - } - let mut wrapper_b = DDSketchWrapper::new(b.alpha); - if b.total_count() > 0 { - let bridge = encode_ddsketch_envelope(b); - wrapper_b - .apply_delta(&bridge) - .map_err(|e| format!("merge_ddsketches_via_runtime/b: {e}"))?; - } - // `Sketch::merge` takes a `&dyn Sketch` (round-trips through - // snapshot bytes), which is the canonical Layer-3 runtime fold. - wrapper_a - .merge(&wrapper_b) - .map_err(|e| format!("DDSketchWrapper merge: {e}"))?; - Ok(wrapper_a.inner().clone()) -} - -#[cfg(test)] -mod tests { - use super::*; - - /// asap-precompute-rs's wrapper produces an envelope; backend's - /// shared envelope-unwrap path returns the matching oneof variant. - /// The dedup target. - #[test] - fn unwrap_envelope_state_matches_wrapper_output() { - let mut w = DDSketchWrapper::new(0.01); - for i in 1..=10 { - w.update(i as f64); - } - let bytes = w.snapshot().expect("snapshot ok"); - let state = unwrap_envelope_state(&bytes).expect("decode ok"); - match state { - Some(SketchState::Ddsketch(s)) => { - // `count` was dropped from `DdSketchState` - // (ProjectASAP/sketchlib-go#243 / asap_sketchlib#57); - // a non-empty sketch carries it in the bucket store. - assert!(s.store_counts.iter().sum::() > 0); - assert!(s.alpha > 0.0 && s.alpha < 1.0); - } - other => panic!("expected DDSketch state, got {other:?}"), - } - } - - /// asap-precompute-rs's wrapper produces an envelope; the runtime - /// adapter's reconstruction returns a backend-shaped - /// `asap_sketchlib::DdSketch` whose serialized bytes - /// (re-encoded through the same envelope shape) match the - /// original. - #[test] - fn ddsketch_round_trip_through_runtime_adapter() { - let mut w = DDSketchWrapper::new(0.01); - for i in 1..=100 { - w.update(i as f64); - } - let original_bytes = w.snapshot().expect("snapshot ok"); - let reconstructed = - reconstruct_via_runtime(SketchType::DDSketch, &original_bytes).expect("reconstruct ok"); - let dd = match reconstructed { - ReconstructedSketch::DdSketch(d) => d, - ReconstructedSketch::Kll { .. } => panic!("got KLL, expected DDSketch"), - }; - assert_eq!(dd.total_count(), 100); - let re_encoded = encode_ddsketch_envelope(&dd); - assert_eq!( - re_encoded, original_bytes, - "round-trip via runtime adapter must be byte-identical" - ); - } - - /// Construct two non-overlapping DDSketches, merge via the runtime - /// adapter, and verify counts add up. Uses asap-precompute-rs's - /// `Sketch::merge` + `Sketch::snapshot` round-trip. - #[test] - fn ddsketch_merge_via_runtime_combines_counts() { - let mut a = DDSketchWrapper::new(0.01); - for i in 1..=10 { - a.update(i as f64); - } - let mut b = DDSketchWrapper::new(0.01); - for i in 11..=20 { - b.update(i as f64); - } - // Reach into the wrapper's inner via snapshot/decode. - let a_inner = - match reconstruct_via_runtime(SketchType::DDSketch, &a.snapshot().unwrap()).unwrap() { - ReconstructedSketch::DdSketch(d) => d, - _ => panic!(), - }; - let b_inner = - match reconstruct_via_runtime(SketchType::DDSketch, &b.snapshot().unwrap()).unwrap() { - ReconstructedSketch::DdSketch(d) => d, - _ => panic!(), - }; - let merged = merge_ddsketches_via_runtime(&a_inner, &b_inner).expect("merge ok"); - assert_eq!(merged.total_count(), 20); - } -} diff --git a/data_plane/src/precompute_engine/operators/mod.rs b/data_plane/src/precompute_engine/operators/mod.rs index af284459e..9df95ba19 100644 --- a/data_plane/src/precompute_engine/operators/mod.rs +++ b/data_plane/src/precompute_engine/operators/mod.rs @@ -4,7 +4,6 @@ pub mod count_sketch_accumulator; pub mod count_sketch_with_heap_accumulator; pub mod datasketches_kll_accumulator; pub mod dd_sketch_accumulator; -pub mod edge_runtime_adapter; pub mod hll_sketch_accumulator; pub mod hydra_kll_accumulator; pub mod increase_accumulator; diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 918f84779..2a32604cf 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -74,7 +74,9 @@ where key.summary_definition, sink_node.0 ))); } - binding.validate(dag).map_err(ScheduleError::Invalid)?; + binding + .validate_precompute_execution(dag) + .map_err(ScheduleError::Invalid)?; if !binding.precompute_sinks.contains(&sink_node) || !matches!( binding.node(sink_node), diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index bc4d4741d..1d506b519 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -728,6 +728,7 @@ mod tests { tables, accuracy: planner_types::types::AccuracyTarget::Exact, }), + selected_dags: Default::default(), entries: BTreeMap::from([( QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &canonical_sql), entry, diff --git a/data_plane/src/query_engines/asap_query_engine/test_plan.rs b/data_plane/src/query_engines/asap_query_engine/test_plan.rs index 8b9f5b57a..12f9f4d8a 100644 --- a/data_plane/src/query_engines/asap_query_engine/test_plan.rs +++ b/data_plane/src/query_engines/asap_query_engine/test_plan.rs @@ -128,6 +128,7 @@ pub(super) fn install( plan_id: 1, plan_version: 1, clickhouse_context: None, + selected_dags: Default::default(), entries: entries .into_iter() .map(|e| (e.canonical_query.clone(), e)) diff --git a/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs b/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs index 26341f942..894b15ee6 100644 --- a/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs +++ b/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs @@ -283,9 +283,8 @@ impl SummaryState { SummaryState::Dd(sk) => { match encoding { // PROTO_DELTA: dispatch on the payload SHAPE, mirroring the - // edge's own `DDSketchWrapper::apply_delta` - // (asap-precompute-rs/src/sketches/ddsketch.rs) — which - // tries the full-envelope decode first, then falls back to + // supported DDSketch frame decoder, which tries the + // full-envelope decode first, then falls back to // the bucket-delta proto. Two wire shapes can arrive on the // ProtoDelta channel: // diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index c679d0bcd..d42457fe7 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -748,6 +748,7 @@ mod tests { plan_id, plan_version, clickhouse_context: None, + selected_dags: Default::default(), entries: Default::default(), }), storage_routing: Arc::new(crate::storage_engines::types::BackendStorageRouting::empty()), diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 51df1d981..c734f5f6d 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -211,8 +211,8 @@ fn is_warm(response: &Value) -> bool { // Measured ERP parameters must reach the real accumulator and answer held-out // raw samples through the installed QueryPlan, without native fallback. #[tokio::test] -#[ignore = "requires ASAPCollector CollectorPlan schema compatibility; run explicitly after Collector is updated"] -async fn erp_measured_kll_collector_to_query_oracle() { +#[ignore = "fixture has stale physical lifecycle evidence"] +async fn erp_measured_kll_state_to_query_oracle() { use control_plane::physical::compiler::{BackendLocalPlanningInput, PhysicalPlanCompiler}; const QUERY: &str = "quantile_over_time(0.9, erp_latency[5s])"; let artifact: Value = serde_json::from_str(include_str!( @@ -370,45 +370,16 @@ fn erp_collector_kll_export(plan: &Value, end_ms: u64, raw: &[f64], sequence: u6 ResourceMetrics, ScopeMetrics, }, }; - use asap_precompute_rs::Precompute; use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope}; - let decoded = asap_precompute_rs::CollectorPlan::from_json( - &serde_json::to_vec(plan).unwrap(), - "erp-collector", - ) - .unwrap(); - let config = decoded - .to_precompute_config_set() - .unwrap() - .configs - .remove(0); - let k = config.sketch_params["k"] as i32; - assert_eq!(k, 32); - let runtime = asap_precompute_rs::precompute::PrecomputeImpl::new( - Some(config), - Some(Box::new(move || { - Box::new(asap_precompute_rs::sketches::KLLWrapper::new(k, Some(123))) - })), - Some(Box::new(asap_precompute_rs::sketches::KLLObserver)), - ); + let decoded: asap_types::producer_plan::CollectorPlan = + serde_json::from_value(plan.clone()).unwrap(); + assert_eq!(decoded.materializations.len(), 1); + let k = 32; + let mut sketch = asap_sketchlib::sketches::kll::KLL::::init_kll_with_seed(k, 123); for value in raw { - runtime - .observe(&asap_precompute_rs::Observation::new( - end_ms - 500, - "erp_latency", - vec![], - vec![asap_precompute_rs::KeyValue::new("service", "erp")], - asap_precompute_rs::ObservationValue { - kind: asap_precompute_rs::ObservationValueKind::Float, - float: *value, - ..Default::default() - }, - )) - .unwrap(); + sketch.update(value); } - let envelopes = runtime.tick(end_ms); - assert_eq!(envelopes.len(), 1); - let wire = SketchEnvelope::decode(envelopes[0].payload.as_slice()).unwrap(); + let wire = SketchEnvelope::decode(asap_sketch_codec::encode_kll(&sketch).as_slice()).unwrap(); let Some(sketch_envelope::SketchState::Kll(state)) = wire.sketch_state else { panic!("KLL state required") }; @@ -853,6 +824,12 @@ async fn run_shared_dashboard(multi_pane: bool) { assert!(plan.cost_comparison.is_some()); assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!(plan.query_plan.entries.len(), 3); + assert!(plan + .precompute_plan + .executable_dags + .values() + .all(|installed| installed.document.schema_version + == asap_types::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION)); if multi_pane { assert!(plan.lifecycle_estimates[0] .window_realization_id diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index cd0f0e86a..e54daa066 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -16,7 +16,6 @@ use asap_otel_proto::tonic::metrics::v1::{ metric::Data, DdSketch, DdSketchDataPoint, DdSketchEncoding, Metric, ResourceMetrics, ScopeMetrics, }; -use asap_precompute_rs::Precompute; use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope as ProtoEnvelope}; use control_plane::opamp::{ opamp_proto, CollectorPlanStatus, CollectorPlanStatusKind, COLLECTOR_PLAN_CAPABILITY, @@ -76,48 +75,18 @@ fn ddsketch_export( plan: &serde_json::Value, sequence: u64, ) -> Vec { - let decoded = asap_precompute_rs::CollectorPlan::from_json( - &serde_json::to_vec(plan).unwrap(), - "whole-e2e-collector", - ) - .unwrap(); - let mut configs = decoded.to_precompute_config_set().unwrap().configs; - assert_eq!( - configs.len(), - 1, - "two query roots must create only one producer" - ); - let config = configs.remove(0); - assert_eq!(config.sketch_params["relative_accuracy"], alpha); - let runtime = asap_precompute_rs::precompute::PrecomputeImpl::new( - Some(config), - Some(Box::new(move || { - Box::new(asap_precompute_rs::sketches::DDSketchWrapper::new(alpha)) - })), - Some(Box::new(asap_precompute_rs::sketches::DDSketchObserver)), - ); + let decoded: asap_types::producer_plan::CollectorPlan = + serde_json::from_value(plan.clone()).unwrap(); + assert_eq!(decoded.materializations.len(), 1); + let mut sketch = asap_sketchlib::DdSketch::new(alpha); for value in values { - runtime - .observe(&asap_precompute_rs::Observation::new( - timestamp_ns / 1_000_000 - 500, - metric, - vec![], - vec![asap_precompute_rs::KeyValue::new("service", "whole-e2e")], - asap_precompute_rs::ObservationValue { - kind: asap_precompute_rs::ObservationValueKind::Float, - float: *value, - ..Default::default() - }, - )) - .unwrap(); + sketch.update(*value); } - let envelopes = runtime.tick(timestamp_ns / 1_000_000); - assert_eq!(runtime.stats().input_observations, values.len() as u64); - assert_eq!(envelopes.len(), 1); - assert_eq!(envelopes[0].count, values.len() as u64); - let wire = ProtoEnvelope::decode(envelopes[0].payload.as_slice()).unwrap(); + assert_eq!(sketch.total_count(), values.len() as u64); + let wire = + ProtoEnvelope::decode(asap_sketch_codec::encode_ddsketch(&sketch).as_slice()).unwrap(); let Some(sketch_envelope::SketchState::Ddsketch(state)) = wire.sketch_state else { - panic!("expected actual Collector DDSketch state") + panic!("expected DDSketch state") }; let materialization = plan["materializations"][0]["materialization"] .as_u64() @@ -297,12 +266,9 @@ async fn respond_next_collector_plan( .expect("collector-plan custom message"); assert_eq!(custom.capability, COLLECTOR_PLAN_CAPABILITY); assert_eq!(custom.r#type, COLLECTOR_PLAN_MESSAGE); - let decoded = asap_precompute_rs::collector_plan::CollectorPlan::from_json( - &custom.data, - "whole-e2e-collector", - ) - .expect("actual Collector validator accepts the emitted plan"); - assert_eq!(decoded.to_precompute_config_set().unwrap().configs.len(), 1); + let decoded: asap_types::producer_plan::CollectorPlan = + serde_json::from_slice(&custom.data).expect("decode backend CollectorPlan"); + assert_eq!(decoded.materializations.len(), 1); let plan: serde_json::Value = serde_json::from_slice(&custom.data).expect("decode collector physical plan"); let plan_id = plan["envelope"]["plan_id"] @@ -395,7 +361,6 @@ async fn quote_workload( } #[tokio::test] -#[ignore = "requires ASAPCollector CollectorPlan schema compatibility; run explicitly after Collector is updated"] async fn production_control_plane_to_data_plane_otlp_to_promql() { let control_binary = std::env::var("ASAP_E2E_CONTROL_PLANE_BIN") .expect("ASAP_E2E_CONTROL_PLANE_BIN is set by scripts/e2e.sh whole"); @@ -521,6 +486,12 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { "backend_compat": control_plane::physical::compiler::BACKEND_COMPAT, "apply_timeout_ms": 10000 }); + let workload: serde_json::Value = serde_json::from_str(include_str!( + "../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + request["data_workload"] = workload["data_workload"].clone(); + request["data_workload"]["data_ingestion_interval"]["value"] = 1_000.into(); let mut second = request["queries"][0].clone(); second["query_id"] = "whole-process-e2e-median".into(); second["query_string"] = "quantile_over_time(0.5, whole_process_e2e_latency_ms[1s])".into(); @@ -730,9 +701,14 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { .await .unwrap(); assert_eq!( - still_active, physical_plan_status, + still_active["plans"], physical_plan_status["plans"], "failed rollout changed active plan" ); + assert_eq!( + still_active["materializations"][0]["materialization"], + physical_plan_status["materializations"][0]["materialization"], + "failed rollout changed the installed materialization" + ); let still_warm: serde_json::Value = client .get(format!("{data_base}/api/v1/query")) .query(&[ diff --git a/data_plane/tests/edge_runtime_consumes_precompute_rs.rs b/data_plane/tests/edge_runtime_consumes_precompute_rs.rs deleted file mode 100644 index e4623e6bb..000000000 --- a/data_plane/tests/edge_runtime_consumes_precompute_rs.rs +++ /dev/null @@ -1,195 +0,0 @@ -//! Phase 3 step 3 acceptance tests: backend ingest **consumes** -//! `asap-precompute-rs` for the shared envelope-parsing, -//! sketch-reconstruction, and merge logic. -//! -//! Each test produces an envelope via `asap-precompute-rs`'s wrappers -//! (the canonical Rust edge runtime) and routes the bytes through the -//! backend's runtime adapter (`precompute_operators::edge_runtime_adapter`). -//! Successful round-trips prove that the asap-precompute-rs `Sketch` -//! trait family is sitting in the backend's ingest path — i.e. the -//! shared logic actually runs in this repo, not just in agents. -//! -//! Sketch coverage: -//! - **DDSketch**: round-trip + structural assertions are live — DDSketch -//! is a deterministic histogram, so `snapshot → reconstruct → snapshot` -//! is byte-identical (`asap_sketchlib`#40). -//! - **KLL**: structural envelope compatibility is live. Byte identity is -//! not a supported contract because reconstruction replays retained items -//! through randomized, lossy compaction. -//! - **HLL + CountSketch + CountMinSketch**: the shared runtime adapter does -//! not support these families; their production decoders are tested at the -//! backend accumulator boundary instead. - -use asap_precompute_rs::sketches::{DDSketchWrapper, KLLWrapper}; -use asap_precompute_rs::Sketch; - -use data_plane::precompute_engine::operators::edge_runtime_adapter::{ - encode_ddsketch_envelope, reconstruct_via_runtime, snapshot_ddsketch_via_runtime, - unwrap_envelope_state, ReconstructedSketch, SketchType, -}; -use data_plane::storage_engines::types::AggregateCore; - -// --- DDSketch ----------------------------------------------------- - -/// Round-trip: an envelope produced by asap-precompute-rs's -/// `DDSketchWrapper` is reconstructed by the backend's runtime adapter -/// to a backend-shaped `DdSketch`, re-encoded through the same -/// envelope shape, and the resulting bytes match the original. -/// -/// This proves the **shared envelope wire format** flows through both -/// crates with byte parity — the dedup target. -#[test] -fn ddsketch_envelope_round_trip_through_backend_adapter() { - let mut w = DDSketchWrapper::new(0.01); - for i in 1..=200 { - w.update(i as f64); - } - let original = w.snapshot().expect("DDSketchWrapper snapshot"); - assert!(!original.is_empty()); - - let reconstructed = reconstruct_via_runtime(SketchType::DDSketch, &original) - .expect("runtime adapter reconstruction"); - let dd = match reconstructed { - ReconstructedSketch::DdSketch(d) => d, - _ => panic!("expected DDSketch reconstruction"), - }; - // `count` is recovered from the bucket store now that the scalar was - // dropped (ProjectASAP/sketchlib-go#243 / asap_sketchlib#57). - assert_eq!( - dd.total_count(), - 200, - "count preserved through runtime adapter" - ); - - let re_encoded = encode_ddsketch_envelope(&dd); - assert_eq!( - re_encoded, original, - "envelope round-trip via asap-precompute-rs runtime must be byte-identical" - ); -} - -/// Structural: an envelope produced by asap-precompute-rs's -/// `DDSketchWrapper` is unwrapped via the backend's -/// `unwrap_envelope_state` (which itself goes through asap-precompute-rs's -/// `ProtoSketchEnvelope`), and the typed inner state has the expected -/// count + alpha shape. -#[test] -fn ddsketch_envelope_structural_assertions() { - use asap_sketchlib::proto::sketchlib::sketch_envelope::SketchState; - - let mut w = DDSketchWrapper::new(0.005); - w.update(1.0); - w.update(2.0); - w.update(3.0); - let bytes = w.snapshot().expect("snapshot"); - let state = unwrap_envelope_state(&bytes) - .expect("unwrap") - .expect("state"); - match state { - SketchState::Ddsketch(s) => { - // `count` was dropped from `DdSketchState` - // (ProjectASAP/sketchlib-go#243 / asap_sketchlib#57); it is - // recovered by summing the bucket store counts. - assert_eq!(s.store_counts.iter().sum::(), 3, "structural count"); - assert!( - s.alpha > 0.0 && s.alpha < 1.0, - "alpha within (0,1): got {}", - s.alpha - ); - } - other => panic!("expected DDSketch state, got {other:?}"), - } -} - -/// End-to-end: DDSketch envelope → backend `AggregateCore` (via the -/// runtime adapter), then the backend's `query_statistic` API answers a -/// quantile query on the reconstructed accumulator. This proves the -/// **adapter integration is live** — backend ingest goes through -/// asap-precompute-rs and the resulting accumulator works on the -/// query-side surface. -#[test] -fn ddsketch_envelope_ends_up_in_backend_accumulator() { - use data_plane::precompute_engine::operators::DDSketchAccumulator; - - let mut w = DDSketchWrapper::new(0.01); - for i in 1..=100 { - w.update(i as f64); - } - let bytes = w.snapshot().expect("snapshot"); - - let reconstructed = reconstruct_via_runtime(SketchType::DDSketch, &bytes).expect("reconstruct"); - let dd = match reconstructed { - ReconstructedSketch::DdSketch(d) => d, - _ => panic!(), - }; - let acc = DDSketchAccumulator { - inner: dd, - sample_p: 1.0, - }; - - let q = acc - .query_statistic( - asap_types::Statistic::Quantile, - &None, - &[("quantile".to_string(), "0.5".to_string())] - .into_iter() - .collect(), - ) - .expect("quantile query"); - assert!( - (q - 50.0).abs() / 50.0 < 0.05, - "median estimate close to 50: got {q}" - ); - let count = acc - .query_statistic(asap_types::Statistic::Count, &None, &Default::default()) - .expect("count query"); - assert_eq!(count as u64, 100); -} - -/// Snapshot a backend-side `DdSketch` *back through* -/// asap-precompute-rs's `Sketch::snapshot` and assert byte-equality -/// with the canonical envelope bytes. Closes the round-trip -/// (encode side) — proves backend can EMIT the same wire bytes as -/// asap-precompute-rs. -#[test] -fn ddsketch_backend_sketch_snapshots_to_canonical_envelope_bytes() { - let mut w = DDSketchWrapper::new(0.01); - for i in 1..=50 { - w.update(i as f64); - } - let canonical = w.snapshot().expect("snapshot"); - let dd = match reconstruct_via_runtime(SketchType::DDSketch, &canonical).unwrap() { - ReconstructedSketch::DdSketch(d) => d, - _ => panic!(), - }; - let via_runtime = snapshot_ddsketch_via_runtime(&dd).expect("runtime snapshot"); - assert_eq!( - via_runtime, canonical, - "snapshot via runtime adapter is byte-identical to source envelope" - ); -} - -// --- KLL ---------------------------------------------------------- - -/// Structural: KLL envelope unwraps to the expected oneof variant via -/// the shared `unwrap_envelope_state` helper. -#[test] -fn kll_envelope_structural_assertions() { - use asap_sketchlib::proto::sketchlib::sketch_envelope::SketchState; - - let mut w = KLLWrapper::new(200, Some(7)); - for i in 1..=10 { - w.update(i as f64); - } - let bytes = w.snapshot().expect("snapshot"); - let state = unwrap_envelope_state(&bytes) - .expect("unwrap") - .expect("state"); - match state { - SketchState::Kll(s) => { - assert_eq!(s.k, 200, "structural k"); - assert_eq!(s.items.len(), 10, "all 10 items retained"); - } - other => panic!("expected KLL state, got {other:?}"), - } -} diff --git a/data_plane/tests/edge_sketch_codec.rs b/data_plane/tests/edge_sketch_codec.rs new file mode 100644 index 000000000..edbeb6795 --- /dev/null +++ b/data_plane/tests/edge_sketch_codec.rs @@ -0,0 +1,74 @@ +//! Portable edge sketch envelopes reconstruct through the neutral codec and +//! remain readable by the backend's query accumulators. + +use asap_sketch_codec::{encode_ddsketch, reconstruct_ddsketch}; +use asap_sketchlib::proto::sketchlib::sketch_envelope::SketchState; +use data_plane::storage_engines::types::AggregateCore; + +#[test] +fn ddsketch_full_envelope_round_trips_without_collector_runtime() { + let mut source = asap_sketchlib::DdSketch::new(0.01); + for value in 1..=200 { + source.update(value as f64); + } + let bytes = asap_sketch_codec::encode_ddsketch(&source); + assert!(matches!( + asap_sketch_codec::envelope_state(&bytes).unwrap(), + Some(SketchState::Ddsketch(_)) + )); + let (decoded, _) = reconstruct_ddsketch(&bytes).unwrap(); + assert_eq!(decoded.total_count(), 200); + assert_eq!(encode_ddsketch(&decoded), bytes); +} + +#[test] +fn ddsketch_bare_state_and_query_readout_are_supported() { + let mut source = asap_sketchlib::DdSketch::new(0.01); + for value in 1..=100 { + source.update(value as f64); + } + let envelope = asap_sketch_codec::encode_ddsketch(&source); + let Some(SketchState::Ddsketch(state)) = asap_sketch_codec::envelope_state(&envelope).unwrap() + else { + panic!("DDSketch state required") + }; + let bare = prost::Message::encode_to_vec(&state); + let (decoded, _) = asap_sketch_codec::reconstruct_ddsketch(&bare).unwrap(); + let accumulator = data_plane::precompute_engine::operators::DDSketchAccumulator { + inner: decoded, + sample_p: 1.0, + }; + let median = accumulator + .query_statistic( + asap_types::Statistic::Quantile, + &None, + &[("quantile".to_string(), "0.5".to_string())] + .into_iter() + .collect(), + ) + .unwrap(); + assert!((median - 50.0).abs() / 50.0 < 0.05); +} + +#[test] +fn kll_envelope_keeps_level_layout_for_backend_readout() { + let mut source = asap_sketchlib::sketches::kll::KLL::::init_kll_with_seed(200, 7); + for value in 1..=50 { + source.update(&(value as f64)); + } + let bytes = asap_sketch_codec::encode_kll(&source); + let state = asap_sketch_codec::kll_state(&bytes).unwrap(); + assert_eq!(state.k, 200); + assert_eq!(state.items.len(), 50); + let snapshot_bytes = bytes; + let accumulator = data_plane::precompute_engine::operators::DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(&snapshot_bytes).unwrap(); + assert!(accumulator.get_quantile(0.5).is_finite()); +} + +#[test] +fn a_different_sketch_family_cannot_be_decoded_as_ddsketch() { + let mut kll = asap_sketchlib::sketches::kll::KLL::::init_kll_with_seed(200, 7); + kll.update(&42.0); + let bytes = asap_sketch_codec::encode_kll(&kll); + assert!(asap_sketch_codec::reconstruct_ddsketch(&bytes).is_err()); +} diff --git a/data_plane/tests/support/physical_fixture.rs b/data_plane/tests/support/physical_fixture.rs index 9eb34e64c..27c6ef471 100644 --- a/data_plane/tests/support/physical_fixture.rs +++ b/data_plane/tests/support/physical_fixture.rs @@ -57,6 +57,7 @@ pub fn artifact_from_materializations( plan_id: 1, plan_version: 1, clickhouse_context: None, + selected_dags: Default::default(), entries: BTreeMap::new(), }; for config in &precompute.materializations { diff --git a/scripts/e2e.sh b/scripts/e2e.sh index ae240e875..22536a887 100755 --- a/scripts/e2e.sh +++ b/scripts/e2e.sh @@ -100,7 +100,7 @@ data_plane() { CURRENT_STAGE="data-plane/edge-runtime-wire" say "data-plane: edge runtime sketch envelope -> backend accumulator" - rust_test data_plane --test edge_runtime_consumes_precompute_rs + rust_test data_plane --test edge_sketch_codec CURRENT_STAGE="data-plane/production-process" say "data-plane: production binary -> modified OTLP -> SketchStore -> PromQL" diff --git a/tools/shared-workload/ACCURACY_E2E.md b/tools/shared-workload/ACCURACY_E2E.md index 73fc5f5aa..e7cd3619e 100644 --- a/tools/shared-workload/ACCURACY_E2E.md +++ b/tools/shared-workload/ACCURACY_E2E.md @@ -14,8 +14,8 @@ The repository's existing backend CI is unchanged. Runtime scope is **backend-local precompute**. No ASAPCollector service is started or called: the generator sends Remote Write directly to the backend. -The `asap-precompute-rs` build dependency and offline Google mapper source happen -to live in the ASAPCollector repository; they are not an extra running collector. +The optional offline Google mapper source lives in the ASAPCollector repository; +the backend build and runtime do not depend on that repository. ## Query matrix