diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 0c7e3a75..343454a3 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -643,6 +643,7 @@ impl PrecomputePlan { #[derive(Debug, Clone)] pub struct PhysicalPlan { pub envelope: PlanEnvelope, + pub summary_catalog: super::summary_catalog::SummaryCatalog, pub collector_plans: Vec, pub precompute_plan: PrecomputePlan, pub transmission_plan: TransmissionPlan, @@ -2401,8 +2402,18 @@ impl PhysicalCompiler { } } query_plan.validate(&materialization_fingerprints)?; + let summary_catalog = super::summary_catalog::SummaryCatalog::from_materializations( + envelope.plan_id, + envelope.plan_version, + &precompute_plan.materializations, + ) + .map_err(|error| CompileError::Query { + query_id: "summary-catalog".into(), + reason: error.to_string(), + })?; Ok(PhysicalPlan { envelope, + summary_catalog, collector_plans, precompute_plan, transmission_plan, @@ -4043,6 +4054,27 @@ mod tests { ) .expect("compile"); assert_eq!(bundle.collector_plans.len(), 2); + bundle.summary_catalog.validate().expect("catalog contract"); + assert_eq!(bundle.summary_catalog.plan_id, bundle.envelope.plan_id); + assert_eq!( + bundle.summary_catalog.plan_version, + bundle.envelope.plan_version + ); + assert_eq!( + bundle + .summary_catalog + .materializations + .keys() + .copied() + .collect::>(), + bundle + .backend_plan + .materializations + .keys() + .copied() + .map(asap_types::sds::MaterializationId::from) + .collect::>() + ); assert_eq!(bundle.precompute_plan.envelope, bundle.envelope); assert_eq!(bundle.precompute_plan.materializations.len(), 1); assert_eq!(bundle.precompute_plan.schemas.len(), 1); diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs index e5d3c0c1..b3c168da 100644 --- a/control_plane/src/physical/mod.rs +++ b/control_plane/src/physical/mod.rs @@ -32,6 +32,7 @@ pub mod post_asap; pub mod runtime_capability; pub mod sketch_catalog; pub mod stage_split; +pub mod summary_catalog; pub mod topology; pub mod window_fusion; pub mod workload_cost; diff --git a/control_plane/src/physical/summary_catalog.rs b/control_plane/src/physical/summary_catalog.rs new file mode 100644 index 00000000..dfee576b --- /dev/null +++ b/control_plane/src/physical/summary_catalog.rs @@ -0,0 +1,312 @@ +//! Authoritative descriptor snapshot for a compiled physical plan. +//! +//! Execution plans keep their compatibility fields during migration. This +//! catalog owns semantic definitions, not producer placement or pane state. + +use std::collections::BTreeMap; + +use asap_types::sds::{ + DataDescriptor, DataDescriptorId, MaterializationId, SummaryDescriptor, SummaryDescriptorId, +}; +use asap_types::PolicyFingerprint; +use serde::{Deserialize, Serialize}; + +pub const SUMMARY_CATALOG_SCHEMA_VERSION: u32 = 1; + +/// Stable materialization identity binds operator and population descriptors. +/// Concrete intervals, groups and completeness belong to runtime instances. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct MaterializationIdentity { + pub summary_descriptor_id: SummaryDescriptorId, + pub data_descriptor_id: DataDescriptorId, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct SummaryCatalog { + pub schema_version: u32, + pub plan_id: u64, + pub plan_version: u64, + pub summary_descriptors: BTreeMap, + pub data_descriptors: BTreeMap, + pub materializations: BTreeMap, +} + +#[derive(Debug, thiserror::Error)] +pub enum SummaryCatalogError { + #[error("unsupported summary catalog schema version {0}")] + SchemaVersion(u32), + #[error("invalid summary catalog descriptor: {0}")] + Descriptor(String), + #[error("materialization {0} has conflicting descriptor bindings")] + ConflictingMaterialization(u64), + #[error("materialization {0} references a missing descriptor")] + MissingDescriptor(u64), +} + +impl SummaryCatalog { + pub fn from_materializations( + plan_id: u64, + plan_version: u64, + materializations: &[asap_types::PrecomputeMaterialization], + ) -> Result { + let entries = materializations + .iter() + .map(|config| { + let summary = SummaryDescriptor::from_config(config) + .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; + let data = DataDescriptor::new( + config.metric.clone(), + asap_types::utils::normalize_spatial_filter(&config.spatial_filter), + config.grouping_labels.labels.clone(), + ); + Ok((config.policy_fingerprint(), summary, data)) + }) + .collect::, SummaryCatalogError>>()?; + Self::build(plan_id, plan_version, entries) + } + + pub fn build( + plan_id: u64, + plan_version: u64, + entries: impl IntoIterator, + ) -> Result { + let mut catalog = Self { + schema_version: SUMMARY_CATALOG_SCHEMA_VERSION, + plan_id, + plan_version, + summary_descriptors: BTreeMap::new(), + data_descriptors: BTreeMap::new(), + materializations: BTreeMap::new(), + }; + for (fingerprint, summary, data) in entries { + let materialization = MaterializationId::from(fingerprint); + summary + .validate() + .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; + data.validate() + .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; + let binding = MaterializationIdentity { + summary_descriptor_id: summary.id().clone(), + data_descriptor_id: data.id().clone(), + }; + if catalog + .materializations + .get(&materialization) + .is_some_and(|old| old != &binding) + { + return Err(SummaryCatalogError::ConflictingMaterialization( + materialization.as_u64(), + )); + } + catalog + .summary_descriptors + .insert(binding.summary_descriptor_id.clone(), summary); + catalog + .data_descriptors + .insert(binding.data_descriptor_id.clone(), data); + catalog.materializations.insert(materialization, binding); + } + catalog.validate()?; + Ok(catalog) + } + + pub fn validate(&self) -> Result<(), SummaryCatalogError> { + if self.schema_version != SUMMARY_CATALOG_SCHEMA_VERSION { + return Err(SummaryCatalogError::SchemaVersion(self.schema_version)); + } + for (key, descriptor) in &self.summary_descriptors { + descriptor + .validate() + .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; + if key != descriptor.id() { + return Err(SummaryCatalogError::Descriptor( + "summary table key differs from content identity".into(), + )); + } + } + for (key, descriptor) in &self.data_descriptors { + descriptor + .validate() + .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; + if key != descriptor.id() { + return Err(SummaryCatalogError::Descriptor( + "data table key differs from content identity".into(), + )); + } + } + for (id, binding) in &self.materializations { + if !self + .summary_descriptors + .contains_key(&binding.summary_descriptor_id) + || !self + .data_descriptors + .contains_key(&binding.data_descriptor_id) + { + return Err(SummaryCatalogError::MissingDescriptor(id.as_u64())); + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; + + fn config(metric: &str, filter: &str, window: u64) -> PrecomputeMaterialization { + PrecomputeMaterialization::new( + AggregationType::Sum, + String::new(), + Default::default(), + KeyByLabelNames::new(vec!["job".into()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + window, + window, + WindowKind::Tumbling, + filter.into(), + metric.into(), + None, + None, + None, + ) + } + + // Panes share definitions but retain distinct physical materialization IDs. + #[test] + fn shares_descriptors_across_windows_and_deduplicates_materializations() { + let one = config("requests", "", 60); + let two = config("requests", "", 120); + let catalog = + SummaryCatalog::from_materializations(7, 2, &[one.clone(), two, one]).unwrap(); + assert_eq!(catalog.summary_descriptors.len(), 1); + assert_eq!(catalog.data_descriptors.len(), 1); + assert_eq!(catalog.materializations.len(), 2); + assert_eq!((catalog.plan_id, catalog.plan_version), (7, 2)); + } + + // Source/population changes never alias, while the operator can be reused. + #[test] + fn separates_population_and_operator_identity() { + let catalog = SummaryCatalog::from_materializations( + 1, + 1, + &[ + config("requests", "", 60), + config("errors", "", 60), + config("requests", "job=user", 60), + ], + ) + .unwrap(); + assert_eq!(catalog.summary_descriptors.len(), 1); + assert_eq!(catalog.data_descriptors.len(), 3); + } + + // Operator changes share population metadata without sharing state identity. + #[test] + fn changing_operator_parameters_creates_a_new_summary_descriptor() { + let mut a = config("requests", "", 60); + a.aggregation_type = AggregationType::DatasketchesKLL; + a.parameters.insert("K".into(), serde_json::json!(100)); + let mut b = a.clone(); + b.parameters.insert("K".into(), serde_json::json!(200)); + let catalog = SummaryCatalog::from_materializations(1, 1, &[a, b]).unwrap(); + assert_eq!(catalog.summary_descriptors.len(), 2); + assert_eq!(catalog.data_descriptors.len(), 1); + assert_eq!(catalog.materializations.len(), 2); + } + + // Construction order cannot affect the published snapshot bytes. + #[test] + fn snapshot_is_deterministic_and_round_trips() { + let a = config("requests", "", 60); + let b = config("errors", "", 60); + let forward = SummaryCatalog::from_materializations(7, 2, &[a.clone(), b.clone()]).unwrap(); + let backward = SummaryCatalog::from_materializations(7, 2, &[b, a]).unwrap(); + let bytes = serde_json::to_vec(&forward).unwrap(); + assert_eq!(bytes, serde_json::to_vec(&backward).unwrap()); + let decoded: SummaryCatalog = serde_json::from_slice(&bytes).unwrap(); + decoded.validate().unwrap(); + assert_eq!(decoded, forward); + } + + // The same materialization cannot silently rebind to another population. + #[test] + fn conflicting_materialization_is_rejected() { + let first = config("requests", "", 60); + let second = config("errors", "", 60); + let catalog = + SummaryCatalog::from_materializations(1, 1, &[first.clone(), second]).unwrap(); + let summary = catalog.summary_descriptors.values().next().unwrap().clone(); + let data = catalog + .data_descriptors + .values() + .cloned() + .collect::>(); + let error = SummaryCatalog::build( + 1, + 1, + [ + (first.policy_fingerprint(), summary.clone(), data[0].clone()), + (first.policy_fingerprint(), summary, data[1].clone()), + ], + ) + .unwrap_err(); + assert!(matches!( + error, + SummaryCatalogError::ConflictingMaterialization(_) + )); + } + + // Imported catalogs must resolve every foreign key and content identity. + #[test] + fn rejects_dangling_refs_tampered_keys_and_unknown_schema() { + let catalog = + SummaryCatalog::from_materializations(1, 1, &[config("requests", "", 60)]).unwrap(); + let mut broken = catalog.clone(); + broken.data_descriptors.clear(); + assert!(matches!( + broken.validate(), + Err(SummaryCatalogError::MissingDescriptor(_)) + )); + let mut broken = catalog.clone(); + let (_, descriptor) = broken.summary_descriptors.pop_first().unwrap(); + broken.summary_descriptors.insert( + serde_json::from_value(serde_json::json!("forged")).unwrap(), + descriptor, + ); + assert!(matches!( + broken.validate(), + Err(SummaryCatalogError::Descriptor(_)) + )); + let mut broken = catalog.clone(); + broken + .summary_descriptors + .values_mut() + .next() + .unwrap() + .state_schema_version += 1; + assert!(matches!( + broken.validate(), + Err(SummaryCatalogError::Descriptor(_)) + )); + let mut broken = catalog; + broken.schema_version = SUMMARY_CATALOG_SCHEMA_VERSION + 1; + assert!(matches!( + broken.validate(), + Err(SummaryCatalogError::SchemaVersion(_)) + )); + } + + // Native exact-only plans have a valid empty catalog, not dummy state. + #[test] + fn empty_catalog_is_valid() { + let catalog = SummaryCatalog::from_materializations(1, 1, &[]).unwrap(); + assert!(catalog.materializations.is_empty()); + catalog.validate().unwrap(); + } +} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 4b8b2054..f205af63 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -8,6 +8,7 @@ pub mod policy_fingerprint; pub mod policy_registry; pub mod query_requirements; pub mod routing_index; +pub mod sds; pub mod storage_backend; pub mod traits; pub mod utils; diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs new file mode 100644 index 00000000..8ac512ce --- /dev/null +++ b/crates/asap_types/src/sds.rs @@ -0,0 +1,566 @@ +//! Shared SDS descriptor contracts. Instances, state bytes, and registry lifetimes +//! remain backend-owned. Canonical IDs describe content, never SID or policy IDs. +use crate::{AggregationType, PrecomputeMaterialization}; +use planner_types::post_asap::{SketchAlgorithm, SketchParams, SummaryFamilyType}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::collections::{BTreeMap, BTreeSet}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SdsError(pub String); +impl std::fmt::Display for SdsError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} +impl std::error::Error for SdsError {} + +macro_rules! descriptor_id { + ($name:ident) => { + #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] + #[serde(transparent)] + pub struct $name(String); + impl $name { + pub fn canonical(&self) -> &str { + &self.0 + } + } + }; +} +/// Semantic materialization reference. Wire-compatible with PolicyFingerprint, +/// but distinct from descriptor IDs and runtime instance/SID identity. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(transparent)] +pub struct MaterializationId(pub crate::PolicyFingerprint); +impl MaterializationId { + pub fn fingerprint(self) -> crate::PolicyFingerprint { + self.0 + } + pub fn as_u64(self) -> u64 { + self.0 .0 + } +} +impl From for MaterializationId { + fn from(value: crate::PolicyFingerprint) -> Self { + Self(value) + } +} +impl From for crate::PolicyFingerprint { + fn from(value: MaterializationId) -> Self { + value.0 + } +} + +descriptor_id!(SummaryDescriptorId); +descriptor_id!(DataDescriptorId); + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum SummaryOperator { + /// Compatibility evidence from a legacy backend record, not a complete + /// configured contract. Its ID can never satisfy a Configured descriptor. + LegacyPartial { operator_canonical: String }, + Sketch { + algorithm: SketchAlgorithm, + parameters: BTreeMap, + }, + ExactAgg { + agg_type: AggregationType, + parameters_canonical: String, + }, + /// Complete planner materialization configuration, including heap/Hydra + /// dimensions and readout/update subtype. Never equal to a legacy projection. + Configured { + aggregation_type: AggregationType, + aggregation_sub_type: String, + parameters: BTreeMap, + }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum FidelityGuarantee { + Exact, + KllRankError { + k: u32, + model: String, + }, + DdSketchRelativeError { + alpha: f64, + }, + HllCardinalityError { + precision: u32, + model: String, + }, + CmsFrequencyError { + width: u32, + depth: u32, + model: String, + }, + CountSketchFrequencyError { + width: u32, + depth: u32, + model: String, + }, + Unknown { + reason: String, + }, +} +impl FidelityGuarantee { + /// Model IDs name parameterized error families/scopes, not certified numeric + /// epsilon/confidence values. Heap membership and Hydra cross-cell readouts + /// need separate models; point-frequency/per-cell rank does not attest them. + pub fn from_config(config: &PrecomputeMaterialization) -> Self { + if config.aggregation_type == AggregationType::HLL { + let precision = config + .parameters + .get("precision") + .or_else(|| config.parameters.get("p")) + .and_then(Value::as_u64) + .and_then(|v| u32::try_from(v).ok()) + .unwrap_or(14); + return Self::HllCardinalityError { + precision, + model: "asap.hll.relative-cardinality.v1".into(), + }; + } + match config.accumulator_spec().map(|s| s.family) { + Ok(SummaryFamilyType::ExactAggregate(..)) => Self::Exact, + Ok(SummaryFamilyType::Sketch(kind, _)) => match kind.params() { + SketchParams::Kll { k } => Self::KllRankError { + k: *k, + model: if config.aggregation_type == AggregationType::HydraKLL { + "asap.hydra-kll.per-cell-rank.v1" + } else { + "asap.kll.normalized-rank.v1" + } + .into(), + }, + SketchParams::DDSketch { alpha } => Self::DdSketchRelativeError { alpha: *alpha }, + SketchParams::Hll { precision } => Self::HllCardinalityError { + precision: (*precision).into(), + model: "asap.hll.relative-cardinality.v1".into(), + }, + SketchParams::Cms { width, depth } + | SketchParams::CmsWithHeap { width, depth, .. } => Self::CmsFrequencyError { + width: *width, + depth: *depth, + model: "asap.cms.point-frequency.v1".into(), + }, + SketchParams::CountSketch { width, depth } + | SketchParams::CountSketchWithHeap { width, depth, .. } => { + Self::CountSketchFrequencyError { + width: *width, + depth: *depth, + model: "asap.count-sketch.point-frequency.v1".into(), + } + } + _ => Self::Unknown { + reason: "No shared parameterized error model for this sketch family".into(), + }, + }, + _ => Self::Unknown { + reason: "Physical accumulator family is unavailable".into(), + }, + } + } + fn validate(&self) -> Result<(), SdsError> { + let valid = match self { + Self::Exact => true, + Self::KllRankError { k, model } => *k > 0 && !model.is_empty(), + Self::DdSketchRelativeError { alpha } => { + alpha.is_finite() && *alpha > 0.0 && *alpha < 1.0 + } + Self::HllCardinalityError { precision, model } => { + *precision > 0 && *precision < 64 && !model.is_empty() + } + Self::CmsFrequencyError { + width, + depth, + model, + } + | Self::CountSketchFrequencyError { + width, + depth, + model, + } => *width > 0 && *depth > 0 && !model.is_empty(), + Self::Unknown { reason } => !reason.is_empty(), + }; + if valid { + Ok(()) + } else { + Err(SdsError("invalid parameterized fidelity contract".into())) + } + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SummaryDescriptor { + pub id: SummaryDescriptorId, + pub operator: SummaryOperator, + pub fidelity: FidelityGuarantee, + pub state_schema_version: u32, +} + +/// Sort every JSON object, including nested configuration values. Arrays retain +/// order; opaque canonical strings are used verbatim, not reparsed as PromQL. +fn canonical(value: &Value) -> String { + match value { + Value::Object(map) => { + let sorted: BTreeMap<_, _> = map.iter().collect(); + format!( + "{{{}}}", + sorted + .into_iter() + .map(|(k, v)| format!("{}:{}", serde_json::to_string(k).unwrap(), canonical(v))) + .collect::>() + .join(",") + ) + } + Value::Array(values) => format!( + "[{}]", + values.iter().map(canonical).collect::>().join(",") + ), + _ => value.to_string(), + } +} +impl SummaryDescriptor { + pub fn new( + operator: SummaryOperator, + fidelity: FidelityGuarantee, + state_schema_version: u32, + ) -> Result { + if state_schema_version == 0 { + return Err(SdsError("state schema version must be positive".into())); + } + fidelity.validate()?; + let content = json!({"operator":operator,"fidelity":fidelity,"state_schema_version":state_schema_version}); + let id = SummaryDescriptorId(format!("summary:v2:{}", canonical(&content))); + Ok(Self { + id, + operator, + fidelity, + state_schema_version, + }) + } + pub fn id(&self) -> &SummaryDescriptorId { + &self.id + } + pub fn validate(&self) -> Result<(), SdsError> { + let rebuilt = Self::new( + self.operator.clone(), + self.fidelity.clone(), + self.state_schema_version, + )?; + if self.id != rebuilt.id { + return Err(SdsError("summary descriptor ID/content mismatch".into())); + } + Ok(()) + } + + /// Preserve every configured state/update parameter. Omitted defaults remain + /// distinct from explicit defaults (conservative identity, never false sharing). + /// Legacy AggKind projections intentionally have different operator variants: + /// they cannot attest heap, Hydra, or aggregation-subtype semantics they lost. + pub fn from_config(config: &PrecomputeMaterialization) -> Result { + let fidelity = FidelityGuarantee::from_config(config); + Self::new( + SummaryOperator::Configured { + aggregation_type: config.aggregation_type, + aggregation_sub_type: config.aggregation_sub_type.clone(), + parameters: config + .parameters + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + }, + fidelity, + 1, + ) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DataDescriptor { + pub id: DataDescriptorId, + pub metric_name: String, + pub population_filter_canonical: String, + pub group_by_keys: BTreeSet, + /// Versioned contract for value projection, timestamp interpretation and + /// missing/duplicate/invalid observation handling. + pub observation_semantics: String, +} +impl DataDescriptor { + pub fn new( + metric: impl Into, + filter: impl Into, + group_by: impl IntoIterator, + ) -> Self { + Self::new_with_semantics( + metric, + filter, + group_by, + "asap.timestamped-metric-samples.v1", + ) + } + + pub fn new_with_semantics( + metric: impl Into, + filter: impl Into, + group_by: impl IntoIterator, + observation_semantics: impl Into, + ) -> Self { + let metric_name = metric.into(); + let population_filter_canonical = filter.into(); + let group_by_keys = group_by.into_iter().collect(); + let observation_semantics = observation_semantics.into(); + let id = data_descriptor_id( + &metric_name, + &population_filter_canonical, + &group_by_keys, + &observation_semantics, + ); + Self { + id, + metric_name, + population_filter_canonical, + group_by_keys, + observation_semantics, + } + } + pub fn id(&self) -> &DataDescriptorId { + &self.id + } + pub fn validate(&self) -> Result<(), SdsError> { + if self.id + != data_descriptor_id( + &self.metric_name, + &self.population_filter_canonical, + &self.group_by_keys, + &self.observation_semantics, + ) + { + return Err(SdsError("data descriptor ID/content mismatch".into())); + } + Ok(()) + } +} +fn data_descriptor_id( + metric: &str, + filter: &str, + group_by: &BTreeSet, + observation_semantics: &str, +) -> DataDescriptorId { + // Preserve the existing v1 length-framed data identity, now normalizing the + // grouping set at the shared contract boundary. + let mut key = format!( + "data:v1|{}:{metric}|{}:{filter}", + metric.len(), + filter.len() + ); + for name in group_by { + key.push_str(&format!("|{}:{name}", name.len())); + } + key.push_str(&format!( + "|{}:{observation_semantics}", + observation_semantics.len() + )); + DataDescriptorId(key) +} + +#[cfg(test)] +mod tests { + use super::*; + fn descriptor(k: u32, fidelity: FidelityGuarantee, version: u32) -> SummaryDescriptor { + SummaryDescriptor::new( + SummaryOperator::Sketch { + algorithm: SketchAlgorithm::Kll, + parameters: BTreeMap::from([("k".into(), json!(k))]), + }, + fidelity, + version, + ) + .unwrap() + } + #[test] + fn identity_includes_configuration_fidelity_and_state_schema() { + let base = descriptor( + 200, + FidelityGuarantee::Unknown { + reason: "not supplied".into(), + }, + 1, + ); + assert_ne!( + base.id, + descriptor( + 201, + FidelityGuarantee::Unknown { + reason: "not supplied".into() + }, + 1 + ) + .id + ); + assert_ne!(base.id, descriptor(200, FidelityGuarantee::Exact, 1).id); + assert_ne!( + base.id, + descriptor( + 200, + FidelityGuarantee::Unknown { + reason: "not supplied".into() + }, + 2 + ) + .id + ); + } + #[test] + fn group_order_and_duplicates_do_not_change_identity() { + let a = DataDescriptor::new("cpu", "{job=\"a\"}", ["z".into(), "a".into(), "a".into()]); + let b = DataDescriptor::new("cpu", "{job=\"a\"}", ["a".into(), "z".into()]); + assert_eq!(a, b); + assert_ne!( + a.id, + DataDescriptor::new("cpu", "{job=\"b\"}", ["a".into(), "z".into()]).id + ); + } + #[test] + fn length_framing_distinguishes_delimiters_and_unicode() { + assert_ne!( + DataDescriptor::new("a|b", "c", []).id, + DataDescriptor::new("a", "b|c", []).id + ); + assert_ne!( + DataDescriptor::new("π", "", ["x|y".into()]).id, + DataDescriptor::new("π", "", ["x".into(), "y".into()]).id + ); + } + #[test] + fn observation_semantics_is_part_of_data_identity() { + assert_ne!( + DataDescriptor::new_with_semantics("cpu", "", [], "samples.v1").id, + DataDescriptor::new_with_semantics("cpu", "", [], "samples.v2").id + ); + } + #[test] + fn wire_roundtrip_and_tampered_id_validation() { + let original = descriptor( + 200, + FidelityGuarantee::Unknown { + reason: "not supplied".into(), + }, + 1, + ); + let mut decoded: SummaryDescriptor = + serde_json::from_str(&serde_json::to_string(&original).unwrap()).unwrap(); + decoded.validate().unwrap(); + assert_eq!(decoded, original); + decoded.state_schema_version = 2; + assert!(decoded.validate().is_err()); + let mut data = DataDescriptor::new("cpu", "", []); + data.metric_name = "other".into(); + assert!(data.validate().is_err()); + } + #[test] + fn invalid_fidelity_is_rejected() { + assert!(SummaryDescriptor::new( + SummaryOperator::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new() + }, + FidelityGuarantee::DdSketchRelativeError { alpha: f64::NAN }, + 1 + ) + .is_err()); + } + #[test] + fn configured_identity_preserves_heap_hydra_and_subtype_and_excludes_population() { + let yaml:serde_yaml::Value=serde_yaml::from_str("aggregationType: DDSketch\naggregationSubType: ''\nmetric: m\nlabels:\n grouping: []\n rollup: []\n aggregated: []\nparameters:\n relative_accuracy: 0.01\nwindowSize: 30\nwindowType: tumbling\nspatialFilter: ''\n").unwrap(); + let mut config = + PrecomputeMaterialization::from_yaml_data(&yaml, None, crate::QueryLanguage::promql) + .unwrap(); + assert!(matches!( + SummaryDescriptor::from_config(&config).unwrap().fidelity, + FidelityGuarantee::DdSketchRelativeError { .. } + )); + for (kind, key) in [ + (AggregationType::CountMinSketchWithHeap, "heap_size"), + (AggregationType::HydraKLL, "row"), + (AggregationType::HydraKLL, "col"), + (AggregationType::HydraKLL, "k"), + ] { + config.aggregation_type = kind; + config.parameters.insert(key.into(), json!(10)); + let before = SummaryDescriptor::from_config(&config).unwrap(); + config.parameters.insert(key.into(), json!(11)); + let after = SummaryDescriptor::from_config(&config).unwrap(); + assert_ne!(before.id, after.id, "{key}"); + config.metric = "other".into(); + config.spatial_filter_normalized = "job=a".into(); + assert_eq!( + after.id, + SummaryDescriptor::from_config(&config).unwrap().id + ); + } + let before = SummaryDescriptor::from_config(&config).unwrap(); + config.aggregation_sub_type = "max".into(); + assert_ne!( + before.id, + SummaryDescriptor::from_config(&config).unwrap().id + ); + } + #[test] + fn canonical_nested_parameters_and_model_versions_are_identity() { + let a = SummaryOperator::Configured { + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: BTreeMap::from([("nested".into(), json!({"z":1,"a":2}))]), + }; + let b = SummaryOperator::Configured { + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: BTreeMap::from([("nested".into(), json!({"a":2,"z":1}))]), + }; + assert_eq!( + SummaryDescriptor::new(a.clone(), FidelityGuarantee::Exact, 1) + .unwrap() + .id, + SummaryDescriptor::new(b, FidelityGuarantee::Exact, 1) + .unwrap() + .id + ); + let first = SummaryDescriptor::new( + a.clone(), + FidelityGuarantee::KllRankError { + k: 200, + model: "rank.v1".into(), + }, + 1, + ) + .unwrap(); + let second = SummaryDescriptor::new( + a, + FidelityGuarantee::KllRankError { + k: 200, + model: "rank.v2".into(), + }, + 1, + ) + .unwrap(); + assert_ne!(first.id, second.id); + } + #[test] + fn materialization_id_preserves_legacy_wire_identity() { + let fingerprint = crate::PolicyFingerprint(42); + let id = MaterializationId::from(fingerprint); + assert_eq!(id.fingerprint(), fingerprint); + assert_eq!(id.as_u64(), 42); + assert_eq!(crate::PolicyFingerprint::from(id), fingerprint); + assert_eq!(serde_json::to_value(id).unwrap(), serde_json::to_value(fingerprint).unwrap()); + assert_eq!(serde_json::from_str::("42").unwrap(), id); + } + +} diff --git a/data_plane/src/storage_engines/sketch_db/sds.rs b/data_plane/src/storage_engines/sketch_db/sds.rs index 0b1ae2fe..5e761cfa 100644 --- a/data_plane/src/storage_engines/sketch_db/sds.rs +++ b/data_plane/src/storage_engines/sketch_db/sds.rs @@ -7,81 +7,72 @@ use std::collections::{BTreeSet, HashMap}; use std::sync::{Arc, RwLock, Weak}; -use super::data::{AccuracyBound, AggKind, SketchAlgorithm, SketchConfig}; +use super::data::{AggKind, SketchConfig}; use super::index::SketchInstanceMetadata; +#[cfg(test)] use crate::storage_engines::types::AggregationType; - -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub struct SummaryDescriptorId(Arc); - -impl SummaryDescriptorId { - pub fn canonical(&self) -> &str { - &self.0 - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub struct DataDescriptorId(Arc); - -impl DataDescriptorId { - pub fn canonical(&self) -> &str { - &self.0 - } -} - -#[derive(Debug, Clone)] -pub enum SummaryOperator { - Sketch { - algorithm: SketchAlgorithm, - config: SketchConfig, - }, - ExactAgg { - agg_type: AggregationType, - parameters_canonical: Arc, - }, -} - -impl SummaryOperator { - fn from_agg_kind(kind: &AggKind) -> Self { - match kind { - AggKind::Sketch { - algorithm, config, .. - } => Self::Sketch { - algorithm: algorithm.clone(), - config: config.clone(), +pub use asap_types::sds::{ + DataDescriptor, DataDescriptorId, FidelityGuarantee, SummaryDescriptor, SummaryDescriptorId, + SummaryOperator, +}; + +// Legacy records lack some state-shape dimensions (heap/Hydra/subtype). Do not +// let these projections masquerade as an authoritative Configured descriptor. +fn legacy_summary(kind: &AggKind) -> SummaryDescriptor { + let fidelity = match kind { + AggKind::ExactAgg { .. } => FidelityGuarantee::Exact, + AggKind::Sketch { config, .. } => match config { + SketchConfig::Kll { k } => FidelityGuarantee::KllRankError { + k: *k, + model: "asap.kll.normalized-rank.v1".into(), }, - AggKind::ExactAgg { - agg_type, - parameters_canonical, - .. - } => Self::ExactAgg { - agg_type: *agg_type, - parameters_canonical: Arc::from(parameters_canonical.as_str()), + SketchConfig::DDSketch { relative_accuracy } => { + FidelityGuarantee::DdSketchRelativeError { + alpha: *relative_accuracy, + } + } + SketchConfig::Hll { precision } => FidelityGuarantee::HllCardinalityError { + precision: *precision, + model: "asap.hll.relative-cardinality.v1".into(), }, - } - } -} - -#[derive(Debug, Clone)] -pub enum FidelityGuarantee { - Exact, - Approximate(AccuracyBound), -} - -#[derive(Debug)] -pub struct SummaryDescriptor { - pub id: SummaryDescriptorId, - pub operator: SummaryOperator, - pub fidelity: FidelityGuarantee, - pub state_schema_version: u32, -} - -#[derive(Debug)] -pub struct DataDescriptor { - pub id: DataDescriptorId, - pub metric_name: Arc, - pub population_filter_canonical: Arc, - pub group_by_keys: Arc>, + SketchConfig::CountMin { rows, cols } if *rows > 0 && *cols > 0 => { + FidelityGuarantee::CmsFrequencyError { + width: *cols as u32, + depth: *rows as u32, + model: "asap.cms.point-frequency.v1".into(), + } + } + SketchConfig::CountSketch { rows, cols } if *rows > 0 && *cols > 0 => { + FidelityGuarantee::CountSketchFrequencyError { + width: *cols as u32, + depth: *rows as u32, + model: "asap.count-sketch.point-frequency.v1".into(), + } + } + _ => FidelityGuarantee::Unknown { + reason: "Legacy configuration has invalid dimensions".into(), + }, + }, + }; + SummaryDescriptor::new( + SummaryOperator::LegacyPartial { + operator_canonical: kind.operator_canonical_string(), + }, + fidelity, + 1, + ) + .unwrap_or_else(|_| { + SummaryDescriptor::new( + SummaryOperator::LegacyPartial { + operator_canonical: kind.operator_canonical_string(), + }, + FidelityGuarantee::Unknown { + reason: "Legacy configuration has invalid fidelity parameters".into(), + }, + 1, + ) + .expect("unknown legacy descriptor has valid version and reason") + }) } /// Runtime foreign-key binding from one SID to shared descriptors. Every pane @@ -120,16 +111,7 @@ impl SummaryDescriptorRegistry { if let Some(existing) = summaries.get(&summary_id).and_then(Weak::upgrade) { existing } else { - let fidelity = match metadata.agg_kind.capability_and_accuracy().1 { - Some(bound) => FidelityGuarantee::Approximate(bound), - None => FidelityGuarantee::Exact, - }; - let descriptor = Arc::new(SummaryDescriptor { - id: summary_id.clone(), - operator: SummaryOperator::from_agg_kind(&metadata.agg_kind), - fidelity, - state_schema_version: 1, - }); + let descriptor = Arc::new(legacy_summary(&metadata.agg_kind)); summaries.insert(summary_id, Arc::downgrade(&descriptor)); descriptor } @@ -146,12 +128,11 @@ impl SummaryDescriptorRegistry { if let Some(existing) = data.get(&data_id).and_then(Weak::upgrade) { existing } else { - let descriptor = Arc::new(DataDescriptor { - id: data_id, - metric_name: Arc::from(metadata.metric_name.as_str()), - population_filter_canonical: Arc::from(filter), - group_by_keys: Arc::new(metadata.group_by_keys.clone()), - }); + let descriptor = Arc::new(DataDescriptor::new( + metadata.metric_name.clone(), + filter, + metadata.group_by_keys.iter().cloned(), + )); data.insert(descriptor.id.clone(), Arc::downgrade(&descriptor)); descriptor } @@ -201,14 +182,8 @@ impl SummaryDescriptorRegistry { * (std::mem::size_of::() + std::mem::size_of::>()); for descriptor in summaries.values().filter_map(Weak::upgrade) { - total += std::mem::size_of::() + descriptor.id.0.len(); - if let SummaryOperator::ExactAgg { - parameters_canonical, - .. - } = &descriptor.operator - { - total += parameters_canonical.len(); - } + total += std::mem::size_of::() + descriptor.id.canonical().len(); + total += serde_json::to_string(&descriptor.operator).map_or(0, |value| value.len()); } drop(summaries); @@ -217,7 +192,7 @@ impl SummaryDescriptorRegistry { * (std::mem::size_of::() + std::mem::size_of::>()); for descriptor in data.values().filter_map(Weak::upgrade) { - total += std::mem::size_of::() + descriptor.id.0.len(); + total += std::mem::size_of::() + descriptor.id.canonical().len(); total += descriptor.metric_name.len() + descriptor.population_filter_canonical.len(); total += descriptor .group_by_keys @@ -230,7 +205,7 @@ impl SummaryDescriptorRegistry { } pub(crate) fn summary_descriptor_id(kind: &AggKind) -> SummaryDescriptorId { - SummaryDescriptorId(kind.operator_canonical_string().into()) + legacy_summary(kind).id } pub(crate) fn data_descriptor_id<'a>( @@ -238,27 +213,7 @@ pub(crate) fn data_descriptor_id<'a>( filter: &str, group_by: impl Iterator, ) -> DataDescriptorId { - DataDescriptorId(canonical_data_key(metric, filter, group_by).into()) -} - -fn canonical_data_key<'a>( - metric: &str, - filter: &str, - group_by: impl Iterator, -) -> String { - fn push_part(out: &mut String, value: &str) { - use std::fmt::Write; - let _ = write!(out, "{}:{value}", value.len()); - } - let mut out = String::from("data:v1|"); - push_part(&mut out, metric); - out.push('|'); - push_part(&mut out, filter); - for key in group_by { - out.push('|'); - push_part(&mut out, key); - } - out + DataDescriptor::new(metric, filter, group_by.map(str::to_string)).id } #[cfg(test)]