From 955ffe8ca3d33650630e27cbddb8fd35f0953a60 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 08:27:21 -0600 Subject: [PATCH 1/3] feat(catalog): identify derived summary inputs --- crates/asap_types/src/aggregation_config.rs | 48 ++- crates/asap_types/src/derived_input.rs | 322 ++++++++++++++++++ crates/asap_types/src/lib.rs | 1 + crates/asap_types/src/policy_fingerprint.rs | 9 +- crates/asap_types/src/precompute_plan.rs | 11 + .../asap_types/src/precompute_plan/catalog.rs | 11 +- crates/asap_types/src/sds.rs | 16 +- crates/asap_types/src/summary_catalog.rs | 57 +++- .../drivers/ingest/prometheus_remote_write.rs | 2 + data_plane/src/drivers/query/servers/http.rs | 1 + .../src/precompute_engine/output_sink.rs | 1 + .../sketch_db/lifecycle/eviction.rs | 1 + .../storage_engines/types/streaming_config.rs | 16 + .../tests/test_utilities/engine_factories.rs | 8 + .../summary-catalog-sds-architecture.md | 21 ++ 15 files changed, 503 insertions(+), 22 deletions(-) create mode 100644 crates/asap_types/src/derived_input.rs diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 3e9f5cab3..8254b65b1 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -103,6 +103,8 @@ pub struct PrecomputeMaterialization { pub grouping_labels: crate::GroupingProjection, #[serde(default, skip_serializing_if = "Option::is_none")] pub partitioning: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub derived_input: Option, pub aggregated_labels: KeyByLabelNames, pub rollup_labels: KeyByLabelNames, pub original_yaml: String, @@ -200,9 +202,39 @@ impl PrecomputeMaterialization { .saturating_mul(1_000) } + pub fn source_identity(&self) -> crate::sds::DataSourceIdentity { + use crate::sds::DataSourceIdentity; + if let Some(input) = &self.derived_input { + DataSourceIdentity::Derived { + input: input.clone(), + } + } else if let Some(table_ref) = &self.table_name { + DataSourceIdentity::Table { + table_ref: table_ref.clone(), + } + } else { + DataSourceIdentity::TimeSeries { + metric: self.metric.clone(), + } + } + } + pub fn population_filter_canonical(&self) -> Result { + if let Some(input) = &self.derived_input { + input.validate()?; + if self.table_name.is_some() + || self.table_population.is_some() + || self.table_timestamp_column.is_some() + || !self.spatial_filter.is_empty() + { + return Err("derived inputs cannot also declare a raw source/filter".into()); + } + } self.effective_value_projection().validate()?; - if self.value_projection.is_some() && self.table_name.is_none() { + if self.value_projection.is_some() + && self.table_name.is_none() + && self.derived_input.is_none() + { return Err("explicit table value projection requires a table source".into()); } if let Some(column) = &self.table_timestamp_column { @@ -255,6 +287,7 @@ impl PrecomputeMaterialization { parameters, grouping_labels: grouping_labels.into(), partitioning: None, + derived_input: None, aggregated_labels, rollup_labels, original_yaml, @@ -400,6 +433,11 @@ impl PrecomputeMaterialization { table_name, value_column, ); + config.derived_input = data + .get("derived_input") + .filter(|v| !v.is_null()) + .map(|v| serde_json::from_value(v.clone())) + .transpose()?; config.partitioning = data .get("partitioning") .filter(|value| !value.is_null()) @@ -595,6 +633,11 @@ impl PrecomputeMaterialization { table_name, value_column, ); + config.derived_input = aggregation_data + .get("derived_input") + .filter(|v| !v.is_null()) + .map(|v| serde_yaml::from_value(v.clone())) + .transpose()?; config.partitioning = aggregation_data .get("partitioning") .filter(|value| !value.is_null()) @@ -640,6 +683,9 @@ impl SerializableToSink for PrecomputeMaterialization { "metric": self.metric, }); + if let Some(input) = &self.derived_input { + json["derived_input"] = serde_json::json!(input); + } // Only include numAggregatesToRetain if it's Some if let Some(num_aggregates) = self.num_aggregates_to_retain { json["numAggregatesToRetain"] = serde_json::json!(num_aggregates); diff --git a/crates/asap_types/src/derived_input.rs b/crates/asap_types/src/derived_input.rs new file mode 100644 index 000000000..5aa86fbb2 --- /dev/null +++ b/crates/asap_types/src/derived_input.rs @@ -0,0 +1,322 @@ +//! Content identity for a maintenance sub-DAG cut at existing summary inputs. +//! The executable program remains in OwnedPostAsapDag; this is its catalog key. +use std::collections::{BTreeMap, BTreeSet}; + +use planner_types::post_asap::PostAsapNodeId; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::{executable_plan::OwnedPostAsapDag, sds::SummaryDefinitionId}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DerivedInputIdentity { + pub inputs: BTreeSet, + pub program_sha256: String, +} + +impl DerivedInputIdentity { + pub fn validate(&self) -> Result<(), String> { + if self.inputs.is_empty() + || self.inputs.iter().any(|id| id.fingerprint().is_unset()) + || self.program_sha256.len() != 64 + || !self + .program_sha256 + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + { + return Err("derived input requires summary references and a canonical SHA-256".into()); + } + Ok(()) + } + + /// Hash semantic nodes and edge roles, replacing input frontiers with stable + /// catalog IDs. Query IDs, node numbering, and catalog generations are absent. + pub fn from_dag( + document: &OwnedPostAsapDag, + root: PostAsapNodeId, + frontiers: &BTreeMap, + ) -> Result { + if document.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION { + return Err("unsupported derived program document version".into()); + } + document.decode()?; + let mut incoming: BTreeMap<_, Vec<_>> = BTreeMap::new(); + for edge in &document.edges { + incoming.entry(edge.consumer).or_default().push(edge); + } + let nodes: BTreeMap<_, _> = document.nodes.iter().map(|n| (n.id, n)).collect(); + if frontiers.contains_key(&root) { + return Err("derived program requires a non-frontier root".into()); + } + let mut hashes = BTreeMap::new(); + let mut visiting = BTreeSet::new(); + let mut inputs = BTreeSet::new(); + let mut stack = vec![(root, false)]; + while let Some((id, finish)) = stack.pop() { + if hashes.contains_key(&id) { + continue; + } + let node = nodes + .get(&id) + .ok_or("derived program references missing node")?; + if let Some(summary) = frontiers.get(&id) { + inputs.insert(*summary); + hashes.insert(id, serde_json::json!({"summary": summary})); + continue; + } + if !finish { + if !visiting.insert(id) { + return Err("derived program has a cycle".into()); + } + stack.push((id, true)); + let edges = incoming + .get(&id) + .ok_or("derived program has an unbound input leaf")?; + for edge in edges { + stack.push((edge.producer, false)); + } + continue; + } + let mut edges = Vec::new(); + for edge in incoming.get(&id).into_iter().flatten() { + let input = hashes + .get(&edge.producer) + .ok_or("derived input was not evaluated")?; + edges.push( + serde_json::to_vec(&serde_json::json!({ + "input": input, "role": edge.role, "schema": edge.intermediate_schema, + "state": edge.data_state, "grouping": edge.grouping, "window": edge.window, + })) + .map_err(|e| e.to_string())?, + ); + } + edges.sort(); + let bytes = serde_json::to_vec(&serde_json::json!({ + "version": 1, "operator": node.operator, "payload": node.payload, + "state": node.output_state, "schema": node.output_schema, + "guarantee": node.guarantee, "inputs": edges, + })) + .map_err(|e| e.to_string())?; + hashes.insert( + id, + serde_json::json!(format!("{:x}", Sha256::digest(bytes))), + ); + visiting.remove(&id); + } + let identity = Self { + inputs, + program_sha256: hashes[&root] + .as_str() + .ok_or("derived root has no semantic hash")? + .into(), + }; + identity.validate()?; + Ok(identity) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::summary_catalog::SummaryCatalog; + use crate::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; + + fn config() -> PrecomputeMaterialization { + PrecomputeMaterialization::new( + AggregationType::Sum, + String::new(), + Default::default(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 10, + 10, + WindowKind::Tumbling, + String::new(), + "m".into(), + None, + None, + None, + ) + } + + #[test] + fn derived_source_is_distinct_and_generation_independent() { + let raw = config(); + let raw_id = SummaryDefinitionId::from(raw.policy_fingerprint()); + let mut derived = raw.clone(); + derived.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([raw_id]), + program_sha256: "a".repeat(64), + }); + assert_ne!(raw.policy_fingerprint(), derived.policy_fingerprint()); + let a = + SummaryCatalog::from_materializations(1, 1, &[raw.clone(), derived.clone()]).unwrap(); + let b = SummaryCatalog::from_materializations(2, 9, &[raw, derived.clone()]).unwrap(); + assert_eq!(a.materializations, b.materializations); + assert_eq!(a.data_descriptors, b.data_descriptors); + let mut renamed = derived.clone(); + renamed.metric = "output_alias".into(); + assert_eq!(renamed.policy_fingerprint(), derived.policy_fingerprint()); + let json = serde_json::to_value(&derived).unwrap(); + let decoded: PrecomputeMaterialization = serde_json::from_value(json).unwrap(); + assert_eq!(decoded.policy_fingerprint(), derived.policy_fingerprint()); + } + + #[test] + fn catalog_rejects_missing_derived_dependencies_and_raw_source_conflicts() { + let mut derived = config(); + derived.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(derived.policy_fingerprint())]), + program_sha256: "b".repeat(64), + }); + assert!(SummaryCatalog::from_materializations(1, 1, &[derived.clone()]).is_err()); + derived.table_name = Some("table".into()); + assert!(derived.population_filter_canonical().is_err()); + } + fn program(source: u32, root: u32) -> OwnedPostAsapDag { + use crate::executable_plan::{OwnedPostAsapEdge, OwnedPostAsapNode}; + use planner_types::post_asap::{ + EdgeRole, ExecutableOperator, ExecutionDataState, GroupingEdgeCompatibility, + WindowEdgeCompatibility, + }; + let state = ExecutionDataState::MAINTENANCE_SUMMARY; + OwnedPostAsapDag { + schema_version: 1, + query_id: "query-a".into(), + root: PostAsapNodeId(root), + nodes: [source, root] + .into_iter() + .map(|id| OwnedPostAsapNode { + id: PostAsapNodeId(id), + operator: ExecutableOperator::SummaryMerge, + payload: serde_json::json!({"kind":"summary_merge"}), + output_state: state, + output_schema: serde_json::json!({"fields":[],"time_index":null}), + guarantee: None, + }) + .collect(), + edges: vec![OwnedPostAsapEdge { + producer: PostAsapNodeId(source), + consumer: PostAsapNodeId(root), + role: EdgeRole::Input, + intermediate_schema: serde_json::json!({"fields":[],"time_index":null}), + data_state: state, + grouping: GroupingEdgeCompatibility::Identical, + window: WindowEdgeCompatibility::NotApplicable, + }], + } + } + + #[test] + fn semantic_signature_ignores_node_and_query_numbering_but_not_inputs() { + let source = SummaryDefinitionId::from(config().policy_fingerprint()); + let a = program(1, 2); + let first = DerivedInputIdentity::from_dag( + &a, + a.root, + &BTreeMap::from([(PostAsapNodeId(1), source)]), + ) + .unwrap(); + let mut b = program(900, 42); + b.query_id = "another-query".into(); + b.nodes.reverse(); + let second = DerivedInputIdentity::from_dag( + &b, + b.root, + &BTreeMap::from([(PostAsapNodeId(900), source)]), + ) + .unwrap(); + assert_eq!(first, second); + b.nodes + .iter_mut() + .find(|n| n.id == b.root) + .unwrap() + .output_schema = serde_json::json!({"fields":[],"time_index":0}); + assert_ne!( + first, + DerivedInputIdentity::from_dag( + &b, + b.root, + &BTreeMap::from([(PostAsapNodeId(900), source)]) + ) + .unwrap() + ); + assert!(DerivedInputIdentity::from_dag(&a, a.root, &BTreeMap::new()).is_err()); + let mut unsupported = a.clone(); + unsupported.schema_version = 999; + assert!(DerivedInputIdentity::from_dag( + &unsupported, + unsupported.root, + &BTreeMap::from([(PostAsapNodeId(1), source)]) + ) + .is_err()); + unsupported = a.clone(); + unsupported.nodes[1].payload = serde_json::json!({"kind":"unknown_operator"}); + assert!(DerivedInputIdentity::from_dag( + &unsupported, + unsupported.root, + &BTreeMap::from([(PostAsapNodeId(1), source)]) + ) + .is_err()); + let mut cycle = a.clone(); + cycle.edges[0].producer = cycle.root; + assert!(DerivedInputIdentity::from_dag(&cycle, cycle.root, &BTreeMap::new()).is_err()); + cycle.edges[0].producer = PostAsapNodeId(999); + assert!(DerivedInputIdentity::from_dag(&cycle, cycle.root, &BTreeMap::new()).is_err()); + } + + #[test] + fn catalog_rejects_self_referential_summary() { + use crate::sds::{ + DataDescriptor, DataSourceIdentity, SummaryDescriptor, ValueProjectionIdentity, + }; + let config = config(); + let input = DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(config.policy_fingerprint())]), + program_sha256: "f".repeat(64), + }; + let data = DataDescriptor::new_typed( + DataSourceIdentity::Derived { input }, + ValueProjectionIdentity::SampleValue, + "", + Vec::::new(), + "derived", + ); + assert!(SummaryCatalog::build( + 1, + 1, + [( + config.policy_fingerprint(), + SummaryDescriptor::from_config(&config).unwrap(), + data, + config.window_layout + )] + ) + .is_err()); + } + #[test] + fn installation_rejects_derived_inputs_until_immutable_consumer_is_enabled() { + use crate::precompute_plan::{PlanEnvelope, PrecomputePlan}; + let mut config = config(); + config.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(config.policy_fingerprint())]), + program_sha256: "c".repeat(64), + }); + let envelope = PlanEnvelope { + plan_id: 1, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "asap-query-backend.v1".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let error = + PrecomputePlan::build(envelope, vec![config], &["producer".into()]).unwrap_err(); + assert!(error.to_string().contains("immutable maintenance consumer")); + } +} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 3c0a23997..0832dfb12 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -2,6 +2,7 @@ pub mod accumulator_spec; pub mod accuracy; pub mod aggregation_config; pub mod aggregation_type; +pub mod derived_input; pub mod enums; pub mod executable_plan; pub mod grouping_projection; diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index f9e751651..46837a95e 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -90,7 +90,14 @@ impl PolicyFingerprint { let mut buf: Vec = Vec::with_capacity(512); // 1. metric name - buf.extend_from_slice(cfg.metric.as_bytes()); + if cfg.derived_input.is_some() { + buf.extend_from_slice(b"derived-input-v1:"); + buf.extend_from_slice( + &serde_json::to_vec(&cfg.source_identity()).expect("typed source identity"), + ); + } else { + buf.extend_from_slice(cfg.metric.as_bytes()); + } buf.push(0); // 2. aggregation_type (Serialize impl is the stable form) diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 257d99dd4..da7b07b0c 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -334,6 +334,17 @@ impl PrecomputePlan { if !valid_ingest { return Err(PrecomputePlanError::UnsupportedIngestEndpoint); } + // Derived input identity is portable, but raw ingress cannot execute it. + // Installation stays closed until the immutable maintenance consumer is wired. + if self + .materializations + .iter() + .any(|config| config.derived_input.is_some()) + { + return Err(PrecomputePlanError::CatalogContract( + "derived summary input requires an immutable maintenance consumer".into(), + )); + } for (query_id, installed) in &self.executable_dags { if query_id != &installed.document.query_id { return Err(PrecomputePlanError::CatalogContract( diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index 9ee907798..e9dd2a599 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -1,6 +1,6 @@ //! Catalog consistency checks for the precompute execution plan. use super::*; -use crate::sds::{DataSourceIdentity, SummaryDefinitionId, SummaryDescriptor}; +use crate::sds::{SummaryDefinitionId, SummaryDescriptor}; use crate::summary_catalog::SummaryCatalog; use planner_types::pre_asap::Source; use std::collections::BTreeSet; @@ -70,14 +70,7 @@ impl PrecomputePlan { return Err(invalid("pane origin differs from catalog definition")); } let data = &catalog.data_descriptors[&binding.data_descriptor_id]; - let expected_source = config.table_name.as_ref().map_or_else( - || DataSourceIdentity::TimeSeries { - metric: config.metric.clone(), - }, - |table_ref| DataSourceIdentity::Table { - table_ref: table_ref.clone(), - }, - ); + let expected_source = config.source_identity(); if config.table_name.is_some() && !config.grouping_labels.is_empty() && data.observation_semantics diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 23c3cc48b..b75fd88ca 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -729,8 +729,15 @@ impl FidelityGuarantee { #[serde(deny_unknown_fields)] #[serde(rename_all = "snake_case")] pub enum DataSourceIdentity { - TimeSeries { metric: String }, - Table { table_ref: String }, + TimeSeries { + metric: String, + }, + Table { + table_ref: String, + }, + Derived { + input: crate::derived_input::DerivedInputIdentity, + }, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -829,7 +836,7 @@ impl DataDescriptor { pub fn time_series_metric(&self) -> Option<&str> { match &self.source { DataSourceIdentity::TimeSeries { metric } => Some(metric), - DataSourceIdentity::Table { .. } => None, + DataSourceIdentity::Table { .. } | DataSourceIdentity::Derived { .. } => None, } } @@ -929,6 +936,9 @@ impl DataDescriptor { &self.id } pub fn validate(&self) -> Result<(), SdsError> { + if let DataSourceIdentity::Derived { input } = &self.source { + input.validate().map_err(SdsError)?; + } self.value_projection.validate().map_err(SdsError)?; self.group_by_keys.validate().map_err(SdsError)?; if matches!(self.source, DataSourceIdentity::TimeSeries { .. }) diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 5f31ee43d..36468724d 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -103,14 +103,7 @@ impl SummaryCatalog { .map(|config| { let summary = SummaryDescriptor::from_config(config) .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; - let source = config.table_name.as_ref().map_or_else( - || DataSourceIdentity::TimeSeries { - metric: config.metric.clone(), - }, - |table_ref| DataSourceIdentity::Table { - table_ref: table_ref.clone(), - }, - ); + let source = config.source_identity(); let value_projection = config.effective_value_projection().clone(); let data = DataDescriptor::new_typed( source, @@ -253,6 +246,54 @@ impl SummaryCatalog { return Err(SummaryCatalogError::MissingDescriptor(id.as_u64())); } } + if !self + .data_descriptors + .values() + .any(|data| matches!(data.source, DataSourceIdentity::Derived { .. })) + { + return Ok(()); + } + // Dependencies must refer to this snapshot and form an acyclic graph. + let mut pending = std::collections::BTreeMap::new(); + let mut consumers: std::collections::BTreeMap<_, Vec<_>> = + std::collections::BTreeMap::new(); + for (id, binding) in &self.materializations { + let dependencies = match &self.data_descriptors[&binding.data_descriptor_id].source { + DataSourceIdentity::Derived { input } => input.inputs.clone(), + _ => Default::default(), + }; + for source in &dependencies { + if !self.materializations.contains_key(source) { + return Err(SummaryCatalogError::Descriptor( + "derived input references missing summary".into(), + )); + } + consumers.entry(*source).or_default().push(*id); + } + pending.insert(*id, dependencies.len()); + } + let mut ready: Vec<_> = pending + .iter() + .filter_map(|(id, count)| (*count == 0).then_some(*id)) + .collect(); + let mut visited = 0; + while let Some(id) = ready.pop() { + visited += 1; + for consumer in consumers.get(&id).into_iter().flatten() { + let count = pending + .get_mut(consumer) + .expect("catalog dependency target"); + *count -= 1; + if *count == 0 { + ready.push(*consumer); + } + } + } + if visited != pending.len() { + return Err(SummaryCatalogError::Descriptor( + "derived summary dependencies have a cycle".into(), + )); + } Ok(()) } } diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index c1ec49919..d0aea8b9e 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -988,6 +988,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -1059,6 +1060,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index c456d8e3c..af0d3a07a 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3681,6 +3681,7 @@ aggregations: table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index b908f2a8b..46a83a42c 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -422,6 +422,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, } diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs index 174c4ae5c..30713e75a 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -276,6 +276,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, } diff --git a/data_plane/src/storage_engines/types/streaming_config.rs b/data_plane/src/storage_engines/types/streaming_config.rs index 91a2dc1b3..2298955db 100644 --- a/data_plane/src/storage_engines/types/streaming_config.rs +++ b/data_plane/src/storage_engines/types/streaming_config.rs @@ -133,6 +133,11 @@ impl StreamingConfig { num_aggregates_to_retain, QueryLanguage::PromQl, )?; + if config.derived_input.is_some() { + anyhow::bail!( + "legacy streaming input cannot execute a derived summary program" + ); + } // PR 5: the map key IS the policy-fingerprint u64. // `AggregationConfig::policy_fp_u64()` is the canonical // accessor for this value. @@ -191,6 +196,17 @@ mod tests { assert_eq!(cfg.storage_backend(), StorageBackend::GorillaObjectStore); } + #[test] + fn legacy_yaml_rejects_derived_summary_input() { + let data = serde_yaml::from_str::(&format!( + "aggregations:\n- aggregationType: Sum\n aggregationSubType: ''\n metric: outer\n labels: {{grouping: [], rollup: [], aggregated: []}}\n parameters: {{}}\n windowSize: 10\n windowType: tumbling\n spatialFilter: ''\n derived_input:\n inputs: [1]\n program_sha256: '{}'\n", "a".repeat(64) + )).unwrap(); + let error = StreamingConfig::from_yaml_data(&data).unwrap_err(); + assert!(error + .to_string() + .contains("legacy streaming input cannot execute")); + } + /// PR 5: a streaming-config YAML that omits `aggregationId` /// parses correctly — the backend derives identity from content /// via `PolicyFingerprint::from_config`. The map key is the diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index 3fce2d44b..564d21d7c 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -109,6 +109,7 @@ pub fn create_engine_single_pop_with_aggregated( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -206,6 +207,7 @@ pub fn create_engine_dual_input( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -233,6 +235,7 @@ pub fn create_engine_dual_input( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -325,6 +328,7 @@ pub fn create_engine_two_metrics( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -351,6 +355,7 @@ pub fn create_engine_two_metrics( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -453,6 +458,7 @@ pub fn create_engine_three_metrics( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -532,6 +538,7 @@ pub fn create_engine_multi_timestamp( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -603,6 +610,7 @@ pub fn create_engine_multi_timestamp_with_window( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 048dcbac4..39956f7fe 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -429,3 +429,24 @@ This is an explicit lifetime transition, not cross-generation recovery of arbitr summary state. Legacy records without trustworthy catalog provenance remain unbound. Tombstone reclamation still requires coordinated removal of old physical parts and is not implemented by this transition. + +### Derived summary input identity + +A summary computed from another summary has a different data source from the +original raw table or metric. `PrecomputeMaterialization.derived_input` and +`DataSourceIdentity::Derived` use the same `DerivedInputIdentity`: the referenced +`SummaryDefinitionId`s and a SHA-256 of the maintenance program. The executable +program remains in `OwnedPostAsapDag`; the catalog does not retain another copy. + +The signature replaces materialized input frontiers with stable summary IDs and +hashes the remaining node payloads, schemas, guarantees, and edge semantics. It +excludes query names, plan-local node numbering, and catalog generations. A changed +input definition or transformation creates a new identity. Existing raw-source +identities retain their previous byte representation. Catalog validation rejects +missing input definitions and dependency cycles. + +This contract is a prerequisite, not enabled summary-over-summary execution. +Installation currently rejects derived inputs so they cannot accidentally receive +raw samples through the legacy metric router. Enabling them requires the immutable +maintenance consumer and durable output deduplication protocol; neither raw-table +substitution nor treating late correction fragments as new observations is valid. From bcbb5a40129493e6b724c18916979704a4028537 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 08:44:47 -0600 Subject: [PATCH 2/3] fix(catalog): separate derived policy domain from raw metric names --- crates/asap_types/src/derived_input.rs | 71 +++++++++++++++++-- crates/asap_types/src/policy_fingerprint.rs | 4 +- .../summary-catalog-sds-architecture.md | 3 +- 3 files changed, 71 insertions(+), 7 deletions(-) diff --git a/crates/asap_types/src/derived_input.rs b/crates/asap_types/src/derived_input.rs index 5aa86fbb2..a5f6d6dce 100644 --- a/crates/asap_types/src/derived_input.rs +++ b/crates/asap_types/src/derived_input.rs @@ -40,7 +40,20 @@ impl DerivedInputIdentity { if document.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION { return Err("unsupported derived program document version".into()); } - document.decode()?; + let decoded = document.decode()?; + let literals: BTreeSet<_> = decoded + .nodes + .iter() + .filter_map(|node| { + matches!( + &node.payload, + planner_types::post_asap::ExecutableOperatorPayload::Fallback { + expression: planner_types::pre_asap::QueryExpr::Literal(_), + } + ) + .then_some(node.id) + }) + .collect(); let mut incoming: BTreeMap<_, Vec<_>> = BTreeMap::new(); for edge in &document.edges { incoming.entry(edge.consumer).or_default().push(edge); @@ -70,10 +83,10 @@ impl DerivedInputIdentity { return Err("derived program has a cycle".into()); } stack.push((id, true)); - let edges = incoming - .get(&id) - .ok_or("derived program has an unbound input leaf")?; - for edge in edges { + if !incoming.contains_key(&id) && !literals.contains(&id) { + return Err("derived program has an unbound input leaf".into()); + } + for edge in incoming.get(&id).into_iter().flatten() { stack.push((edge.producer, false)); } continue; @@ -165,6 +178,22 @@ mod tests { assert_eq!(decoded.policy_fingerprint(), derived.policy_fingerprint()); } + #[test] + fn raw_utf8_metric_cannot_impersonate_derived_policy_domain() { + let mut derived = config(); + derived.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(derived.policy_fingerprint())]), + program_sha256: "d".repeat(64), + }); + let mut raw = derived.clone(); + raw.metric = format!( + "derived-input-v1:{}", + serde_json::to_string(&derived.source_identity()).unwrap() + ); + raw.derived_input = None; + assert_ne!(raw.policy_fingerprint(), derived.policy_fingerprint()); + } + #[test] fn catalog_rejects_missing_derived_dependencies_and_raw_source_conflicts() { let mut derived = config(); @@ -319,4 +348,36 @@ mod tests { PrecomputePlan::build(envelope, vec![config], &["producer".into()]).unwrap_err(); assert!(error.to_string().contains("immutable maintenance consumer")); } + #[test] + fn literal_leaves_are_hashed_without_inventing_materialization_references() { + use planner_types::{ + post_asap::{ExecutableOperator, ExecutableOperatorPayload}, + pre_asap::{QueryExpr, ScalarValue}, + }; + let mut dag = program(1, 2); + let mut literal = dag.nodes[0].clone(); + literal.id = PostAsapNodeId(3); + literal.operator = ExecutableOperator::Fallback; + literal.payload = serde_json::to_value(ExecutableOperatorPayload::Fallback { + expression: QueryExpr::Literal(ScalarValue::Int64(2)), + }) + .unwrap(); + dag.nodes.push(literal); + let mut edge = dag.edges[0].clone(); + edge.producer = PostAsapNodeId(3); + dag.edges.push(edge); + let source = SummaryDefinitionId::from(config().policy_fingerprint()); + let frontiers = BTreeMap::from([(PostAsapNodeId(1), source)]); + let first = DerivedInputIdentity::from_dag(&dag, dag.root, &frontiers).unwrap(); + assert_eq!(first.inputs, BTreeSet::from([source])); + dag.nodes[2].payload = serde_json::to_value(ExecutableOperatorPayload::Fallback { + expression: QueryExpr::Literal(ScalarValue::Int64(3)), + }) + .unwrap(); + assert_ne!( + first, + DerivedInputIdentity::from_dag(&dag, dag.root, &frontiers).unwrap() + ); + assert!(DerivedInputIdentity::from_dag(&dag, PostAsapNodeId(3), &frontiers).is_err()); + } } diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 46837a95e..38afc45ff 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -91,7 +91,9 @@ impl PolicyFingerprint { // 1. metric name if cfg.derived_input.is_some() { - buf.extend_from_slice(b"derived-input-v1:"); + // Raw policies start with UTF-8 metric bytes; 0xff is impossible + // there, so a metric cannot impersonate this source domain. + buf.extend_from_slice(b"\xffderived-input-v1:"); buf.extend_from_slice( &serde_json::to_vec(&cfg.source_identity()).expect("typed source identity"), ); diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 39956f7fe..4196f079a 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -440,7 +440,8 @@ program remains in `OwnedPostAsapDag`; the catalog does not retain another copy. The signature replaces materialized input frontiers with stable summary IDs and hashes the remaining node payloads, schemas, guarantees, and edge semantics. It -excludes query names, plan-local node numbering, and catalog generations. A changed +excludes query names, plan-local node numbering, and catalog generations. Literal +leaves are hashed directly; raw input leaves still require catalog frontiers. A changed input definition or transformation creates a new identity. Existing raw-source identities retain their previous byte representation. Catalog validation rejects missing input definitions and dependency cycles. From 2fc455a5d24a847a4047ed01de9fdf0a12070224 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 09:24:19 -0600 Subject: [PATCH 3/3] docs(sds): separate immutable window section --- docs/design_docs/summary-catalog-sds-architecture.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index e2fa87306..14b388e5c 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -462,6 +462,7 @@ Installation currently rejects derived inputs so they cannot accidentally receiv raw samples through the legacy metric router. Enabling them requires the immutable maintenance consumer and durable output deduplication protocol; neither raw-table substitution nor treating late correction fragments as new observations is valid. + ### Immutable completed windows Finite Remote Write completion now fences the SummaryStore append boundary,