diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index b49aadff..bc2e2db4 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -13,6 +13,43 @@ use std::collections::{BTreeMap, BTreeSet}; use std::sync::{Arc, Mutex}; type SummaryState = Arc; + +#[derive(Clone)] +enum MaintenanceValue { + Summary { + state: SummaryState, + family: Option, + }, + // A collection is retained until the DAG explicitly reduces it. Evaluating + // the whole DAG once per source pane would change nested reductions. + SummaryWindows { + states: Arc<[(i64, SummaryState)]>, + family: planner_types::post_asap::SummaryFamilyType, + }, + Rows { + values: Vec<(i64, f64)>, + name: String, + }, +} + +impl MaintenanceValue { + fn summary(state: SummaryState) -> Self { + Self::Summary { + state, + family: None, + } + } + + fn state(&self) -> Result<&SummaryState, String> { + match self { + Self::Summary { state, .. } => Ok(state), + Self::SummaryWindows { states, .. } if states.len() == 1 => Ok(&states[0].1), + Self::Rows { .. } | Self::SummaryWindows { .. } => { + Err("maintenance sink requires an explicit reduction to one summary state".into()) + } + } + } +} type PendingOutput = ( Option<(MaterializationCommitKey, u64)>, PrecomputedOutput, @@ -23,30 +60,132 @@ struct OperatorAdapter<'a> { binding: &'a BackendExecutableBinding, source_definition: asap_types::sds::SummaryDefinitionId, source: SummaryState, + configs: &'a [asap_types::aggregation_config::AggregationConfig], + immutable_windows: Option>, } -impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { +impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { type Error = String; - fn materialized_input(&self, node: &ExecutableDagNode) -> Result, String> { - Ok(matches!( + fn materialized_input( + &self, + node: &ExecutableDagNode, + ) -> Result, String> { + if !matches!( self.binding.node(node.id), Some(BackendNodeBinding::Materialization { summary_definition }) if *summary_definition == self.source_definition - ) - .then(|| Arc::clone(&self.source))) + ) { + return Ok(None); + } + let family = node.output_schema.fields.iter().find_map(|field| { + (!matches!( + field.dtype, + planner_types::post_asap::SummaryFamilyType::Plain(_) + )) + .then(|| field.dtype.clone()) + }); + if let Some(states) = &self.immutable_windows { + return Ok(Some(MaintenanceValue::SummaryWindows { + states: Arc::clone(states), + family: family.ok_or("immutable source lacks a summary schema")?, + })); + } + Ok(Some(MaintenanceValue::Summary { + state: Arc::clone(&self.source), + family, + })) } fn execute( &self, node: &ExecutableDagNode, - inputs: &[Arc], - ) -> Result { + inputs: &[Arc], + ) -> Result { match &node.payload { ExecutableOperatorPayload::SummaryMerge => merge_inputs(inputs), - ExecutableOperatorPayload::SummaryAgg { .. } => Err( - "maintenance SummaryAgg requires a typed update evaluator; merging input state does not execute its update expression".into(), - ), + ExecutableOperatorPayload::Value { + operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, + timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, + } => { + if self.immutable_windows.is_none() { + return Err( + "maintenance finalization requires immutable completed input windows" + .into(), + ); + } + finalize_exact(node, inputs) + } + ExecutableOperatorPayload::SummaryAgg { family, input, .. } => { + let [value] = inputs else { + return Err("maintenance SummaryAgg requires exactly one row input".into()); + }; + let MaintenanceValue::Rows { values, name } = value.as_ref() else { + return Err("maintenance SummaryAgg requires a typed update evaluator; finalize summary state before applying an update".into()); + }; + if self.immutable_windows.is_none() { + return Err( + "maintenance aggregation requires immutable completed input windows".into(), + ); + } + let target = match self.binding.node(node.id) { + Some(BackendNodeBinding::Materialization { summary_definition }) => { + summary_definition + } + _ => { + return Err( + "maintenance SummaryAgg lacks installed materialization binding".into(), + ) + } + }; + let config = self + .configs + .iter() + .find(|config| config.policy_fingerprint() == target.fingerprint()) + .ok_or("maintenance SummaryAgg lacks installed accumulator configuration")?; + let source_config = self + .configs + .iter() + .find(|config| { + config.policy_fingerprint() == self.source_definition.fingerprint() + }) + .ok_or("maintenance input lacks installed source configuration")?; + if config.grouping_labels != source_config.grouping_labels + || config.partitioning != source_config.partitioning + { + return Err( + "maintenance transform requires explicit target window/group routing" + .into(), + ); + } + if config.accumulator_spec().map_err(|e| e.to_string())?.family != *family { + return Err( + "maintenance SummaryAgg family differs from installed configuration".into(), + ); + } + if input.item.is_some() { + return Err( + "keyed maintenance updates require explicit row identity routing".into(), + ); + } + let mut updater = super::accumulator_factory::create_accumulator_updater(config); + if updater.is_keyed() { + return Err("keyed maintenance accumulator requires an item expression".into()); + } + for (timestamp_ms, value) in values { + let weight = evaluate_weight(&input.weight, *value, name)?; + updater.update_single(weight, *timestamp_ms); + } + let timestamp = values + .iter() + .map(|(timestamp, _)| *timestamp) + .max() + .ok_or("maintenance aggregation has no input rows")?; + Ok(MaintenanceValue::SummaryWindows { + states: vec![(timestamp, Arc::from(updater.into_accumulator()))].into(), + family: family.clone(), + }) + } payload => Err(format!( "maintenance operator {:?} has no summary-state implementation", payload.operator() @@ -55,21 +194,418 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { } } -fn merge_inputs(inputs: &[Arc]) -> Result { - let Some(first) = inputs.first() else { +fn evaluate_weight( + expression: &planner_types::post_asap::SummaryInputExpr, + value: f64, + name: &str, +) -> Result { + use planner_types::{post_asap::SummaryInputExpr, pre_asap::ColumnRef}; + let weight = match expression { + SummaryInputExpr::Constant(value) => *value, + SummaryInputExpr::Column(ColumnRef::SampleValue) => value, + SummaryInputExpr::Column(ColumnRef::Named(column)) if column == name => value, + _ => { + return Err("maintenance update does not resolve against the supplied typed row".into()) + } + }; + if !weight.is_finite() { + return Err("maintenance update weight is not finite".into()); + } + Ok(weight) +} + +fn finalize_exact( + node: &ExecutableDagNode, + inputs: &[Arc], +) -> Result { + use planner_types::post_asap::{ExactKind, SummaryFamilyType}; + let [input] = inputs else { + return Err("exact maintenance finalization requires one summary input".into()); + }; + let (states, family): (Vec<(i64, &SummaryState)>, _) = match input.as_ref() { + MaintenanceValue::Summary { + state, + family: Some(family), + } => { + // Single-state merge execution has no row timestamp. Immutable + // execution supplies SummaryWindows with the actual window ends. + (vec![(0, state)], family) + } + MaintenanceValue::SummaryWindows { states, family } => ( + states.iter().map(|(time, state)| (*time, state)).collect(), + family, + ), + _ => { + return Err("exact maintenance finalization requires a typed exact accumulator".into()) + } + }; + let SummaryFamilyType::ExactAggregate(kind, _) = family else { + return Err("exact maintenance finalization requires a typed exact accumulator".into()); + }; + let statistic = match kind { + ExactKind::Sum => asap_types::Statistic::Sum, + ExactKind::Count => asap_types::Statistic::Count, + _ => { + return Err( + "exact maintenance readout requires explicit operator/time semantics".into(), + ) + } + }; + let [field] = node.output_schema.fields.as_slice() else { + return Err("exact maintenance finalization requires one scalar output column".into()); + }; + if !matches!( + field.dtype, + SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Float64) + ) { + return Err( + "exact maintenance finalization currently requires a Float64 output column".into(), + ); + } + let values = states + .into_iter() + .map(|(timestamp, state)| { + let value = state + .query_statistic(statistic, &None, &std::collections::HashMap::new()) + .map_err(|error| error.to_string())?; + if !value.is_finite() { + return Err("exact maintenance finalization produced a non-finite value".into()); + } + Ok((timestamp, value)) + }) + .collect::, String>>()?; + Ok(MaintenanceValue::Rows { + values, + name: field.name.clone(), + }) +} + +fn merge_inputs(inputs: &[Arc]) -> Result { + let mut states = Vec::new(); + let mut family = None; + let mut end_timestamp = None; + for input in inputs { + let input_family = match input.as_ref() { + MaintenanceValue::Summary { state, family } => { + states.push(state); + family.as_ref() + } + MaintenanceValue::SummaryWindows { + states: windows, + family, + } => { + states.extend(windows.iter().map(|(_, state)| state)); + end_timestamp = end_timestamp + .into_iter() + .chain(windows.iter().map(|(time, _)| *time)) + .max(); + Some(family) + } + MaintenanceValue::Rows { .. } => return Err("summary merge cannot consume rows".into()), + }; + if let Some(input_family) = input_family { + if family.as_ref().is_some_and(|family| family != input_family) { + return Err("summary merge input families differ".into()); + } + family = Some(input_family.clone()); + } + } + let Some((first, rest)) = states.split_first() else { return Err("summary maintenance node has no input state".into()); }; - let mut merged: Box = (**first).clone_boxed_core(); - for input in &inputs[1..] { + let mut merged = first.clone_boxed_core(); + for state in rest { merged = merged - .merge_with(input.as_ref().as_ref()) - .map_err(|error| error.to_string())?; + .merge_with(state.as_ref()) + .map_err(|e| e.to_string())?; + } + if let Some(timestamp) = end_timestamp { + return Ok(MaintenanceValue::SummaryWindows { + states: vec![(timestamp, Arc::from(merged))].into(), + family: family.ok_or("merged immutable state lacks a family")?, + }); + } + Ok(MaintenanceValue::Summary { + state: Arc::from(merged), + family, + }) +} + +/// Evaluate one installed maintenance sink over an immutable physical source +/// incarnation. Durable publication is a separate existing-store transaction; +/// the in-memory scheduler cache here never claims durable exactly-once writes. +fn prepare_frozen_maintenance_sink( + installed: &asap_types::executable_plan::InstalledPostAsapDag, + configs: &[asap_types::PrecomputeMaterialization], + sink: PostAsapNodeId, + input: &crate::storage_engines::sketch_db::index::FrozenExactWindows, + output_window: (u64, u64), +) -> Result< + ( + planner_types::post_asap::ExecutableDag, + MaterializationCommitKey, + Vec<(i64, SummaryState)>, + ), + String, +> { + installed.validate()?; + let target = match installed.binding.node(sink) { + Some(BackendNodeBinding::Materialization { summary_definition }) => *summary_definition, + _ => return Err("immutable sink lacks a materialization binding".into()), + }; + let config = configs + .iter() + .find(|config| config.policy_fingerprint() == target.fingerprint()) + .ok_or("immutable sink lacks its installed configuration")?; + let expected_input = config + .derived_input + .as_ref() + .ok_or("immutable sink is not a derived materialization")?; + if expected_input.inputs != BTreeSet::from([input.definition]) { + return Err("immutable sink requires synchronized input definitions".into()); + } + if output_window.0 >= output_window.1 + || output_window.1 - output_window.0 != config.stored_window_ms() + || input + .windows + .keys() + .any(|(start, end)| *start < output_window.0 || *end > output_window.1) + { + return Err("immutable inputs do not fit the installed output window".into()); + } + let dag = installed.document.decode()?; + let producers: Vec<_> = dag + .edges + .iter() + .filter(|edge| edge.consumer == sink) + .collect(); + let [producer] = producers.as_slice() else { + return Err("immutable sink requires one input relation".into()); + }; + let frontiers = installed + .binding + .nodes + .iter() + .filter_map(|(node, binding)| match binding { + BackendNodeBinding::Materialization { summary_definition } + if *summary_definition == input.definition => + { + Some((*node, *summary_definition)) + } + _ => None, + }) + .collect(); + let actual_input = asap_types::derived_input::DerivedInputIdentity::from_dag( + &installed.document, + producer.producer, + &frontiers, + )?; + if &actual_input != expected_input { + return Err("immutable sink input program differs from its catalog identity".into()); + } + let mut lineage = Sha256::new(); + lineage.update(b"immutable-maintenance-input-v1"); + lineage.update(input.sid.to_be_bytes()); + lineage.update( + serde_json::to_vec(&( + &input.definition, + &input.generation, + &input.group, + expected_input, + )) + .map_err(|error| error.to_string())?, + ); + let mut states = Vec::with_capacity(input.windows.len()); + for ((start, end), state) in &input.windows { + lineage.update(start.to_be_bytes()); + lineage.update(end.to_be_bytes()); + let bytes = state.serialize_to_bytes(); + lineage.update((bytes.len() as u64).to_be_bytes()); + lineage.update(&bytes); + states.push((*end as i64, Arc::clone(state))); + } + let digest: [u8; 32] = lineage.finalize().into(); + let key = MaterializationCommitKey { + plan_id: input.generation.plan_id, + plan_version: input.generation.plan_version, + summary_definition: target, + window_start_ms: i64::try_from(output_window.0) + .map_err(|_| "output window exceeds timestamp range")?, + window_end_ms: i64::try_from(output_window.1) + .map_err(|_| "output window exceeds timestamp range")?, + input_lineage: digest.to_vec(), + }; + Ok((dag, key, states)) +} + +fn execute_prepared_frozen_sink( + installed: &asap_types::executable_plan::InstalledPostAsapDag, + configs: &[asap_types::PrecomputeMaterialization], + sink: PostAsapNodeId, + input: &crate::storage_engines::sketch_db::index::FrozenExactWindows, + dag: &planner_types::post_asap::ExecutableDag, + key: MaterializationCommitKey, + states: Vec<(i64, SummaryState)>, +) -> Result { + let first = states.first().ok_or("immutable input is empty")?; + let adapter = OperatorAdapter { + binding: &installed.binding, + source_definition: input.definition, + source: Arc::clone(&first.1), + configs, + immutable_windows: Some(states.into()), + }; + let value = execute_precompute_sink( + dag, + &installed.binding, + sink, + key, + &adapter, + &CommitRegistry::default(), + ) + .map_err(schedule_error)?; + Ok(Arc::clone(value.state()?)) +} + +#[cfg(test)] +fn evaluate_frozen_maintenance_sink( + installed: &asap_types::executable_plan::InstalledPostAsapDag, + configs: &[asap_types::PrecomputeMaterialization], + sink: PostAsapNodeId, + input: &crate::storage_engines::sketch_db::index::FrozenExactWindows, + output_window: (u64, u64), +) -> Result<(SummaryState, [u8; 32]), String> { + let (dag, key, states) = + prepare_frozen_maintenance_sink(installed, configs, sink, input, output_window)?; + let digest = key + .input_lineage + .as_slice() + .try_into() + .map_err(|_| "invalid input digest")?; + let state = execute_prepared_frozen_sink(installed, configs, sink, input, &dag, key, states)?; + Ok((state, digest)) +} + +/// Execute an installed, single-population maintenance subDAG from frozen +/// base panes and publish its complete output through the durable part path. +/// This initial entry point accepts non-overlapping output windows; sliding +/// replacement and cross-population shuffles require their own scheduling +/// proof and are rejected, rather than treating corrections as observations. +#[allow(clippy::too_many_arguments)] +pub fn execute_completed_maintenance( + store: &crate::storage_engines::sketch_db::index::SketchStore, + installed: &asap_types::executable_plan::InstalledPostAsapDag, + configs: &[asap_types::PrecomputeMaterialization], + sink: planner_types::post_asap::PostAsapNodeId, + source_sid: u64, + target_sid: u64, + window: (u64, u64), + group: &BTreeMap, +) -> Result { + use asap_types::executable_plan::BackendNodeBinding; + let target = match installed.binding.node(sink) { + Some(BackendNodeBinding::Materialization { summary_definition }) => *summary_definition, + _ => return Err("maintenance sink lacks an installed output identity".into()), + }; + let target_config = configs + .iter() + .find(|config| config.policy_fingerprint() == target.fingerprint()) + .ok_or("maintenance output configuration is absent")?; + let derived = target_config + .derived_input + .as_ref() + .ok_or("maintenance output has no derived input")?; + if derived.inputs.len() != 1 { + return Err("maintenance execution requires synchronized multi-source scheduling".into()); + } + let source = *derived.inputs.first().unwrap(); + let source_config = configs + .iter() + .find(|config| config.policy_fingerprint() == source.fingerprint()) + .ok_or("maintenance source configuration is absent")?; + let source_width = source_config.stored_window_ms(); + let target_width = target_config.stored_window_ms(); + let origin = source_config.pane_origin_ms.unwrap_or(0); + if source_width == 0 + || target_width == 0 + || window.0 >= window.1 + || window.1 - window.0 != target_width + || target_width % source_width != 0 + || target_width / source_width > 65_536 + || source_config + .slide_interval + .checked_mul(1000) + .is_none_or(|slide| slide < source_width) + || target_config + .slide_interval + .checked_mul(1000) + .is_none_or(|slide| slide < target_width) + || window.1 > i64::MAX as u64 + || (window.0 as i128 - origin as i128).rem_euclid(source_width as i128) != 0 + || (window.0 as i128 - target_config.pane_origin_ms.unwrap_or(0) as i128) + .rem_euclid(target_width as i128) + != 0 + { + return Err("maintenance window requires unsupported overlap, phase, or extent".into()); } - Ok(Arc::from(merged)) + let expected = (0..target_width / source_width) + .map(|index| { + let start = window.0 + index * source_width; + (start, start + source_width) + }) + .collect(); + let generation = store + .active_catalog_generation() + .ok_or("maintenance requires an authoritative catalog")?; + let frozen = + store.read_frozen_exact_windows(source_sid, source, &generation, &expected, group)?; + let (dag, key, states) = + prepare_frozen_maintenance_sink(installed, configs, sink, &frozen, window)?; + let digest = key + .input_lineage + .as_slice() + .try_into() + .map_err(|_| "invalid input digest")?; + if store.recover_frozen_maintenance_output( + target_sid, + target_config, + &frozen, + digest, + window, + )? { + return Ok(false); + } + let state = execute_prepared_frozen_sink(installed, configs, sink, &frozen, &dag, key, states)?; + let mut output = crate::storage_engines::types::PrecomputedOutput::new( + window.0, + window.1, + Some(crate::storage_engines::types::KeyByLabelValues { + labels: target_config + .grouping_labels + .iter() + .map(|name| { + group + .get(name) + .cloned() + .ok_or("maintenance population is missing an output grouping key") + }) + .collect::, _>>()?, + }), + target.fingerprint(), + ); + output.catalog_generation = Some(generation); + store.publish_frozen_maintenance_output( + target_sid, + target_config, + &output, + state.as_ref(), + &frozen, + digest, + ) } struct CommittedState { - value: Option>, + value: Option>, published: bool, } @@ -245,13 +781,13 @@ impl CommitRegistry { } } -impl IdempotentCommitSink for CommitRegistry { +impl IdempotentCommitSink for CommitRegistry { type Error = String; fn get( &self, key: &MaterializationCommitKey, - ) -> Result>, Self::Error> { + ) -> Result>, Self::Error> { let state = self.0.lock().map_err(|_| "commit registry poisoned")?; state.validate_key(key)?; Ok(state @@ -263,8 +799,8 @@ impl IdempotentCommitSink for CommitRegistry { fn commit_if_absent( &self, key: MaterializationCommitKey, - value: Arc, - ) -> Result, Self::Error> { + value: Arc, + ) -> Result, Self::Error> { let mut commits = self.0.lock().map_err(|_| "commit registry poisoned")?; commits.validate_key(&key)?; let committed = commits @@ -347,6 +883,8 @@ impl MaintenanceDagSink { binding: &installed.binding, source_definition, source: Arc::clone(&source), + configs: &plan.precompute_plan.materializations, + immutable_windows: None, }; for sink_node in &installed.binding.precompute_sinks { if !depends_on_any(&dag, *sink_node, &source_nodes) { @@ -435,7 +973,7 @@ impl MaintenanceDagSink { derived.push(( Some((key, horizon_ms)), target_output, - value.as_ref().as_ref().clone_boxed_core(), + value.state()?.clone_boxed_core(), )); } } @@ -644,6 +1182,479 @@ mod tests { Arc::new(accumulator) } + #[test] + fn finalized_summary_update_builds_a_different_installed_family() { + use planner_types::post_asap::{ + ExactKind, ExactParams, GroupingStrategy, SummaryFamilyType, SummaryField, + SummaryInputExpr, SummaryUpdate, + }; + use planner_types::pre_asap::{DataType, Reduction}; + let mut snapshot: serde_json::Value = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + snapshot["query_workload"]["repeating_queries"][0]["query"] = + "sum(sum_over_time(m[1m]))".into(); + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_value(snapshot).unwrap(); + let bundle = snapshot.compile().unwrap(); + let mut source_config = bundle.precompute_plan.materializations[0].clone(); + source_config.window_size = 2; + source_config.slide_interval = 2; + source_config.window_layout = + asap_types::aggregation_config::WindowMaterializationLayout::Pane { pane_secs: 1 }; + let source_definition = source_config.policy_fingerprint().into(); + let mut target_config = source_config.clone(); + target_config.window_layout = + asap_types::aggregation_config::WindowMaterializationLayout::FullWindow; + target_config.aggregation_type = asap_types::AggregationType::DatasketchesKLL; + target_config.aggregation_sub_type = "quantile".into(); + target_config + .parameters + .insert("k".into(), serde_json::json!(200)); + let target = target_config.policy_fingerprint().into(); + let target_family = target_config.accumulator_spec().unwrap().family; + let configs = [source_config, target_config]; + let binding = BackendExecutableBinding { + nodes: BTreeMap::from([ + ( + PostAsapNodeId(1), + BackendNodeBinding::Materialization { + summary_definition: source_definition, + }, + ), + (PostAsapNodeId(2), BackendNodeBinding::MaintenanceInput), + ( + PostAsapNodeId(3), + BackendNodeBinding::Materialization { + summary_definition: target, + }, + ), + ]), + query_sink: PostAsapNodeId(3), + query_plan_sink: control_plane::query_plan::QueryNodeId(3), + precompute_sinks: vec![PostAsapNodeId(3)], + }; + let adapter = OperatorAdapter { + binding: &binding, + source_definition, + source: sum(7.0), + configs: &configs, + immutable_windows: Some(vec![(1000, sum(7.0))].into()), + }; + let mut read = node(2); + read.payload = ExecutableOperatorPayload::Value { + operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, + timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, + }; + read.output_schema.fields = vec![SummaryField { + name: "value".into(), + dtype: SummaryFamilyType::Plain(DataType::Float64), + nullable: false, + }]; + let source = Arc::new(MaintenanceValue::Summary { + state: sum(7.0), + family: Some(SummaryFamilyType::ExactAggregate( + ExactKind::Sum, + ExactParams::Sum, + )), + }); + let row = adapter.execute(&read, &[source]).unwrap(); + let mut aggregate = node(3); + aggregate.payload = ExecutableOperatorPayload::SummaryAgg { + family: target_family, + input: SummaryUpdate { + item: None, + weight: SummaryInputExpr::Constant(3.0), + weight_domain: Default::default(), + }, + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + }; + let result = adapter.execute(&aggregate, &[Arc::new(row)]).unwrap(); + let mut kwargs = std::collections::HashMap::new(); + kwargs.insert("quantile".into(), "0.5".into()); + assert_eq!( + result + .state() + .unwrap() + .query_statistic(asap_types::Statistic::Quantile, &None, &kwargs) + .unwrap(), + 3.0 + ); + // Exercise the same registry through the production topological + // scheduler, including the precomputed source frontier and commit. + let mut source_node = node(1); + source_node.output_schema.fields = vec![SummaryField { + name: "state".into(), + dtype: SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + nullable: false, + }]; + read.operator = read.payload.operator(); + read.output_state = planner_types::post_asap::ExecutionDataState::MAINTENANCE_ROWS; + aggregate.operator = aggregate.payload.operator(); + aggregate.output_schema.fields = vec![SummaryField { + name: "state".into(), + dtype: configs[1].accumulator_spec().unwrap().family, + nullable: false, + }]; + let mut query = node(4); + query.output_state = planner_types::post_asap::ExecutionDataState::READ_ROWS; + let mut first_edge = edge(1, 2); + first_edge.intermediate_schema = source_node.output_schema.clone(); + let mut second_edge = edge(2, 3); + second_edge.intermediate_schema = read.output_schema.clone(); + second_edge.data_state = read.output_state; + let mut query_edge = edge(3, 4); + query_edge.intermediate_schema = aggregate.output_schema.clone(); + let dag = ExecutableDag { + nodes: vec![source_node, read, aggregate, query], + edges: vec![first_edge, second_edge, query_edge], + root: PostAsapNodeId(4), + }; + let mut scheduled_binding = binding.clone(); + scheduled_binding.nodes.insert( + PostAsapNodeId(1), + BackendNodeBinding::Materialization { + summary_definition: source_definition, + }, + ); + scheduled_binding + .nodes + .insert(PostAsapNodeId(2), BackendNodeBinding::MaintenanceInput); + scheduled_binding.nodes.insert( + PostAsapNodeId(4), + BackendNodeBinding::Query { + query_node: control_plane::query_plan::QueryNodeId(4), + }, + ); + scheduled_binding.query_sink = PostAsapNodeId(4); + scheduled_binding.query_plan_sink = control_plane::query_plan::QueryNodeId(4); + let scheduled_adapter = OperatorAdapter { + binding: &scheduled_binding, + ..adapter + }; + let key = MaterializationCommitKey { + plan_id: 1, + plan_version: 1, + summary_definition: target, + window_start_ms: 0, + window_end_ms: 1000, + input_lineage: vec![1], + }; + let committed = execute_precompute_sink( + &dag, + &scheduled_binding, + PostAsapNodeId(3), + key, + &scheduled_adapter, + &CommitRegistry::default(), + ) + .unwrap(); + assert_eq!( + committed + .state() + .unwrap() + .query_statistic(asap_types::Statistic::Quantile, &None, &kwargs) + .unwrap(), + 3.0 + ); + // The real store provides the completion proof. Execute, persist, + // restart, and retry the same installed subDAG without additive append. + use crate::storage_engines::sketch_db::index::{ + persistence::config::SketchStorePersistenceConfig, SketchStore, + }; + use asap_types::executable_plan::{InstalledPostAsapDag, OwnedPostAsapDag}; + let document = OwnedPostAsapDag::from_executable("immutable-chain".into(), &dag).unwrap(); + let mut durable_configs = configs.to_vec(); + durable_configs[1].derived_input = Some( + asap_types::derived_input::DerivedInputIdentity::from_dag( + &document, + PostAsapNodeId(2), + &BTreeMap::from([(PostAsapNodeId(1), source_definition)]), + ) + .unwrap(), + ); + let mut durable_binding = scheduled_binding.clone(); + durable_binding.nodes.insert( + PostAsapNodeId(3), + BackendNodeBinding::Materialization { + summary_definition: durable_configs[1].policy_fingerprint().into(), + }, + ); + let installed = InstalledPostAsapDag { + document, + binding: durable_binding, + }; + let catalog = Arc::new( + asap_types::summary_catalog::SummaryCatalog::from_materializations( + 1, + 1, + &durable_configs, + ) + .unwrap(), + ); + let directory = tempfile::tempdir().unwrap(); + let persistence_config = || { + let mut config = SketchStorePersistenceConfig::with_memory_limit( + 1 << 24, + directory.path().to_path_buf(), + ); + config.delete_older_than_ms = None; + config.hot_window_ms = None; + config.flush_interval = std::time::Duration::from_millis(5); + config + }; + let expected = BTreeSet::from([(0, 1000), (1000, 2000)]); + let mut store = Arc::new(SketchStore::new()); + store.install_summary_catalog(Arc::clone(&catalog)).unwrap(); + let mut persistence = store.start_persistence(persistence_config()).unwrap(); + let generation = store.active_catalog_generation().unwrap(); + for ((start, end), value) in [((0, 1000), 2.0), ((1000, 2000), 7.0)] { + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: source_definition, + time_range: asap_types::sds::HalfOpenTimeRange { + start_ms: start, + end_ms: end, + }, + group_values: BTreeMap::new(), + }; + let revision = store + .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()])) + .unwrap(); + let mut output = PrecomputedOutput::new( + start as u64, + end as u64, + None, + durable_configs[0].policy_fingerprint(), + ); + output.catalog_generation = Some(Arc::clone(&generation)); + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 120_000, + || { + store.ingest_precompute_with_series_id( + 600, + &durable_configs[0], + &output, + sum(value).as_ref(), + ) + }, + ) + .unwrap(); + } + assert!(store + .read_frozen_exact_windows( + 600, + source_definition, + &generation, + &expected, + &BTreeMap::new() + ) + .is_err()); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !store.seal_finite_summary_input(&generation).unwrap() { + assert!(std::time::Instant::now() < deadline); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + let frozen = store + .read_frozen_exact_windows( + 600, + source_definition, + &generation, + &expected, + &BTreeMap::new(), + ) + .unwrap(); + let (result, digest) = evaluate_frozen_maintenance_sink( + &installed, + &durable_configs, + PostAsapNodeId(3), + &frozen, + (0, 2000), + ) + .unwrap(); + let mut output = + PrecomputedOutput::new(0, 2000, None, durable_configs[1].policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + let log = persistence.manifest.log_path(); + let backup = log.with_extension("saved"); + std::fs::rename(&log, &backup).unwrap(); + std::fs::create_dir(&log).unwrap(); + assert!(execute_completed_maintenance( + &store, + &installed, + &durable_configs, + PostAsapNodeId(3), + 600, + 601, + (0, 2000), + &BTreeMap::new() + ) + .is_err()); + std::fs::remove_dir(&log).unwrap(); + std::fs::rename(&backup, &log).unwrap(); + persistence.shutdown(); + drop(store); + store = Arc::new(SketchStore::new()); + store.install_summary_catalog(Arc::clone(&catalog)).unwrap(); + persistence = store.start_persistence(persistence_config()).unwrap(); + // The already durable pending KLL part is completed before a new + // randomized sketch can be built after restart. + assert!(!execute_completed_maintenance( + &store, + &installed, + &durable_configs, + PostAsapNodeId(3), + 600, + 601, + (0, 2000), + &BTreeMap::new(), + ) + .unwrap()); + assert!(!execute_completed_maintenance( + &store, + &installed, + &durable_configs, + PostAsapNodeId(3), + 600, + 601, + (0, 2000), + &BTreeMap::new() + ) + .unwrap()); + assert_eq!( + result + .query_statistic(asap_types::Statistic::Quantile, &None, &kwargs) + .unwrap(), + 3.0 + ); + let mut correction = + PrecomputedOutput::new(0, 1000, None, durable_configs[0].policy_fingerprint()); + correction.catalog_generation = Some(Arc::clone(&generation)); + assert!(store + .ingest_precompute_with_series_id( + 600, + &durable_configs[0], + &correction, + sum(100.0).as_ref() + ) + .is_none()); + persistence.shutdown(); + drop(store); + let restored = Arc::new(SketchStore::new()); + restored.install_summary_catalog(catalog).unwrap(); + let mut persistence = restored.start_persistence(persistence_config()).unwrap(); + let frozen = restored + .read_frozen_exact_windows( + 600, + source_definition, + &generation, + &expected, + &BTreeMap::new(), + ) + .unwrap(); + let (_result, replay_digest) = evaluate_frozen_maintenance_sink( + &installed, + &durable_configs, + PostAsapNodeId(3), + &frozen, + (0, 2000), + ) + .unwrap(); + assert_eq!(digest, replay_digest); + assert!(!execute_completed_maintenance( + &restored, + &installed, + &durable_configs, + PostAsapNodeId(3), + 600, + 601, + (0, 2000), + &BTreeMap::new() + ) + .unwrap()); + assert!(restored + .ingest_precompute_with_series_id( + 600, + &durable_configs[0], + &correction, + sum(100.0).as_ref() + ) + .is_none()); + let target_entries = persistence + .manifest + .live_parts() + .iter() + .map(|part| { + let path = crate::storage_engines::sketch_db::persistence::part::part_dir_path( + &persistence.parts_root, + part.part_id, + ); + crate::storage_engines::sketch_db::persistence::part::PartReader::open(&path) + .unwrap() + .index_records() + .into_iter() + .filter(|entry| entry.agg_id == 601) + .count() + }) + .sum::(); + assert_eq!(target_entries, 1); + persistence.shutdown(); + assert!(evaluate_weight( + &SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::Named("missing".into())), + 7.0, + "value" + ) + .is_err()); + } + + #[test] + fn finalization_preserves_windows_until_an_explicit_merge() { + use planner_types::post_asap::{ExactKind, ExactParams, SummaryFamilyType, SummaryField}; + let family = SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum); + let inputs = Arc::new(MaintenanceValue::SummaryWindows { + states: vec![(1_000, sum(2.0)), (2_000, sum(7.0))].into(), + family, + }); + let mut read = node(2); + read.output_schema.fields = vec![SummaryField { + name: "value".into(), + dtype: SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Float64), + nullable: false, + }]; + let MaintenanceValue::Rows { values, .. } = + finalize_exact(&read, &[inputs.clone()]).unwrap() + else { + panic!("expected finalized rows") + }; + assert_eq!(values, vec![(1_000, 2.0), (2_000, 7.0)]); + // Merge is a semantic DAG operation, not an implicit batch optimization. + // Finalizing after it emits exactly one value instead of two updates. + let merged = Arc::new(merge_inputs(&[inputs]).unwrap()); + let MaintenanceValue::Rows { values, .. } = finalize_exact(&read, &[merged]).unwrap() + else { + panic!("expected finalized row") + }; + assert_eq!(values, vec![(2_000, 9.0)]); + read.output_schema.fields[0].dtype = + SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Int64); + let integer_state = Arc::new(MaintenanceValue::Summary { + state: sum(9.0), + family: Some(SummaryFamilyType::ExactAggregate( + ExactKind::Sum, + ExactParams::Sum, + )), + }); + assert!( + matches!(finalize_exact(&read, &[integer_state]), Err(error) if error.contains("Float64")) + ); + } + #[test] fn summary_aggregation_does_not_silently_reuse_input_family() { use planner_types::post_asap::{ @@ -661,6 +1672,8 @@ mod tests { binding: &binding, source_definition: definition(1), source: sum(7.0), + configs: &[], + immutable_windows: None, }; let mut aggregate = node(1); aggregate.operator = ExecutableOperator::SummaryAgg; @@ -670,7 +1683,7 @@ mod tests { reduction: Reduction::by(vec![]), grouping: GroupingStrategy::default(), }; - let error = adapter.execute(&aggregate, &[Arc::new(sum(7.0))]); + let error = adapter.execute(&aggregate, &[Arc::new(MaintenanceValue::summary(sum(7.0)))]); assert!(matches!(error, Err(reason) if reason.contains("typed update evaluator"))); } @@ -689,7 +1702,7 @@ mod tests { let fast = key(1000); commits.begin_batch([1; 32]).unwrap(); commits - .commit_if_absent(fast.clone(), Arc::new(sum(2.0))) + .commit_if_absent(fast.clone(), Arc::new(MaintenanceValue::summary(sum(2.0)))) .unwrap(); commits.publish(&fast, || Ok(())).unwrap(); commits.complete_batch([1; 32], &[(fast, 30)]).unwrap(); @@ -698,7 +1711,7 @@ mod tests { commits.begin_batch([2; 32]).unwrap(); commits.pin_admitted(&slow).unwrap(); commits - .commit_if_absent(slow.clone(), Arc::new(sum(3.0))) + .commit_if_absent(slow.clone(), Arc::new(MaintenanceValue::summary(sum(3.0)))) .unwrap(); commits.publish(&slow, || Ok(())).unwrap(); commits @@ -725,7 +1738,7 @@ mod tests { let key = key(end); commits.begin_batch([0; 32]).unwrap(); commits - .commit_if_absent(key.clone(), Arc::new(sum(2.0))) + .commit_if_absent(key.clone(), Arc::new(MaintenanceValue::summary(sum(2.0)))) .unwrap(); commits.publish(&key, || Ok(())).unwrap(); commits.complete_batch([0; 32], &[(key, 30)]).unwrap(); @@ -742,7 +1755,7 @@ mod tests { assert!(commits.is_published(&key(980)).unwrap()); commits.0.lock().unwrap().generation = Some((7, 2)); assert!(commits - .commit_if_absent(key(1_000), Arc::new(sum(2.0))) + .commit_if_absent(key(1_000), Arc::new(MaintenanceValue::summary(sum(2.0)))) .is_err()); } @@ -762,7 +1775,7 @@ mod tests { }; assert!(!commits.is_published(&key).unwrap()); commits - .commit_if_absent(key.clone(), Arc::new(sum(2.0))) + .commit_if_absent(key.clone(), Arc::new(MaintenanceValue::summary(sum(2.0)))) .unwrap(); commits.publish(&key, || Ok(())).unwrap(); } @@ -983,6 +1996,8 @@ mod tests { binding: &binding, source_definition: definition(1), source, + configs: &[], + immutable_windows: None, }; let commits = CommitRegistry::default(); let key = MaterializationCommitKey { @@ -1002,7 +2017,7 @@ mod tests { &commits, ) .unwrap(); - assert_eq!(result.as_ref().aux_stats().sum, Some(4.0)); + assert_eq!(result.state().unwrap().aux_stats().sum, Some(4.0)); commits.publish(&key, || Ok(())).unwrap(); assert!( commits.get(&key).unwrap().is_none(), @@ -1055,6 +2070,8 @@ mod tests { binding: &binding, source_definition: definition(1), source: sum(2.0), + configs: &[], + immutable_windows: None, }; let commits = CommitRegistry::default(); let key = MaterializationCommitKey { diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs new file mode 100644 index 00000000..f8a3403d --- /dev/null +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -0,0 +1,389 @@ +//! Immutable maintenance inputs read from the existing durable summary tier. +//! +//! Unlike query fallback helpers, this path must propagate corrupt/missing part +//! errors: silently omitting a source window would permanently corrupt an outer +//! summary. Completion is an admission barrier, not merely an emitted flag. +use super::*; +use crate::storage_engines::types::AggregateCore; + +pub(crate) struct FrozenExactWindows { + pub(crate) sid: u64, + pub(crate) definition: SummaryDefinitionId, + pub(crate) generation: Arc, + pub(crate) group: BTreeMap, + pub(crate) windows: BTreeMap<(u64, u64), Arc>, +} + +impl SketchStore { + pub(crate) fn read_frozen_exact_windows( + &self, + sid: u64, + definition: SummaryDefinitionId, + generation: &Arc, + expected_windows: &BTreeSet<(u64, u64)>, + group: &BTreeMap, + ) -> Result { + let start_ms = expected_windows + .iter() + .map(|window| window.0) + .min() + .ok_or("immutable input has no requested windows")?; + let end_ms = expected_windows + .iter() + .map(|window| window.1) + .max() + .unwrap(); + if expected_windows.iter().any(|(start, end)| start >= end) || end_ms > i64::MAX as u64 { + return Err("invalid immutable input window".into()); + } + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + // Do not retain either lock while opening parts. The immutable frontier + // is monotone, and final publication rechecks the catalog incarnation. + let keys = { + let bindings = self + .instances + .read() + .map_err(|_| "instance registry poisoned")?; + let binding = bindings.get(&sid).ok_or("immutable input SID is absent")?; + if binding.metadata.policy_fp != definition.fingerprint() + || binding.catalog_generation.as_deref() != Some(generation.as_ref()) + || binding.metadata.status() == AggStatus::Expired + { + return Err("immutable input identity or lifetime differs".into()); + } + binding.metadata.group_by_keys.clone() + }; + if group.keys().cloned().collect::>() != keys { + return Err("immutable input population does not match its descriptor".into()); + } + let keys: Vec<_> = keys.into_iter().collect(); + if self + .completed_windows + .read() + .map_err(|_| "completion registry poisoned")? + .get(&sid) + .is_none_or(|frontier| *frontier < end_ms) + { + return Err("immutable input window is not complete".into()); + } + let handle = self + .persistence_read + .read() + .map_err(|_| "persistence registry poisoned")? + .clone() + .ok_or("immutable maintenance requires durable input state")?; + let mut windows = BTreeMap::new(); + for part in handle.manifest.live_parts_overlapping(start_ms, end_ms) { + let reader = handle + .part_cache + .get_or_load(part.part_id) + .map_err(|e| e.to_string())?; + for record in reader.index_records() { + if record.agg_id != sid || record.start_ts < start_ms || record.end_ts > end_ms { + continue; + } + let entry = reader.load_entry(&record).map_err(|e| e.to_string())?; + if entry.label.as_ref().map_or(0, |label| label.labels.len()) != keys.len() { + return Err("immutable input label arity differs from its descriptor".into()); + } + if Self::rebuild_label_map(&keys, &entry.label) != *group { + continue; + } + let state = reconstruct_exact_agg(&entry.sketch_type_name, &entry.sketch_bytes) + .ok_or("immutable input accumulator cannot be decoded")?; + if windows + .insert((record.start_ts, record.end_ts), Arc::from(state)) + .is_some() + { + // No last-writer-wins or implicit merge at a frozen source + // boundary: the producer must publish one complete state. + return Err("immutable input has multiple states for one window".into()); + } + } + } + if windows.keys().copied().collect::>() != *expected_windows { + return Err("immutable input window coverage is missing, expired, or ambiguous".into()); + } + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + Ok(FrozenExactWindows { + sid, + definition, + generation: Arc::clone(generation), + group: group.clone(), + windows, + }) + } +} + +impl SketchStore { + pub(crate) fn recover_frozen_maintenance_output( + &self, + sid: u64, + config: &asap_types::PrecomputeMaterialization, + source: &FrozenExactWindows, + digest: [u8; 32], + window: (u64, u64), + ) -> Result { + self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?; + let instances = self + .instances + .read() + .map_err(|_| "instance registry poisoned")?; + let Some(binding) = instances.get(&sid) else { + return Ok(false); + }; + if binding.metadata.policy_fp != config.policy_fingerprint() + || binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) + || !binding.metadata.is_writable() + { + return Err("immutable output identity or lifetime changed".into()); + } + let record = self + .metadata_record_without_completion(binding) + .ok_or("immutable output has no catalog metadata")?; + let publisher = self + .immutable_publisher + .read() + .map_err(|_| "publisher registry poisoned")? + .upgrade() + .ok_or("immutable publication requires active persistence")?; + let _mutation = self.begin_state_mutation(); + let mut completed = self + .completed_windows + .write() + .map_err(|_| "completion registry poisoned")?; + if publisher + .resume_matching_immutable_window(&record, digest, window.0, window.1) + .map_err(|error| error.to_string())? + .is_some() + { + completed + .entry(sid) + .and_modify(|end| *end = (*end).max(window.1)) + .or_insert(window.1); + return Ok(true); + } + let found = publisher + .lookup_immutable_window(&record, digest, window.0, window.1) + .map_err(|error| error.to_string())? + .is_some(); + if found { + completed + .entry(sid) + .and_modify(|end| *end = (*end).max(window.1)) + .or_insert(window.1); + } + Ok(found) + } + + /// Publish a derived result without passing it through additive hot-state + /// append. The existing part reservation commits it once and seals it. + pub(crate) fn publish_frozen_maintenance_output( + &self, + sid: u64, + config: &asap_types::PrecomputeMaterialization, + output: &crate::storage_engines::types::PrecomputedOutput, + state: &dyn AggregateCore, + source: &FrozenExactWindows, + input_digest: [u8; 32], + ) -> Result { + use persistence::source::{EpochSnapshot, EpochSnapshotEntry}; + if config + .derived_input + .as_ref() + .is_none_or(|derived| derived.inputs != BTreeSet::from([source.definition])) + || output.policy_fp != config.policy_fingerprint() + || output.catalog_generation.as_deref() != Some(source.generation.as_ref()) + { + return Err("derived publication differs from its installed input identity".into()); + } + let labels = self + .register_precompute_output(sid, config, output) + .ok_or("derived output registration failed")?; + let _mutation = self.begin_state_mutation(); + let instances = self + .instances + .read() + .map_err(|_| "instance registry poisoned")?; + let source_binding = instances + .get(&source.sid) + .ok_or("immutable source was removed")?; + let binding = instances.get(&sid).ok_or("derived output was removed")?; + if source_binding.metadata.policy_fp != source.definition.fingerprint() + || source_binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) + || source_binding.metadata.status() == AggStatus::Expired + || !binding.metadata.is_writable() + || binding.metadata.policy_fp != output.policy_fp + || binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) + { + return Err("derived publication physical lifetime changed".into()); + } + self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?; + let mut completed = self + .completed_windows + .write() + .map_err(|_| "completion registry poisoned")?; + let record = self + .metadata_record_without_completion(binding) + .ok_or("derived publication lacks catalog metadata")?; + let publisher = self + .immutable_publisher + .read() + .map_err(|_| "publisher registry poisoned")? + .upgrade() + .ok_or("immutable publication requires active persistence")?; + let bytes = state.serialize_to_bytes(); + let snapshot = EpochSnapshot { + agg_id: sid, + epoch_id: 0, + min_ts: output.start_timestamp, + max_ts: output.end_timestamp, + approx_bytes: bytes.len(), + entries: vec![EpochSnapshotEntry { + start_ts: output.start_timestamp, + end_ts: output.end_timestamp, + label: Some(crate::storage_engines::types::KeyByLabelValues { + labels: labels.into_values().collect(), + }), + sketch_type_name: state.type_name().to_string(), + encoding_tag: 0, + sketch_bytes: bytes, + }], + }; + // Another executor may have completed the same input after our first + // lookup. Reuse its durable payload instead of comparing a newly built, + // potentially randomized sketch byte representation. + if publisher + .lookup_immutable_window( + &record, + input_digest, + output.start_timestamp, + output.end_timestamp, + ) + .map_err(|error| error.to_string())? + .is_some() + { + completed + .entry(sid) + .and_modify(|end| *end = (*end).max(output.end_timestamp)) + .or_insert(output.end_timestamp); + return Ok(false); + } + let published = publisher + .publish_immutable_window(&record, input_digest, &snapshot) + .map_err(|error| error.to_string())?; + completed + .entry(sid) + .and_modify(|end| *end = (*end).max(output.end_timestamp)) + .or_insert(output.end_timestamp); + if !published.already_published { + crate::precompute_engine::metrics::record_materialized_outputs(1); + } + Ok(!published.already_published) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::precompute_engine::operators::SumAccumulator; + use crate::storage_engines::types::PrecomputedOutput; + use asap_types::traits::SerializableToSink; + + #[test] + fn committed_recovery_restores_live_seal_and_rejects_additive_derived_writes() { + // A durable sidecar may survive an I/O error before the caller updates + // its live seal. Recovery must repair admission, not only return a hit. + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_str(include_str!( + "../../../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + let plan = snapshot.compile().unwrap(); + let mut source_config = plan.precompute_plan.materializations[0].clone(); + source_config.aggregation_type = asap_types::AggregationType::Sum; + source_config.aggregation_sub_type = "sum".into(); + let source_id = source_config.policy_fingerprint().into(); + let mut target = source_config.clone(); + target.derived_input = Some(asap_types::derived_input::DerivedInputIdentity { + inputs: BTreeSet::from([source_id]), + program_sha256: "0".repeat(64), + }); + let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations( + 1, + 1, + &[source_config, target.clone()], + ) + .unwrap(); + let store = Arc::new(SketchStore::new()); + store.install_summary_catalog(Arc::new(catalog)).unwrap(); + let directory = tempfile::tempdir().unwrap(); + let mut config = persistence::config::SketchStorePersistenceConfig::with_memory_limit( + 1 << 24, + directory.path().to_path_buf(), + ); + config.delete_older_than_ms = None; + config.hot_window_ms = None; + let mut persistence = store.start_persistence(config).unwrap(); + let generation = store.active_catalog_generation().unwrap(); + let mut output = PrecomputedOutput::new(0, 1000, None, target.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + store + .register_precompute_output(601, &target, &output) + .unwrap(); + let record = store + .metadata_record(&store.instances.read().unwrap()[&601]) + .unwrap(); + let state = SumAccumulator::new(); + let snapshot = persistence::source::EpochSnapshot { + agg_id: 601, + epoch_id: 0, + min_ts: 0, + max_ts: 1000, + approx_bytes: 0, + entries: vec![persistence::source::EpochSnapshotEntry { + start_ts: 0, + end_ts: 1000, + label: None, + sketch_type_name: state.type_name().into(), + encoding_tag: 0, + sketch_bytes: state.serialize_to_bytes(), + }], + }; + persistence + .flusher + .publish_immutable_window(&record, [7; 32], &snapshot) + .unwrap(); + assert!(!store.completed_windows.read().unwrap().contains_key(&601)); + let input = FrozenExactWindows { + sid: 600, + definition: source_id, + generation, + group: BTreeMap::new(), + windows: BTreeMap::new(), + }; + assert!(store + .recover_frozen_maintenance_output(601, &target, &input, [7; 32], (0, 1000)) + .unwrap()); + assert_eq!( + store.completed_windows.read().unwrap().get(&601), + Some(&1000) + ); + assert!(!store.append_precompute( + 601, + BTreeMap::new(), + (2000, 3000), + Box::new(SumAccumulator::new()) + )); + assert!(!store.append_sample( + 601, + BTreeMap::new(), + (2000, 3000), + SketchSampleState { + bytes: vec![], + encoding: SketchEncoding::MsgpackFull, + } + )); + persistence.shutdown(); + } +} diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 1066e9f7..20f0cfc2 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -693,6 +693,7 @@ pub struct SketchStore { /// flush-then-evict loop is the memory bound). persistence_read: RwLock>>, persistence_metadata: RwLock>>, + immutable_publisher: RwLock>, removed_sids: RwLock< BTreeMap< u64, @@ -1264,6 +1265,27 @@ impl SketchStore { series_label_values: BTreeMap, window: TimestampRange, sample: SketchSampleState, + ) -> bool { + let instances = self.instances.read().unwrap(); + if instances.get(&sid).is_some_and(|binding| { + matches!( + binding.data_descriptor.source, + asap_types::sds::DataSourceIdentity::Derived { .. } + ) + }) { + return false; + } + self.append_sample_with_binding(sid, series_label_values, window, sample) + } + + // Caller retains the existing metadata guard and has rejected derived + // definitions. Avoid recursively acquiring it when a writer is waiting. + fn append_sample_with_binding( + &self, + sid: u64, + series_label_values: BTreeMap, + window: TimestampRange, + sample: SketchSampleState, ) -> bool { let completed = self.completed_windows.read().unwrap(); if completed.get(&sid).is_some_and(|end| window.1 <= *end) { @@ -1312,6 +1334,27 @@ impl SketchStore { series_label_values: BTreeMap, window: TimestampRange, payload: Box, + ) -> bool { + let instances = self.instances.read().unwrap(); + if instances.get(&sid).is_some_and(|binding| { + matches!( + binding.data_descriptor.source, + asap_types::sds::DataSourceIdentity::Derived { .. } + ) + }) { + return false; + } + self.append_precompute_with_binding(sid, series_label_values, window, payload) + } + + // Caller retains the existing metadata guard and has rejected derived + // definitions. Avoid recursively acquiring it when a writer is waiting. + fn append_precompute_with_binding( + &self, + sid: u64, + series_label_values: BTreeMap, + window: TimestampRange, + payload: Box, ) -> bool { let completed = self.completed_windows.read().unwrap(); if completed.get(&sid).is_some_and(|end| window.1 <= *end) { @@ -2472,6 +2515,15 @@ impl SketchStore { } fn metadata_record(&self, m: &SdsBinding) -> Option { + let mut record = self.metadata_record_without_completion(m)?; + record.completed_through_ms = self.completed_windows.read().unwrap().get(&m.sid).copied(); + Some(record) + } + + fn metadata_record_without_completion( + &self, + m: &SdsBinding, + ) -> Option { let mut record = crate::storage_engines::sketch_db::index::persistence::metadata::SidMetaRecord::new( m.sid, @@ -2484,7 +2536,6 @@ impl SketchStore { record.summary_definition_id = Some(SummaryDefinitionId::from(m.policy_fp)); record.catalog_generation = Some(Arc::clone(m.catalog_generation.as_ref()?)); } - record.completed_through_ms = self.completed_windows.read().unwrap().get(&m.sid).copied(); record.retired_at_ms = m.retired_at_ms; record.expires_at_ms = m.expires_at_ms; Some(record) @@ -2752,27 +2803,12 @@ impl SketchStore { self.ingest_precompute_with_series_id(sid, agg_cfg, output, accumulator) } - /// B7.7 sid-direct sibling of [`Self::ingest_precompute_for_agg_config`]. - /// - /// Callers that already hold the bucket sid (the live worker after - /// B7.6 reshaped its `WorkerMessage`, and the backfill processor - /// after B7.7 rekeyed its per-window grouping from group_key to - /// sid) skip the mint round-trip by handing the sid in directly. - /// The mint-driven [`Self::ingest_precompute_for_agg_config`] is - /// content-addressed and idempotent with this method — passing the - /// resolver-minted sid here yields the same state under the same - /// sid — so both methods can coexist while migration finishes. - /// - /// The §6.3 ingest barrier (`Retired` / `Expired` sids reject - /// writes) and first-sight metadata registration are identical to - /// the mint-driven path. - pub fn ingest_precompute_with_series_id( + fn register_precompute_output( &self, sid: u64, - agg_cfg: &asap_types::aggregation_config::AggregationConfig, + agg_cfg: &asap_types::PrecomputeMaterialization, output: &crate::storage_engines::types::PrecomputedOutput, - accumulator: &dyn crate::storage_engines::types::AggregateCore, - ) -> Option { + ) -> Option> { let (_attrs_fp, label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); let key_names = &agg_cfg.grouping_labels.names(); let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg); @@ -2819,12 +2855,44 @@ impl SketchStore { Some(_) => {} } + Some(label_values_map) + } + + /// B7.7 sid-direct sibling of [`Self::ingest_precompute_for_agg_config`]. + /// + /// Callers that already hold the bucket sid (the live worker after + /// B7.6 reshaped its `WorkerMessage`, and the backfill processor + /// after B7.7 rekeyed its per-window grouping from group_key to + /// sid) skip the mint round-trip by handing the sid in directly. + /// The mint-driven [`Self::ingest_precompute_for_agg_config`] is + /// content-addressed and idempotent with this method — passing the + /// resolver-minted sid here yields the same state under the same + /// sid — so both methods can coexist while migration finishes. + /// + /// The §6.3 ingest barrier (`Retired` / `Expired` sids reject + /// writes) and first-sight metadata registration are identical to + /// the mint-driven path. + pub fn ingest_precompute_with_series_id( + &self, + sid: u64, + agg_cfg: &asap_types::aggregation_config::AggregationConfig, + output: &crate::storage_engines::types::PrecomputedOutput, + accumulator: &dyn crate::storage_engines::types::AggregateCore, + ) -> Option { + let label_values_map = self.register_precompute_output(sid, agg_cfg, output)?; + // Keep the physical lifetime alive through publication. Removal takes // this same lock exclusively, so it cannot race metadata validation and // recreate orphan payload after the tombstone commits. let instances = self.instances.read().ok()?; let binding = instances.get(&sid)?; - if !binding.metadata.is_writable() || binding.metadata.policy_fp != output.policy_fp { + if !binding.metadata.is_writable() + || binding.metadata.policy_fp != output.policy_fp + || matches!( + binding.data_descriptor.source, + asap_types::sds::DataSourceIdentity::Derived { .. } + ) + { return None; } if binding.catalog_generation.is_some() && output.catalog_generation.is_none() { @@ -2863,7 +2931,7 @@ impl SketchStore { let window = (output.start_timestamp, output.end_timestamp); let accepted = match crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg) { - AggKind::Sketch { .. } => self.append_sample( + AggKind::Sketch { .. } => self.append_sample_with_binding( sid, label_values_map, window, @@ -2872,7 +2940,7 @@ impl SketchStore { encoding: SketchEncoding::MsgpackFull, }, ), - AggKind::ExactAgg { .. } => self.append_precompute( + AggKind::ExactAgg { .. } => self.append_precompute_with_binding( sid, label_values_map, window, @@ -3041,6 +3109,7 @@ impl SketchStore { metadata_writer, )?; + *self.immutable_publisher.write().unwrap() = Arc::downgrade(&flusher.publication_handle()); Ok(SketchIndexPersistence { manifest, part_cache, @@ -5775,6 +5844,8 @@ mod tests { // 2026-05 reorg: generic epoch-partitioned columnar storage lives // alongside the store that uses it. mod admission; +mod maintenance; +pub(crate) use maintenance::FrozenExactWindows; pub mod epoch_columnar; // `persistence` moved up to `sketch_db::persistence`. Re-exported here diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 14b388e5..53ed95e3 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -457,11 +457,11 @@ 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. +Installation still rejects derived inputs so they cannot accidentally receive +raw samples through the legacy metric router. The explicit immutable maintenance +entry point below must be wired into compiler construction and automatic +scheduling before this guard is removed. Neither raw-table substitution nor +treating late correction fragments as new observations is valid. ### Immutable completed windows @@ -486,3 +486,39 @@ must atomically publish their output identity before claiming replay-safe consum The existing finite-source completeness proof still rejects untracked writes or pending admitted work. Continuous producer watermarks and derived-state commit transactions are separate from this finite-input boundary. +### Executing an immutable maintenance sink + +`precompute_engine::maintenance_runtime::execute_completed_maintenance` executes +one installed semantic subDAG from a physical source whose required base windows +are durably complete. SummaryStore validates the catalog generation, physical +SeriesId, population, exact window coverage, and each part read. Missing, corrupt, +or duplicate source windows are errors; this path cannot silently omit a pane as +a query fallback helper might. + +The existing maintenance operator registry preserves a collection of source +states until the DAG explicitly merges or finalizes it. Exact Sum/Count +finalization with a declared Float64 output produces one row per source window; an unkeyed SummaryAgg consumes +those rows together. Consequently `Finalize -> SummaryAgg` does not accidentally +become one complete DAG evaluation per correction fragment. Live worker fragments +remain ineligible for finalization. + +The engine resumes a matching durable pending part and looks up the stored input +digest before computing a potentially randomized sketch. The existing flusher publishes a new result through its part +reservation protocol; SummaryStore fences query reads and physical lifetime +changes during publication. A concurrent identical completion reuses the durable +result instead of comparing newly randomized bytes. Both pending recovery and a +committed lookup restore the live completion boundary. Catalog-derived definitions +reject additive sketch/precompute writes even beyond that boundary; only reserved +publication may create their output state. The latest committed window can be +retried after restart without adding another part. + +This is an explicit maintenance entry point, not automatic workload coverage. +The initial consumer supports one source definition and population, complete +non-overlapping base panes, Sum/Count finalization, and unkeyed aggregate updates. +Compiler construction and automatic scheduling must use this entry point before +derived installation is enabled. Cross-population reductions, synchronized +multiple sources, general row operators, overlapping output-window replacement, +and continuous producer watermarks remain unsupported. In particular, the SQL +subquery's timestamp grouping and sampling predicate must not be replaced with an +arbitrary tumbling aggregate. Historical completion-metadata GC and pinning source +parts for recovery before a reserved output part exists remain lifecycle work.