From 2752c31fb3a768fb00ddcb375bc130717ca4063e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 11:57:20 -0600 Subject: [PATCH 01/21] Acquire complete immutable input cohorts before maintenance publication --- .../precompute_engine/maintenance_runtime.rs | 8 +- .../sketch_db/index/maintenance.rs | 180 ++++++++++++++++++ 2 files changed, 186 insertions(+), 2 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index cbdaf19f..08b0298f 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -608,8 +608,12 @@ pub fn execute_completed_maintenance( 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 mut cohort = store.read_frozen_exact_cohort( + &generation, + &derived.inputs, + &[(source_sid, source, expected, group.clone())], + )?; + let frozen = cohort.pop().ok_or("immutable input cohort is empty")?; let (dag, key, states) = prepare_frozen_maintenance_sink(installed, configs, sink, &frozen, window)?; let digest = key diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index 53e6064f..c49af73a 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -16,6 +16,49 @@ pub(crate) struct FrozenExactWindows { } impl SketchStore { + /// Acquire a complete immutable read set before any output is reserved. + /// Individual payloads are immutable; checking the shared generation again + /// after all reads prevents a catalog transition from mixing incarnations. + pub(crate) fn read_frozen_exact_cohort( + &self, + generation: &Arc, + expected_definitions: &BTreeSet, + requests: &[( + u64, + SummaryDefinitionId, + BTreeSet<(u64, u64)>, + BTreeMap, + )], + ) -> Result, String> { + if requests.is_empty() || requests.len() > 65_536 { + return Err("immutable input cohort has invalid population count".into()); + } + let supplied = requests + .iter() + .map(|(_, definition, _, _)| *definition) + .collect(); + if expected_definitions != &supplied { + return Err("immutable input cohort differs from installed input definitions".into()); + } + let mut ordered: Vec<_> = requests.iter().collect(); + ordered.sort_by(|left, right| (left.1, left.0, &left.3).cmp(&(right.1, right.0, &right.3))); + if ordered + .windows(2) + .any(|pair| (pair[0].1, pair[0].0, &pair[0].3) == (pair[1].1, pair[1].0, &pair[1].3)) + { + return Err("immutable input cohort repeats a physical population".into()); + } + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + let inputs = ordered + .into_iter() + .map(|(sid, definition, windows, group)| { + self.read_frozen_exact_windows(*sid, *definition, generation, windows, group) + }) + .collect::, _>>()?; + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + Ok(inputs) + } + fn durable_maintenance_population_ids( &self, definition: SummaryDefinitionId, @@ -411,6 +454,102 @@ mod tests { use crate::storage_engines::types::PrecomputedOutput; use asap_types::traits::SerializableToSink; + #[test] + fn cohort_requires_every_durable_source_in_one_catalog_generation() { + // Neither a missing second window nor a new catalog may yield a + // partially acquired cohort, even when the first source is complete. + 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 first = plan.precompute_plan.materializations[0].clone(); + first.aggregation_type = asap_types::AggregationType::Sum; + first.aggregation_sub_type = "sum".into(); + first.grouping_labels = std::iter::empty::().collect(); + let mut second = first.clone(); + second.metric = "cohort_second".into(); + let configs = [first, second]; + let catalog = + asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs) + .unwrap(); + let store = Arc::new(SketchStore::new()); + store + .install_summary_catalog(Arc::new(catalog.clone())) + .unwrap(); + let generation = store.active_catalog_generation().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 mut requests = Vec::new(); + for (index, config) in configs.iter().enumerate() { + let definition = config.policy_fingerprint().into(); + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: definition, + time_range: HalfOpenTimeRange { + start_ms: 0, + end_ms: 1000, + }, + group_values: BTreeMap::new(), + }; + let revision = store + .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()])) + .unwrap(); + let mut output = PrecomputedOutput::new(0, 1000, None, config.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + let mut sum = SumAccumulator::new(); + sum.update(5.0 + index as f64); + let sid = 900 + index as u64; + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 2000, + |writer| writer.ingest_precompute_with_series_id(sid, config, &output, &sum), + ) + .unwrap(); + requests.push(( + sid, + definition, + BTreeSet::from([(0, 1000)]), + BTreeMap::new(), + )); + } + 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 definitions = requests.iter().map(|request| request.1).collect(); + let parts = persistence.manifest.live_parts().len(); + let cohort = store + .read_frozen_exact_cohort(&generation, &definitions, &requests) + .unwrap(); + assert_eq!(cohort.len(), 2); + assert!(cohort.iter().all(|input| input.generation == generation)); + requests[1].2.insert((1000, 2000)); + assert!(store + .read_frozen_exact_cohort(&generation, &definitions, &requests) + .is_err()); + assert_eq!(persistence.manifest.live_parts().len(), parts); + let mut next = catalog; + next.plan_version += 1; + store.install_summary_catalog(Arc::new(next)).unwrap(); + requests[1].2.remove(&(1000, 2000)); + assert!(store + .read_frozen_exact_cohort(&generation, &definitions, &requests) + .is_err()); + persistence.shutdown(); + } + #[test] fn one_sid_with_two_populations_cannot_publish_a_partial_global_summary() { let mut fixture: serde_json::Value = serde_json::from_str(include_str!( @@ -502,6 +641,47 @@ mod tests { ) .unwrap(); assert!(!frozen.singleton_population_complete); + // The read set is deterministic and all-or-nothing, independently of + // the later routing decision (which still rejects this global reduce). + let request = |name: &str| { + ( + 700, + source_id, + BTreeSet::from([(0, 60_000)]), + BTreeMap::from([("instance".to_string(), name.to_string())]), + ) + }; + let cohort = store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id]), + &[request("b"), request("a")], + ) + .unwrap(); + assert_eq!(cohort.len(), 2); + assert_eq!(cohort[0].group["instance"], "a"); + assert_eq!(cohort[1].group["instance"], "b"); + assert!(store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id]), + &[request("a"), request("a")] + ) + .is_err()); + assert!(store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id, target.policy_fingerprint().into()]), + &[request("a")] + ) + .is_err()); + assert!(store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id]), + &[request("a"), request("absent")] + ) + .is_err()); let parts_before = persistence.manifest.live_parts().len(); let resolver = crate::drivers::ingest::series_resolver::SeriesIdResolver::new(); for _ in 0..2 { From 190964c8a2c0eeb6306225d6a862bf4b7f281bda Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:05:10 -0600 Subject: [PATCH 02/21] Hold cohort lifetimes through immutable registration and publication --- .../precompute_engine/maintenance_runtime.rs | 4 +- .../sketch_db/index/maintenance.rs | 149 ++++++++++++++---- .../storage_engines/sketch_db/index/mod.rs | 67 +++++--- 3 files changed, 169 insertions(+), 51 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 08b0298f..290f064a 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -635,7 +635,7 @@ pub fn execute_completed_maintenance( if store.recover_frozen_maintenance_output( target_sid, target_config, - &frozen, + std::slice::from_ref(&frozen), digest, window, )? { @@ -665,7 +665,7 @@ pub fn execute_completed_maintenance( target_config, &output, state.as_ref(), - &frozen, + std::slice::from_ref(&frozen), digest, ) } diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index c49af73a..c48680b4 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -277,24 +277,71 @@ impl SketchStore { } impl SketchStore { + fn validate_frozen_cohort_bindings<'a>( + &self, + config: &asap_types::PrecomputeMaterialization, + sources: &'a [FrozenExactWindows], + instances: &HashMap, + ) -> Result<&'a Arc, String> { + let generation = &sources + .first() + .ok_or("immutable publication input cohort is empty")? + .generation; + let definitions = sources.iter().map(|source| source.definition).collect(); + if config + .derived_input + .as_ref() + .is_none_or(|input| input.inputs != definitions) + { + return Err( + "immutable publication input definitions differ from installed identity".into(), + ); + } + let mut populations = BTreeSet::new(); + for source in sources { + if &source.generation != generation + || !populations.insert((source.definition, source.sid, &source.group)) + { + return Err( + "immutable publication input cohort has mixed generations or duplicates".into(), + ); + } + let binding = instances + .get(&source.sid) + .ok_or("immutable source was removed")?; + if binding.metadata.policy_fp != source.definition.fingerprint() + || binding.catalog_generation.as_deref() != Some(generation.as_ref()) + || !binding.metadata.is_writable() + { + return Err("immutable source identity or lifetime changed".into()); + } + } + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + Ok(generation) + } + pub(crate) fn recover_frozen_maintenance_output( &self, sid: u64, config: &asap_types::PrecomputeMaterialization, - source: &FrozenExactWindows, + sources: &[FrozenExactWindows], digest: [u8; 32], window: (u64, u64), ) -> Result { - self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?; + let _admission = self + .admission + .read() + .map_err(|_| "admission registry poisoned")?; let instances = self .instances .read() .map_err(|_| "instance registry poisoned")?; + let generation = self.validate_frozen_cohort_bindings(config, sources, &instances)?; 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.catalog_generation.as_deref() != Some(generation.as_ref()) || !binding.metadata.is_writable() { return Err("immutable output identity or lifetime changed".into()); @@ -345,41 +392,37 @@ impl SketchStore { config: &asap_types::PrecomputeMaterialization, output: &crate::storage_engines::types::PrecomputedOutput, state: &dyn AggregateCore, - source: &FrozenExactWindows, + sources: &[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()) + // One guard excludes source retirement through target registration + // and commit; admission excludes catalog transitions in the same span. + let _admission = self + .admission + .read() + .map_err(|_| "admission registry poisoned")?; + let mut instances = self + .instances + .write() + .map_err(|_| "instance registry poisoned")?; + let generation = self.validate_frozen_cohort_bindings(config, sources, &instances)?; + if output.policy_fp != config.policy_fingerprint() + || output.catalog_generation.as_deref() != Some(generation.as_ref()) { return Err("derived publication differs from its installed input identity".into()); } let labels = self - .register_precompute_output(sid, config, output) + .register_precompute_output_with_instances(sid, config, output, &mut instances) .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() + if !binding.metadata.is_writable() || binding.metadata.policy_fp != output.policy_fp - || binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) + || binding.catalog_generation.as_deref() != Some(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() @@ -470,7 +513,15 @@ mod tests { first.grouping_labels = std::iter::empty::().collect(); let mut second = first.clone(); second.metric = "cohort_second".into(); - let configs = [first, second]; + let mut target = first.clone(); + target.derived_input = Some(asap_types::derived_input::DerivedInputIdentity { + inputs: BTreeSet::from([ + first.policy_fingerprint().into(), + second.policy_fingerprint().into(), + ]), + program_sha256: "0".repeat(64), + }); + let configs = [first, second, target]; let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs) .unwrap(); @@ -488,7 +539,7 @@ mod tests { config.hot_window_ms = None; let mut persistence = store.start_persistence(config).unwrap(); let mut requests = Vec::new(); - for (index, config) in configs.iter().enumerate() { + for (index, config) in configs.iter().take(2).enumerate() { let definition = config.policy_fingerprint().into(); let coordinate = asap_types::sds::SummaryInstanceCoordinates { summary_definition_id: definition, @@ -540,6 +591,37 @@ mod tests { .read_frozen_exact_cohort(&generation, &definitions, &requests) .is_err()); assert_eq!(persistence.manifest.live_parts().len(), parts); + let target = &configs[2]; + let mut output = PrecomputedOutput::new(0, 1000, None, target.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + let mut sum = SumAccumulator::new(); + sum.update(11.0); + assert!(store + .publish_frozen_maintenance_output(902, target, &output, &sum, &cohort, [42; 32]) + .unwrap()); + assert!(store + .recover_frozen_maintenance_output(902, target, &cohort, [42; 32], (0, 1000)) + .unwrap()); + let published_parts = persistence.manifest.live_parts().len(); + store.force_expire(901).unwrap(); + assert!(store + .publish_frozen_maintenance_output(903, target, &output, &sum, &cohort, [42; 32]) + .is_err()); + assert!(store + .recover_frozen_maintenance_output(902, target, &cohort, [42; 32], (0, 1000)) + .is_err()); + assert!(!store.instances.read().unwrap().contains_key(&903)); + assert_eq!(persistence.manifest.live_parts().len(), published_parts); + assert!(!store + .persistence_metadata + .read() + .unwrap() + .as_ref() + .unwrap() + .load_strict() + .unwrap() + .iter() + .any(|record| record.sid == 903)); let mut next = catalog; next.plan_version += 1; store.install_summary_catalog(Arc::new(next)).unwrap(); @@ -831,7 +913,7 @@ mod tests { let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations( 1, 1, - &[source_config, target.clone()], + &[source_config.clone(), target.clone()], ) .unwrap(); let store = Arc::new(SketchStore::new()); @@ -847,6 +929,11 @@ mod tests { 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)); + let mut source_output = output.clone(); + source_output.policy_fp = source_config.policy_fingerprint(); + store + .register_precompute_output(600, &source_config, &source_output) + .unwrap(); store .register_precompute_output(601, &target, &output) .unwrap(); @@ -883,7 +970,13 @@ mod tests { singleton_population_complete: false, }; assert!(store - .recover_frozen_maintenance_output(601, &target, &input, [7; 32], (0, 1000)) + .recover_frozen_maintenance_output( + 601, + &target, + std::slice::from_ref(&input), + [7; 32], + (0, 1000) + ) .unwrap()); assert_eq!( store.completed_windows.read().unwrap().get(&601), 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 e9ef1234..c7c29c90 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -818,6 +818,15 @@ impl SketchStore { /// where the two indexes were written under separate sequential /// locks). See the index-field doc comments for the full invariant. pub fn register(&self, meta: SketchInstanceMetadata) { + let mut instances = self.instances.write().unwrap(); + self.register_with_instances(meta, &mut instances); + } + + fn register_with_instances( + &self, + meta: SketchInstanceMetadata, + instances: &mut HashMap, + ) -> bool { let sid = meta.sid; let policy_fp = meta.policy_fp; // A non-legacy materialization must resolve through the installed @@ -827,7 +836,7 @@ impl SketchStore { Ok(instance) => instance, Err(error) => { tracing::warn!(sid, %error, "rejecting SummaryStore registration outside the active SummaryCatalog"); - return; + return false; } }; let metric_name = instance @@ -835,10 +844,9 @@ impl SketchStore { .time_series_metric() .map(str::to_owned); // Fixed lock order: instances → policy_to_series_ids → metric_to_series_ids. - let mut instances = self.instances.write().unwrap(); if self.removed_sids.read().unwrap().contains_key(&sid) { tracing::warn!(sid, "rejecting reuse of a removed summary instance ID"); - return; + return false; } let mut policy_idx = self.policy_to_series_ids.write().unwrap(); let mut metric_idx = self.metric_to_series_ids.write().unwrap(); @@ -849,6 +857,7 @@ impl SketchStore { if let Some(metric_name) = metric_name { metric_idx.entry(metric_name).or_default().insert(sid); } + true } /// Install one authoritative catalog snapshot for future registrations. @@ -2973,13 +2982,24 @@ impl SketchStore { sid: u64, agg_cfg: &asap_types::PrecomputeMaterialization, output: &crate::storage_engines::types::PrecomputedOutput, + ) -> Option> { + let mut instances = self.instances.write().ok()?; + self.register_precompute_output_with_instances(sid, agg_cfg, output, &mut instances) + } + + fn register_precompute_output_with_instances( + &self, + sid: u64, + agg_cfg: &asap_types::PrecomputeMaterialization, + output: &crate::storage_engines::types::PrecomputedOutput, + instances: &mut HashMap, ) -> 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); let (capability, accuracy) = agg_kind.capability_and_accuracy(); - match self.instance(sid) { + match instances.get(&sid) { None => { if self.active_catalog_generation().as_deref() != output.catalog_generation.as_deref() @@ -3009,23 +3029,28 @@ impl SketchStore { // unconditionally `None`, which meant ASAP-tier ExactAgg // state was reachable only through the legacy precompute // query path; capability-matching couldn't see it. - self.register(SketchInstanceMetadata { - sid, - metric_name: agg_cfg.metric.clone(), - group_by_keys, - capability: Some(capability), - agg_kind, - accuracy, - first_seen_unix_ms: output.start_timestamp as i64, - retired_at_ms: None, - expires_at_ms: None, - // Trust the caller's output — it carries the - // policy fingerprint computed at emit time - // (precompute worker / backfill processor). - // Falling back to `from_config(&agg_cfg)` here - // would also be correct but redundant. - policy_fp: output.policy_fp, - }); + if !self.register_with_instances( + SketchInstanceMetadata { + sid, + metric_name: agg_cfg.metric.clone(), + group_by_keys, + capability: Some(capability), + agg_kind, + accuracy, + first_seen_unix_ms: output.start_timestamp as i64, + retired_at_ms: None, + expires_at_ms: None, + // Trust the caller's output — it carries the + // policy fingerprint computed at emit time + // (precompute worker / backfill processor). + // Falling back to `from_config(&agg_cfg)` here + // would also be correct but redundant. + policy_fp: output.policy_fp, + }, + instances, + ) { + return None; + } } Some(existing) if !existing.is_writable() || existing.policy_fp != output.policy_fp => { return None; From 537bc86265b0ce0f0a8729285c20d43cbfe7d437 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:09:30 -0600 Subject: [PATCH 03/21] Bind immutable receipts to canonical complete input lineage --- .../precompute_engine/maintenance_runtime.rs | 151 +++++++++++++++--- 1 file changed, 129 insertions(+), 22 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 290f064a..bb8d3079 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -381,6 +381,67 @@ fn merge_inputs(inputs: &[Arc]) -> Result Result<[u8; 32], String> { + let generation = &inputs + .first() + .ok_or("immutable input cohort is empty")? + .generation; + if inputs + .iter() + .map(|input| input.definition) + .collect::>() + != expected.inputs + { + return Err("immutable lineage differs from installed input definitions".into()); + } + let mut ordered: Vec<_> = inputs.iter().collect(); + ordered.sort_by(|left, right| { + (left.definition, left.sid, &left.group).cmp(&(right.definition, right.sid, &right.group)) + }); + if ordered.windows(2).any(|pair| { + (pair[0].definition, pair[0].sid, &pair[0].group) + == (pair[1].definition, pair[1].sid, &pair[1].group) + }) { + return Err("immutable lineage repeats a physical population".into()); + } + let multiple = ordered.len() > 1; + let mut lineage = Sha256::new(); + if multiple { + lineage.update(b"immutable-maintenance-input-v2"); + lineage.update((ordered.len() as u64).to_be_bytes()); + } else { + // Preserve the existing durable single-input receipt identity. + lineage.update(b"immutable-maintenance-input-v1"); + } + for input in ordered { + if &input.generation != generation || input.windows.is_empty() { + return Err("immutable lineage has mixed generations or empty windows".into()); + } + lineage.update(input.sid.to_be_bytes()); + let metadata = + serde_json::to_vec(&(&input.definition, &input.generation, &input.group, expected)) + .map_err(|error| error.to_string())?; + if multiple { + lineage.update((metadata.len() as u64).to_be_bytes()); + } + lineage.update(metadata); + if multiple { + lineage.update((input.windows.len() as u64).to_be_bytes()); + } + 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); + } + } + Ok(lineage.finalize().into()) +} + /// 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. @@ -453,28 +514,12 @@ fn prepare_frozen_maintenance_sink( 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 digest = frozen_cohort_lineage(std::slice::from_ref(input), expected_input)?; + let states = input + .windows + .iter() + .map(|((_, end), state)| (*end as i64, Arc::clone(state))) + .collect(); let key = MaterializationCommitKey { plan_id: input.generation.plan_id, plan_version: input.generation.plan_version, @@ -1356,6 +1401,68 @@ mod tests { asap_types::PolicyFingerprint(value).into() } + #[test] + fn cohort_lineage_is_order_independent_and_binds_every_input() { + use crate::storage_engines::sketch_db::index::FrozenExactWindows; + let make = |sid, id, value| { + let mut state = crate::precompute_engine::operators::SumAccumulator::new(); + state.update(value); + FrozenExactWindows { + sid, + definition: definition(id), + generation: Arc::new(asap_types::sds::CatalogGeneration { + schema_version: 2, + plan_id: 1, + plan_version: 1, + snapshot_sha256: "0".repeat(64), + }), + group: BTreeMap::from([("instance".into(), sid.to_string())]), + windows: BTreeMap::from([((0, 1000), Arc::new(state) as SummaryState)]), + singleton_population_complete: false, + } + }; + let expected = asap_types::derived_input::DerivedInputIdentity { + inputs: BTreeSet::from([definition(1), definition(2)]), + program_sha256: "1".repeat(64), + }; + let baseline = + frozen_cohort_lineage(&[make(10, 1, 3.0), make(20, 2, 5.0)], &expected).unwrap(); + assert_eq!( + baseline, + frozen_cohort_lineage(&[make(20, 2, 5.0), make(10, 1, 3.0)], &expected).unwrap() + ); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), make(20, 2, 6.0)], &expected).unwrap() + ); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), make(21, 2, 5.0)], &expected).unwrap() + ); + let mut changed = make(20, 2, 5.0); + changed.group.insert("instance".into(), "other".into()); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).unwrap() + ); + let mut changed = make(20, 2, 5.0); + let state = changed.windows.remove(&(0, 1000)).unwrap(); + changed.windows.insert((1000, 2000), state); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).unwrap() + ); + let mut changed = make(20, 2, 5.0); + Arc::make_mut(&mut changed.generation).plan_version += 1; + assert!(frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).is_err()); + assert!(frozen_cohort_lineage(&[make(10, 1, 3.0)], &expected).is_err()); + assert!(frozen_cohort_lineage( + &[make(10, 1, 3.0), make(10, 1, 3.0), make(20, 2, 5.0)], + &expected + ) + .is_err()); + } + fn node(id: u32) -> ExecutableDagNode { ExecutableDagNode { id: PostAsapNodeId(id), From b761197ce55835e4836b8953d6dcb774fa51ae40 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 11:54:45 -0600 Subject: [PATCH 04/21] feat: validate shared derived source window cohorts --- control_plane/src/physical/compiler.rs | 20 ++--- crates/asap_types/src/precompute_plan.rs | 108 +++++++++++++++++++++-- 2 files changed, 110 insertions(+), 18 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index f5d91043..b3788499 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1239,18 +1239,18 @@ impl PhysicalCompiler { query_id: query.query_id.clone(), reason: "derived source config missing".into(), })?; - if runtime_materialization.window_size != source_config.window_size - || runtime_materialization.slide_interval != source_config.slide_interval - || runtime_materialization.pane_origin_ms != source_config.pane_origin_ms - || runtime_materialization.window_size - != runtime_materialization.slide_interval - || runtime_materialization.stored_window_ms() - != source_config.stored_window_ms() - || source_config.stored_window_ms() - != source_config.window_size.saturating_mul(1_000) + asap_types::precompute_plan::validated_source_window_cohort( + &runtime_materialization, + &[source_config], + ) + .map_err(|error| CompileError::Query { + query_id: query.query_id.clone(), + reason: error.to_string(), + })?; + if runtime_materialization.window_size != runtime_materialization.slide_interval { return Err(CompileError::Query { query_id: query.query_id.clone(), - reason: "immutable scalar composition requires matching full nonoverlapping windows".into() }); + reason: "immutable scalar composition runtime requires nonoverlapping windows".into() }); } let SummaryExpr::SummaryAgg { child, .. } = &selected.node.expr else { unreachable!() diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 543fae8d..efc95582 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -9,6 +9,55 @@ use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet, HashMap}; use thiserror::Error; +/// Validate the physical windows consumed by one derived input program. +/// This returns the existing source definitions, not a second serialized +/// contract. It does not prove completion or authorize runtime execution. +/// Full-window sliding is lossless here; pane merging requires a separate +/// explicit operator and is deliberately not inferred from window sizes. +pub fn validated_source_window_cohort<'a>( + target: &crate::PrecomputeMaterialization, + sources: &[&'a crate::PrecomputeMaterialization], +) -> Result, PrecomputePlanError> { + let invalid = || { + PrecomputePlanError::CatalogContract( + "derived inputs require matching explicit full stored windows".into(), + ) + }; + let full_ms = target.window_size.checked_mul(1000).ok_or_else(invalid)?; + if sources.is_empty() + || full_ms == 0 + || target.slide_interval == 0 + || target.slide_interval > target.window_size + || target.pane_origin_ms.is_none() + || target.stored_window_ms() != full_ms + || (target.slide_interval < target.window_size + && !matches!( + target.window_layout, + crate::WindowMaterializationLayout::FullWindow + )) + { + return Err(invalid()); + } + let mut identities = BTreeSet::new(); + for source in sources { + if source.derived_input.is_some() + || !identities.insert(source.policy_fingerprint()) + || source.window_size != target.window_size + || source.slide_interval != target.slide_interval + || source.pane_origin_ms != target.pane_origin_ms + || source.stored_window_ms() != full_ms + || (source.slide_interval < source.window_size + && !matches!( + source.window_layout, + crate::WindowMaterializationLayout::FullWindow + )) + { + return Err(invalid()); + } + } + Ok(sources.to_vec()) +} + pub const BACKEND_COMPAT: &str = "asap-query-backend.v1"; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -381,14 +430,11 @@ impl PrecomputePlan { .iter() .find(|candidate| candidate.policy_fingerprint() == source_id.fingerprint()) .ok_or_else(invalid)?; - if source.derived_input.is_some() - || source.window_size != config.window_size - || source.slide_interval != config.slide_interval - || source.window_size != source.slide_interval - || source.pane_origin_ms != config.pane_origin_ms - || source.stored_window_ms() != config.stored_window_ms() - || source.window_size.checked_mul(1000) != Some(source.stored_window_ms()) - { + validated_source_window_cohort(config, &[source])?; + // Current installed runtime capability remains nonoverlapping. + // The shared cohort contract also describes explicit full-window + // sliding for consumers which separately prove its completion. + if source.window_size != source.slide_interval { return Err(invalid()); } let mut matched = false; @@ -700,3 +746,49 @@ pub(crate) fn state_encodings(family: &SummaryFamilyType) -> Vec _ => Vec::new(), } } + +#[cfg(test)] +mod source_window_cohort_tests { + use super::*; + fn full_window() -> crate::PrecomputeMaterialization { + serde_json::from_value(serde_json::json!({ + "aggregation_type":"Sum", "aggregation_sub_type":"", "parameters":{}, + "grouping_labels":{"labels":[]}, "aggregated_labels":{"labels":[]}, + "rollup_labels":{"labels":[]}, "original_yaml":"", + "window_size":60, "slide_interval":10, "window_type":"sliding", + "window_layout":{"kind":"full_window"}, "pane_origin_ms":0, + "spatial_filter":"", "spatial_filter_normalized":"", "metric":"m", + "num_aggregates_to_retain":null, "table_name":null, "value_projection":null + })) + .unwrap() + } + #[test] + fn full_sliding_cohort_preserves_explicit_windows_and_identity() { + let target = full_window(); + let source = full_window(); + let result = validated_source_window_cohort(&target, &[&source]).unwrap(); + assert!(std::ptr::eq(result[0], &source)); + let mut other = source.clone(); + other.metric = "other".into(); + assert_eq!( + validated_source_window_cohort(&target, &[&source, &other]) + .unwrap() + .len(), + 2 + ); + assert!(validated_source_window_cohort(&target, &[&source, &source]).is_err()); + for mutation in 0..3 { + let mut changed = source.clone(); + match mutation { + 0 => changed.pane_origin_ms = Some(1), + 1 => changed.slide_interval = 20, + _ => { + changed.window_layout = + crate::WindowMaterializationLayout::Pane { pane_secs: 10 } + } + } + assert_ne!(source.policy_fingerprint(), changed.policy_fingerprint()); + assert!(validated_source_window_cohort(&target, &[&changed]).is_err()); + } + } +} From 0e96c09e06c68ad92a7b20c6383f8f77bca75f9d Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 11:55:57 -0600 Subject: [PATCH 05/21] fix: retain legacy nonoverlapping origin contract --- crates/asap_types/src/precompute_plan.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index efc95582..f48df61a 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -28,7 +28,7 @@ pub fn validated_source_window_cohort<'a>( || full_ms == 0 || target.slide_interval == 0 || target.slide_interval > target.window_size - || target.pane_origin_ms.is_none() + || (target.slide_interval < target.window_size && target.pane_origin_ms.is_none()) || target.stored_window_ms() != full_ms || (target.slide_interval < target.window_size && !matches!( From 91f380b9edea083ff63f6e228a3d0d003e7cb590 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 11:57:01 -0600 Subject: [PATCH 06/21] test: distinguish explicit sliding and legacy origins --- crates/asap_types/src/precompute_plan.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index f48df61a..818195a3 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -766,6 +766,11 @@ mod source_window_cohort_tests { fn full_sliding_cohort_preserves_explicit_windows_and_identity() { let target = full_window(); let source = full_window(); + let mut missing_origin = source.clone(); + missing_origin.pane_origin_ms = None; + assert!(validated_source_window_cohort(&missing_origin, &[&missing_origin]).is_err()); + missing_origin.slide_interval = missing_origin.window_size; + assert!(validated_source_window_cohort(&missing_origin, &[&missing_origin]).is_ok()); let result = validated_source_window_cohort(&target, &[&source]).unwrap(); assert!(std::ptr::eq(result[0], &source)); let mut other = source.clone(); From 8db331968027b3a27b07a0939168bac17be624af Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:16:55 -0600 Subject: [PATCH 07/21] Resolve frozen maintenance inputs at each materialized frontier --- .../precompute_engine/maintenance_runtime.rs | 307 +++++++++++++----- 1 file changed, 222 insertions(+), 85 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index bb8d3079..9e0c7467 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -56,13 +56,18 @@ type PendingOutput = ( Box, ); +enum MaintenanceInputs<'a> { + Live { + definition: asap_types::sds::SummaryDefinitionId, + state: SummaryState, + }, + Frozen(&'a [crate::storage_engines::sketch_db::index::FrozenExactWindows]), +} + struct OperatorAdapter<'a> { binding: &'a BackendExecutableBinding, - source_definition: asap_types::sds::SummaryDefinitionId, - source: SummaryState, + inputs: MaintenanceInputs<'a>, configs: &'a [asap_types::aggregation_config::AggregationConfig], - immutable_windows: Option>, - singleton_population_complete: bool, } impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { @@ -72,13 +77,10 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { &self, node: &ExecutableDagNode, ) -> Result, String> { - if !matches!( - self.binding.node(node.id), - Some(BackendNodeBinding::Materialization { summary_definition }) - if *summary_definition == self.source_definition - ) { - return Ok(None); - } + let definition = match self.binding.node(node.id) { + Some(BackendNodeBinding::Materialization { summary_definition }) => *summary_definition, + _ => return Ok(None), + }; let family = node.output_schema.fields.iter().find_map(|field| { (!matches!( field.dtype, @@ -86,16 +88,34 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { )) .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")?, - })); + match &self.inputs { + MaintenanceInputs::Live { + definition: source, + state, + } if definition == *source => Ok(Some(MaintenanceValue::Summary { + state: Arc::clone(state), + family, + })), + MaintenanceInputs::Live { .. } => Ok(None), + MaintenanceInputs::Frozen(inputs) => { + let mut matching = inputs.iter().filter(|input| input.definition == definition); + let Some(input) = matching.next() else { + return Ok(None); + }; + if matching.next().is_some() { + return Err("maintenance frontier requires explicit population routing".into()); + } + Ok(Some(MaintenanceValue::SummaryWindows { + states: input + .windows + .iter() + .map(|((_, end), state)| (*end as i64, Arc::clone(state))) + .collect::>() + .into(), + family: family.ok_or("immutable source lacks a summary schema")?, + })) + } } - Ok(Some(MaintenanceValue::Summary { - state: Arc::clone(&self.source), - family, - })) } fn execute( @@ -109,7 +129,7 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, } => { - if self.immutable_windows.is_none() { + if !matches!(&self.inputs, MaintenanceInputs::Frozen(_)) { return Err( "maintenance finalization requires immutable completed input windows" .into(), @@ -124,7 +144,7 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { 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() { + if !matches!(&self.inputs, MaintenanceInputs::Frozen(_)) { return Err( "maintenance aggregation requires immutable completed input windows".into(), ); @@ -144,19 +164,24 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { .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")?; - validate_maintenance_grouping( - config, - source_config, - node, - self.singleton_population_complete, - )?; + let MaintenanceInputs::Frozen(sources) = &self.inputs else { + return Err("maintenance aggregation requires frozen sources".into()); + }; + for source in *sources { + let source_config = self + .configs + .iter() + .find(|config| { + config.policy_fingerprint() == source.definition.fingerprint() + }) + .ok_or("maintenance input lacks installed source configuration")?; + validate_maintenance_grouping( + config, + source_config, + node, + source.singleton_population_complete, + )?; + } if config.accumulator_spec().map_err(|e| e.to_string())?.family != *family { return Err( "maintenance SummaryAgg family differs from installed configuration".into(), @@ -449,13 +474,12 @@ 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, + inputs: &[crate::storage_engines::sketch_db::index::FrozenExactWindows], output_window: (u64, u64), ) -> Result< ( planner_types::post_asap::ExecutableDag, MaterializationCommitKey, - Vec<(i64, SummaryState)>, ), String, > { @@ -472,15 +496,17 @@ fn prepare_frozen_maintenance_sink( .derived_input .as_ref() .ok_or("immutable sink is not a derived materialization")?; - if expected_input.inputs != BTreeSet::from([input.definition]) { + if expected_input.inputs != inputs.iter().map(|input| input.definition).collect() { 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) + || inputs.iter().any(|input| { + 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()); } @@ -499,7 +525,7 @@ fn prepare_frozen_maintenance_sink( .iter() .filter_map(|(node, binding)| match binding { BackendNodeBinding::Materialization { summary_definition } - if *summary_definition == input.definition => + if expected_input.inputs.contains(summary_definition) => { Some((*node, *summary_definition)) } @@ -514,15 +540,14 @@ fn prepare_frozen_maintenance_sink( if &actual_input != expected_input { return Err("immutable sink input program differs from its catalog identity".into()); } - let digest = frozen_cohort_lineage(std::slice::from_ref(input), expected_input)?; - let states = input - .windows - .iter() - .map(|((_, end), state)| (*end as i64, Arc::clone(state))) - .collect(); + let digest = frozen_cohort_lineage(inputs, expected_input)?; + let generation = &inputs + .first() + .ok_or("immutable input cohort is empty")? + .generation; let key = MaterializationCommitKey { - plan_id: input.generation.plan_id, - plan_version: input.generation.plan_version, + plan_id: generation.plan_id, + plan_version: generation.plan_version, summary_definition: target, window_start_ms: i64::try_from(output_window.0) .map_err(|_| "output window exceeds timestamp range")?, @@ -530,26 +555,21 @@ fn prepare_frozen_maintenance_sink( .map_err(|_| "output window exceeds timestamp range")?, input_lineage: digest.to_vec(), }; - Ok((dag, key, states)) + Ok((dag, key)) } 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, + inputs: &[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), + inputs: MaintenanceInputs::Frozen(inputs), configs, - immutable_windows: Some(states.into()), - singleton_population_complete: input.singleton_population_complete, }; let value = execute_precompute_sink( dag, @@ -571,14 +591,26 @@ fn evaluate_frozen_maintenance_sink( 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 (dag, key) = prepare_frozen_maintenance_sink( + installed, + configs, + sink, + std::slice::from_ref(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)?; + let state = execute_prepared_frozen_sink( + installed, + configs, + sink, + std::slice::from_ref(input), + &dag, + key, + )?; Ok((state, digest)) } @@ -659,8 +691,13 @@ pub fn execute_completed_maintenance( &[(source_sid, source, expected, group.clone())], )?; let frozen = cohort.pop().ok_or("immutable input cohort is empty")?; - let (dag, key, states) = - prepare_frozen_maintenance_sink(installed, configs, sink, &frozen, window)?; + let (dag, key) = prepare_frozen_maintenance_sink( + installed, + configs, + sink, + std::slice::from_ref(&frozen), + window, + )?; let digest = key .input_lineage .as_slice() @@ -686,7 +723,14 @@ pub fn execute_completed_maintenance( )? { return Ok(false); } - let state = execute_prepared_frozen_sink(installed, configs, sink, &frozen, &dag, key, states)?; + let state = execute_prepared_frozen_sink( + installed, + configs, + sink, + std::slice::from_ref(&frozen), + &dag, + key, + )?; let mut output = crate::storage_engines::types::PrecomputedOutput::new( window.0, window.1, @@ -1124,11 +1168,11 @@ impl MaintenanceDagSink { } let adapter = OperatorAdapter { binding: &installed.binding, - source_definition, - source: Arc::clone(&source), + inputs: MaintenanceInputs::Live { + definition: source_definition, + state: Arc::clone(&source), + }, configs: &plan.precompute_plan.materializations, - immutable_windows: None, - singleton_population_complete: false, }; for sink_node in &installed.binding.precompute_sinks { // Derived summaries consume complete immutable windows at the @@ -1498,6 +1542,87 @@ mod tests { Arc::new(accumulator) } + #[test] + fn frozen_adapter_resolves_each_materialized_frontier_without_aliasing() { + use planner_types::post_asap::{ExactKind, ExactParams, SummaryFamilyType, SummaryField}; + let make = |id| crate::storage_engines::sketch_db::index::FrozenExactWindows { + sid: id, + definition: definition(id), + generation: Arc::new(asap_types::sds::CatalogGeneration { + schema_version: 2, + plan_id: 1, + plan_version: 1, + snapshot_sha256: "0".repeat(64), + }), + group: BTreeMap::new(), + windows: BTreeMap::from([((0, 1000), sum(id as f64))]), + singleton_population_complete: false, + }; + let inputs = [make(1), make(2)]; + let binding = BackendExecutableBinding { + nodes: [(1, definition(1)), (2, definition(2)), (3, definition(3))] + .into_iter() + .map(|(id, summary_definition)| { + ( + PostAsapNodeId(id), + BackendNodeBinding::Materialization { summary_definition }, + ) + }) + .collect(), + query_sink: PostAsapNodeId(3), + query_plan_sink: asap_types::query_plan::QueryNodeId(3), + precompute_sinks: vec![PostAsapNodeId(3)], + }; + let adapter = OperatorAdapter { + binding: &binding, + inputs: MaintenanceInputs::Frozen(&inputs), + configs: &[], + }; + let source_node = |id| { + let mut source = node(id); + source.output_schema.fields = vec![SummaryField { + name: "state".into(), + dtype: SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + nullable: false, + }]; + source + }; + let first = adapter + .materialized_input(&source_node(1)) + .unwrap() + .unwrap(); + let second = adapter + .materialized_input(&source_node(2)) + .unwrap() + .unwrap(); + assert!(adapter + .materialized_input(&source_node(3)) + .unwrap() + .is_none()); + let merged = adapter + .execute(&node(3), &[Arc::new(first), Arc::new(second)]) + .unwrap(); + assert_eq!( + merged + .state() + .unwrap() + .query_statistic( + asap_types::Statistic::Sum, + &None, + &std::collections::HashMap::new() + ) + .unwrap(), + 3.0 + ); + let ambiguous = [make(1), make(1)]; + let adapter = OperatorAdapter { + binding: &binding, + inputs: MaintenanceInputs::Frozen(&ambiguous), + configs: &[], + }; + assert!(adapter.materialized_input(&source_node(1)).is_err()); + } + #[test] fn finalized_summary_update_builds_a_different_installed_family() { use planner_types::post_asap::{ @@ -1551,13 +1676,25 @@ mod tests { query_plan_sink: control_plane::query_plan::QueryNodeId(3), precompute_sinks: vec![PostAsapNodeId(3)], }; + let frozen_inputs = [ + crate::storage_engines::sketch_db::index::FrozenExactWindows { + sid: 1, + definition: source_definition, + generation: Arc::new(asap_types::sds::CatalogGeneration { + schema_version: 2, + plan_id: 1, + plan_version: 1, + snapshot_sha256: "0".repeat(64), + }), + group: BTreeMap::new(), + windows: BTreeMap::from([((0, 1000), sum(7.0))]), + singleton_population_complete: false, + }, + ]; let adapter = OperatorAdapter { binding: &binding, - source_definition, - source: sum(7.0), + inputs: MaintenanceInputs::Frozen(&frozen_inputs), configs: &configs, - immutable_windows: Some(vec![(1000, sum(7.0))].into()), - singleton_population_complete: false, }; let mut read = node(2); read.payload = ExecutableOperatorPayload::Value { @@ -2106,11 +2243,11 @@ mod tests { }; let adapter = OperatorAdapter { binding: &binding, - source_definition: definition(1), - source: sum(7.0), + inputs: MaintenanceInputs::Live { + definition: definition(1), + state: sum(7.0), + }, configs: &[], - immutable_windows: None, - singleton_population_complete: false, }; let mut aggregate = node(1); aggregate.operator = ExecutableOperator::SummaryAgg; @@ -2431,11 +2568,11 @@ mod tests { let source = sum(2.0); let adapter = OperatorAdapter { binding: &binding, - source_definition: definition(1), - source, + inputs: MaintenanceInputs::Live { + definition: definition(1), + state: source, + }, configs: &[], - immutable_windows: None, - singleton_population_complete: false, }; let commits = CommitRegistry::default(); let key = MaterializationCommitKey { @@ -2506,11 +2643,11 @@ mod tests { }; let adapter = OperatorAdapter { binding: &binding, - source_definition: definition(1), - source: sum(2.0), + inputs: MaintenanceInputs::Live { + definition: definition(1), + state: sum(2.0), + }, configs: &[], - immutable_windows: None, - singleton_population_complete: false, }; let commits = CommitRegistry::default(); let key = MaterializationCommitKey { From c71c801cfee5a8dab577e4a33f9f5c7fbc4d17c9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:22:55 -0600 Subject: [PATCH 08/21] Execute complete aligned source cohorts through the installed maintenance DAG --- .../precompute_engine/maintenance_runtime.rs | 425 +++++++++++++++--- 1 file changed, 359 insertions(+), 66 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 9e0c7467..0cfbcdb7 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -629,6 +629,43 @@ pub fn execute_completed_maintenance( target_sid: u64, window: (u64, u64), group: &BTreeMap, +) -> Result { + 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 derived = configs + .iter() + .find(|config| config.policy_fingerprint() == target.fingerprint()) + .and_then(|config| 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()); + } + execute_completed_maintenance_cohort( + store, + installed, + configs, + sink, + &BTreeMap::from([(*derived.inputs.first().unwrap(), source_sid)]), + target_sid, + window, + group, + ) +} + +/// Execute one completed population per input definition at matching full +/// windows. All source populations must share the same explicit label map. +#[allow(clippy::too_many_arguments)] +pub(crate) fn execute_completed_maintenance_cohort( + store: &crate::storage_engines::sketch_db::index::SketchStore, + installed: &asap_types::executable_plan::InstalledPostAsapDag, + configs: &[asap_types::PrecomputeMaterialization], + sink: PostAsapNodeId, + source_sids: &BTreeMap, + target_sid: u64, + window: (u64, u64), + group: &BTreeMap, ) -> Result { use asap_types::executable_plan::BackendNodeBinding; let target = match installed.binding.node(sink) { @@ -643,61 +680,73 @@ pub fn execute_completed_maintenance( .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()); + if derived.inputs != source_sids.keys().copied().collect() { + return Err("maintenance source set differs from installed input definitions".into()); } - let expected = (0..target_width / source_width) - .map(|index| { - let start = window.0 + index * source_width; - (start, start + source_width) + let source_configs = source_sids + .keys() + .map(|source| { + configs + .iter() + .find(|config| config.policy_fingerprint() == source.fingerprint()) + .ok_or("maintenance source configuration is absent") }) - .collect(); + .collect::, _>>()?; + if source_configs.len() > 1 { + asap_types::precompute_plan::validated_source_window_cohort(target_config, &source_configs) + .map_err(|error| error.to_string())?; + } + let mut requests = Vec::with_capacity(source_sids.len()); + for source_config in &source_configs { + let source = source_config.policy_fingerprint().into(); + let source_sid = source_sids[&source]; + 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()); + } + let expected = (0..target_width / source_width) + .map(|index| { + let start = window.0 + index * source_width; + (start, start + source_width) + }) + .collect(); + requests.push((source_sid, source, expected, group.clone())); + } let generation = store .active_catalog_generation() .ok_or("maintenance requires an authoritative catalog")?; - let mut cohort = store.read_frozen_exact_cohort( - &generation, - &derived.inputs, - &[(source_sid, source, expected, group.clone())], - )?; - let frozen = cohort.pop().ok_or("immutable input cohort is empty")?; - let (dag, key) = prepare_frozen_maintenance_sink( - installed, - configs, - sink, - std::slice::from_ref(&frozen), - window, - )?; + let cohort = store.read_frozen_exact_cohort(&generation, &derived.inputs, &requests)?; + if cohort.len() > 1 + && cohort + .iter() + .any(|source| !source.singleton_population_complete) + { + return Err( + "multi-source maintenance requires a complete single population per source".into(), + ); + } + let (dag, key) = prepare_frozen_maintenance_sink(installed, configs, sink, &cohort, window)?; let digest = key .input_lineage .as_slice() @@ -708,29 +757,28 @@ pub fn execute_completed_maintenance( .iter() .find(|node| node.id == sink) .ok_or("maintenance target node is absent")?; - validate_maintenance_grouping( - target_config, - source_config, - target_node, - frozen.singleton_population_complete, - )?; + for frozen in &cohort { + let source_config = source_configs + .iter() + .find(|config| config.policy_fingerprint() == frozen.definition.fingerprint()) + .ok_or("maintenance source configuration is absent")?; + validate_maintenance_grouping( + target_config, + source_config, + target_node, + frozen.singleton_population_complete, + )?; + } if store.recover_frozen_maintenance_output( target_sid, target_config, - std::slice::from_ref(&frozen), + &cohort, digest, window, )? { return Ok(false); } - let state = execute_prepared_frozen_sink( - installed, - configs, - sink, - std::slice::from_ref(&frozen), - &dag, - key, - )?; + let state = execute_prepared_frozen_sink(installed, configs, sink, &cohort, &dag, key)?; let mut output = crate::storage_engines::types::PrecomputedOutput::new( window.0, window.1, @@ -754,7 +802,7 @@ pub fn execute_completed_maintenance( target_config, &output, state.as_ref(), - std::slice::from_ref(&frozen), + &cohort, digest, ) } @@ -2115,6 +2163,7 @@ mod tests { .series_ids_for_policy(durable_configs[1].policy_fingerprint()) .is_empty()); persistence.shutdown(); + exercise_two_source_completed_sink(&dag, &configs, &scheduled_binding); assert!(evaluate_weight( &SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::Named("missing".into())), 7.0, @@ -2123,6 +2172,250 @@ mod tests { .is_err()); } + fn exercise_two_source_completed_sink( + template: &ExecutableDag, + configs: &[asap_types::PrecomputeMaterialization], + binding: &BackendExecutableBinding, + ) { + // A bound operator fixture, not a claim that a frontend selected this + // composition. Both actual durable sources are required before output. + use crate::storage_engines::sketch_db::index::{ + persistence::config::SketchStorePersistenceConfig, SketchStore, + }; + use asap_types::executable_plan::{InstalledPostAsapDag, OwnedPostAsapDag}; + let mut first = configs[0].clone(); + first.window_layout = asap_types::WindowMaterializationLayout::FullWindow; + let mut second = first.clone(); + second.metric = "second_maintenance_source".into(); + let first_id = first.policy_fingerprint().into(); + let second_id = second.policy_fingerprint().into(); + let mut dag = template.clone(); + let mut second_node = dag + .nodes + .iter() + .find(|node| node.id == PostAsapNodeId(1)) + .unwrap() + .clone(); + second_node.id = PostAsapNodeId(5); + let mut merge = second_node.clone(); + merge.id = PostAsapNodeId(6); + merge.operator = ExecutableOperator::SummaryMerge; + merge.payload = ExecutableOperatorPayload::SummaryMerge; + dag.nodes.extend([second_node, merge]); + let original = dag + .edges + .iter() + .find(|edge| edge.producer == PostAsapNodeId(1) && edge.consumer == PostAsapNodeId(2)) + .unwrap() + .clone(); + dag.edges.retain(|edge| { + !(edge.producer == PostAsapNodeId(1) && edge.consumer == PostAsapNodeId(2)) + }); + for (producer, consumer) in [(1, 6), (5, 6), (6, 2)] { + let mut edge = original.clone(); + edge.producer = PostAsapNodeId(producer); + edge.consumer = PostAsapNodeId(consumer); + dag.edges.push(edge); + } + if let ExecutableOperatorPayload::SummaryAgg { input, .. } = &mut dag + .nodes + .iter_mut() + .find(|node| node.id == PostAsapNodeId(3)) + .unwrap() + .payload + { + input.weight = planner_types::post_asap::SummaryInputExpr::Column( + planner_types::pre_asap::ColumnRef::SampleValue, + ); + } + let document = + OwnedPostAsapDag::from_executable("two-source-fixture".into(), &dag).unwrap(); + let mut target = configs[1].clone(); + target.derived_input = Some( + asap_types::derived_input::DerivedInputIdentity::from_dag( + &document, + PostAsapNodeId(2), + &BTreeMap::from([ + (PostAsapNodeId(1), first_id), + (PostAsapNodeId(5), second_id), + ]), + ) + .unwrap(), + ); + let mut binding = binding.clone(); + for (node, summary_definition) in [ + (1, first_id), + (5, second_id), + (3, target.policy_fingerprint().into()), + ] { + binding.nodes.insert( + PostAsapNodeId(node), + BackendNodeBinding::Materialization { summary_definition }, + ); + } + binding + .nodes + .insert(PostAsapNodeId(6), BackendNodeBinding::MaintenanceInput); + let installed = InstalledPostAsapDag { document, binding }; + let configs = [first, second, target]; + let catalog = Arc::new( + asap_types::summary_catalog::SummaryCatalog::from_materializations(2, 1, &configs) + .unwrap(), + ); + let directory = tempfile::tempdir().unwrap(); + let persistence_config = || { + let mut config = + SketchStorePersistenceConfig::with_memory_limit(1 << 24, directory.path().into()); + config.delete_older_than_ms = None; + config.hot_window_ms = None; + config + }; + let 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(); + let sources = BTreeMap::from([(first_id, 800), (second_id, 801)]); + for (index, value) in [2.0, 7.0].into_iter().enumerate() { + let config = &configs[index]; + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: config.policy_fingerprint().into(), + time_range: asap_types::sds::HalfOpenTimeRange { + start_ms: 0, + end_ms: 2000, + }, + group_values: BTreeMap::new(), + }; + let revision = store + .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()])) + .unwrap(); + let mut output = PrecomputedOutput::new(0, 2000, None, config.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 4000, + |writer| { + writer.ingest_precompute_with_series_id( + 800 + index as u64, + config, + &output, + sum(value).as_ref(), + ) + }, + ) + .unwrap(); + } + assert!(execute_completed_maintenance_cohort( + &store, + &installed, + &configs, + PostAsapNodeId(3), + &sources, + 802, + (0, 2000), + &BTreeMap::new() + ) + .is_err()); + assert!(store + .series_ids_for_policy(configs[2].policy_fingerprint()) + .is_empty()); + 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 requests = sources + .iter() + .map(|(definition, sid)| { + ( + *sid, + *definition, + BTreeSet::from([(0, 2000)]), + BTreeMap::new(), + ) + }) + .collect::>(); + let cohort = store + .read_frozen_exact_cohort( + &generation, + &configs[2].derived_input.as_ref().unwrap().inputs, + &requests, + ) + .unwrap(); + let (dag, key) = prepare_frozen_maintenance_sink( + &installed, + &configs, + PostAsapNodeId(3), + &cohort, + (0, 2000), + ) + .unwrap(); + let result = execute_prepared_frozen_sink( + &installed, + &configs, + PostAsapNodeId(3), + &cohort, + &dag, + key, + ) + .unwrap(); + assert_eq!( + result + .query_statistic( + asap_types::Statistic::Quantile, + &None, + &std::collections::HashMap::from([("quantile".into(), "0.5".into())]) + ) + .unwrap(), + 9.0 + ); + assert!(execute_completed_maintenance_cohort( + &store, + &installed, + &configs, + PostAsapNodeId(3), + &sources, + 802, + (0, 2000), + &BTreeMap::new() + ) + .unwrap()); + let parts = persistence.manifest.live_parts().len(); + assert!(!execute_completed_maintenance_cohort( + &store, + &installed, + &configs, + PostAsapNodeId(3), + &sources, + 802, + (0, 2000), + &BTreeMap::new() + ) + .unwrap()); + assert_eq!(persistence.manifest.live_parts().len(), parts); + 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(); + assert!(!execute_completed_maintenance_cohort( + &restored, + &installed, + &configs, + PostAsapNodeId(3), + &sources, + 802, + (0, 2000), + &BTreeMap::new() + ) + .unwrap()); + assert_eq!(persistence.manifest.live_parts().len(), parts); + persistence.shutdown(); + } + #[test] fn finalization_preserves_windows_until_an_explicit_merge() { use planner_types::post_asap::{ExactKind, ExactParams, SummaryFamilyType, SummaryField}; From be1ec6c56db3fe06607ecc29430c888fe3366038 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:29:38 -0600 Subject: [PATCH 09/21] refactor: share the data-plane Float64 arithmetic kernel --- .../asap_query_engine/post_asap_readout.rs | 15 ++------------- data_plane/src/utils/arithmetic.rs | 19 +++++++++++++++++++ data_plane/src/utils/mod.rs | 1 + 3 files changed, 22 insertions(+), 13 deletions(-) create mode 100644 data_plane/src/utils/arithmetic.rs diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index 8e118473..6a4c6268 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -4,6 +4,8 @@ use std::collections::BTreeMap; +use crate::utils::arithmetic::evaluate_float64_arithmetic as arithmetic; + use crate::query_engines::asap_query_engine::summary_exec::{execute, ExecOutcome}; use asap_types::query_plan::{QueryNodeId, QueryPlanNode}; use control_plane::types_v2::AccuracyTarget; @@ -410,19 +412,6 @@ fn expand_item_readout( Ok((expand_item_rows(group_key, value, item_labels)?, coverage)) } -fn arithmetic(operator: &planner_types::pre_asap::ArithmeticOpKind, left: f64, right: f64) -> f64 { - use planner_types::pre_asap::ArithmeticOpKind::*; - match operator { - Add => left + right, - Sub => left - right, - Mul => left * right, - Div => left / right, - Mod => left % right, - Pow => left.powf(right), - Atan2 => left.atan2(right), - } -} - fn binary_values( operator: &planner_types::pre_asap::ArithmeticOpKind, lhs: &PhysicalQueryOutput, diff --git a/data_plane/src/utils/arithmetic.rs b/data_plane/src/utils/arithmetic.rs new file mode 100644 index 00000000..94aba683 --- /dev/null +++ b/data_plane/src/utils/arithmetic.rs @@ -0,0 +1,19 @@ +//! Float64 arithmetic shared by data-plane execution engines. +//! Preserve IEEE non-finite results; callers own their output policies. + +pub(crate) fn evaluate_float64_arithmetic( + operator: &planner_types::pre_asap::ArithmeticOpKind, + left: f64, + right: f64, +) -> f64 { + use planner_types::pre_asap::ArithmeticOpKind::*; + match operator { + Add => left + right, + Sub => left - right, + Mul => left * right, + Div => left / right, + Mod => left % right, + Pow => left.powf(right), + Atan2 => left.atan2(right), + } +} diff --git a/data_plane/src/utils/mod.rs b/data_plane/src/utils/mod.rs index 5d620636..72f331c5 100644 --- a/data_plane/src/utils/mod.rs +++ b/data_plane/src/utils/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod arithmetic; pub mod file_io; pub mod http; From 4c0b4ac5e5cdc50960f1ca47ddbf5de8dcfcf99e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:40:55 -0600 Subject: [PATCH 10/21] Preserve binary operand roles in maintenance scheduling --- .../src/precompute_engine/subdag_scheduler.rs | 98 +++++++++++++++++-- 1 file changed, 92 insertions(+), 6 deletions(-) diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 6a52b88d..f3ba0840 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -1,6 +1,8 @@ use asap_types::executable_plan::{BackendExecutableBinding, BackendNodeBinding}; use planner_types::post_asap::PostAsapNodeId; -use planner_types::post_asap::{ExecutableDag, ExecutableDagNode, ExecutionDataState}; +use planner_types::post_asap::{ + EdgeRole, ExecutableDag, ExecutableDagNode, ExecutableOperatorPayload, ExecutionDataState, +}; use std::{ collections::{BTreeMap, BTreeSet}, sync::Arc, @@ -91,12 +93,36 @@ where .iter() .map(|n| (n.id.0, n)) .collect::>(); - let mut inputs = BTreeMap::>::new(); + let mut incoming = BTreeMap::>::new(); for edge in &dag.edges { - inputs - .entry(edge.consumer.0) - .or_default() - .push(edge.producer.0); + incoming.entry(edge.consumer.0).or_default().push(edge); + } + let mut inputs = BTreeMap::>::new(); + for (consumer, edges) in incoming { + let ordered = if nodes + .get(&consumer) + .is_some_and(|node| matches!(node.payload, ExecutableOperatorPayload::Binary { .. })) + { + // Wire order is not operand order. Preserve noncommutative binary + // semantics even when a valid transport reorders its edge list. + let left = edges + .iter() + .filter(|edge| edge.role == EdgeRole::Left) + .collect::>(); + let right = edges + .iter() + .filter(|edge| edge.role == EdgeRole::Right) + .collect::>(); + if edges.len() != 2 || left.len() != 1 || right.len() != 1 { + return Err(ScheduleError::Invalid( + "binary maintenance input roles must be exactly Left and Right".into(), + )); + } + vec![left[0].producer.0, right[0].producer.0] + } else { + edges.iter().map(|edge| edge.producer.0).collect() + }; + inputs.insert(consumer, ordered); } let mut active = BTreeSet::new(); let mut values = BTreeMap::>::new(); @@ -272,6 +298,66 @@ mod tests { } } + #[test] + fn binary_operand_roles_survive_edge_reordering_and_reject_duplicates() { + use planner_types::post_asap::BinaryOperator; + use planner_types::pre_asap::{ArithmeticOpKind, BinaryOpKind}; + struct Subtract; + impl PrecomputeOperatorRegistry for Subtract { + type Error = String; + fn execute( + &self, + node: &ExecutableDagNode, + inputs: &[Arc], + ) -> Result { + match node.id.0 { + 0 => Ok(10), + 1 => Ok(3), + 3 => Ok(*inputs[0] - *inputs[1]), + _ => Err("unexpected node".into()), + } + } + } + let mut binary = node(3); + binary.operator = ExecutableOperator::Binary; + binary.payload = ExecutableOperatorPayload::Binary { + operator: BinaryOperator { + kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Sub), + vector_match: None, + }, + }; + let mut left = edge(0, 3); + left.role = EdgeRole::Left; + let mut right = edge(1, 3); + right.role = EdgeRole::Right; + let mut dag = ExecutableDag { + nodes: vec![node(0), node(1), node(2), binary, { + let mut query = node(4); + query.output_state = ExecutionDataState::READ_ROWS; + query + }], + edges: vec![right, left], + root: PostAsapNodeId(3), + }; + let execute = |dag: &ExecutableDag| { + execute_precompute_sink( + dag, + &binding(), + PostAsapNodeId(3), + key(3), + &Subtract, + &Sink::default(), + ) + }; + assert_eq!(*execute(&dag).unwrap(), 7); + dag.edges.reverse(); + assert_eq!(*execute(&dag).unwrap(), 7); + dag.edges[1].role = EdgeRole::Left; + assert!(matches!(execute(&dag), Err(ScheduleError::Invalid(_)))); + dag.edges.pop(); + assert!(matches!(execute(&dag), Err(ScheduleError::Invalid(_)))); + } + #[test] fn shared_dependency_executes_once_and_replay_reads_committed_value() { // 0 is shared by 1 and 2; 3 consumes both branches. From a38262e91db901677e247b680a7c657314fb93a6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:41:35 -0600 Subject: [PATCH 11/21] Evaluate aligned frozen Float64 rows with explicit timestamp provenance --- .../precompute_engine/maintenance_runtime.rs | 221 +++++++++++++++--- 1 file changed, 185 insertions(+), 36 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 0cfbcdb7..7e0ec79d 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -29,6 +29,7 @@ enum MaintenanceValue { Rows { values: Vec<(i64, f64)>, name: String, + timestamped: bool, }, } @@ -141,7 +142,7 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { let [value] = inputs else { return Err("maintenance SummaryAgg requires exactly one row input".into()); }; - let MaintenanceValue::Rows { values, name } = value.as_ref() else { + 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 !matches!(&self.inputs, MaintenanceInputs::Frozen(_)) { @@ -263,6 +264,119 @@ fn evaluate_weight( Ok(weight) } +// Rows carry only a value and its window timestamp. Reject any schema that +// would require silently dropping another column or manufacturing a timestamp. +fn maintenance_float64_column( + node: &ExecutableDagNode, +) -> Result<&planner_types::post_asap::SummaryField, String> { + use planner_types::post_asap::SummaryFamilyType; + let fields = &node.output_schema.fields; + let field = match node.output_schema.time_index { + None if fields.len() == 1 => &fields[0], + Some(time_index) if fields.len() == 2 && time_index < 2 => { + let timestamp = &fields[time_index]; + let value = &fields[1 - time_index]; + if timestamp.nullable + || timestamp.name == value.name + || !matches!( + timestamp.dtype, + SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Timestamp) + ) + { + return Err("finalization timestamp column differs from its typed schema".into()); + } + value + } + _ => { + return Err( + "exact maintenance finalization requires one value and optional declared timestamp" + .into(), + ) + } + }; + if field.nullable + || !matches!( + field.dtype, + SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Float64) + ) + { + return Err( + "exact maintenance finalization currently requires a Float64 output column".into(), + ); + } + Ok(field) +} + +fn evaluate_aligned_binary( + node: &ExecutableDagNode, + operator: &planner_types::post_asap::BinaryOperator, + inputs: &[Arc], +) -> Result { + use planner_types::pre_asap::BinaryOpKind; + let BinaryOpKind::Arithmetic(arithmetic) = &operator.kind else { + return Err("maintenance binary currently requires arithmetic".into()); + }; + if operator.vector_match.is_some() { + return Err( + "maintenance binary requires explicit population routing for vector matching".into(), + ); + } + let name = maintenance_float64_column(node)?.name.clone(); + if node.output_schema.time_index.is_none() { + return Err("maintenance binary requires declared window timestamps".into()); + } + let [left, right] = inputs else { + return Err("maintenance binary requires two row inputs".into()); + }; + let ( + MaintenanceValue::Rows { + values: left, + timestamped: true, + .. + }, + MaintenanceValue::Rows { + values: right, + timestamped: true, + .. + }, + ) = (left.as_ref(), right.as_ref()) + else { + return Err("maintenance binary requires finalized row inputs".into()); + }; + if left.is_empty() || left.len() != right.len() { + return Err("maintenance binary requires matching nonempty timestamp sets".into()); + } + // Canonical timestamp maps accept arrival-order differences, but never + // collapse duplicate updates or pair unrelated source windows by position. + let mut left_rows = BTreeMap::new(); + let mut right_rows = BTreeMap::new(); + for (rows, index) in [(left, &mut left_rows), (right, &mut right_rows)] { + for &(timestamp, value) in rows { + if !value.is_finite() || index.insert(timestamp, value).is_some() { + return Err( + "maintenance binary input has duplicate timestamps or non-finite values".into(), + ); + } + } + } + let mut values = Vec::with_capacity(left.len()); + for (timestamp, left) in left_rows { + let right = right_rows + .get(×tamp) + .ok_or("maintenance binary requires matching timestamp sets")?; + let value = crate::utils::arithmetic::evaluate_float64_arithmetic(arithmetic, left, *right); + if !value.is_finite() { + return Err("maintenance binary produced a non-finite update".into()); + } + values.push((timestamp, value)); + } + Ok(MaintenanceValue::Rows { + values, + name, + timestamped: true, + }) +} + fn finalize_exact( node: &ExecutableDagNode, inputs: &[Arc], @@ -303,40 +417,7 @@ fn finalize_exact( ) } }; - let fields = &node.output_schema.fields; - let field = match node.output_schema.time_index { - None if fields.len() == 1 => &fields[0], - Some(time_index) if fields.len() == 2 && time_index < 2 => { - let timestamp = &fields[time_index]; - let value = &fields[1 - time_index]; - if timestamp.nullable - || timestamp.name == value.name - || !matches!( - timestamp.dtype, - SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Timestamp) - ) - { - return Err("finalization timestamp column differs from its typed schema".into()); - } - value - } - _ => { - return Err( - "exact maintenance finalization requires one value and optional declared timestamp" - .into(), - ) - } - }; - if field.nullable - || !matches!( - field.dtype, - SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Float64) - ) - { - return Err( - "exact maintenance finalization currently requires a Float64 output column".into(), - ); - } + let field = maintenance_float64_column(node)?; let values = states .into_iter() .map(|(timestamp, state)| { @@ -352,6 +433,8 @@ fn finalize_exact( Ok(MaintenanceValue::Rows { values, name: field.name.clone(), + timestamped: node.output_schema.time_index.is_some() + && matches!(input.as_ref(), MaintenanceValue::SummaryWindows { .. }), }) } @@ -2416,6 +2499,72 @@ mod tests { persistence.shutdown(); } + #[test] + fn maintenance_binary_aligns_windows_and_rejects_incomplete_or_ambiguous_rows() { + use planner_types::post_asap::{BinaryOperator, SummaryFamilyType, SummaryField}; + use planner_types::pre_asap::{ArithmeticOpKind, BinaryOpKind, DataType}; + let mut operation = node(10); + operation.output_schema.fields = vec![ + SummaryField { + name: "ts".into(), + dtype: SummaryFamilyType::Plain(DataType::Timestamp), + nullable: false, + }, + SummaryField { + name: "value".into(), + dtype: SummaryFamilyType::Plain(DataType::Float64), + nullable: false, + }, + ]; + operation.output_schema.time_index = Some(0); + let mut operator = BinaryOperator { + kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Sub), + vector_match: None, + }; + let rows = |values: Vec<(i64, f64)>| { + Arc::new(MaintenanceValue::Rows { + values, + name: "value".into(), + timestamped: true, + }) + }; + let left = rows(vec![(2_000, 7.0), (1_000, 5.0)]); + let right = rows(vec![(1_000, 2.0), (2_000, 3.0)]); + // Arrival order cannot exchange windows, and subtraction retains edge order. + let MaintenanceValue::Rows { values, .. } = + evaluate_aligned_binary(&operation, &operator, &[left.clone(), right.clone()]).unwrap() + else { + panic!("expected rows") + }; + assert_eq!(values, vec![(1_000, 3.0), (2_000, 4.0)]); + for invalid in [ + rows(vec![]), + rows(vec![(1_000, 2.0)]), + rows(vec![(1_000, 2.0), (3_000, 3.0)]), + rows(vec![(1_000, 2.0), (1_000, 3.0)]), + rows(vec![(1_000, f64::NAN), (2_000, 3.0)]), + Arc::new(MaintenanceValue::Rows { + values: vec![(1_000, 2.0), (2_000, 3.0)], + name: "value".into(), + timestamped: false, + }), + Arc::new(MaintenanceValue::summary(sum(2.0))), + ] { + assert!( + evaluate_aligned_binary(&operation, &operator, &[left.clone(), invalid]).is_err() + ); + } + operator.kind = BinaryOpKind::Arithmetic(ArithmeticOpKind::Div); + assert!(evaluate_aligned_binary( + &operation, + &operator, + &[left.clone(), rows(vec![(1_000, 0.0), (2_000, 3.0)])] + ) + .is_err()); + operation.output_schema.fields[1].dtype = SummaryFamilyType::Plain(DataType::Int64); + assert!(evaluate_aligned_binary(&operation, &operator, &[left, right]).is_err()); + } + #[test] fn finalization_preserves_windows_until_an_explicit_merge() { use planner_types::post_asap::{ExactKind, ExactParams, SummaryFamilyType, SummaryField}; @@ -2480,7 +2629,7 @@ mod tests { }, ]; read.output_schema.time_index = Some(0); - let MaintenanceValue::Rows { values, name } = + let MaintenanceValue::Rows { values, name, .. } = finalize_exact(&read, &[Arc::clone(&input)]).unwrap() else { panic!("expected typed rows") From f39609f31ba1aff5988f23e0b6fc5bb444edcf8e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:43:17 -0600 Subject: [PATCH 12/21] fix: consume explicit binary timing in query compilation --- control_plane/src/physical/compiler.rs | 1 + control_plane/src/query_plan.rs | 21 ++++++++++++++++++--- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index b3788499..9fb0a03e 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -4602,6 +4602,7 @@ mod tests { let selected = request.queries[0].post_asap.clone(); request.queries[0].post_asap = Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { + timing: planner_types::post_asap::ExecutionTiming::ReadTime, lhs: selected.clone(), rhs: selected.clone(), operator: planner_types::post_asap::BinaryOperator { diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 34fcdbae..475e319a 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -606,7 +606,12 @@ where completeness: completeness.clone(), } } - SummaryExpr::BinaryOp { lhs, rhs, operator } if self.logical_source.is_some() => { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + } if self.logical_source.is_some() => { let operator = logical::binary_operator(operator)?; QueryPlanNode::Logical { operator, @@ -680,7 +685,12 @@ where inputs: vec![self.lower(child)?], } } - SummaryExpr::BinaryOp { lhs, rhs, operator } if exact_value_executable(node) => { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + } if exact_value_executable(node) => { let planner_types::pre_asap::BinaryOpKind::Arithmetic(operator) = &operator.kind else { unreachable!() @@ -931,7 +941,12 @@ pub(crate) fn exact_value_executable(node: &SummaryNode) -> bool { exact_accumulator_value_source(node).is_some_and(exact_value_executable) } SummaryExpr::KeepPreAsap(expr) => scalar_literal(expr).is_some(), - SummaryExpr::BinaryOp { lhs, rhs, operator } => { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + } => { matches!( operator.kind, planner_types::pre_asap::BinaryOpKind::Arithmetic(_) From 91402c808f375fa6b0ae816b358cf3c28b71ccea Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:43:17 -0600 Subject: [PATCH 13/21] fix: consume explicit binary timing in query compilation --- control_plane/src/physical/compiler.rs | 1 + control_plane/src/query_plan.rs | 21 ++++++++++++++++++--- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index b3788499..9fb0a03e 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -4602,6 +4602,7 @@ mod tests { let selected = request.queries[0].post_asap.clone(); request.queries[0].post_asap = Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { + timing: planner_types::post_asap::ExecutionTiming::ReadTime, lhs: selected.clone(), rhs: selected.clone(), operator: planner_types::post_asap::BinaryOperator { diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 34fcdbae..475e319a 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -606,7 +606,12 @@ where completeness: completeness.clone(), } } - SummaryExpr::BinaryOp { lhs, rhs, operator } if self.logical_source.is_some() => { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + } if self.logical_source.is_some() => { let operator = logical::binary_operator(operator)?; QueryPlanNode::Logical { operator, @@ -680,7 +685,12 @@ where inputs: vec![self.lower(child)?], } } - SummaryExpr::BinaryOp { lhs, rhs, operator } if exact_value_executable(node) => { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + } if exact_value_executable(node) => { let planner_types::pre_asap::BinaryOpKind::Arithmetic(operator) = &operator.kind else { unreachable!() @@ -931,7 +941,12 @@ pub(crate) fn exact_value_executable(node: &SummaryNode) -> bool { exact_accumulator_value_source(node).is_some_and(exact_value_executable) } SummaryExpr::KeepPreAsap(expr) => scalar_literal(expr).is_some(), - SummaryExpr::BinaryOp { lhs, rhs, operator } => { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + } => { matches!( operator.kind, planner_types::pre_asap::BinaryOpKind::Arithmetic(_) From 37e24641bea174a3d62cb755cb419f1ad3b8c8d6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:47:29 -0600 Subject: [PATCH 14/21] Execute explicitly timed binary maintenance over frozen rows --- Cargo.lock | 10 ++-- control_plane/Cargo.toml | 8 +-- crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 6 +-- .../precompute_engine/maintenance_runtime.rs | 51 +++++++++++++++++++ .../src/precompute_engine/subdag_scheduler.rs | 8 +++ .../asap_query_engine/summary_exec.rs | 1 + 7 files changed, 73 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 532bb94d..776d32ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,7 +373,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=e78572c57e6c3eea2a9017f968fbc65a9196d56e#e78572c57e6c3eea2a9017f968fbc65a9196d56e" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21#1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" dependencies = [ "asap-types", "serde", @@ -384,7 +384,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=e78572c57e6c3eea2a9017f968fbc65a9196d56e#e78572c57e6c3eea2a9017f968fbc65a9196d56e" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21#1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" dependencies = [ "asap-types", "promql-parser", @@ -393,7 +393,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=e78572c57e6c3eea2a9017f968fbc65a9196d56e#e78572c57e6c3eea2a9017f968fbc65a9196d56e" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21#1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -415,12 +415,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=e78572c57e6c3eea2a9017f968fbc65a9196d56e#e78572c57e6c3eea2a9017f968fbc65a9196d56e" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21#1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=e78572c57e6c3eea2a9017f968fbc65a9196d56e#e78572c57e6c3eea2a9017f968fbc65a9196d56e" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21#1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 6f928922..eb0032b1 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -76,8 +76,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -85,8 +85,8 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } -asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } +asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index b7f65f90..b55a8e9e 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -34,4 +34,4 @@ sha2 = "0.10" # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index bc967e4f..9c9b1de0 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -39,8 +39,8 @@ sha2 = "0.10" # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } # Shared external (workspace) serde.workspace = true @@ -133,7 +133,7 @@ fs2 = "0.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "e78572c57e6c3eea2a9017f968fbc65a9196d56e" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "1149b4e40e2caeddbd107eb5e1c7280f2e6dfc21" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 7e0ec79d..2b322ef2 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -126,6 +126,19 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { ) -> Result { match &node.payload { ExecutableOperatorPayload::SummaryMerge => merge_inputs(inputs), + ExecutableOperatorPayload::Binary { + operator, + timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, + } => { + if !matches!(&self.inputs, MaintenanceInputs::Frozen(_)) + || node.output_state + != planner_types::post_asap::ExecutionDataState::MAINTENANCE_ROWS + { + return Err("maintenance binary requires immutable completed row inputs".into()); + } + evaluate_aligned_binary(node, operator, inputs) + } + ExecutableOperatorPayload::Value { operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, @@ -2537,6 +2550,44 @@ mod tests { panic!("expected rows") }; assert_eq!(values, vec![(1_000, 3.0), (2_000, 4.0)]); + let binding = BackendExecutableBinding { + nodes: BTreeMap::new(), + query_sink: PostAsapNodeId(10), + query_plan_sink: asap_types::query_plan::QueryNodeId(10), + precompute_sinks: vec![], + }; + let frozen = OperatorAdapter { + binding: &binding, + inputs: MaintenanceInputs::Frozen(&[]), + configs: &[], + }; + operation.payload = ExecutableOperatorPayload::Binary { + operator: operator.clone(), + timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, + }; + operation.output_state = planner_types::post_asap::ExecutionDataState::MAINTENANCE_ROWS; + assert!(frozen + .execute(&operation, &[left.clone(), right.clone()]) + .is_ok()); + let live = OperatorAdapter { + binding: &binding, + inputs: MaintenanceInputs::Live { + definition: definition(1), + state: sum(1.0), + }, + configs: &[], + }; + assert!(live + .execute(&operation, &[left.clone(), right.clone()]) + .is_err()); + operation.payload = ExecutableOperatorPayload::Binary { + operator: operator.clone(), + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + }; + assert!(frozen + .execute(&operation, &[left.clone(), right.clone()]) + .is_err()); + for invalid in [ rows(vec![]), rows(vec![(1_000, 2.0)]), diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index f3ba0840..69d67c40 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -97,6 +97,11 @@ where for edge in &dag.edges { incoming.entry(edge.consumer.0).or_default().push(edge); } + for node in &dag.nodes { + if matches!(node.payload, ExecutableOperatorPayload::Binary { .. }) { + incoming.entry(node.id.0).or_default(); + } + } let mut inputs = BTreeMap::>::new(); for (consumer, edges) in incoming { let ordered = if nodes @@ -321,6 +326,7 @@ mod tests { let mut binary = node(3); binary.operator = ExecutableOperator::Binary; binary.payload = ExecutableOperatorPayload::Binary { + timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, operator: BinaryOperator { kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Sub), vector_match: None, @@ -356,6 +362,8 @@ mod tests { assert!(matches!(execute(&dag), Err(ScheduleError::Invalid(_)))); dag.edges.pop(); assert!(matches!(execute(&dag), Err(ScheduleError::Invalid(_)))); + dag.edges.clear(); + assert!(matches!(execute(&dag), Err(ScheduleError::Invalid(_)))); } #[test] diff --git a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs index 4ab734ea..672363d1 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs @@ -544,6 +544,7 @@ mod tests { let child = logical_node(); let tree = SummaryNode { expr: SummaryExpr::BinaryOp { + timing: planner_types::post_asap::ExecutionTiming::ReadTime, lhs: child.clone(), rhs: child.clone(), operator: planner_types::post_asap::BinaryOperator { From dfcd7b198a81700b50c08163531b16e6b5311853 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:59:16 -0600 Subject: [PATCH 15/21] test: preserve read-time binary evidence fixture --- control_plane/tests/offline_evidence.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/control_plane/tests/offline_evidence.rs b/control_plane/tests/offline_evidence.rs index b36bd3fd..0c4a936b 100644 --- a/control_plane/tests/offline_evidence.rs +++ b/control_plane/tests/offline_evidence.rs @@ -351,6 +351,7 @@ fn binary_summary_has_explicit_warm_tier_fallback() { let child = bound(&model()); let root = std::rc::Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { + timing: planner_types::post_asap::ExecutionTiming::ReadTime, lhs: child.clone(), rhs: child.clone(), operator: BinaryOperator { From 640799263353b8a4b3d6889b842d10362d344e34 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 13:01:59 -0600 Subject: [PATCH 16/21] Schedule complete aligned source cohorts under the captured catalog generation --- .../precompute_engine/maintenance_runtime.rs | 306 ++++++++++++++++-- 1 file changed, 287 insertions(+), 19 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 2b322ef2..c6db71c8 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -738,8 +738,12 @@ pub fn execute_completed_maintenance( if derived.inputs.len() != 1 { return Err("maintenance execution requires synchronized multi-source scheduling".into()); } + let generation = store + .active_catalog_generation() + .ok_or("maintenance requires an authoritative catalog")?; execute_completed_maintenance_cohort( store, + &generation, installed, configs, sink, @@ -755,6 +759,7 @@ pub fn execute_completed_maintenance( #[allow(clippy::too_many_arguments)] pub(crate) fn execute_completed_maintenance_cohort( store: &crate::storage_engines::sketch_db::index::SketchStore, + generation: &Arc, installed: &asap_types::executable_plan::InstalledPostAsapDag, configs: &[asap_types::PrecomputeMaterialization], sink: PostAsapNodeId, @@ -829,10 +834,7 @@ pub(crate) fn execute_completed_maintenance_cohort( .collect(); requests.push((source_sid, source, expected, group.clone())); } - let generation = store - .active_catalog_generation() - .ok_or("maintenance requires an authoritative catalog")?; - let cohort = store.read_frozen_exact_cohort(&generation, &derived.inputs, &requests)?; + let cohort = store.read_frozen_exact_cohort(generation, &derived.inputs, &requests)?; if cohort.len() > 1 && cohort .iter() @@ -892,7 +894,7 @@ pub(crate) fn execute_completed_maintenance_cohort( }), target.fingerprint(), ); - output.catalog_generation = Some(generation); + output.catalog_generation = Some(Arc::clone(generation)); store.publish_frozen_maintenance_output( target_sid, target_config, @@ -903,6 +905,165 @@ pub(crate) fn execute_completed_maintenance_cohort( ) } +/// Schedule a complete common population across all raw input definitions. +/// Missing windows are unavailable; they are never interpreted as zero values. +fn execute_finite_source_cohort( + store: &crate::storage_engines::sketch_db::index::SketchStore, + resolver: &crate::drivers::ingest::series_resolver::SeriesIdResolver, + plan: &asap_types::precompute_plan::PrecomputePlan, + installed: &asap_types::executable_plan::InstalledPostAsapDag, + sink: PostAsapNodeId, +) -> Result<(), String> { + let generation = plan + .summary_catalog + .as_ref() + .ok_or("finite maintenance requires a catalog generation")?; + let Some(BackendNodeBinding::Materialization { + summary_definition: target, + }) = installed.binding.node(sink) + else { + return Err("finite maintenance sink has no installed definition".into()); + }; + let config = plan + .materializations + .iter() + .find(|config| config.policy_fingerprint() == target.fingerprint()) + .ok_or("finite maintenance target configuration is absent")?; + let derived = config + .derived_input + .as_ref() + .ok_or("finite maintenance target has no input program")?; + let source_configs = derived + .inputs + .iter() + .map(|source| { + plan.materializations + .iter() + .find(|config| config.policy_fingerprint() == source.fingerprint()) + .ok_or("finite maintenance source configuration is absent") + }) + .collect::, _>>()?; + asap_types::precompute_plan::validated_source_window_cohort(config, &source_configs) + .map_err(|error| error.to_string())?; + let width = config.stored_window_ms(); + if config.slide_interval.checked_mul(1000) != Some(width) { + return Err("finite source cohorts currently require non-overlapping full windows".into()); + } + let active_generation = store + .active_catalog_generation() + .ok_or("finite source cohort requires an active catalog")?; + if active_generation.as_ref() != generation { + return Err("finite source cohort catalog generation changed".into()); + } + let mut source_sids = BTreeMap::new(); + let mut common_group = None; + let mut common_windows: Option> = None; + for source in &derived.inputs { + let populations = store.completed_maintenance_coordinates(*source, generation)?; + if populations.is_empty() { + return Ok(()); + } + if populations.len() != 1 || populations.values().any(|groups| groups.len() != 1) { + return Err( + "finite source cohort requires one physical population per definition".into(), + ); + } + let (sid, groups) = populations.into_iter().next().unwrap(); + let (group, windows) = groups.into_iter().next().unwrap(); + if common_group + .as_ref() + .is_some_and(|expected| expected != &group) + { + return Err("finite source cohort requires matching explicit populations".into()); + } + if windows + .iter() + .any(|(start, end)| end.checked_sub(*start) != Some(width)) + { + return Err("finite source cohort contains a non-full source window".into()); + } + common_group = Some(group); + source_sids.insert(*source, sid); + match &mut common_windows { + Some(common) => common.retain(|window| windows.contains(window)), + None => common_windows = Some(windows), + } + } + let Some(group) = common_group else { + return Ok(()); + }; + let output_group: BTreeMap<_, _> = config + .grouping_labels + .iter() + .map(|key| { + group + .get(key) + .cloned() + .map(|value| (key.clone(), value)) + .ok_or("finite maintenance output grouping key is absent") + }) + .collect::>()?; + let existing = store.completed_maintenance_coordinates(*target, generation)?; + for window in common_windows.unwrap_or_default() { + if (window.0 as i128 - config.pane_origin_ms.unwrap_or(0) as i128).rem_euclid(width as i128) + != 0 + { + continue; + } + // All definitions and populations are proven before resolving any + // target. The publication transaction revalidates every lifetime. + let requests = source_sids + .iter() + .map(|(definition, sid)| (*sid, *definition, BTreeSet::from([window]), group.clone())) + .collect::>(); + let cohort = + store.read_frozen_exact_cohort(&active_generation, &derived.inputs, &requests)?; + if cohort + .iter() + .any(|input| !input.singleton_population_complete) + { + return Err("finite source cohort has incomplete population proof".into()); + } + let pairs = output_group + .iter() + .map(|(key, value)| (key.as_str(), value.as_str())) + .collect::>(); + let attrs = crate::drivers::ingest::canonical_attrs_fingerprint(&pairs); + let kind = crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); + let target_sid = + resolver.resolve_with_reactivation(&config.metric, &attrs, &kind, |sid| { + store.validate_routed_catalog_generation(Some(generation))?; + let activation = store.authorize_series_reactivation(sid, *target)?; + if activation + .as_deref() + .is_some_and(|actual| actual != generation) + { + return Err("finite maintenance generation changed".into()); + } + Ok(activation) + })?; + if existing + .get(&target_sid) + .and_then(|groups| groups.get(&output_group)) + .is_some_and(|windows| windows.contains(&window)) + { + continue; + } + execute_completed_maintenance_cohort( + store, + &active_generation, + installed, + &plan.materializations, + sink, + &source_sids, + target_sid, + window, + &group, + )?; + } + Ok(()) +} + /// Schedule retained, aligned completed windows from the installed DAG after /// the finite source barrier. Missing panes remain unavailable to query reads. pub(crate) fn execute_finite_maintenance( @@ -914,6 +1075,12 @@ pub(crate) fn execute_finite_maintenance( .summary_catalog .as_ref() .ok_or("finite maintenance requires a catalog generation")?; + let active_generation = store + .active_catalog_generation() + .ok_or("finite maintenance requires an active catalog")?; + if active_generation.as_ref() != generation { + return Err("finite maintenance catalog generation changed".into()); + } for installed in plan.executable_dags.values() { for sink in &installed.binding.precompute_sinks { let Some(BackendNodeBinding::Materialization { @@ -930,8 +1097,12 @@ pub(crate) fn execute_finite_maintenance( let Some(derived) = &config.derived_input else { continue; }; - if derived.inputs.len() != 1 { - return Err("finite maintenance requires one source definition".into()); + if derived.inputs.len() > 1 { + execute_finite_source_cohort(store, resolver, plan, installed, *sink)?; + continue; + } + if derived.inputs.is_empty() { + return Err("finite maintenance requires a source definition".into()); } let source = *derived.inputs.first().unwrap(); let source_config = plan @@ -1017,12 +1188,13 @@ pub(crate) fn execute_finite_maintenance( { continue; } - execute_completed_maintenance( - &store, + execute_completed_maintenance_cohort( + store, + &active_generation, installed, &plan.materializations, *sink, - source_sid, + &BTreeMap::from([(source, source_sid)]), target_sid, (*start, end), &group, @@ -2259,7 +2431,8 @@ mod tests { .series_ids_for_policy(durable_configs[1].policy_fingerprint()) .is_empty()); persistence.shutdown(); - exercise_two_source_completed_sink(&dag, &configs, &scheduled_binding); + exercise_two_source_completed_sink(&dag, &configs, &scheduled_binding, true); + exercise_two_source_completed_sink(&dag, &configs, &scheduled_binding, false); assert!(evaluate_weight( &SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::Named("missing".into())), 7.0, @@ -2272,6 +2445,7 @@ mod tests { template: &ExecutableDag, configs: &[asap_types::PrecomputeMaterialization], binding: &BackendExecutableBinding, + matching_windows: bool, ) { // A bound operator fixture, not a claim that a frontend selected this // composition. Both actual durable sources are required before output. @@ -2371,20 +2545,53 @@ mod tests { let mut persistence = store.start_persistence(persistence_config()).unwrap(); let generation = store.active_catalog_generation().unwrap(); let sources = BTreeMap::from([(first_id, 800), (second_id, 801)]); + // A bound scheduler fixture; actual frontend selection is tested separately. + let mut plan = asap_types::precompute_plan::PrecomputePlan::build_backend_local( + asap_types::precompute_plan::PlanEnvelope { + plan_id: 2, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: asap_types::precompute_plan::BACKEND_COMPAT.into(), + planner_revision: "scheduler-fixture".into(), + capability_snapshot_id: "scheduler-fixture".into(), + }, + configs[..2].to_vec(), + ) + .unwrap(); + plan.materializations = configs.to_vec(); + plan.summary_catalog = Some(generation.as_ref().clone()); + plan.executable_dags = BTreeMap::from([("cohort-fixture".into(), installed.clone())]); + let resolver_path = directory.path().join("resolver.wal"); + let resolver = + crate::drivers::ingest::series_resolver::SeriesIdResolver::open(resolver_path.clone()) + .unwrap(); + for (index, value) in [2.0, 7.0].into_iter().enumerate() { let config = &configs[index]; + let start = if !matching_windows && index == 1 { + 2000 + } else { + 0 + }; let coordinate = asap_types::sds::SummaryInstanceCoordinates { summary_definition_id: config.policy_fingerprint().into(), time_range: asap_types::sds::HalfOpenTimeRange { - start_ms: 0, - end_ms: 2000, + start_ms: start, + end_ms: start + 2000, }, group_values: BTreeMap::new(), }; let revision = store .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()])) .unwrap(); - let mut output = PrecomputedOutput::new(0, 2000, None, config.policy_fingerprint()); + let mut output = PrecomputedOutput::new( + start as u64, + (start + 2000) as u64, + None, + config.policy_fingerprint(), + ); output.catalog_generation = Some(Arc::clone(&generation)); store .publish_admitted_summary_update( @@ -2406,11 +2613,12 @@ mod tests { } assert!(execute_completed_maintenance_cohort( &store, + &generation, &installed, &configs, PostAsapNodeId(3), &sources, - 802, + 1, (0, 2000), &BTreeMap::new() ) @@ -2423,6 +2631,28 @@ mod tests { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); } + let source_parts = persistence.manifest.live_parts().len(); + execute_finite_maintenance(&store, &resolver, &plan).unwrap(); + if !matching_windows { + // Complete inputs at different windows cannot create any target. + assert!(store + .series_ids_for_policy(configs[2].policy_fingerprint()) + .is_empty()); + assert_eq!(persistence.manifest.live_parts().len(), source_parts); + assert!(persistence + .flusher + .metadata_store() + .load_strict() + .unwrap() + .iter() + .all(|record| record.sid != 1)); + persistence.shutdown(); + return; + } + assert_eq!( + store.series_ids_for_policy(configs[2].policy_fingerprint()), + vec![1] + ); let requests = sources .iter() .map(|(definition, sid)| { @@ -2468,13 +2698,14 @@ mod tests { .unwrap(), 9.0 ); - assert!(execute_completed_maintenance_cohort( + assert!(!execute_completed_maintenance_cohort( &store, + &generation, &installed, &configs, PostAsapNodeId(3), &sources, - 802, + 1, (0, 2000), &BTreeMap::new() ) @@ -2482,11 +2713,12 @@ mod tests { let parts = persistence.manifest.live_parts().len(); assert!(!execute_completed_maintenance_cohort( &store, + &generation, &installed, &configs, PostAsapNodeId(3), &sources, - 802, + 1, (0, 2000), &BTreeMap::new() ) @@ -2497,18 +2729,54 @@ mod tests { let restored = Arc::new(SketchStore::new()); restored.install_summary_catalog(catalog).unwrap(); let mut persistence = restored.start_persistence(persistence_config()).unwrap(); + drop(resolver); + let resolver = + crate::drivers::ingest::series_resolver::SeriesIdResolver::open(resolver_path).unwrap(); + execute_finite_maintenance(&restored, &resolver, &plan).unwrap(); + assert!(!execute_completed_maintenance_cohort( &restored, + &generation, &installed, &configs, PostAsapNodeId(3), &sources, - 802, + 1, (0, 2000), &BTreeMap::new() ) .unwrap()); assert_eq!(persistence.manifest.live_parts().len(), parts); + let next_catalog = Arc::new( + asap_types::summary_catalog::SummaryCatalog::from_materializations(2, 2, &configs) + .unwrap(), + ); + restored.install_summary_catalog(next_catalog).unwrap(); + assert_ne!(generation, restored.active_catalog_generation().unwrap()); + // Stale scheduling fails at the captured catalog boundary, before any + // new generation's completion state or payload can be substituted. + let stale = execute_completed_maintenance_cohort( + &restored, + &generation, + &installed, + &configs, + PostAsapNodeId(3), + &sources, + 2, + (0, 2000), + &BTreeMap::new(), + ) + .unwrap_err(); + assert!(stale.contains("stale producer"), "{stale}"); + assert!(execute_finite_maintenance(&restored, &resolver, &plan).is_err()); + assert_eq!(persistence.manifest.live_parts().len(), parts); + assert!(persistence + .flusher + .metadata_store() + .load_strict() + .unwrap() + .iter() + .all(|record| record.sid != 2)); persistence.shutdown(); } From 6dc0567cb54ed04fd36b764b47f99e5c8b0eb594 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 13:04:41 -0600 Subject: [PATCH 17/21] fix: retain binary timing in calibration candidate exports --- control_plane/examples/calibration_candidates.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/control_plane/examples/calibration_candidates.rs b/control_plane/examples/calibration_candidates.rs index 3b119bb7..65fcb136 100644 --- a/control_plane/examples/calibration_candidates.rs +++ b/control_plane/examples/calibration_candidates.rs @@ -27,10 +27,15 @@ fn planner_forest(queries: &[control_plane::physical::compiler::PlanningQuery]) vec![], json!({"query_expr_debug":format!("{expr:#?}")}), ), - SummaryExpr::BinaryOp { lhs, rhs, operator } => ( + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing, + } => ( "BinaryOp", vec![lhs, rhs], - json!({"operator_debug":format!("{operator:?}")}), + json!({"operator_debug":format!("{operator:?}"),"timing_debug":format!("{timing:?}")}), ), SummaryExpr::CandidateTopK { candidates, From 8ec8b4192743453915031d0775ee061ee1c1c8d6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 13:08:29 -0600 Subject: [PATCH 18/21] feat: bind actual finite multi-source maintenance plans --- control_plane/src/physical/compiler.rs | 169 +++++++++++++----- crates/asap_types/src/precompute_plan.rs | 118 +++++++++--- .../support/immutable_maintenance_process.rs | 61 +++++-- 3 files changed, 259 insertions(+), 89 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 9fb0a03e..da414638 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1204,7 +1204,7 @@ impl PhysicalCompiler { query_id: query.query_id.clone(), reason, })?; - if let Some(source) = immutable_materialization_source(&selected.node) { + if let Some(source) = immutable_materialization_sources(&selected.node) { if environment.target != PhysicalDeploymentTarget::BackendLocalRemoteWrite { return Err(CompileError::Query { query_id: query.query_id.clone(), @@ -1213,24 +1213,24 @@ impl PhysicalCompiler { }); } let compiled = executable_dags[query_index].as_ref().expect("compiled DAG"); - let source_node = - compiled - .node_ids - .node_id(&source) - .ok_or_else(|| CompileError::Query { + let mut frontiers = BTreeMap::new(); + let mut source_configs = Vec::new(); + for source in source { + let source_node = compiled.node_ids.node_id(&source).ok_or_else(|| { + CompileError::Query { query_id: query.query_id.clone(), reason: "derived source absent from selected DAG".into(), - })?; - let source_id = node_bindings - .get(&(query_index, source_node)) - .copied() - .ok_or_else(|| CompileError::Query { - query_id: query.query_id.clone(), - reason: "derived input source was not installed before its consumer" - .into(), + } })?; - let source_config: &asap_types::PrecomputeMaterialization = - compiled_materializations + let source_id = node_bindings + .get(&(query_index, source_node)) + .copied() + .ok_or_else(|| CompileError::Query { + query_id: query.query_id.clone(), + reason: "derived source was not installed before consumer".into(), + })?; + frontiers.insert(source_node, source_id.into()); + let config = compiled_materializations .iter() .find(|config: &&asap_types::PrecomputeMaterialization| { config.policy_fingerprint() == source_id @@ -1239,9 +1239,17 @@ impl PhysicalCompiler { query_id: query.query_id.clone(), reason: "derived source config missing".into(), })?; + if !source_configs.iter().any( + |existing: &&asap_types::PrecomputeMaterialization| { + existing.policy_fingerprint() == source_id + }, + ) { + source_configs.push(config); + } + } asap_types::precompute_plan::validated_source_window_cohort( &runtime_materialization, - &[source_config], + &source_configs, ) .map_err(|error| CompileError::Query { query_id: query.query_id.clone(), @@ -1269,9 +1277,7 @@ impl PhysicalCompiler { })?; runtime_materialization.derived_input = Some( asap_types::derived_input::DerivedInputIdentity::from_dag( - &document, - input_node, - &BTreeMap::from([(source_node, source_id.into())]), + &document, input_node, &frontiers, ) .map_err(|reason| CompileError::Query { query_id: query.query_id.clone(), @@ -1435,11 +1441,13 @@ impl PhysicalCompiler { reason: format!("promql-compatible identity: {error}"), })?; let binding = |node: &Rc, node_family: &SummaryFamilyType| -> Result { - summary_agg_metric(node).ok_or_else(|| { - crate::query_plan::QueryPlanError::Invalid( - "materialized node has no unique time-series source".into(), - ) - })?; + if immutable_materialization_sources(node).is_none() { + summary_agg_metric(node).ok_or_else(|| { + crate::query_plan::QueryPlanError::Invalid( + "materialized node has no unique time-series source".into(), + ) + })?; + } let node_id = executable_dags[query_index] .as_ref() .ok_or_else(|| { @@ -2585,7 +2593,7 @@ fn raw_time_series_input_contract( /// The first immutable-input capability accepts one exact accumulator readout. /// Population/window closure is checked by the installed runtime, not inferred /// from the presence of this syntax. -fn immutable_materialization_source(node: &SummaryNode) -> Option> { +fn immutable_materialization_sources(node: &SummaryNode) -> Option>> { use planner_types::post_asap::{ExactKind, ExecutionTiming, SummaryInputExpr}; let SummaryExpr::SummaryAgg { child, @@ -2607,27 +2615,46 @@ fn immutable_materialization_source(node: &SummaryNode) -> Option, sources: &mut Vec>) -> Option<()> { + match &node.expr { + SummaryExpr::BinaryOp { + lhs, + rhs, + operator, + timing: ExecutionTiming::MaintenanceTime, + } if operator.vector_match.is_none() + && matches!( + operator.kind, + planner_types::pre_asap::BinaryOpKind::Arithmetic(_) + ) => + { + collect(lhs, sources)?; + collect(rhs, sources)?; + } + SummaryExpr::ValueOperation { + child: source, + operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, + timing: ExecutionTiming::MaintenanceTime, + } if matches!(&source.expr, + SummaryExpr::SummaryAgg { family: SummaryFamilyType::ExactAggregate(ExactKind::Sum | ExactKind::Count, _), child, .. } + if matches!(child.expr, SummaryExpr::KeepPreAsap(_))) => + { + if !sources.iter().any(|old| Rc::ptr_eq(old, source)) { + sources.push(source.clone()); + } + } + _ => return None, + } + Some(()) } - Some(Rc::clone(source)) + let mut sources = Vec::new(); + collect(child, &mut sources)?; + Some(sources) } fn selected_input_contract(node: &SummaryNode) -> Result<(String, Option, String), String> { - if let Some(source) = immutable_materialization_source(node) { - materialization_leaf_contract(&source) + if let Some(source) = immutable_materialization_sources(node) { + materialization_leaf_contract(&source[0]) } else { materialization_leaf_contract(node) } @@ -2778,7 +2805,7 @@ fn materialization_consumers( &state.node, )?; let program = - immutable_materialization_source(&state.node).map(|_| Rc::as_ptr(&state.node)); + immutable_materialization_sources(&state.node).map(|_| Rc::as_ptr(&state.node)); if let Some(previous) = cohort_programs.insert(config.policy_fingerprint(), program) { if previous != program && (previous.is_some() || program.is_some()) { return Err(CompileError::Query { @@ -2894,8 +2921,10 @@ fn collect_selected_materializations( } else { None }; - if let Some(source) = immutable_materialization_source(node) { - walk(&source, None, composable, None, selected)?; + if let Some(source) = immutable_materialization_sources(node) { + for source in source { + walk(&source, None, composable, None, selected)?; + } } match &node.expr { SummaryExpr::CandidateTopK { @@ -2936,7 +2965,7 @@ fn collect_selected_materializations( } SummaryExpr::SummaryAgg { child, .. } if !matches!(child.expr, SummaryExpr::KeepPreAsap(_)) - && immutable_materialization_source(node).is_none() => {} + && immutable_materialization_sources(node).is_none() => {} SummaryExpr::SummaryAgg { family: SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Count, _), @@ -3714,8 +3743,8 @@ mod tests { let states = collect_selected_materializations(&workload.queries[0].post_asap, true).unwrap(); assert_eq!(states.len(), 2, "source and consumer must both be selected"); - assert!(immutable_materialization_source(&states[0].node).is_none()); - assert!(immutable_materialization_source(&states[1].node).is_some()); + assert!(immutable_materialization_sources(&states[0].node).is_none()); + assert!(immutable_materialization_sources(&states[1].node).is_some()); let mut deployment = environment(10_000); deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; deployment.collector_ids.clear(); @@ -3747,6 +3776,48 @@ mod tests { .values() .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. }))); } + #[test] + fn immutable_two_sources_keep_actual_frontiers_and_bindings() { + let mut workload = request( + "nested", + "quantile(0.9, sum_over_time(m[1m]) + sum_over_time(n[1m]))", + ); + workload.hybrid_execution = true; + let states = + collect_selected_materializations(&workload.queries[0].post_asap, true).unwrap(); + assert_eq!(states.len(), 3, "source and consumer must both be selected"); + assert!(immutable_materialization_sources(&states[0].node).is_none()); + assert!(immutable_materialization_sources(&states[2].node).is_some()); + let mut deployment = environment(10_000); + deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; + deployment.collector_ids.clear(); + let plan = PhysicalCompiler.compile(workload, deployment).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 3); + let derived = plan + .precompute_plan + .materializations + .iter() + .find(|m| m.derived_input.is_some()) + .unwrap(); + let sources = plan + .precompute_plan + .materializations + .iter() + .filter(|m| m.derived_input.is_none()) + .map(|m| m.policy_fingerprint().into()) + .collect::>(); + assert_eq!(derived.derived_input.as_ref().unwrap().inputs, sources); + assert_eq!(sources.len(), 2); + let entry = plan.query_plan.entries.values().next().unwrap(); + assert!(entry.nodes.values().any(|node| matches!( + node, + crate::query_plan::QueryPlanNode::ReadMaterialization { .. } + ))); + assert!(!entry + .nodes + .values() + .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. }))); + } /// Distinct range queries retain a per-series HLL selected by Planner. #[test] diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 818195a3..ef4621ba 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -420,21 +420,30 @@ impl PrecomputePlan { ) }; if self.ingest.protocol != IngestProtocol::PrometheusRemoteWriteV1 - || derived.inputs.len() != 1 + || derived.inputs.is_empty() { return Err(invalid()); } - let source_id = *derived.inputs.first().unwrap(); - let source = self - .materializations + let sources = derived + .inputs .iter() - .find(|candidate| candidate.policy_fingerprint() == source_id.fingerprint()) - .ok_or_else(invalid)?; - validated_source_window_cohort(config, &[source])?; - // Current installed runtime capability remains nonoverlapping. - // The shared cohort contract also describes explicit full-window - // sliding for consumers which separately prove its completion. - if source.window_size != source.slide_interval { + .map(|id| { + self.materializations + .iter() + .find(|candidate| candidate.policy_fingerprint() == id.fingerprint()) + .ok_or_else(invalid) + }) + .collect::, _>>()?; + validated_source_window_cohort(config, &sources)?; + if sources.iter().any(|source| { + !matches!( + source.aggregation_type, + crate::AggregationType::Sum + ) + }) { + return Err(invalid()); + } + if config.window_size != config.slide_interval { return Err(invalid()); } let mut matched = false; @@ -458,10 +467,19 @@ impl PrecomputePlan { let [edge] = inputs.as_slice() else { return Err(invalid()); }; - let frontiers = installed.binding.nodes.iter().filter_map(|(node,binding)| { - matches!(binding, crate::executable_plan::BackendNodeBinding::Materialization { summary_definition } - if *summary_definition == source_id).then_some((*node,source_id)) - }).collect(); + let frontiers = installed + .binding + .nodes + .iter() + .filter_map(|(node, binding)| match binding { + crate::executable_plan::BackendNodeBinding::Materialization { + summary_definition, + } if derived.inputs.contains(summary_definition) => { + Some((*node, *summary_definition)) + } + _ => None, + }) + .collect(); let actual = crate::derived_input::DerivedInputIdentity::from_dag( &installed.document, edge.producer, @@ -471,20 +489,64 @@ impl PrecomputePlan { if &actual != derived { return Err(invalid()); } - let input = dag - .nodes - .iter() - .find(|node| node.id == edge.producer) - .ok_or_else(invalid)?; - if !matches!( - input.payload, - planner_types::post_asap::ExecutableOperatorPayload::Value { - operation: - planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, - timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, + let mut pending = vec![edge.producer]; + let mut visited = BTreeSet::new(); + while let Some(id) = pending.pop() { + if !visited.insert(id) { + continue; + } + let node = dag + .nodes + .iter() + .find(|node| node.id == id) + .ok_or_else(invalid)?; + let children: Vec<_> = dag + .edges + .iter() + .filter(|edge| edge.consumer == id) + .collect(); + use planner_types::post_asap::{ + ExecutableOperatorPayload as Payload, ExecutionTiming, ValueOperation, + }; + if node.output_state + != planner_types::post_asap::ExecutionDataState::MAINTENANCE_ROWS + { + return Err(invalid()); + } + match &node.payload { + Payload::Value { + operation: ValueOperation::FinalizeExactAccumulator, + timing: ExecutionTiming::MaintenanceTime, + } if children.len() == 1 + && frontiers.contains_key(&children[0].producer) => {} + Payload::Binary { + operator, + timing: ExecutionTiming::MaintenanceTime, + } if children.len() == 2 + && children + .iter() + .filter(|edge| { + edge.role == planner_types::post_asap::EdgeRole::Left + }) + .count() + == 1 + && children + .iter() + .filter(|edge| { + edge.role == planner_types::post_asap::EdgeRole::Right + }) + .count() + == 1 + && operator.vector_match.is_none() + && matches!( + operator.kind, + planner_types::pre_asap::BinaryOpKind::Arithmetic(_) + ) => + { + pending.extend(children.iter().map(|edge| edge.producer)); + } + _ => return Err(invalid()), } - ) { - return Err(invalid()); } matched = true; } diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs index f7357bc9..e329e9a2 100644 --- a/data_plane/tests/support/immutable_maintenance_process.rs +++ b/data_plane/tests/support/immutable_maintenance_process.rs @@ -3,13 +3,27 @@ use super::*; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn single_source_maintenance_is_automatic_and_durable() { - const QUERY: &str = "quantile(0.9, sum_over_time(immutable_value[1m]))"; + run_maintenance_process(false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn two_source_maintenance_is_automatic_and_durable() { + run_maintenance_process(true).await; +} + +async fn run_maintenance_process(multi_source: bool) { + let query = if multi_source { + "quantile(0.9, sum_over_time(immutable_value[1m]) + sum_over_time(immutable_other[1m]))" + } else { + "quantile(0.9, sum_over_time(immutable_value[1m]))" + }; + let expected = if multi_source { 20.0 } else { 10.0 }; let mut fixture: Value = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); let mut entry = fixture["query_workload"]["repeating_queries"][3].clone(); - entry["query"] = QUERY.into(); + entry["query"] = query.into(); entry["demand"]["fixed_interval_at"]["interval"] = 60_000.into(); entry["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into(); entry["time_selection"]["lookback"] = 60_000.into(); @@ -17,7 +31,10 @@ async fn single_source_maintenance_is_automatic_and_durable() { let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); let plan = snapshot.compile().unwrap(); - assert_eq!(plan.precompute_plan.materializations.len(), 2); + assert_eq!( + plan.precompute_plan.materializations.len(), + if multi_source { 3 } else { 2 } + ); let source = plan .precompute_plan .materializations @@ -36,7 +53,12 @@ async fn single_source_maintenance_is_automatic_and_durable() { assert_eq!(derived.pane_origin_ms, Some(0)); assert_eq!( derived.derived_input.as_ref().unwrap().inputs, - std::collections::BTreeSet::from([source.policy_fingerprint().into()]) + plan.precompute_plan + .materializations + .iter() + .filter(|m| m.derived_input.is_none()) + .map(|m| m.policy_fingerprint().into()) + .collect() ); // The production cost model may choose DDSketch or KLL. Preserve that // choice and use its actual value contract for this singleton oracle. @@ -63,7 +85,10 @@ async fn single_source_maintenance_is_automatic_and_durable() { }; // Two independent deployments: singleton is supported; a second physical // input series must never be mistaken for a complete singleton population. - for count in [1, 2] { + for (count, missing_source) in [(1, false), (2, false), (1, true)] { + if missing_source && !multi_source { + continue; + } let mut directory = tempfile::tempdir().unwrap(); eprintln!("IMMUTABLE_PROCESS_ARTIFACT {}", directory.path().display()); directory.disable_cleanup(true); @@ -113,7 +138,7 @@ async fn single_source_maintenance_is_automatic_and_durable() { let backend = format!("http://127.0.0.1:{port}"); let mut first = spawn(port); wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut first.0).await; - let series = (0..count) + let mut series: Vec<_> = (0..count) .map(|i| { series_with_labels( "immutable_value", @@ -125,6 +150,18 @@ async fn single_source_maintenance_is_automatic_and_durable() { ) }) .collect(); + if multi_source && !missing_source { + for i in 0..count { + series.push(series_with_labels( + "immutable_other", + &[ + ("instance", if i == 0 { "a" } else { "b" }), + ("job", "worker"), + ], + &[(1_000, 2.0), (2_000, 3.0), (60_000, 5.0)], + )); + } + } assert_eq!( remote_write(&client, &backend, &WriteRequest { timeseries: series }).await, 204 @@ -134,7 +171,7 @@ async fn single_source_maintenance_is_automatic_and_durable() { .send() .await .unwrap(); - if count == 1 { + if count == 1 && !missing_source { assert!( drain.status().is_success(), "{}", @@ -143,14 +180,14 @@ async fn single_source_maintenance_is_automatic_and_durable() { } let response: Value = client .get(format!("{backend}/api/v1/query")) - .query(&[("query", QUERY), ("time", "60")]) + .query(&[("query", query), ("time", "60")]) .send() .await .unwrap() .json() .await .unwrap(); - if count == 2 { + if count == 2 || missing_source { assert!( !is_warm(&response), "multi-series population was incorrectly admitted: {response}" @@ -173,7 +210,7 @@ async fn single_source_maintenance_is_automatic_and_durable() { .parse::() .unwrap(); assert!( - estimate.is_finite() && (estimate - 10.0).abs() / 10.0 <= max_relative_error, + estimate.is_finite() && (estimate - expected).abs() / expected <= max_relative_error, "selected singleton quantile exceeded its value contract: {response}" ); drop(first); @@ -188,7 +225,7 @@ async fn single_source_maintenance_is_automatic_and_durable() { .await; let after: Value = client .get(format!("{backend}/api/v1/query")) - .query(&[("query", QUERY), ("time", "60")]) + .query(&[("query", query), ("time", "60")]) .send() .await .unwrap() @@ -245,7 +282,7 @@ async fn single_source_maintenance_is_automatic_and_durable() { .await; let stale: Value = client .get(format!("{backend}/api/v1/query")) - .query(&[("query", QUERY), ("time", "60")]) + .query(&[("query", query), ("time", "60")]) .send() .await .unwrap() From 2271bc0ac639d7554bdd3fee181f5615b1403e75 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 13:11:52 -0600 Subject: [PATCH 19/21] docs: describe finite arithmetic input cohorts --- control_plane/src/physical/compiler.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index da414638..c4967cc9 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2590,9 +2590,9 @@ fn raw_time_series_input_contract( } } -/// The first immutable-input capability accepts one exact accumulator readout. -/// Population/window closure is checked by the installed runtime, not inferred -/// from the presence of this syntax. +/// Immutable inputs may combine explicit exact accumulator readouts using +/// maintenance-time arithmetic. Population and window closure are checked by +/// the installed runtime, not inferred from the presence of this syntax. fn immutable_materialization_sources(node: &SummaryNode) -> Option>> { use planner_types::post_asap::{ExactKind, ExecutionTiming, SummaryInputExpr}; let SummaryExpr::SummaryAgg { From c8a94673020c4cd504bb75303c5ad26474b7ee6b Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 13:16:27 -0600 Subject: [PATCH 20/21] test: assert empty local routing errors without warm provenance --- data_plane/src/drivers/query/servers/http.rs | 38 ++++++++++++++++---- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 73b54134..914ec4c3 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -4295,8 +4295,7 @@ aggregations: async fn http_routes_asap_tier_metric_to_simple_engine() { // Default (no hot-reload) → `SketchStore`. The handler // takes the direct `ASAPQueryEngine::handle_query` path; the - // response's `infos` array carries `data_source: asap_query` - // so callers can byte-compare which engine answered. + // empty local engine reports no result without a warm annotation. let server_port = setup_test_server_with_router(StorageBackend::SketchStore, Vec::new()).await; let client = Client::new(); @@ -4312,7 +4311,12 @@ aggregations: resp.status() ); let body: serde_json::Value = resp.json().await.unwrap(); - assert_data_source(&body, "asap_query"); + // Empty local fixtures exercise routing, not successful execution. + assert_eq!( + body, + serde_json::json!({"status":"error", "data":null, + "errorType":"bad_data", "error":"No result for query"}) + ); } #[tokio::test] @@ -4375,7 +4379,12 @@ aggregations: resp.status() ); let body: serde_json::Value = resp.json().await.unwrap(); - assert_data_source(&body, "asap_query"); + // Empty local fixtures exercise routing, not successful execution. + assert_eq!( + body, + serde_json::json!({"status":"error", "data":null, + "errorType":"bad_data", "error":"No result for query"}) + ); } #[tokio::test] @@ -4614,7 +4623,12 @@ aggregations: resp.status(), ); let body: serde_json::Value = resp.json().await.unwrap(); - assert_data_source(&body, "asap_query"); + // Empty local fixtures exercise routing, not successful execution. + assert_eq!( + body, + serde_json::json!({"status":"error", "data":null, + "errorType":"bad_data", "error":"No result for query"}) + ); } #[tokio::test] @@ -4762,7 +4776,12 @@ aggregations: resp.status() ); let body: serde_json::Value = resp.json().await.unwrap(); - assert_data_source(&body, "asap_query"); + // Empty local fixtures exercise routing, not successful execution. + assert_eq!( + body, + serde_json::json!({"status":"error", "data":null, + "errorType":"bad_data", "error":"No result for query"}) + ); assert_eq!( gorilla_calls.load(Ordering::SeqCst), 0, @@ -4941,7 +4960,12 @@ aggregations: .expect("Failed to send request"); assert!(resp.status().is_success(), "default routing must still 2xx"); let body: serde_json::Value = resp.json().await.unwrap(); - assert_data_source(&body, "asap_query"); + // Empty local fixtures exercise routing, not successful execution. + assert_eq!( + body, + serde_json::json!({"status":"error", "data":null, + "errorType":"bad_data", "error":"No result for query"}) + ); assert_eq!( gorilla_calls.load(Ordering::SeqCst), 0, From f46aef4ca0456f8b5b2dca432c32e6819be48595 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 13:27:01 -0600 Subject: [PATCH 21/21] feat: enumerate complete durable raw maintenance populations --- .../sketch_db/index/maintenance.rs | 174 +++++++++++++++++- 1 file changed, 173 insertions(+), 1 deletion(-) diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index c48680b4..9465ec57 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -87,7 +87,36 @@ impl SketchStore { generation: &CatalogGeneration, ) -> Result, BTreeSet<(u64, u64)>>>, String> { + self.maintenance_coordinates(definition, generation, false) + } + + /// Prove the complete raw population before a group-aware consumer joins it. + /// Historical durable SIDs cannot disappear merely because restart did not + /// bind them to the current catalog. + pub(crate) fn complete_raw_maintenance_population( + &self, + definition: SummaryDefinitionId, + generation: &CatalogGeneration, + ) -> Result, BTreeSet<(u64, u64)>>>, String> + { + self.maintenance_coordinates(definition, generation, true) + } + + fn maintenance_coordinates( + &self, + definition: SummaryDefinitionId, + generation: &CatalogGeneration, + require_complete_population: bool, + ) -> Result, BTreeSet<(u64, u64)>>>, String> + { + let admission = self + .admission + .read() + .map_err(|_| "admission registry poisoned")?; self.validate_routed_catalog_generation(Some(generation))?; + if require_complete_population && !admission.is_finite_complete() { + return Err("complete raw population requires a finite source closure".into()); + } let handle = self .persistence_read .read() @@ -105,13 +134,33 @@ impl SketchStore { .filter(|(_, binding)| binding.metadata.policy_fp == definition.fingerprint()) .map(|(sid, _)| *sid), ); - if population_ids.len() > 1 { + if !require_complete_population && population_ids.len() > 1 { return Err("immutable maintenance requires one durable physical population".into()); } let completed = self .completed_windows .read() .map_err(|_| "completion registry poisoned")?; + if require_complete_population { + if population_ids.is_empty() { + return Err("raw population has no durable physical instances".into()); + } + for sid in &population_ids { + let binding = instances + .get(sid) + .ok_or("durable population SID is not bound in this catalog")?; + if binding.catalog_generation.as_deref() != Some(generation) + || !binding.metadata.is_writable() + || matches!( + binding.data_descriptor.source, + asap_types::sds::DataSourceIdentity::Derived { .. } + ) + || !completed.contains_key(sid) + { + return Err("raw population contains an incomplete or foreign lifetime".into()); + } + } + } let mut coordinates = BTreeMap::new(); for (sid, binding) in instances.iter() { if binding.metadata.policy_fp != definition.fingerprint() @@ -148,6 +197,11 @@ impl SketchStore { } } } + if require_complete_population + && coordinates.keys().copied().collect::>() != population_ids + { + return Err("completed raw population is missing durable payload coordinates".into()); + } Ok(coordinates) } @@ -497,6 +551,114 @@ mod tests { use crate::storage_engines::types::PrecomputedOutput; use asap_types::traits::SerializableToSink; + #[test] + fn complete_population_keeps_every_sid_and_rejects_missing_live_binding() { + // Multiple SIDs are inventory entries, never an implicit singleton; + // removing a live binding cannot hide its retained durable population. + 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 first = plan.precompute_plan.materializations[0].clone(); + first.aggregation_type = asap_types::AggregationType::Sum; + first.aggregation_sub_type = "sum".into(); + first.grouping_labels = ["instance".to_string()].into_iter().collect(); + let mut target = first.clone(); + target.derived_input = Some(asap_types::derived_input::DerivedInputIdentity { + inputs: BTreeSet::from([first.policy_fingerprint().into()]), + program_sha256: "0".repeat(64), + }); + let configs = [first.clone(), first, target]; + let catalog = + asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs) + .unwrap(); + let store = Arc::new(SketchStore::new()); + store + .install_summary_catalog(Arc::new(catalog.clone())) + .unwrap(); + let generation = store.active_catalog_generation().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(); + for (index, config) in configs.iter().take(2).enumerate() { + let definition = config.policy_fingerprint().into(); + let population = BTreeMap::from([("instance".to_string(), index.to_string())]); + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: definition, + time_range: HalfOpenTimeRange { + start_ms: 0, + end_ms: 1000, + }, + group_values: population.clone(), + }; + let revision = store + .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()])) + .unwrap(); + let mut output = PrecomputedOutput::new(0, 1000, None, config.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + output.population_labels = Some(population); + let mut sum = SumAccumulator::new(); + sum.update(5.0 + index as f64); + let sid = 900 + index as u64; + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 2000, + |writer| writer.ingest_precompute_with_series_id(sid, config, &output, &sum), + ) + .unwrap(); + } + 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 definition = configs[0].policy_fingerprint().into(); + let parts = persistence.manifest.live_parts().len(); + let metadata = persistence + .flusher + .metadata_store() + .load_strict() + .unwrap() + .len(); + let inventory = store + .complete_raw_maintenance_population(definition, &generation) + .unwrap(); + assert_eq!( + inventory.keys().copied().collect::>(), + BTreeSet::from([900, 901]) + ); + assert!(inventory.values().all(|groups| groups.len() == 1)); + assert!(store + .completed_maintenance_coordinates(definition, &generation) + .is_err()); + store.instances.write().unwrap().remove(&901); + assert!(store + .complete_raw_maintenance_population(definition, &generation) + .is_err()); + assert_eq!(persistence.manifest.live_parts().len(), parts); + assert_eq!( + persistence + .flusher + .metadata_store() + .load_strict() + .unwrap() + .len(), + metadata + ); + persistence.shutdown(); + } + #[test] fn cohort_requires_every_durable_source_in_one_catalog_generation() { // Neither a missing second window nor a new catalog may yield a @@ -701,6 +863,9 @@ mod tests { ) .unwrap(); } + assert!(store + .complete_raw_maintenance_population(source.policy_fingerprint().into(), &generation) + .is_err()); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); while !store.seal_finite_summary_input(&generation).unwrap() { assert!(std::time::Instant::now() < deadline); @@ -712,6 +877,13 @@ mod tests { .unwrap(); assert_eq!(coordinates.len(), 1); assert_eq!(coordinates[&700].len(), 2); + assert_eq!( + store + .complete_raw_maintenance_population(source_id, &generation) + .unwrap(), + coordinates + ); + let group = BTreeMap::from([("instance".into(), "a".into())]); let frozen = store .read_frozen_exact_windows(