diff --git a/crates/integration-tests/tests/joint_search_equivalence.rs b/crates/integration-tests/tests/joint_search_equivalence.rs index 63ee2bda5..fe7c8f038 100644 --- a/crates/integration-tests/tests/joint_search_equivalence.rs +++ b/crates/integration-tests/tests/joint_search_equivalence.rs @@ -267,33 +267,28 @@ fn bounded_search_matches_complete_tiny_catalog_under_every_objective() { } } -// Repeated queries retain all outputs and per-query accounting without creating -// additional implementation scopes or producer-to-consumer choices. +// Bounded runs can stop at different progressive pane-width domains. Compare +// validated output semantics here, not inventory sizes from different catalogs. +// Fixed-catalog terminal/producer scaling is tested in candidates::scopes. #[test] -fn repeated_queries_preserve_outputs_without_quadratic_candidate_growth() { +fn repeated_queries_preserve_all_outputs_and_exact_guarantees() { let settings = PlanSelectionSettings::default(); let deployment = DeploymentCapabilities::default(); - let mut counts = Vec::new(); for count in [2, 8] { let workload = workload_of( &vec![("sum_over_time(m[1m])", 60_000); count], AccuracyTarget::exact(), ); let run = run_workload(&workload, &settings, &deployment); - let report = run - .stage3_plan_selection - .as_ref() - .unwrap() - .search - .candidate_connections - .as_ref() - .unwrap(); - counts.push(( - report.physical_configurations, - report.compatible_connections, - )); + let selection = run.stage3_plan_selection.as_ref().unwrap(); + let accepted = &selection.accepted[&selection.selected]; + assert_eq!(accepted.len(), count); + for accuracy in accepted { + assert!(!accuracy.accepted_regardless_of_guarantee); + assert_eq!(accuracy.guarantee_check_rejection, None); + assert_eq!(accuracy.accepted_metric, None, "every output remains exact"); + } } - assert_eq!(counts[0], counts[1]); } // Different windows over the same source remain legal with retained raw data diff --git a/crates/logical-asap-optimization/src/memo/state_configurations.rs b/crates/logical-asap-optimization/src/memo/state_configurations.rs index 30b1d1ee0..f3b000971 100644 --- a/crates/logical-asap-optimization/src/memo/state_configurations.rs +++ b/crates/logical-asap-optimization/src/memo/state_configurations.rs @@ -14,6 +14,12 @@ pub enum PaneWidthDomain { #[default] Complete, SeedGrids, + /// Seed grids plus widths obtained by dividing each grid into at most + /// `max_parts` equal pieces. Increasing the limit monotonically exposes + /// every legal divisor; only `Complete` certifies full-domain search. + Expanding { + max_parts: u64, + }, } pub fn pane_width_domain( @@ -44,6 +50,7 @@ pub(crate) fn pane_width_domain_before( } return Ok(complete.into_iter().collect()); } + let grids = widths.clone(); loop { let mut additional = BTreeSet::new(); for &a in &widths { @@ -60,10 +67,95 @@ pub(crate) fn pane_width_domain_before( } widths.extend(additional); } + if let PaneWidthDomain::Expanding { max_parts } = domain { + // Work is proportional to the requested frontier, not the complete + // divisor domain. Do not introduce the finest reciprocal width early. + for grid in grids { + for parts in 2..=max_parts.min(grid) { + if parts % 1024 == 2 { + super::check_generation_deadline(deadline)?; + } + if grid.is_multiple_of(parts) { + widths.insert(grid / parts); + } + } + } + } super::check_generation_deadline(deadline)?; Ok(widths.into_iter().collect()) } +/// Integer-domain frontier, independent of expensive candidate graph expansion. +/// Complete divisor lists are computed once under the caller's deadline. +pub struct PaneWidthFrontier { + grids: Vec, + complete: Vec>, + previous: Vec>, + max_parts: u64, + finished: bool, +} + +impl PaneWidthFrontier { + pub fn new( + grids: impl IntoIterator, + deadline: Option, + ) -> Result { + let grids: Vec<_> = grids + .into_iter() + .filter(|grid| *grid > 0) + .collect::>() + .into_iter() + .collect(); + let complete = grids + .iter() + .map(|&grid| pane_width_divisors_before(grid, deadline)) + .collect::, _>>()?; + let previous = grids.iter().map(|&grid| vec![grid]).collect(); + Ok(Self { + grids, + complete, + previous, + max_parts: 4, + finished: false, + }) + } + + /// Skip frontiers which expose no new width for any individual reader grid. + /// Comparing per-grid sets also protects group-specific supplier choices. + pub fn next_domain( + &mut self, + deadline: Option, + ) -> Result, super::CandidateGenerationTimeout> { + while !self.finished { + super::check_generation_deadline(deadline)?; + let current: Vec> = self + .grids + .iter() + .zip(&self.complete) + .map(|(&grid, widths)| { + widths + .iter() + .copied() + .filter(|width| grid / width <= self.max_parts) + .collect() + }) + .collect(); + self.max_parts = self.max_parts.saturating_mul(2); + if current == self.complete { + self.finished = true; + return Ok(Some(PaneWidthDomain::Complete)); + } + if current != self.previous { + self.previous = current; + return Ok(Some(PaneWidthDomain::Expanding { + max_parts: self.max_parts / 2, + })); + } + } + Ok(None) + } +} + #[derive(Debug, Clone)] pub struct PaneRequirement { pub pane_width_ms: u64, @@ -387,6 +479,62 @@ mod tests { // Every integer-millisecond width that divides the grid is offered, not // just the coarsest width or GCDs of the readers' original configurations. + #[test] + fn width_frontier_skips_empty_rounds_and_detects_actual_completion() { + let mut frontier = PaneWidthFrontier::new([97], None).unwrap(); + assert_eq!( + frontier.next_domain(None).unwrap(), + Some(PaneWidthDomain::Complete) + ); + assert_eq!(frontier.next_domain(None).unwrap(), None); + let mut frontier = PaneWidthFrontier::new([12], None).unwrap(); + assert_eq!( + frontier.next_domain(None).unwrap(), + Some(PaneWidthDomain::Expanding { max_parts: 4 }) + ); + assert_eq!( + frontier.next_domain(None).unwrap(), + Some(PaneWidthDomain::Expanding { max_parts: 8 }) + ); + assert_eq!( + frontier.next_domain(None).unwrap(), + Some(PaneWidthDomain::Complete) + ); + assert!(PaneWidthFrontier::new([u64::MAX], Some(std::time::Instant::now())).is_err()); + } + + #[test] + fn expanding_widths_are_nested_and_eventually_complete() { + let grids = [60, 90]; + let mut previous = pane_width_domain(grids, PaneWidthDomain::SeedGrids); + for max_parts in [2, 4, 8, 16, 32, 64, 128] { + let next = pane_width_domain(grids, PaneWidthDomain::Expanding { max_parts }); + assert!(previous.iter().all(|width| next.contains(width))); + assert!(next + .iter() + .all(|width| grids.iter().any(|grid| grid % width == 0))); + previous = next; + } + assert_eq!( + previous, + pane_width_domain(grids, PaneWidthDomain::Complete) + ); + } + + #[test] + fn expanding_domain_defers_fine_panes_and_obeys_deadline() { + let widths = pane_width_domain([60_000], PaneWidthDomain::Expanding { max_parts: 4 }); + assert_eq!(widths, vec![15_000, 20_000, 30_000, 60_000]); + assert!(pane_width_domain_before( + [u64::MAX], + PaneWidthDomain::Expanding { + max_parts: u64::MAX + }, + Some(std::time::Instant::now()), + ) + .is_err()); + } + #[test] fn pane_width_domain_contains_every_legal_divisor() { let reader = PaneRequirement { diff --git a/crates/logical-asap-optimization/src/memo/window_candidates.rs b/crates/logical-asap-optimization/src/memo/window_candidates.rs index 135b94541..bc76b5d9f 100644 --- a/crates/logical-asap-optimization/src/memo/window_candidates.rs +++ b/crates/logical-asap-optimization/src/memo/window_candidates.rs @@ -37,6 +37,18 @@ pub fn try_add_local_window_composition_in_domain( entries: &[QueryWorkloadEntry], domain: PaneWidthDomain, deadline: Option, +) -> Result<(), CandidateGenerationTimeout> { + extend_local_window_composition(memo, entries, None, domain, deadline) +} + +/// Extend a pre-state-expansion memo with only newly exposed pane widths. +/// `previous` must describe widths already generated for this memo. +pub fn extend_local_window_composition( + memo: &mut Memo, + entries: &[QueryWorkloadEntry], + previous: Option, + domain: PaneWidthDomain, + deadline: Option, ) -> Result<(), CandidateGenerationTimeout> { check_generation_deadline(deadline)?; // Flatten each existing/generated graph once. Repeatedly flattening the @@ -51,7 +63,9 @@ pub fn try_add_local_window_composition_in_domain( } } } - for (group, candidate) in local_pane_candidates_before(memo, entries, domain, deadline)? { + for (group, candidate) in + local_pane_candidates_delta(memo, entries, previous, domain, deadline)? + { check_generation_deadline(deadline)?; let Ok(signature) = candidate_signature(&candidate) else { memo.candidates[group].push(candidate); @@ -88,6 +102,16 @@ fn local_pane_candidates_before( entries: &[QueryWorkloadEntry], domain: PaneWidthDomain, deadline: Option, +) -> Result, CandidateGenerationTimeout> { + local_pane_candidates_delta(memo, entries, None, domain, deadline) +} + +fn local_pane_candidates_delta( + memo: &Memo, + entries: &[QueryWorkloadEntry], + previous: Option, + domain: PaneWidthDomain, + deadline: Option, ) -> Result, CandidateGenerationTimeout> { check_generation_deadline(deadline)?; let group_of = group_of_node(&memo.groups); @@ -102,7 +126,23 @@ fn local_pane_candidates_before( domain, deadline, )?; + let previous_widths = previous + .map(|previous| { + pane_width_domain_before( + group + .queries + .iter() + .filter_map(|&query| entries.get(query)?.evaluation_grid_ms()), + previous, + deadline, + ) + }) + .transpose()? + .unwrap_or_default(); for grid in grids { + if previous_widths.binary_search(&grid).is_ok() { + continue; + } check_generation_deadline(deadline)?; for candidate in &memo.candidates[g] { check_generation_deadline(deadline)?; @@ -308,6 +348,46 @@ mod tests { ); } + #[test] + fn incremental_local_widths_match_one_complete_pass() { + let (base, entries) = fixture(&[repeated(60)]); + let mut incremental = base.clone(); + let mut previous = None; + for domain in [ + PaneWidthDomain::SeedGrids, + PaneWidthDomain::Expanding { max_parts: 4 }, + PaneWidthDomain::Expanding { max_parts: 8 }, + PaneWidthDomain::Complete, + ] { + extend_local_window_composition(&mut incremental, &entries, previous, domain, None) + .unwrap(); + assert!(local_pane_candidates_delta( + &incremental, + &entries, + Some(domain), + domain, + None + ) + .unwrap() + .is_empty()); + previous = Some(domain); + } + let mut complete = base; + add_local_window_composition(&mut complete, &entries); + let signatures = |memo: &Memo| { + memo.candidates + .iter() + .enumerate() + .flat_map(|(group, candidates)| { + candidates + .iter() + .map(move |candidate| (group, candidate_signature(candidate).unwrap())) + }) + .collect::>() + }; + assert_eq!(signatures(&incremental), signatures(&complete)); + } + fn repeated(interval: u32) -> QueryRecurrence { QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(interval))) } diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 1c4eefbcb..46c9e32e2 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -1024,13 +1024,30 @@ pub struct SearchReport { #[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)] pub struct CandidateConnectionReport { + /// Pane-width domain of the most recently reported complete inventory. + /// Inventory counts form one snapshot from this domain, even if a later + /// expansion timed out. This does not certify full-domain optimality. + #[serde(default)] + pub inventory_pane_width_domain: Option, + /// Width domains attempted in order; state/physical catalogs are rebuilt per round. + #[serde(default)] + pub pane_width_rounds: Vec, + /// Some legal widths remain unsearched; restricted optima are not global certificates. + #[serde(default)] + pub pane_widths_pending: bool, + /// Number of distinct allocated compatible-supplier lists. + #[serde(default)] + pub supplier_sets: usize, + /// Supplier IDs physically stored across shared lists (not logical edges). + #[serde(default)] + pub stored_supplier_ids: usize, /// Conservative zero bounds used to stay within the solver coefficient range. #[serde(default)] pub rounded_down_cost_bounds: usize, /// Time spent obtaining a validated incumbent from the seed width domain. #[serde(default)] pub seed_planning_ms: f64, - /// Complete-domain phase wall time, including interrupted generation. + /// Post-seed expansion/selection wall time, including interrupted rounds. #[serde(default)] pub full_planning_ms: f64, #[serde(default)] diff --git a/crates/plan-selection/src/search/candidates/configurations.rs b/crates/plan-selection/src/search/candidates/configurations.rs index b85e3b4f4..beed24f37 100644 --- a/crates/plan-selection/src/search/candidates/configurations.rs +++ b/crates/plan-selection/src/search/candidates/configurations.rs @@ -6,6 +6,8 @@ //! Stage 2 validation still checks operator legality and upstream timing. //! Potential shared non-summary nodes also receive query materialization; Stage 2 //! checks highest-shared eligibility on the selected graph, not all possible consumers. +use super::suppliers::{SupplierCandidates, SupplierSetInterner}; +use std::collections::HashMap; use std::rc::Rc; use asap_logical_asap_optimization::deployment_capabilities::DeploymentCapabilities; @@ -297,24 +299,54 @@ pub(in crate::search) fn enumerate_candidate_configurations_before( } } } + let mut supplier_sets = SupplierSetInterner::default(); + // Compatibility depends on consumer timing and producer mode/timing only. + // Evaluate each source configuration once per timing, then reuse this exact + // index; reader eligibility remains expressed by separate graph constraints. + let mut timing_index = Vec::with_capacity(source_configurations.len()); + for configurations in &source_configurations { + super::check_deadline(deadline)?; + let mut by_timing = [SupplierCandidates::default(), SupplierCandidates::default()]; + for (timing_index, timing) in [ExecutionTiming::QueryTime, ExecutionTiming::IngestionTime] + .into_iter() + .enumerate() + { + let mut compatible = Vec::new(); + for &producer in configurations { + super::check_deadline(deadline)?; + if compatible_timing(timing, &result.configurations[producer.0]) { + compatible.push(producer); + } + } + by_timing[timing_index] = supplier_sets.intern(compatible); + } + timing_index.push(by_timing); + } + let mut input_cache: HashMap<(SupplierCandidates, ExecutionTiming), SupplierCandidates> = + HashMap::new(); for (id, configuration) in result.configurations.iter().enumerate() { super::check_deadline(deadline)?; let mut requirements = Vec::new(); for input in &source.graph.candidates[configuration.source_candidate.0].producer_requirements { - let mut compatible_producers = Vec::new(); - for source in &input.compatible_producers { - super::check_deadline(deadline)?; - compatible_producers.extend( - source_configurations[source.0] - .iter() - .copied() - .filter(|producer| { - compatible_timing(configuration, &result.configurations[producer.0]) - }), - ); - } + let key = (input.compatible_producers.clone(), configuration.timing); + let compatible_producers = if let Some(suppliers) = input_cache.get(&key) { + suppliers.clone() + } else { + let timing = match configuration.timing { + ExecutionTiming::QueryTime => 0, + ExecutionTiming::IngestionTime => 1, + }; + let mut compatible = Vec::new(); + for producer in &input.compatible_producers { + super::check_deadline(deadline)?; + compatible.extend(timing_index[producer.0][timing].iter().copied()); + } + let suppliers = supplier_sets.intern(compatible); + input_cache.insert(key, suppliers.clone()); + suppliers + }; requirements.push(ProducerRequirement { compatible_producers, }); @@ -434,9 +466,9 @@ pub(in crate::search) fn enumerate_candidate_configurations_before( ) } -fn compatible_timing(consumer: &CandidateConfiguration, producer: &CandidateConfiguration) -> bool { - consumer.timing == producer.timing - || (consumer.timing == ExecutionTiming::QueryTime +fn compatible_timing(consumer_timing: ExecutionTiming, producer: &CandidateConfiguration) -> bool { + consumer_timing == producer.timing + || (consumer_timing == ExecutionTiming::QueryTime && matches!(producer.materialization, Materialization::MatAtIngestion(_))) } @@ -558,6 +590,70 @@ mod tests { matches!(node.asap(), Some(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. }) if *kind.algorithm() == SketchAlgorithm::Kll) } + // Indexed compatibility must reproduce every original ordered edge while + // storing repeated supplier lists once, including query/ingestion modes. + #[test] + fn supplier_index_preserves_all_edges_and_shares_storage() { + let (source, entries, data) = fixture(); + let expanded = enumerate_candidate_configurations( + &source, + &[0, 1], + &entries, + &data, + &DeploymentCapabilities::default(), + ) + .unwrap(); + for (consumer, configuration) in expanded.configurations.iter().enumerate() { + let original = &source.graph.candidates[configuration.source_candidate.0]; + let generated = &expanded.inventory.graph.candidates[consumer]; + assert_eq!( + generated.producer_requirements.len(), + original.producer_requirements.len() + ); + for (original, generated) in original + .producer_requirements + .iter() + .zip(&generated.producer_requirements) + { + let expected: Vec<_> = original + .compatible_producers + .iter() + .flat_map(|source| { + expanded.configurations.iter().enumerate().filter_map( + move |(id, producer)| { + (producer.source_candidate == *source + && (configuration.timing == producer.timing + || (configuration.timing == ExecutionTiming::QueryTime + && matches!( + producer.materialization, + Materialization::MatAtIngestion(_) + )))) + .then_some(PhysicalCandidateId(id)) + }, + ) + }) + .collect(); + assert_eq!( + generated.compatible_producers.as_slice(), + expected.as_slice() + ); + } + } + let metrics = super::super::suppliers::supplier_storage_metrics(&expanded.inventory.graph); + let requirements = expanded + .inventory + .graph + .candidates + .iter() + .map(|candidate| candidate.producer_requirements.len()) + .sum::(); + assert!(metrics.unique_supplier_sets < requirements, "{metrics:?}"); + assert!( + metrics.stored_supplier_ids < metrics.logical_connections, + "{metrics:?}" + ); + } + // Configuration expansion is bounded per node, and never crosses ephemeral timing. #[test] fn real_quantiles_offer_local_and_shared_timing_configurations() { @@ -621,7 +717,7 @@ mod tests { .flat_map(|input| &input.compatible_producers) { assert!(compatible_timing( - &expanded.configurations[consumer], + expanded.configurations[consumer].timing, &expanded.configurations[producer.0] )); } @@ -656,7 +752,7 @@ mod tests { let mut requirements = Vec::new(); node.operator.map_children(|child| { requirements.push(ProducerRequirement { - compatible_producers: vec![append(child, source)], + compatible_producers: vec![append(child, source)].into(), }); child.clone() }); diff --git a/crates/plan-selection/src/search/candidates/factoring.rs b/crates/plan-selection/src/search/candidates/factoring.rs new file mode 100644 index 000000000..1e9359fe1 --- /dev/null +++ b/crates/plan-selection/src/search/candidates/factoring.rs @@ -0,0 +1,505 @@ +//! Memoize identical supplier interfaces and quotient equivalent terminal readers. +//! Scope copies preserve independent bindings; internal computations are never merged. +use std::collections::HashMap; + +use asap_logical_asap_optimization::memo::MemoGroupId; +use asap_types::ir::OperatorNode; + +use super::graph::PhysicalCandidateId; +use super::suppliers::{SupplierCandidates, SupplierSetInterner}; + +#[derive(Hash, PartialEq, Eq)] +pub(super) struct CompatibilityKey(Vec); + +impl CompatibilityKey { + /// Include everything output_contract_satisfied reads from its expected + /// node. Byte-exact keys preserve parameters that PartialEq or JSON may + /// erase, and include closedness and every required uniqueness key. + pub(super) fn new(group: MemoGroupId, expected: &OperatorNode) -> Result { + rmp_serde::to_vec(&(group.0, &expected.result_kind, &expected.schema)) + .map(Self) + .map_err(|error| error.to_string()) + } +} + +#[derive(Default)] +pub(super) struct CompatibilityMemo { + results: HashMap, + lists: SupplierSetInterner, + #[cfg(test)] + disabled: bool, + #[cfg(test)] + hits: std::cell::Cell, +} + +impl CompatibilityMemo { + pub(super) fn get(&self, key: &CompatibilityKey) -> Option { + #[cfg(test)] + if self.disabled { + return None; + } + let found = self.results.get(key).cloned(); + #[cfg(test)] + if found.is_some() { + self.hits.set(self.hits.get() + 1); + } + found + } + + #[cfg(test)] + pub(super) fn hits(&self) -> usize { + self.hits.get() + } + + #[cfg(test)] + pub(super) fn disabled() -> Self { + Self { + disabled: true, + ..Self::default() + } + } + + pub(super) fn remember( + &mut self, + key: CompatibilityKey, + suppliers: Vec, + ) -> SupplierCandidates { + let suppliers = self.lists.intern(suppliers); + self.results.insert(key, suppliers.clone()); + suppliers + } +} + +/// Quotient only terminal unary summary readers. Internal consumers and merge +/// operators retain their original identities. Scope expansion must retain one +/// independently bound reader copy per possible output reference; then every +/// old selection maps to these copies, and each new binding expands to an old +/// reader with the same complete interface and fixed local cost. +/// +/// These operators bypass ConcreteRows' witness-based row certification. +/// certified_price checks every supplied producer; template-based proposal +/// prices remain advisory. Execution and accuracy are recomputed from selected +/// suppliers, never from the representative template's child witness. +pub(super) fn factor_terminal_readers( + inventory: &mut super::inventory::CandidateInventory, + deadline: Option, +) -> Result { + use super::inventory::CandidateInventoryError; + use asap_types::ir::ASAPOp; + let invalid = |error: String| CandidateInventoryError::InvalidMemo(error); + if !inventory.graph.reader_constraints.is_empty() + || !inventory.graph.resource_budgets.is_empty() + { + return Ok(0); + } + let count = inventory.graph.candidates.len(); + let mut internal = vec![false; count]; + for candidate in &inventory.graph.candidates { + super::check_deadline(deadline)?; + for input in &candidate.producer_requirements { + for producer in &input.compatible_producers { + internal[producer.0] = true; + } + } + } + let mut outputs = vec![Vec::new(); count]; + for (output, root) in inventory.graph.query_roots.iter().enumerate() { + for id in &root.compatible_candidates { + outputs[id.0].push(output); + } + } + let mut classes = HashMap::, usize>::new(); + let mut representative: Vec<_> = (0..count).collect(); + let mut eligible = vec![false; count]; + for id in 0..count { + super::check_deadline(deadline)?; + let candidate = &inventory.graph.candidates[id]; + if internal[id] || outputs[id].is_empty() || candidate.producer_requirements.len() != 1 { + continue; + } + let Some(template) = &inventory.templates[id] else { + continue; + }; + if !matches!( + template.asap(), + Some(ASAPOp::SummaryEstimate { .. } | ASAPOp::FinalizeExactAccumulator { .. }) + ) { + continue; + } + let mut inputs = Vec::new(); + let operator = template.operator.map_children(|child| { + inputs.push(child.clone()); + 0u8 + }); + let [input] = inputs.as_slice() else { + continue; + }; + eligible[id] = true; + let key = rmp_serde::to_vec(&( + candidate.group.0, + candidate.local_cost.to_bits(), + &outputs[id], + operator, + &template.schema, + template.result_kind, + &template.guarantee, + template.timing, + template.materialization, + &input.schema, + input.result_kind, + &input.guarantee, + input.timing, + input.materialization, + )) + .map_err(|error| invalid(error.to_string()))?; + if let Some(&previous) = classes.get(&key) { + representative[id] = previous; + } else { + classes.insert(key, id); + } + } + let removed = representative + .iter() + .enumerate() + .filter(|(id, rep)| *id != **rep) + .count(); + if removed == 0 { + return Ok(0); + } + let mut unions = HashMap::>::new(); + let mut known = HashMap::>::new(); + for (id, &rep) in representative.iter().enumerate() { + super::check_deadline(deadline)?; + if eligible[id] { + let suppliers = unions.entry(rep).or_default(); + let known = known.entry(rep).or_default(); + for &supplier in + &inventory.graph.candidates[id].producer_requirements[0].compatible_producers + { + if known.insert(supplier) { + suppliers.push(supplier); + } + } + } + } + let mut new_ids = vec![PhysicalCandidateId(0); count]; + let mut next = 0; + for (id, &rep) in representative.iter().enumerate() { + if id == rep { + new_ids[id] = PhysicalCandidateId(next); + next += 1; + } + } + for (id, &rep) in representative.iter().enumerate() { + new_ids[id] = new_ids[rep]; + } + let mut lists = SupplierSetInterner::default(); + let mut candidates = Vec::with_capacity(next); + let mut templates = Vec::with_capacity(next); + for (id, candidate) in inventory.graph.candidates.iter().enumerate() { + super::check_deadline(deadline)?; + if representative[id] != id { + continue; + } + let mut candidate = candidate.clone(); + if let Some(suppliers) = unions.remove(&id) { + candidate.producer_requirements[0].compatible_producers = suppliers.into(); + } + for input in &mut candidate.producer_requirements { + input.compatible_producers = lists.intern( + input + .compatible_producers + .iter() + .map(|id| new_ids[id.0]) + .collect::>(), + ); + } + candidates.push(candidate); + templates.push(inventory.templates[id].clone()); + } + let remap = |ids: &mut Vec| { + let mut seen = std::collections::HashSet::new(); + *ids = ids + .iter() + .map(|id| new_ids[id.0]) + .filter(|id| seen.insert(*id)) + .collect(); + }; + for root in &mut inventory.graph.query_roots { + remap(&mut root.compatible_candidates); + } + for group in &mut inventory.group_candidates { + remap(group); + } + inventory.graph.candidates = candidates; + inventory.templates = templates; + inventory + .graph + .validate() + .map_err(|error| invalid(error.to_string()))?; + Ok(removed) +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_logical_asap_optimization::test_support::lower_promql; + use asap_types::ir::properties::AccuracyTarget; + + fn terminal_fixture() -> super::super::inventory::CandidateInventory { + use super::super::graph::{ + CandidateDependencyGraph, PhysicalCandidate, ProducerRequirement, QueryRootCandidates, + }; + use asap_logical_asap_optimization::test_support::{cms_summary, estimate_over}; + use asap_types::ir::properties::AccuracyGuarantee; + use std::rc::Rc; + let state = Rc::new(cms_summary("count(m)")); + let reader = estimate_over(state.clone(), AccuracyGuarantee::exact("fixture")); + super::super::inventory::CandidateInventory { + graph: CandidateDependencyGraph { + candidates: (0..4) + .map(|id| PhysicalCandidate { + group: MemoGroupId(if id < 2 { 0 } else { 1 }), + local_cost: [8.0, 12.0, 1.0, 1.0][id], + producer_requirements: if id < 2 { + vec![] + } else { + vec![ProducerRequirement { + compatible_producers: vec![PhysicalCandidateId(id - 2)].into(), + }] + }, + }) + .collect(), + query_roots: vec![ + QueryRootCandidates { + compatible_candidates: vec![PhysicalCandidateId(2), PhysicalCandidateId(3)] + }; + 2 + ], + ..Default::default() + }, + templates: vec![ + Some(state.clone()), + Some(state), + Some(reader.clone()), + Some(reader), + ], + group_candidates: vec![ + vec![PhysicalCandidateId(0), PhysicalCandidateId(1)], + vec![PhysicalCandidateId(2), PhysicalCandidateId(3)], + ], + } + } + + fn with_scopes( + inventory: super::super::inventory::CandidateInventory, + ) -> super::super::configurations::CandidateConfigurationInventory { + use super::super::configurations::{ + CandidateConfiguration, CandidateConfigurationInventory, + }; + use asap_types::ir::properties::ExecutionTiming; + use asap_types::physical::{Materialization, Medium}; + let count = inventory.graph.candidates.len(); + let configurations = (0..count) + .map(|id| CandidateConfiguration { + source_candidate: PhysicalCandidateId(id), + timing: if id < 2 { + ExecutionTiming::IngestionTime + } else { + ExecutionTiming::QueryTime + }, + materialization: if id < 2 { + Materialization::MatAtIngestion(Medium::Memory) + } else { + Materialization::NotMaterialized + }, + }) + .collect(); + super::super::scopes::expand_execution_scopes( + CandidateConfigurationInventory { + inventory, + configurations, + }, + &vec![1; count], + &[0, 0], + ) + .unwrap() + } + + // Enumerate every output-to-state assignment, including split bindings for + // two references of one query. State 1 models a higher-retention producer: + // its different cost and identity must survive despite equal reader types. + #[test] + fn terminal_quotient_preserves_all_bindings_and_pay_once_costs() { + use super::super::graph::ReaderConstraint; + use crate::search::milp::model::select_candidates; + for assignment in [[0, 0], [0, 1], [1, 0], [1, 1]] { + let original = with_scopes(terminal_fixture()); + let mut factored = terminal_fixture(); + assert_eq!(factor_terminal_readers(&mut factored, None).unwrap(), 1); + assert_eq!(factored.graph.candidates.len(), 3); + let compressed = with_scopes(factored); + assert!( + compressed.inventory.graph.candidates.len() + < original.inventory.graph.candidates.len() + ); + for mut configured in [original, compressed] { + for state in 0..2 { + configured.inventory.graph.reader_constraints.push( + ReaderConstraint::AllowedOutputs { + candidate: PhysicalCandidateId(state), + outputs: assignment + .iter() + .enumerate() + .filter_map(|(output, &producer)| { + (producer == state).then_some(output) + }) + .collect(), + }, + ); + } + let selected = select_candidates(&configured.inventory.graph) + .unwrap() + .expect("every original assignment remains feasible"); + let expected = if assignment[0] == assignment[1] { + [8.0, 12.0][assignment[0]] + 1.0 + } else { + 22.0 + }; + assert_eq!(selected.modeled_cost, expected, "assignment {assignment:?}"); + } + } + } + + #[test] + fn duplicate_terminal_interfaces_reduce_scoped_candidates_and_connections() { + let build = || { + let mut inventory = terminal_fixture(); + for reader in &mut inventory.graph.candidates[2..] { + reader.producer_requirements[0].compatible_producers = + vec![PhysicalCandidateId(0), PhysicalCandidateId(1)].into(); + } + inventory + }; + let original = with_scopes(build()); + let mut compact = build(); + assert_eq!(factor_terminal_readers(&mut compact, None).unwrap(), 1); + let compact = with_scopes(compact); + let counts = |graph: &super::super::graph::CandidateDependencyGraph| { + ( + graph.candidates.len(), + graph + .candidates + .iter() + .flat_map(|candidate| &candidate.producer_requirements) + .map(|input| input.compatible_producers.len()) + .sum::(), + ) + }; + assert_eq!(counts(&original.inventory.graph), (6, 8)); + assert_eq!(counts(&compact.inventory.graph), (4, 4)); + } + + #[test] + fn terminal_quotient_preserves_interfaces_and_internal_consumers() { + use super::super::graph::ProducerRequirement; + use asap_types::ir::properties::AccuracyGuarantee; + use std::rc::Rc; + let mut accuracy = terminal_fixture(); + let mut other = (**accuracy.templates[3].as_ref().unwrap()).clone(); + other.guarantee = Some(AccuracyGuarantee::exact("different proof contract")); + accuracy.templates[3] = Some(Rc::new(other)); + assert_eq!(factor_terminal_readers(&mut accuracy, None).unwrap(), 0); + let mut eligibility = terminal_fixture(); + eligibility.graph.query_roots[1].compatible_candidates = vec![PhysicalCandidateId(3)]; + assert_eq!(factor_terminal_readers(&mut eligibility, None).unwrap(), 0); + let mut internal = terminal_fixture(); + internal.graph.candidates[0] + .producer_requirements + .push(ProducerRequirement { + compatible_producers: vec![PhysicalCandidateId(2)].into(), + }); + assert_eq!(factor_terminal_readers(&mut internal, None).unwrap(), 0); + } + + // Logical ownership (including window semantics), closedness and required + // keys remain distinct even when their fields have identical types. + #[test] + fn compatibility_cache_keeps_every_expected_output_requirement() { + let expected = lower_promql("latency", AccuracyTarget::exact()); + let group = MemoGroupId(1); + let key = CompatibilityKey::new(group, &expected).unwrap(); + let mut memo = CompatibilityMemo::default(); + let suppliers = vec![PhysicalCandidateId(3), PhysicalCandidateId(7)]; + memo.remember(key, suppliers.clone()); + assert_eq!( + &*memo + .get(&CompatibilityKey::new(group, &expected).unwrap()) + .unwrap(), + &suppliers + ); + assert!(memo + .get(&CompatibilityKey::new(MemoGroupId(2), &expected).unwrap()) + .is_none()); + let mut closed = (*expected).clone(); + closed.schema.closed = !closed.schema.closed; + assert!(memo + .get(&CompatibilityKey::new(group, &closed).unwrap()) + .is_none()); + let mut keys = (*expected).clone(); + keys.schema.unique_keys.push(vec![]); + assert!(memo + .get(&CompatibilityKey::new(group, &keys).unwrap()) + .is_none()); + } + + // Storage sharing never couples the two consumers' supplier decisions. + #[test] + fn two_reader_bindings_keep_independent_supplier_lists() { + let expected = lower_promql("latency", AccuracyTarget::exact()); + let key = || CompatibilityKey::new(MemoGroupId(1), &expected).unwrap(); + let mut memo = CompatibilityMemo::default(); + memo.remember(key(), vec![PhysicalCandidateId(1), PhysicalCandidateId(2)]); + let mut first = memo.get(&key()).unwrap(); + let second = memo.get(&key()).unwrap(); + first.retain(|id| id.0 == 1); + assert_eq!(&*first, &[PhysicalCandidateId(1)]); + assert_eq!(&*second, &[PhysicalCandidateId(1), PhysicalCandidateId(2)]); + assert_eq!(memo.get(&key()).unwrap(), second); + } + // Exhaust every two-reader binding, charging each retained producer once. + // A longer retention requirement has its own logical group and cannot + // accidentally inherit the shorter reader's supplier set from the cache. + #[test] + fn cached_interfaces_preserve_exhaustive_sharing_and_retention_choices() { + let expected = lower_promql("latency", AccuracyTarget::exact()); + let short = vec![PhysicalCandidateId(0), PhysicalCandidateId(1)]; + let long = vec![PhysicalCandidateId(1)]; + let mut memo = CompatibilityMemo::default(); + let key = |group| CompatibilityKey::new(MemoGroupId(group), &expected).unwrap(); + memo.remember(key(0), short.clone()); + memo.remember(key(1), long.clone()); + let enumerate = |left: &[PhysicalCandidateId], right: &[PhysicalCandidateId]| { + let costs = [8, 12]; + left.iter() + .flat_map(|a| { + right.iter().map(move |b| { + let cost = costs[a.0] + if a == b { 0 } else { costs[b.0] } + 2; + ((a.0, b.0), cost) + }) + }) + .collect::>() + }; + for (right, group, optimum) in [(&short, 0, 10), (&long, 1, 14)] { + let reference = enumerate(&short, right); + let factored = enumerate(&memo.get(&key(0)).unwrap(), &memo.get(&key(group)).unwrap()); + assert_eq!(factored, reference); + assert_eq!(factored.values().min(), Some(&optimum)); + assert_eq!( + factored.get(&(0, 1)), + Some(&22), + "two distinct producer bindings remain feasible" + ); + } + } +} diff --git a/crates/plan-selection/src/search/candidates/graph.rs b/crates/plan-selection/src/search/candidates/graph.rs index 9171b25ae..f8b55e82a 100644 --- a/crates/plan-selection/src/search/candidates/graph.rs +++ b/crates/plan-selection/src/search/candidates/graph.rs @@ -1,4 +1,5 @@ //! Candidate dependency graph, declared constraints, and structural validation. +use super::suppliers::SupplierCandidates; use asap_logical_asap_optimization::memo::MemoGroupId; use std::collections::HashSet; @@ -9,7 +10,7 @@ pub(crate) struct PhysicalCandidateId(pub usize); /// One required input, distinct even when another input accepts the same producers. #[derive(Debug, Clone)] pub(crate) struct ProducerRequirement { - pub compatible_producers: Vec, + pub compatible_producers: SupplierCandidates, } /// One executable choice with costs already expressed over the same fixed horizon. diff --git a/crates/plan-selection/src/search/candidates/inventory.rs b/crates/plan-selection/src/search/candidates/inventory.rs index 7cc54563d..e8a53b772 100644 --- a/crates/plan-selection/src/search/candidates/inventory.rs +++ b/crates/plan-selection/src/search/candidates/inventory.rs @@ -14,6 +14,7 @@ use asap_logical_asap_optimization::memo::MemoGroupId; use asap_types::ir::OperatorNode; use asap_types::physical::Materialization; +use crate::search::candidates::factoring::{CompatibilityKey, CompatibilityMemo}; use crate::search::candidates::graph::{ CandidateDependencyGraph, PhysicalCandidate, PhysicalCandidateId, ProducerRequirement, QueryRootCandidates, @@ -64,10 +65,28 @@ pub(in crate::search) fn build_candidate_inventory( } pub(in crate::search) fn build_candidate_inventory_before( + memo: &CandidateMemo, + output_groups: &[MemoGroupId], + node_local_cost: impl FnMut(&OperatorNode) -> Result, + deadline: Option, +) -> Result { + let mut inventory = build_with_compatibility_memo( + memo, + output_groups, + node_local_cost, + deadline, + &mut CompatibilityMemo::default(), + )?; + crate::search::candidates::factoring::factor_terminal_readers(&mut inventory, deadline)?; + Ok(inventory) +} + +fn build_with_compatibility_memo( memo: &CandidateMemo, output_groups: &[MemoGroupId], mut node_local_cost: impl FnMut(&OperatorNode) -> Result, deadline: Option, + compatibility: &mut CompatibilityMemo, ) -> Result { super::check_deadline(deadline)?; let invalid = |message: &str| CandidateInventoryError::InvalidMemo(message.into()); @@ -178,6 +197,7 @@ pub(in crate::search) fn build_candidate_inventory_before( } let mut builder = ImmediateCandidateBuilder { inventory: &mut inventory, + compatibility: &mut *compatibility, boundaries, private_nodes: HashMap::new(), used_boundaries: HashSet::new(), @@ -228,6 +248,7 @@ pub(in crate::search) fn build_candidate_inventory_before( struct ImmediateCandidateBuilder<'a, F> { inventory: &'a mut CandidateInventory, + compatibility: &'a mut CompatibilityMemo, boundaries: HashMap<*const OperatorNode, MemoGroupId>, private_nodes: HashMap<*const OperatorNode, PhysicalCandidateId>, used_boundaries: HashSet, @@ -263,22 +284,30 @@ impl Result> ImmediateCandidateBuilder<' for input in inputs { let producers = if let Some(&group) = self.boundaries.get(&Rc::as_ptr(&input)) { self.used_boundaries.insert(group); - let mut compatible = Vec::new(); - for &candidate in &self.inventory.group_candidates[group.0 as usize] { - super::check_deadline(self.deadline)?; - let template = - self.inventory.templates[candidate.0] - .as_ref() - .ok_or_else(|| { - CandidateInventoryError::InvalidMemo("missing root template".into()) - })?; - if compatible_output(&input, template)? { - compatible.push(candidate); + let key = CompatibilityKey::new(group, &input) + .map_err(CandidateInventoryError::InvalidMemo)?; + if let Some(compatible) = self.compatibility.get(&key) { + compatible + } else { + let mut compatible = Vec::new(); + for &candidate in &self.inventory.group_candidates[group.0 as usize] { + super::check_deadline(self.deadline)?; + let template = + self.inventory.templates[candidate.0] + .as_ref() + .ok_or_else(|| { + CandidateInventoryError::InvalidMemo( + "missing root template".into(), + ) + })?; + if compatible_output(&input, template)? { + compatible.push(candidate); + } } + self.compatibility.remember(key, compatible) } - compatible } else { - vec![self.add(&input, None)?] + vec![self.add(&input, None)?].into() }; requirements.push(ProducerRequirement { compatible_producers: producers, @@ -365,6 +394,44 @@ mod tests { ) } + // Reusing a typed compatibility result must leave every activation and + // connection decision unchanged, including shared-producer cost ownership. + #[test] + fn compatibility_memo_preserves_complete_candidate_graph_and_optimum() { + let (memo, outputs) = quantile_memo(); + let cost = |node: &OperatorNode| Ok(if is_kll(node) { 8.0 } else { 1.0 }); + let mut compatibility = CompatibilityMemo::default(); + let cached = + build_with_compatibility_memo(&memo, &outputs, cost, None, &mut compatibility).unwrap(); + assert!( + compatibility.hits() > 0, + "repeated typed supplier checks were actually reused" + ); + let uncached = build_with_compatibility_memo( + &memo, + &outputs, + cost, + None, + &mut CompatibilityMemo::disabled(), + ) + .unwrap(); + assert_eq!( + format!("{:?}", cached.graph), + format!("{:?}", uncached.graph) + ); + assert_eq!(cached.group_candidates, uncached.group_candidates); + assert_eq!(cached.templates.len(), uncached.templates.len()); + for (cached, uncached) in cached.templates.iter().zip(&uncached.templates) { + assert!( + Rc::ptr_eq(cached.as_ref().unwrap(), uncached.as_ref().unwrap()) + || cached.as_ref().unwrap() == uncached.as_ref().unwrap() + ); + } + let cached_plan = select_candidates(&cached.graph).unwrap().unwrap(); + let uncached_plan = select_candidates(&uncached.graph).unwrap().unwrap(); + assert_eq!(cached_plan.modeled_cost, uncached_plan.modeled_cost); + } + fn is_kll(node: &OperatorNode) -> bool { matches!(node.asap(), Some(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. }) if *kind.algorithm() == SketchAlgorithm::Kll) } @@ -1144,7 +1211,41 @@ mod tests { (try_populate_pass1(initial, &generated).unwrap(), outputs) } + #[test] + fn terminal_reader_factoring_reconstructs_the_same_minimum_cost_computation() { + let (memo, outputs) = nested_sum_memo(false); + let original = build_with_compatibility_memo( + &memo, + &outputs, + |_| Ok(1.0), + None, + &mut CompatibilityMemo::default(), + ) + .unwrap(); + let compact = build_candidate_inventory(&memo, &outputs, |_| Ok(1.0)).unwrap(); + assert!(compact.graph.candidates.len() < original.graph.candidates.len()); + let before = select_candidates(&original.graph).unwrap().unwrap(); + let after = select_candidates(&compact.graph).unwrap().unwrap(); + assert_eq!(before.modeled_cost, after.modeled_cost); + for (inventory, selected) in [(&original, before), (&compact, after)] { + let roots = + reconstruct_candidate_outputs(&inventory.graph, &selected, &inventory.templates) + .unwrap(); + assert_eq!(roots.len(), 1); + roots[0].validate_structure().unwrap(); + assert!( + crate::search::candidates::reconstruction::output_contract_satisfied( + &memo.groups[outputs[0].0 as usize].node, + &roots[0] + ) + .unwrap() + ); + } + } + // Fusion owns its private work but does not publish another logical result. + // Inspect the uncompressed catalog here because the fixture identifies its + // particular fusion rule by its original memo ordinal. #[test] fn fused_nested_sum_activates_only_its_actual_input_groups() { let (memo, outputs) = nested_sum_memo(false); @@ -1154,7 +1255,14 @@ mod tests { .position(|candidate| candidate.implements.len() > 1) .expect("nested sum has an absorbed computation"); let absorbed = memo.candidates[owner][fused_index].implements[1]; - let mut inventory = build_candidate_inventory(&memo, &outputs, |_| Ok(1.0)).unwrap(); + let mut inventory = build_with_compatibility_memo( + &memo, + &outputs, + |_| Ok(1.0), + None, + &mut CompatibilityMemo::default(), + ) + .unwrap(); let fused = inventory.group_candidates[owner][fused_index]; assert!(!inventory.group_candidates[absorbed.0 as usize].contains(&fused)); assert!( @@ -1194,7 +1302,14 @@ mod tests { candidate.node.contains_asap() && !candidate.children.contains(&inner) }) .expect("fusion can privately recompute the shared inner work"); - let mut inventory = build_candidate_inventory(&memo, &outputs, |_| Ok(1.0)).unwrap(); + let mut inventory = build_with_compatibility_memo( + &memo, + &outputs, + |_| Ok(1.0), + None, + &mut CompatibilityMemo::default(), + ) + .unwrap(); let fused = inventory.group_candidates[owner][fused_index]; inventory.graph.query_roots[0].compatible_candidates = vec![fused]; assert!(!inventory.graph.query_roots[1] diff --git a/crates/plan-selection/src/search/candidates/mod.rs b/crates/plan-selection/src/search/candidates/mod.rs index 71f45ada4..21cc2ea45 100644 --- a/crates/plan-selection/src/search/candidates/mod.rs +++ b/crates/plan-selection/src/search/candidates/mod.rs @@ -1,10 +1,12 @@ //! Candidate graph representation and physical implementation construction. pub(crate) mod configurations; +mod factoring; pub(crate) mod graph; pub(crate) mod inventory; pub(crate) mod physical; pub(crate) mod reconstruction; pub(crate) mod scopes; +pub(crate) mod suppliers; /// Discard partially built catalogs on timeout; they cannot certify coverage. fn check_deadline( diff --git a/crates/plan-selection/src/search/candidates/physical.rs b/crates/plan-selection/src/search/candidates/physical.rs index a337d53a6..994abfc7d 100644 --- a/crates/plan-selection/src/search/candidates/physical.rs +++ b/crates/plan-selection/src/search/candidates/physical.rs @@ -328,7 +328,7 @@ mod tests { group: MemoGroupId(1), local_cost: 0.0, producer_requirements: vec![ProducerRequirement { - compatible_producers: vec![PhysicalCandidateId(0)], + compatible_producers: vec![PhysicalCandidateId(0)].into(), }], }); } diff --git a/crates/plan-selection/src/search/candidates/reconstruction.rs b/crates/plan-selection/src/search/candidates/reconstruction.rs index 71642266b..81389bc1f 100644 --- a/crates/plan-selection/src/search/candidates/reconstruction.rs +++ b/crates/plan-selection/src/search/candidates/reconstruction.rs @@ -299,7 +299,7 @@ mod tests { let producer_requirements = ordered_inputs(node) .iter() .map(|child| ProducerRequirement { - compatible_producers: vec![add(child, graph, templates)], + compatible_producers: vec![add(child, graph, templates)].into(), }) .collect(); let id = templates.len(); @@ -366,8 +366,9 @@ mod tests { template.operator.map_children(|_| Rc::clone(&child)), template.schema.clone(), ))); - graph.candidates[root].producer_requirements[0].compatible_producers = vec![first]; - graph.candidates[root].producer_requirements[1].compatible_producers = vec![duplicate]; + graph.candidates[root].producer_requirements[0].compatible_producers = vec![first].into(); + graph.candidates[root].producer_requirements[1].compatible_producers = + vec![duplicate].into(); let selected = select_candidates(&graph).unwrap().unwrap(); let outputs = reconstruct_candidate_outputs(&graph, &selected, &templates).unwrap(); let children = outputs[0].children(); diff --git a/crates/plan-selection/src/search/candidates/scopes.rs b/crates/plan-selection/src/search/candidates/scopes.rs index 707a8b745..83353c869 100644 --- a/crates/plan-selection/src/search/candidates/scopes.rs +++ b/crates/plan-selection/src/search/candidates/scopes.rs @@ -2,6 +2,7 @@ //! Copies are interchangeable, not assigned to disjoint reader sets: one query //! may consume multiple copies while different subsets share other inputs. //! Activation-prefix constraints remove permutations of unused copies only. +use super::suppliers::{SupplierCandidates, SupplierSetInterner}; use crate::search::candidates::configurations::{ CandidateConfigurationInventory, ConfigurationError, }; @@ -10,6 +11,7 @@ use crate::search::candidates::graph::{ ReaderConstraint, }; use crate::search::candidates::inventory::CandidateInventory; +use std::collections::HashMap; #[cfg(test)] pub(in crate::search) fn expand_execution_scopes( @@ -42,6 +44,18 @@ pub(in crate::search) fn expand_execution_scopes_before( "execution-scope output contexts are incomplete".into(), )); } + let mut internally_consumed = vec![false; count]; + for input in graph + .candidates + .iter() + .flat_map(|candidate| &candidate.producer_requirements) + { + super::check_deadline(deadline)?; + for producer in &input.compatible_producers { + super::check_deadline(deadline)?; + internally_consumed[producer.0] = true; + } + } let mut output_uses = vec![std::collections::HashSet::new(); count]; let mut output_positions = vec![0usize; count]; for (output, root) in graph.query_roots.iter().enumerate() { @@ -103,6 +117,32 @@ pub(in crate::search) fn expand_execution_scopes_before( } else { contexts.max(uses) }; + let terminal_unary_reader = !internally_consumed[id] + && candidate.producer_requirements.len() == 1 + && source + .inventory + .templates + .get(id) + .and_then(Option::as_ref) + .is_some_and(|node| { + matches!( + node.asap(), + Some( + asap_types::ir::ASAPOp::SummaryEstimate { .. } + | asap_types::ir::ASAPOp::FinalizeExactAccumulator { .. } + ) + ) && node.children().len() == 1 + }); + // Factored terminal readouts may need a different binding per output + // reference, including two references in one scalar query. Every active + // terminal copy must serve a root, so the reference count supplies a + // finite sufficient capacity. Copies remain optional and can be shared. + // Internal consumers retain the original occurrence/context rule. + let multiplicity = if terminal_unary_reader { + multiplicity.max(output_positions[id]) + } else { + multiplicity + }; for _ in 0..multiplicity { let copy = PhysicalCandidateId(result.configurations.len()); if let Some(&previous) = copies[id].last() { @@ -122,22 +162,34 @@ pub(in crate::search) fn expand_execution_scopes_before( result.inventory.graph.candidates.push(candidate.clone()); } } + let mut supplier_sets = SupplierSetInterner::default(); + let mut input_cache: HashMap = HashMap::new(); for (id, candidate) in graph.candidates.iter().enumerate() { super::check_deadline(deadline)?; + let mut requirements = Vec::new(); + for input in &candidate.producer_requirements { + let compatible_producers = + if let Some(suppliers) = input_cache.get(&input.compatible_producers) { + suppliers.clone() + } else { + let mut compatible = Vec::new(); + for producer in &input.compatible_producers { + super::check_deadline(deadline)?; + compatible.extend(copies[producer.0].iter().copied()); + } + let suppliers = supplier_sets.intern(compatible); + input_cache.insert(input.compatible_producers.clone(), suppliers.clone()); + suppliers + }; + requirements.push(ProducerRequirement { + compatible_producers, + }); + } + // Consumer copies retain distinct requirement slots and MILP edges; + // only their immutable compatible-supplier storage is shared. for © in &copies[id] { super::check_deadline(deadline)?; - let mut requirements = Vec::new(); - for input in &candidate.producer_requirements { - let mut compatible_producers = Vec::new(); - for producer in &input.compatible_producers { - super::check_deadline(deadline)?; - compatible_producers.extend(copies[producer.0].iter().copied()); - } - requirements.push(ProducerRequirement { - compatible_producers, - }); - } - result.inventory.graph.candidates[copy.0].producer_requirements = requirements; + result.inventory.graph.candidates[copy.0].producer_requirements = requirements.clone(); } } result.inventory.graph.query_roots = graph @@ -225,6 +277,203 @@ mod tests { use asap_types::ir::properties::ExecutionTiming; use asap_types::physical::Materialization; + fn terminal_reader_inventory(separate_suppliers: bool) -> CandidateConfigurationInventory { + use asap_logical_asap_optimization::test_support::{cms_summary, estimate_over}; + use asap_types::ir::properties::AccuracyGuarantee; + use std::rc::Rc; + let state = Rc::new(cms_summary("count(m)")); + let reader = estimate_over(state, AccuracyGuarantee::exact("scope fixture")); + let mut graph = CandidateDependencyGraph { + candidates: vec![ + PhysicalCandidate { + group: MemoGroupId(0), + local_cost: 2.0, + producer_requirements: vec![], + }, + PhysicalCandidate { + group: MemoGroupId(0), + local_cost: 3.0, + producer_requirements: vec![], + }, + PhysicalCandidate { + group: MemoGroupId(1), + local_cost: 1.0, + producer_requirements: vec![ProducerRequirement { + compatible_producers: vec![PhysicalCandidateId(0), PhysicalCandidateId(1)] + .into(), + }], + }, + ], + query_roots: vec![ + QueryRootCandidates { + compatible_candidates: vec![PhysicalCandidateId(2)] + }; + 2 + ], + ..Default::default() + }; + if separate_suppliers { + for output in 0..2 { + graph + .reader_constraints + .push(ReaderConstraint::AllowedOutputs { + candidate: PhysicalCandidateId(output), + outputs: vec![output], + }); + } + } + CandidateConfigurationInventory { + inventory: CandidateInventory { + graph, + templates: vec![None, None, Some(reader)], + group_candidates: vec![ + vec![PhysicalCandidateId(0), PhysicalCandidateId(1)], + vec![PhysicalCandidateId(2)], + ], + }, + configurations: (0..3) + .map(|id| CandidateConfiguration { + source_candidate: PhysicalCandidateId(id), + materialization: Materialization::NotMaterialized, + timing: ExecutionTiming::QueryTime, + }) + .collect(), + } + } + + // Fix the producer catalog independently of planning budgets and progressive + // pane domains. Root-only readers need one optional binding per output; + // producers stay constant and producer connections grow exactly linearly. + // Output-to-copy assignment choices are intentionally N*N, not claimed linear. + #[test] + fn repeated_outputs_have_constant_producers_and_linear_reader_connections() { + for output_count in [2, 4, 8] { + let mut source = terminal_reader_inventory(false); + source.inventory.graph.query_roots = vec![ + QueryRootCandidates { + compatible_candidates: vec![PhysicalCandidateId(2)] + }; + output_count + ]; + let expanded = + expand_execution_scopes(source, &[1, 1, 1], &vec![0; output_count]).unwrap(); + let graph = &expanded.inventory.graph; + let producer_count = expanded + .configurations + .iter() + .filter(|configuration| configuration.source_candidate.0 < 2) + .count(); + let terminal_count = expanded + .configurations + .iter() + .filter(|configuration| configuration.source_candidate == PhysicalCandidateId(2)) + .count(); + let producer_connections: usize = graph + .candidates + .iter() + .flat_map(|candidate| &candidate.producer_requirements) + .map(|input| input.compatible_producers.len()) + .sum(); + assert_eq!(producer_count, 2); + assert_eq!(terminal_count, output_count); + assert_eq!(graph.candidates.len(), 2 + output_count); + assert_eq!(producer_connections, 2 * output_count); + assert_eq!( + graph + .query_roots + .iter() + .map(|output| output.compatible_candidates.len()) + .sum::(), + output_count * output_count + ); + let selected = select_candidates(graph).unwrap().unwrap(); + assert_eq!(selected.query_roots.len(), output_count); + assert!(selected + .query_roots + .iter() + .all(|output| output.candidate == selected.query_roots[0].candidate)); + assert_eq!( + selected.modeled_cost, 3.0, + "additional capacity does not force duplicate work" + ); + } + } + + // Terminal-reader factoring must preserve independent bindings for two + // output references of the same scalar query, not only distinct queries. + #[test] + fn terminal_reader_copies_preserve_same_query_independent_bindings() { + let source = terminal_reader_inventory(true); + assert!(select_candidates(&source.inventory.graph) + .unwrap() + .is_none()); + let expanded = expand_execution_scopes(source, &[1, 1, 1], &[0, 0]).unwrap(); + let selected = select_candidates(&expanded.inventory.graph) + .unwrap() + .expect("each root reference can bind a separate reader"); + assert_ne!( + selected.query_roots[0].candidate, + selected.query_roots[1].candidate + ); + let suppliers = selected + .connections + .iter() + .map(|edge| expanded.configurations[edge.producer.0].source_candidate) + .collect::>(); + assert_eq!( + suppliers, + std::collections::HashSet::from([PhysicalCandidateId(0), PhysicalCandidateId(1)]) + ); + assert_eq!(selected.modeled_cost, 7.0); + + // Capacity is optional: without conflicting eligibility, both root + // references may still consume one reader and pay its cost once. + let shared = + expand_execution_scopes(terminal_reader_inventory(false), &[1, 1, 1], &[0, 0]).unwrap(); + let selected = select_candidates(&shared.inventory.graph).unwrap().unwrap(); + assert_eq!( + selected.query_roots[0].candidate, + selected.query_roots[1].candidate + ); + assert_eq!(selected.modeled_cost, 3.0); + } + + // The terminal-readout bound does not alter capacity for a reader that + // supplies another candidate; that requires a separate dependency proof. + #[test] + fn terminal_capacity_rule_leaves_internal_readers_unchanged() { + let mut source = terminal_reader_inventory(false); + source.inventory.graph.candidates.push(PhysicalCandidate { + group: MemoGroupId(2), + local_cost: 1.0, + producer_requirements: vec![ProducerRequirement { + compatible_producers: vec![PhysicalCandidateId(2)].into(), + }], + }); + source.inventory.templates.push(None); + source + .inventory + .group_candidates + .push(vec![PhysicalCandidateId(3)]); + source.configurations.push(CandidateConfiguration { + source_candidate: PhysicalCandidateId(3), + materialization: Materialization::NotMaterialized, + timing: ExecutionTiming::QueryTime, + }); + for output in &mut source.inventory.graph.query_roots { + output.compatible_candidates = vec![PhysicalCandidateId(3)]; + } + let expanded = expand_execution_scopes(source, &[1; 4], &[0, 0]).unwrap(); + assert_eq!( + expanded + .configurations + .iter() + .filter(|configuration| configuration.source_candidate == PhysicalCandidateId(2)) + .count(), + 1 + ); + } + // The unknown-phase query reads both states; the two known-phase queries // share different copies. Another independent state remains shared by all. #[test] @@ -240,10 +489,10 @@ mod tests { } else { vec![ ProducerRequirement { - compatible_producers: vec![id(0)], + compatible_producers: vec![id(0)].into(), }, ProducerRequirement { - compatible_producers: vec![id(1)], + compatible_producers: vec![id(1)].into(), }, ] }, @@ -263,6 +512,7 @@ mod tests { right: 3, }); assert!(select_candidates(&graph).unwrap().is_none()); + let original_graph = graph.clone(); let source = CandidateConfigurationInventory { inventory: CandidateInventory { graph, @@ -278,6 +528,43 @@ mod tests { .collect(), }; let expanded = expand_execution_scopes(source, &[2, 2, 2, 2], &[0, 1, 2, 3]).unwrap(); + // Copy expansion must preserve every slot's complete ordered supplier + // mapping even though repeated copies share the backing allocation. + for (id, configuration) in expanded.configurations.iter().enumerate() { + let original = &original_graph.candidates[configuration.source_candidate.0]; + let copied = &expanded.inventory.graph.candidates[id]; + assert_eq!( + original.producer_requirements.len(), + copied.producer_requirements.len() + ); + for (input, copied_input) in original + .producer_requirements + .iter() + .zip(&copied.producer_requirements) + { + let expected: Vec<_> = input + .compatible_producers + .iter() + .flat_map(|producer| { + expanded.configurations.iter().enumerate().filter_map( + move |(id, configuration)| { + (configuration.source_candidate == *producer) + .then_some(PhysicalCandidateId(id)) + }, + ) + }) + .collect(); + assert_eq!( + copied_input.compatible_producers.as_slice(), + expected.as_slice() + ); + } + } + let metrics = super::super::suppliers::supplier_storage_metrics(&expanded.inventory.graph); + assert!( + metrics.stored_supplier_ids < metrics.logical_connections, + "{metrics:?}" + ); let selected = select_candidates(&expanded.inventory.graph) .unwrap() .unwrap(); diff --git a/crates/plan-selection/src/search/candidates/suppliers.rs b/crates/plan-selection/src/search/candidates/suppliers.rs new file mode 100644 index 000000000..322fce580 --- /dev/null +++ b/crates/plan-selection/src/search/candidates/suppliers.rs @@ -0,0 +1,200 @@ +//! Shared storage for ordered compatible suppliers. Sharing this storage never +//! ties together the independent selections of different consumer input slots. +use super::graph::PhysicalCandidateId; +use std::collections::HashSet; +use std::ops::{Deref, DerefMut}; +use std::rc::Rc; + +#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)] +pub(crate) struct SupplierCandidates(Rc>); + +impl From> for SupplierCandidates { + fn from(values: Vec) -> Self { + Self(Rc::new(values)) + } +} + +impl FromIterator for SupplierCandidates { + fn from_iter>(values: T) -> Self { + Self::from(values.into_iter().collect::>()) + } +} + +impl Deref for SupplierCandidates { + type Target = Vec; + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl DerefMut for SupplierCandidates { + fn deref_mut(&mut self) -> &mut Self::Target { + Rc::make_mut(&mut self.0) + } +} + +impl<'a> IntoIterator for &'a SupplierCandidates { + type Item = &'a PhysicalCandidateId; + type IntoIter = std::slice::Iter<'a, PhysicalCandidateId>; + fn into_iter(self) -> Self::IntoIter { + self.iter() + } +} + +impl<'a> IntoIterator for &'a mut SupplierCandidates { + type Item = &'a mut PhysicalCandidateId; + type IntoIter = std::slice::IterMut<'a, PhysicalCandidateId>; + fn into_iter(self) -> Self::IntoIter { + self.iter_mut() + } +} + +impl IntoIterator for SupplierCandidates { + type Item = PhysicalCandidateId; + type IntoIter = std::vec::IntoIter; + fn into_iter(self) -> Self::IntoIter { + Rc::unwrap_or_clone(self.0).into_iter() + } +} + +/// Exact ordered-content interning. No sorting, deduplication, or compatibility +/// inference occurs here; malformed input remains visible to graph validation. +#[derive(Default)] +pub(in crate::search) struct SupplierSetInterner { + sets: HashSet, +} + +impl SupplierSetInterner { + pub(in crate::search) fn intern( + &mut self, + values: impl Into, + ) -> SupplierCandidates { + let values = values.into(); + if let Some(existing) = self.sets.get(&values) { + return existing.clone(); + } + self.sets.insert(values.clone()); + values + } +} + +/// Count physical supplier-list storage separately from logical connections. +/// Distinct input positions remain distinct even when their storage is shared. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub(in crate::search) struct SupplierStorageMetrics { + pub logical_connections: usize, + pub unique_supplier_sets: usize, + pub stored_supplier_ids: usize, +} + +pub(in crate::search) fn supplier_storage_metrics( + graph: &super::graph::CandidateDependencyGraph, +) -> SupplierStorageMetrics { + let mut allocations = HashSet::new(); + let mut metrics = SupplierStorageMetrics::default(); + for input in graph + .candidates + .iter() + .flat_map(|candidate| &candidate.producer_requirements) + { + metrics.logical_connections += input.compatible_producers.len(); + if allocations.insert(Rc::as_ptr(&input.compatible_producers.0)) { + metrics.unique_supplier_sets += 1; + metrics.stored_supplier_ids += input.compatible_producers.len(); + } + } + metrics +} + +#[cfg(test)] +mod tests { + use super::*; + + // Two slots can choose different producers even when their compatible lists + // share one allocation; storage compression adds no selection equality. + #[test] + fn shared_supplier_storage_keeps_input_choices_independent() { + use crate::search::candidates::graph::{ + validate_selected_graph, CandidateDependencyGraph, PhysicalCandidate, + ProducerConsumerConnection, ProducerRequirement, QueryRootAssignment, + QueryRootCandidates, SelectedCandidateGraph, + }; + use asap_logical_asap_optimization::memo::MemoGroupId; + let mut interner = SupplierSetInterner::default(); + let choices = interner.intern(vec![PhysicalCandidateId(0), PhysicalCandidateId(1)]); + let graph = CandidateDependencyGraph { + candidates: (0..3) + .map(|id| PhysicalCandidate { + group: MemoGroupId(id), + local_cost: 0.0, + producer_requirements: if id == 2 { + vec![ + ProducerRequirement { + compatible_producers: choices.clone(), + }, + ProducerRequirement { + compatible_producers: choices.clone(), + }, + ] + } else { + vec![] + }, + }) + .collect(), + query_roots: vec![QueryRootCandidates { + compatible_candidates: vec![PhysicalCandidateId(2)], + }], + ..Default::default() + }; + let selected = SelectedCandidateGraph { + selected_candidates: vec![ + PhysicalCandidateId(0), + PhysicalCandidateId(1), + PhysicalCandidateId(2), + ], + connections: (0..2) + .map(|requirement| ProducerConsumerConnection { + producer: PhysicalCandidateId(requirement), + consumer: PhysicalCandidateId(2), + requirement, + }) + .collect(), + query_roots: vec![QueryRootAssignment { + query_root: 0, + candidate: PhysicalCandidateId(2), + }], + modeled_cost: 0.0, + }; + graph.validate().unwrap(); + validate_selected_graph(&graph, &selected).unwrap(); + assert_eq!( + supplier_storage_metrics(&graph), + SupplierStorageMetrics { + logical_connections: 4, + unique_supplier_sets: 1, + stored_supplier_ids: 2 + } + ); + } + + // Equal ordered lists share one allocation; mutation of one input detaches + // it and cannot silently change another input or an interner key. + #[test] + fn interned_inputs_detach_on_mutation() { + let mut interner = SupplierSetInterner::default(); + let original = interner.intern(vec![PhysicalCandidateId(2), PhysicalCandidateId(1)]); + let mut changed = interner.intern(vec![PhysicalCandidateId(2), PhysicalCandidateId(1)]); + assert!(Rc::ptr_eq(&original.0, &changed.0)); + changed.retain(|id| id.0 == 1); + assert_eq!( + original.as_slice(), + &[PhysicalCandidateId(2), PhysicalCandidateId(1)] + ); + assert_eq!(changed.as_slice(), &[PhysicalCandidateId(1)]); + assert!(!Rc::ptr_eq(&original.0, &changed.0)); + let repeated = interner.intern(original.clone()); + assert!(Rc::ptr_eq(&original.0, &repeated.0)); + let reversed = interner.intern(vec![PhysicalCandidateId(1), PhysicalCandidateId(2)]); + assert!(!Rc::ptr_eq(&original.0, &reversed.0)); + } +} diff --git a/crates/plan-selection/src/search/costing/demand.rs b/crates/plan-selection/src/search/costing/demand.rs index 1cef0948f..6b1c716aa 100644 --- a/crates/plan-selection/src/search/costing/demand.rs +++ b/crates/plan-selection/src/search/costing/demand.rs @@ -232,7 +232,7 @@ mod tests { producer_requirements: inputs .iter() .map(|&id| ProducerRequirement { - compatible_producers: vec![PhysicalCandidateId(id)], + compatible_producers: vec![PhysicalCandidateId(id)].into(), }) .collect(), }) diff --git a/crates/plan-selection/src/search/costing/lower_bounds.rs b/crates/plan-selection/src/search/costing/lower_bounds.rs index 0e0713d41..f3e70efdf 100644 --- a/crates/plan-selection/src/search/costing/lower_bounds.rs +++ b/crates/plan-selection/src/search/costing/lower_bounds.rs @@ -740,7 +740,7 @@ mod tests { producer_requirements: inputs .into_iter() .map(|input| ProducerRequirement { - compatible_producers: vec![PhysicalCandidateId(input)], + compatible_producers: vec![PhysicalCandidateId(input)].into(), }) .collect(), }); @@ -776,7 +776,7 @@ mod tests { for candidate in &mut configured.inventory.graph.candidates { for input in &mut candidate.producer_requirements { if chains.contains(&input.compatible_producers[0]) { - input.compatible_producers = chains.clone(); + input.compatible_producers = chains.clone().into(); } } } @@ -874,7 +874,7 @@ mod tests { group: MemoGroupId(id.0 as u32), local_cost: 0.0, producer_requirements: vec![ProducerRequirement { - compatible_producers: vec![producer], + compatible_producers: vec![producer].into(), }], }); producer = id; @@ -883,7 +883,7 @@ mod tests { } for chain in chains { configured.inventory.graph.candidates[chain].producer_requirements[0] - .compatible_producers = projects.clone(); + .compatible_producers = projects.clone().into(); let child = configured.inventory.templates[projects[0].0] .as_ref() .unwrap() @@ -951,7 +951,7 @@ mod tests { group: MemoGroupId(id.0 as u32), local_cost: 0.0, producer_requirements: vec![ProducerRequirement { - compatible_producers: vec![*scan], + compatible_producers: vec![*scan].into(), }], }); projects.push(id); @@ -966,7 +966,7 @@ mod tests { // former arbitrary representative selection. let mut suppliers = scans.clone(); suppliers.extend(projects.iter().copied()); - input.compatible_producers = suppliers; + input.compatible_producers = suppliers.into(); } } let models = PlanningModels::builtin(); @@ -1283,7 +1283,7 @@ mod tests { node.as_ref().unwrap().asap(), Some(ASAPOp::PaneChain { .. }) ) { - candidate.producer_requirements[0].compatible_producers = scans.clone(); + candidate.producer_requirements[0].compatible_producers = scans.clone().into(); } } let context = CandidatePlanningContext { diff --git a/crates/plan-selection/src/search/refinement/components.rs b/crates/plan-selection/src/search/refinement/components.rs index 227e17ccd..8141ab858 100644 --- a/crates/plan-selection/src/search/refinement/components.rs +++ b/crates/plan-selection/src/search/refinement/components.rs @@ -11,6 +11,7 @@ use crate::search::candidates::graph::{ ProducerConsumerConnection, QueryRootAssignment, ReaderConstraint, SelectedCandidateGraph, }; use crate::search::candidates::inventory::CandidateInventory; +use crate::search::candidates::suppliers::{SupplierCandidates, SupplierSetInterner}; use crate::search::milp::model::CandidateSolveMetrics; #[cfg(test)] use crate::search::refinement::planner::evaluate_configured_candidate; @@ -198,6 +199,8 @@ fn project( .map(|(local, &global)| (global, local)) .collect(); let graph = &configured.inventory.graph; + let mut supplier_sets = SupplierSetInterner::default(); + let mut remapped: HashMap = HashMap::new(); let mut projected = CandidateDependencyGraph { candidates: component .candidates @@ -205,9 +208,18 @@ fn project( .map(|&id| { let mut candidate = graph.candidates[id].clone(); for input in &mut candidate.producer_requirements { - for id in &mut input.compatible_producers { - id.0 = candidates[&id.0]; - } + input.compatible_producers = remapped + .entry(input.compatible_producers.clone()) + .or_insert_with(|| { + supplier_sets.intern( + input + .compatible_producers + .iter() + .map(|id| PhysicalCandidateId(candidates[&id.0])) + .collect::>(), + ) + }) + .clone(); } candidate }) diff --git a/crates/plan-selection/src/search/refinement/materialization.rs b/crates/plan-selection/src/search/refinement/materialization.rs index ed1325f50..ba2a3620f 100644 --- a/crates/plan-selection/src/search/refinement/materialization.rs +++ b/crates/plan-selection/src/search/refinement/materialization.rs @@ -90,7 +90,7 @@ pub(super) fn restrict( if !requirement.compatible_producers.contains(&producer) { return None; } - requirement.compatible_producers = vec![producer]; + requirement.compatible_producers = vec![producer].into(); } } for root in &selected.query_roots { @@ -131,7 +131,8 @@ mod tests { group: MemoGroupId(query as u32 + 1), local_cost: 0.0, producer_requirements: vec![ProducerRequirement { - compatible_producers: vec![PhysicalCandidateId(0), PhysicalCandidateId(1)], + compatible_producers: vec![PhysicalCandidateId(0), PhysicalCandidateId(1)] + .into(), }], }); graph.query_roots.push(QueryRootCandidates { diff --git a/crates/plan-selection/src/search/refinement/planner.rs b/crates/plan-selection/src/search/refinement/planner.rs index bcbe94152..64b819483 100644 --- a/crates/plan-selection/src/search/refinement/planner.rs +++ b/crates/plan-selection/src/search/refinement/planner.rs @@ -1340,7 +1340,7 @@ mod tests { let id = PhysicalCandidateId(configured.configurations.len()); let mut candidate = configured.inventory.graph.candidates[supplier.0].clone(); candidate.producer_requirements = vec![ProducerRequirement { - compatible_producers: vec![supplier], + compatible_producers: vec![supplier].into(), }]; configured.inventory.group_candidates[candidate.group.0 as usize].push(id); configured.inventory.graph.candidates.push(candidate); diff --git a/crates/plan-selection/src/search/refinement/pruning.rs b/crates/plan-selection/src/search/refinement/pruning.rs index 2744f93c1..6f7aa771b 100644 --- a/crates/plan-selection/src/search/refinement/pruning.rs +++ b/crates/plan-selection/src/search/refinement/pruning.rs @@ -5,6 +5,8 @@ use crate::search::candidates::configurations::CandidateConfigurationInventory; use crate::search::candidates::graph::{ CandidateDependencyGraph, PhysicalCandidateId, ReaderConstraint, }; +use crate::search::candidates::suppliers::{SupplierCandidates, SupplierSetInterner}; +use std::collections::HashMap; #[derive(Debug, Default)] pub(in crate::search) struct PruningMetrics { @@ -37,6 +39,8 @@ pub(in crate::search) fn prune( .map(|input| input.compatible_producers.len()) .sum(); let graph = &mut configured.inventory.graph; + let mut supplier_sets = SupplierSetInterner::default(); + let mut remapped: HashMap = HashMap::new(); graph.candidates = std::mem::take(&mut graph.candidates) .into_iter() .enumerate() @@ -45,11 +49,18 @@ pub(in crate::search) fn prune( return None; } for input in &mut candidate.producer_requirements { - input.compatible_producers = input - .compatible_producers - .iter() - .filter_map(|id| mapping[id.0]) - .collect(); + input.compatible_producers = remapped + .entry(input.compatible_producers.clone()) + .or_insert_with(|| { + supplier_sets.intern( + input + .compatible_producers + .iter() + .filter_map(|id| mapping[id.0]) + .collect::>(), + ) + }) + .clone(); } Some(candidate) }) @@ -241,6 +252,51 @@ mod tests { } } + // Removing an unrelated candidate remaps IDs once per shared supplier + // list; the subsequent solve still sees independent consumer requirements. + #[test] + fn pruning_retains_shared_supplier_storage_after_remapping() { + let graph = CandidateDependencyGraph { + candidates: vec![ + candidate(9.0, &[]), + candidate(1.0, &[]), + candidate(1.0, &[&[1]]), + candidate(1.0, &[&[1]]), + ], + query_roots: vec![ + QueryRootCandidates { + compatible_candidates: vec![PhysicalCandidateId(2)], + }, + QueryRootCandidates { + compatible_candidates: vec![PhysicalCandidateId(3)], + }, + ], + ..Default::default() + }; + let original = graph.clone(); + let mut inventory = configured(graph); + assert_eq!(prune(&mut inventory).unwrap().removed_candidates, 1); + let graph = &inventory.inventory.graph; + assert_eq!( + graph.candidates[1].producer_requirements[0] + .compatible_producers + .as_slice(), + &[PhysicalCandidateId(0)] + ); + assert_eq!( + crate::search::candidates::suppliers::supplier_storage_metrics(graph), + crate::search::candidates::suppliers::SupplierStorageMetrics { + logical_connections: 2, + unique_supplier_sets: 1, + stored_supplier_ids: 1 + } + ); + assert_eq!( + select_candidates(graph).unwrap().unwrap().modeled_cost, + select_candidates(&original).unwrap().unwrap().modeled_cost + ); + } + // Dead dependency cycles and unused nodes disappear; a cycle with a finite // escape remains available. Cost and resource accounting survive remapping. #[test] diff --git a/crates/plan-selection/src/search/refinement/search.rs b/crates/plan-selection/src/search/refinement/search.rs index 8e39d3694..145a60c77 100644 --- a/crates/plan-selection/src/search/refinement/search.rs +++ b/crates/plan-selection/src/search/refinement/search.rs @@ -918,7 +918,7 @@ mod tests { graph.candidates[1] .producer_requirements .push(ProducerRequirement { - compatible_producers: vec![PhysicalCandidateId(0)], + compatible_producers: vec![PhysicalCandidateId(0)].into(), }); let result = search_candidates(&graph, limits(10), |selected| { Ok(Some(if selected.selected_candidates.len() == 1 { @@ -976,7 +976,7 @@ mod tests { graph.candidates[2] .producer_requirements .push(ProducerRequirement { - compatible_producers: vec![PhysicalCandidateId(0), PhysicalCandidateId(1)], + compatible_producers: vec![PhysicalCandidateId(0), PhysicalCandidateId(1)].into(), }); graph.query_roots = (0..3) .map(|id| QueryRootCandidates { diff --git a/crates/plan-selection/src/search/route.rs b/crates/plan-selection/src/search/route.rs index 5d44a33fc..f2e3b6e65 100644 --- a/crates/plan-selection/src/search/route.rs +++ b/crates/plan-selection/src/search/route.rs @@ -51,11 +51,14 @@ struct DomainProgress { catalog_completed: bool, refinements: usize, report: Option, + local_memo: Option, + previous_domain: + Option, } -/// A small catalog obtains a validated incumbent before full-width expansion. -/// Both phases use the same planner and frozen objective; only the complete -/// phase may establish a bound for the full declared domain. +/// Obtain an incumbent on seed grids, then progressively expose finer widths. +/// All rounds use the same planner and frozen objective; only a completed +/// full-domain round can establish a bound for the declared search space. pub(crate) fn candidate_connection_search( initial: LogicalMemo, context: &CandidateSearchContext<'_, Id>, @@ -72,8 +75,26 @@ pub(crate) fn candidate_connection_search( let total_ms = search.milp_time_limit_ms.unwrap_or(30_000); let prior_ms = local_candidate_generation_ms.unwrap_or(0.0); let calibration = CandidateCalibration::for_models(models).map_err(SelectionError::Solver)?; + // Pane rules require a temporal range below a summary build. Pass 1 may + // introduce the build, but cannot invent a missing positive Range window. + // Instant lookback selectors are not splittable panes. Existing pane + // chains also retain the width domain when callers provide lowered state. + // This conservative gate keeps SQL/PromQL windows regardless of arrival + // mode; recurrence alone does not make a window-free workload pane-capable. + let has_pane_window = initial.groups.iter().any(|group| { + matches!( + group.node.non_asap(), + Some(asap_types::ir::NonASAPOp::TimeRange { + kind: asap_types::ir::TimeRangeKind::Range, range, .. + }) if !range.is_zero() + ) || matches!( + group.node.asap(), + Some(asap_types::ir::ASAPOp::PaneChain { .. }) + ) + }); // Exhaustive archives and complete-catalog calibration require one domain. - let seed_enabled = !search.exhaustive + let seed_enabled = has_pane_window + && !search.exhaustive && !search.keep_every_candidate && matches!(calibration, CandidateCalibration::Frozen(_)) && prior_ms.is_finite() @@ -106,92 +127,193 @@ pub(crate) fn candidate_connection_search( &mut seed_archive, PaneWidthDomain::SeedGrids, &mut seed_progress, - ); + ) + .map(|plan| namespace_round(plan, "seed-")); let seed_elapsed = started.elapsed(); if seed.is_err() && !seed_progress.budget_stopped && !seed_progress.infeasible { return seed; } let seed_report = seed_progress.report.clone(); let full_started = Instant::now(); - let mut full_progress = DomainProgress::default(); - let full = search_domain( - initial, - context, - Stage1ToStage3Search { - max_evaluations: Some( - search - .max_evaluations - .unwrap_or(128) - .saturating_sub(seed_progress.refinements), - ), - ..search - }, - Some(prior_ms + seed_elapsed.as_secs_f64() * 1_000.0), - archive, - PaneWidthDomain::Complete, - &mut full_progress, + let mut full_progress = DomainProgress { + local_memo: seed_progress.local_memo.take(), + previous_domain: seed_progress.previous_domain, + ..Default::default() + }; + let global_deadline = started.checked_add( + std::time::Duration::from_millis(total_ms) + .saturating_sub(std::time::Duration::from_secs_f64(prior_ms / 1000.0)), ); + // A literal-only workload has no computation in which pane rules can fire. + let grids = context + .entries + .iter() + .filter_map(|entry| entry.evaluation_grid_ms()) + .filter(|_| !initial.groups.is_empty()); + let mut frontier = + asap_logical_asap_optimization::memo::state_configurations::PaneWidthFrontier::new( + grids, + global_deadline, + ); + let mut rounds = vec!["SeedGrids".to_owned()]; + let mut full = Err(SelectionError::Solver( + "pane-width expansion has not started".into(), + )); + let mut completed_full_domain = false; + loop { + let elapsed_ms = prior_ms + started.elapsed().as_secs_f64() * 1000.0; + let remaining_ms = total_ms.saturating_sub(elapsed_ms.ceil() as u64); + let remaining_evaluations = search + .max_evaluations + .unwrap_or(128) + .saturating_sub(seed_progress.refinements + full_progress.refinements); + if remaining_ms == 0 || remaining_evaluations == 0 { + break; + } + let domain = match &mut frontier { + Ok(frontier) => match frontier.next_domain(global_deadline) { + Ok(Some(domain)) => domain, + Ok(None) => break, + Err(error) => { + if full.is_err() { + full = Err(SelectionError::Solver(error.to_string())); + } + break; + } + }, + Err(error) => { + if full.is_err() { + full = Err(SelectionError::Solver(error.to_string())); + } + break; + } + }; + // Frontier enumeration is part of the same global budget. Never pass + // the pre-enumeration remainder as a fresh per-round allowance. + let remaining_ms = total_ms + .saturating_sub((prior_ms + started.elapsed().as_secs_f64() * 1000.0).ceil() as u64); + if remaining_ms == 0 { + break; + } + rounds.push(format!("{domain:?}")); + let complete = domain == PaneWidthDomain::Complete; + let mut round_progress = DomainProgress { + local_memo: full_progress.local_memo.take(), + previous_domain: full_progress.previous_domain, + ..Default::default() + }; + let round = search_domain( + initial.clone(), + context, + Stage1ToStage3Search { + milp_time_limit_ms: Some(if complete { + remaining_ms + } else { + remaining_ms.min(5_000) + }), + max_evaluations: Some(if complete { + remaining_evaluations + } else { + remaining_evaluations.min(16) + }), + ..search + }, + Some(0.0), + archive, + domain, + &mut round_progress, + ) + .map(|plan| namespace_round(plan, &format!("pane-round-{}-", rounds.len() - 1))); + full_progress.local_memo = round_progress.local_memo.take(); + full_progress.previous_domain = round_progress.previous_domain; + full_progress.refinements += round_progress.refinements; + completed_full_domain = complete && round_progress.catalog_completed; + full_progress.catalog_completed = completed_full_domain; + full_progress.budget_stopped = round_progress.budget_stopped; + full_progress.infeasible = complete && round_progress.infeasible; + if let Some(mut report) = round_progress.report.take() { + if let Some(previous) = full_progress.report.take() { + add_phase_metrics(&mut report, &previous); + } + full_progress.report = Some(report); + } + full = match (full, round) { + (Ok(previous), Ok(next)) => Ok(merge_validated_rounds(previous, next, models).1), + (_, Ok(next)) => Ok(next), + (previous, Err(error)) => { + if complete && round_progress.infeasible { + // The final catalog is infeasible, not a reason to return + // an earlier plan. Preserve prior pruning diagnostics in + // the aggregated failure report below. + full = Err(error); + break; + } + if !round_progress.budget_stopped && !round_progress.infeasible { + return Err(error); + } + match previous { + Ok(previous) => Ok(previous), + Err(_) => Err(error), + } + } + }; + if complete { + break; + } + // A timed-out rule pass leaves no reusable partial memo; stop rather + // than repeating all generation and pretending the domain expanded. + if full_progress.local_memo.is_none() { + break; + } + } + full_progress.catalog_completed = completed_full_domain; + if !completed_full_domain { + full_progress.budget_stopped = true; + if full_progress.report.is_none() { + full_progress.report = Some(crate::CandidateConnectionReport { + status: "time_limit".into(), + pane_width_rounds: rounds.clone(), + pane_widths_pending: true, + catalog_scope: "pane-width expansion interrupted before catalog completion".into(), + time_limit_ms: total_ms, + max_refinements: search.max_evaluations.unwrap_or(128), + ..Default::default() + }); + } + } + if !completed_full_domain { + if let Some(report) = &mut full_progress.report { + report.lower_bound = None; + report.relative_gap = None; + report.catalog_exhausted = Some(false); + report.evaluation_complete = Some(false); + report.full_catalog_completed = false; + report.pane_widths_pending = true; + report.status = "pane_width_domain_incomplete".into(); + } + } + if let Ok((selection, _, _)) = &mut full { + if let Some(report) = &full_progress.report { + selection.search.candidate_connections = Some(report.clone()); + } + } let full_planning_ms = full_started.elapsed().as_secs_f64() * 1_000.0; let mut seed_selected = false; let mut fallback_reason = None; let mut result = match (seed, full) { ( - Ok((seed_selection, mut seed_plan, seed_logical)), + Ok((seed_selection, seed_plan, seed_logical)), Ok((full_selection, full_plan, full_logical)), ) => { - seed_plan.id = format!("seed-{}", seed_plan.id); - seed_plan.from_logical = format!("seed-{}", seed_plan.from_logical); - let ranking = crate::Ranking::of(&models.plan_selection.cost_model); - // Both plans were validated under the same frozen calibration. Use - // PlanSelector's ranking without another unbudgeted evaluator call. - seed_selected = ranking.better( - &seed_selection.costs[&seed_selection.selected], - &full_selection.costs[&full_selection.selected], + let merged = merge_validated_rounds( + (seed_selection, seed_plan, seed_logical), + (full_selection, full_plan, full_logical), + models, ); - let full_search = full_selection.search.clone(); - let mut seed_selection = seed_selection; - seed_selection.selected = seed_plan.id.clone(); - seed_selection.costs = seed_selection - .costs - .into_iter() - .map(|(id, cost)| (format!("seed-{id}"), cost)) - .collect(); - seed_selection.accepted = seed_selection - .accepted - .into_iter() - .map(|(id, accuracy)| (format!("seed-{id}"), accuracy)) - .collect(); - for rejection in &mut seed_selection.rejected { - rejection.id = format!("seed-{}", rejection.id); - } - let (mut selection, other, physical, logical) = if seed_selected { - (seed_selection, full_selection, seed_plan, seed_logical) - } else { - (full_selection, seed_selection, full_plan, full_logical) - }; - // Preserve the validated Pareto front from both searches. - selection.costs.extend(other.costs); - selection.accepted.extend(other.accepted); - let ids = selection.costs.keys().cloned().collect::>(); - let costs = ids - .iter() - .map(|id| &selection.costs[id]) - .collect::>(); - let (_, front) = crate::choose( - crate::Ranking::of(&models.plan_selection.cost_model), - &costs, - ) - .expect("two validated selections have costs"); - selection.front = front.into_iter().map(|index| ids[index].clone()).collect(); - selection.search = full_search; - // A phase-local runner-up need not be second across both catalogs. - selection.search.runner_up_cost_per_second = None; - selection.search.incumbent_cost = selection - .costs - .get(&physical.id) - .and_then(|cost| cost.cost_per_second); - (selection, physical, logical) + seed_selected = merged.0; + merged.1 } + (Ok(seed), Err(error)) => { // The outer report owns phase accounting; do not embed a stale // standalone complete-phase report inside its fallback reason. @@ -208,7 +330,7 @@ pub(crate) fn candidate_connection_search( seed } (_, Err(error)) => { - if full_progress.budget_stopped { + if full_progress.budget_stopped || full_progress.infeasible { if let Some(mut report) = full_progress.report.take() { if let Some(seed) = &seed_report { add_phase_metrics(&mut report, seed); @@ -219,10 +341,17 @@ pub(crate) fn candidate_connection_search( report.time_limit_ms = total_ms; report.max_refinements = search.max_evaluations.unwrap_or(128); report.full_catalog_completed = full_progress.catalog_completed; + report.pane_width_rounds = rounds.clone(); + report.pane_widths_pending = !full_progress.catalog_completed; let metrics = serde_json::to_string(&report) .map_err(|error| SelectionError::Solver(error.to_string()))?; + let reason = if full_progress.infeasible { + "complete candidate catalog is infeasible under the configured model" + } else { + "planning budget ended before either seed or complete search found a validated incumbent" + }; return Err(SelectionError::Solver(format!( - "candidate-connections: planning budget ended before either seed or complete search found a validated incumbent. Search report: {metrics}" + "candidate-connections: {reason}. Search report: {metrics}" ))); } } @@ -246,19 +375,22 @@ pub(crate) fn candidate_connection_search( report.seed_planning_ms = seed_elapsed.as_secs_f64() * 1_000.0; report.seed_selected = seed_selected; report.full_catalog_completed = full_progress.catalog_completed; + report.pane_width_rounds = rounds.clone(); + report.pane_widths_pending = !full_progress.catalog_completed; report.seed_fallback_reason = fallback_reason; report.time_limit_ms = total_ms; report.local_candidate_generation_ms = local_candidate_generation_ms; report.max_refinements = search.max_evaluations.unwrap_or(128); - if report.seed_fallback_reason.is_some() { + if report.seed_fallback_reason.is_some() || !full_progress.catalog_completed { // The restricted seed's optimum is not a certificate for all widths. - report.status = "time_or_evaluation_limit_with_validated_seed".into(); + report.status = "time_or_evaluation_limit_with_validated_incumbent".into(); report.lower_bound = None; report.relative_gap = None; report.catalog_exhausted = Some(false); report.evaluation_complete = Some(false); - } else if seed_selected { - // A full-domain lower bound remains valid; avoid a stale incumbent gap. + } else { + // The selected incumbent may come from any earlier round. Recompute + // its gap against the complete-domain bound, not the last proposal. report.relative_gap = report .lower_bound @@ -283,6 +415,112 @@ pub(crate) fn candidate_connection_search( Ok(result) } +type ValidatedRound = ( + crate::Selection, + asap_physical_asap_optimization::PhysicalWorkloadPlan, + Vec, +); + +/// Assign a readable phase identity once, before combining validated results. +fn namespace_round(mut plan: ValidatedRound, namespace: &str) -> ValidatedRound { + let (selection, physical, _) = &mut plan; + physical.id = format!("{namespace}{}", physical.id); + physical.from_logical = format!("{namespace}{}", physical.from_logical); + selection.selected = format!("{namespace}{}", selection.selected); + selection.costs = std::mem::take(&mut selection.costs) + .into_iter() + .map(|(id, value)| (format!("{namespace}{id}"), value)) + .collect(); + selection.accepted = std::mem::take(&mut selection.accepted) + .into_iter() + .map(|(id, value)| (format!("{namespace}{id}"), value)) + .collect(); + for id in &mut selection.front { + *id = format!("{namespace}{id}"); + } + for rejection in &mut selection.rejected { + rejection.id = format!("{namespace}{}", rejection.id); + } + plan +} + +fn merge_validated_rounds( + seed: ValidatedRound, + full: ValidatedRound, + models: PlanningModels<'_>, +) -> (bool, ValidatedRound) { + let (full_selection, full_plan, full_logical) = full; + let identifiers = + |selection: &crate::Selection, + plan: &asap_physical_asap_optimization::PhysicalWorkloadPlan| { + selection + .costs + .keys() + .chain(selection.accepted.keys()) + .chain(selection.front.iter()) + .chain(selection.rejected.iter().map(|rejection| &rejection.id)) + .chain([&selection.selected, &plan.id, &plan.from_logical]) + .cloned() + .collect::>() + }; + let occupied = identifiers(&full_selection, &full_plan); + let earlier_ids = identifiers(&seed.0, &seed.1); + // Production phases already have disjoint namespaces. Retain a defensive + // fallback for standalone callers combining unnamespaced results. + let seed = if earlier_ids.is_disjoint(&occupied) { + seed + } else { + let mut index = 1; + loop { + let namespace = format!("merged-round-{index}-"); + if earlier_ids + .iter() + .all(|id| !occupied.contains(&format!("{namespace}{id}"))) + { + break namespace_round(seed, &namespace); + } + index += 1; + } + }; + let (seed_selection, seed_plan, seed_logical) = seed; + let ranking = crate::Ranking::of(&models.plan_selection.cost_model); + // Both plans were validated under the same frozen calibration. Use + // PlanSelector's ranking without another unbudgeted evaluator call. + let seed_selected = ranking.better( + &seed_selection.costs[&seed_selection.selected], + &full_selection.costs[&full_selection.selected], + ); + let full_search = full_selection.search.clone(); + let (mut selection, other, physical, logical) = if seed_selected { + (seed_selection, full_selection, seed_plan, seed_logical) + } else { + (full_selection, seed_selection, full_plan, full_logical) + }; + // Preserve the validated Pareto front from both searches. + selection.costs.extend(other.costs); + selection.accepted.extend(other.accepted); + selection.rejected.extend(other.rejected); + let ids = selection.costs.keys().cloned().collect::>(); + let costs = ids + .iter() + .map(|id| &selection.costs[id]) + .collect::>(); + let (_, front) = crate::choose( + crate::Ranking::of(&models.plan_selection.cost_model), + &costs, + ) + .expect("two validated selections have costs"); + selection.front = front.into_iter().map(|index| ids[index].clone()).collect(); + selection.search = full_search; + // A phase-local runner-up need not be second across both catalogs. + selection.search.runner_up_cost_per_second = None; + selection.search.incumbent_cost = selection + .costs + .get(&physical.id) + .and_then(|cost| cost.cost_per_second); + (seed_selected, (selection, physical, logical)) +} + /// An independently validated plan is a witness against a larger global lower /// bound. Keep the executable plan, but expose the inconsistent certificate /// instead of silently clamping it or claiming optimality. @@ -307,6 +545,35 @@ fn add_phase_metrics( report: &mut crate::CandidateConnectionReport, phase: &crate::CandidateConnectionReport, ) { + // All progressive rounds use one configured objective and frozen model. + // Preparation-only reports may lack this metadata without lacking an + // inventory snapshot (or vice versa); inherit it independently of counts. + if report.objective.is_empty() && !phase.objective.is_empty() { + report.objective = phase.objective.clone(); + } + if report.calibration_scope.is_empty() && !phase.calibration_scope.is_empty() { + report.calibration_scope = phase.calibration_scope.clone(); + report.calibration_factor = phase.calibration_factor; + } + // An early timeout has no inventory snapshot. Retain the latest complete + // one as a unit, with explicit provenance; never combine field-wise maxima. + // A genuine empty inventory still has Some(domain) and must replace older + // counts rather than inheriting them because they happen to be nonzero. + if report.inventory_pane_width_domain.is_none() && phase.inventory_pane_width_domain.is_some() { + report.inventory_pane_width_domain = phase.inventory_pane_width_domain.clone(); + report.groups = phase.groups; + report.logical_candidates = phase.logical_candidates; + report.immediate_candidates = phase.immediate_candidates; + report.physical_configurations = phase.physical_configurations; + report.compatible_connections = phase.compatible_connections; + report.supplier_sets = phase.supplier_sets; + report.stored_supplier_ids = phase.stored_supplier_ids; + report.output_references = phase.output_references; + report.structurally_removed_candidates = phase.structurally_removed_candidates; + report.structurally_removed_connections = phase.structurally_removed_connections; + report.rounded_down_cost_bounds = phase.rounded_down_cost_bounds; + report.independent_components = phase.independent_components; + } report.memo_expansion_ms += phase.memo_expansion_ms; report.inventory_construction_ms += phase.inventory_construction_ms; report.search_and_validation_ms += phase.search_and_validation_ms; @@ -314,6 +581,19 @@ fn add_phase_metrics( report.solver_ms += phase.solver_ms; report.validation_ms += phase.validation_ms; report.early_query_pruning_ms += phase.early_query_pruning_ms; + for (kind, count) in &phase.early_query_pruning_counts { + *report + .early_query_pruning_counts + .entry(kind.clone()) + .or_default() += count; + } + for example in &phase.early_query_pruning_examples { + if report.early_query_pruning_examples.len() < 8 + && !report.early_query_pruning_examples.contains(example) + { + report.early_query_pruning_examples.push(example.clone()); + } + } report.solve_calls += phase.solve_calls; report.refinements += phase.refinements; report.certified_query_cuts += phase.certified_query_cuts; @@ -405,6 +685,8 @@ fn search_domain( // measured generation time and budget; leave unfinished catalog // counts unset rather than inventing completed model statistics. let report = crate::CandidateConnectionReport { + pane_width_rounds: vec![format!("{domain:?}")], + pane_widths_pending: domain != asap_logical_asap_optimization::memo::state_configurations::PaneWidthDomain::Complete, status: "time_limit".into(), catalog_scope: format!( "planning stopped after {phase}, before solving; model statistics unavailable" @@ -416,6 +698,7 @@ fn search_domain( local_candidate_generation_ms, ..Default::default() }; + progress.report = Some(report.clone()); let metrics = serde_json::to_string(&report).map_err(|error| unsupported(&error.to_string()))?; Err(unsupported(&format!( @@ -462,17 +745,25 @@ fn search_domain( )); } let started = Instant::now(); - let pruned = crate::search::query_pruning::prune_query_candidates( - inventory, - entries, - data, - models, - if search.exhaustive { 0 } else { 64 }, - ) - .map_err(|error| unsupported(&error))?; + let pruned = if progress.local_memo.is_none() { + Some( + crate::search::query_pruning::prune_query_candidates( + inventory, + entries, + data, + models, + if search.exhaustive { 0 } else { 64 }, + ) + .map_err(|error| unsupported(&error))?, + ) + } else { + None + }; let early_query_pruning_ms = started.elapsed().as_secs_f64() * 1000.0; check_budget("early query pruning", progress)?; - let pruning = pruned.report; + let pruning = pruned.as_ref().map(|pruned| &pruned.report); + let empty_pruning = crate::search::query_pruning::QueryCandidatePruningReport::default(); + let pruning = pruning.unwrap_or(&empty_pruning); let early_query_pruning_counts = [ ("choices_checked".into(), pruning.choices_checked), ("choices_removed".into(), pruning.removed.len()), @@ -507,15 +798,30 @@ fn search_domain( }) .collect(); let started = Instant::now(); - let mut memo = try_populate_pass1(initial, &pruned.inventory) - .map_err(|error| unsupported(&error.to_string()))?; + let mut memo = match progress.local_memo.take() { + Some(memo) => memo, + None => try_populate_pass1( + initial, + &pruned.as_ref().expect("first round pruned").inventory, + ) + .map_err(|error| unsupported(&error.to_string()))?, + }; let deadline = Instant::now().checked_add(remaining()); - asap_logical_asap_optimization::memo::window_candidates::try_add_local_window_composition_in_domain( - &mut memo, entries, domain, deadline, - ).map_err(|error| { + asap_logical_asap_optimization::memo::window_candidates::extend_local_window_composition( + &mut memo, + entries, + progress.previous_domain, + domain, + deadline, + ) + .map_err(|error| { progress.budget_stopped = true; unsupported(&error.to_string()) })?; + // State-group splitting is currently one-shot. Keep the unexpanded memo; + // later rounds append local widths, then rebuild the state/physical catalog. + progress.local_memo = Some(memo.clone()); + progress.previous_domain = Some(domain); let memo = expand_concrete_summary_state_groups_before(memo, domain, deadline) .map_err(|error| { progress.budget_stopped = @@ -701,7 +1007,14 @@ fn search_domain( result.status, CandidateSearchStatus::TimeLimit | CandidateSearchStatus::RefinementLimit ); + let supplier_storage = + crate::search::candidates::suppliers::supplier_storage_metrics(&configured.inventory.graph); let report = crate::CandidateConnectionReport { + inventory_pane_width_domain: Some(format!("{domain:?}")), + supplier_sets: supplier_storage.unique_supplier_sets, + stored_supplier_ids: supplier_storage.stored_supplier_ids, + pane_width_rounds: vec![format!("{domain:?}")], + pane_widths_pending: domain != asap_logical_asap_optimization::memo::state_configurations::PaneWidthDomain::Complete, rounded_down_cost_bounds: bound_metrics.rounded_down_candidates, seed_planning_ms: 0.0, full_planning_ms: planning_started.elapsed().as_secs_f64() * 1_000.0, @@ -914,6 +1227,19 @@ mod tests { let kept = validated_seed_after_stop(seed, error, &full_progress).unwrap(); assert_eq!(kept.0.selected, selected); kept.1.dag.validate().unwrap(); + let infeasible = validated_seed_after_stop( + kept.clone(), + SelectionError::Solver("complete catalog infeasible".into()), + &DomainProgress { + infeasible: true, + catalog_completed: true, + ..Default::default() + }, + ) + .unwrap_err(); + assert!(infeasible + .to_string() + .contains("complete catalog infeasible")); let error = validated_seed_after_stop( kept, SelectionError::Solver("internal reconstruction failure".into()), @@ -970,7 +1296,7 @@ mod tests { // Joint phases retain valid frontier metadata under the exact configured ranking. #[test] - fn seed_and_complete_pareto_fronts_keep_distinct_ids_and_frozen_fit() { + fn pane_free_pareto_uses_complete_domain_and_frozen_fit() { let settings = crate::PlanSelectionSettings { frozen_calibration_factor: Some(3.0), cost_model: CostModel::ClusterCapacity(crate::ClusterLimits { @@ -984,11 +1310,11 @@ mod tests { let run = scalar_run(&settings, Stage1ToStage3Search::default()); let selection = run.stage3_plan_selection.unwrap(); let report = selection.search.candidate_connections.as_ref().unwrap(); - assert!(report.seed_planning_ms > 0.0); + assert_eq!(report.seed_planning_ms, 0.0); assert!(report.full_catalog_completed); assert_eq!(report.calibration_factor, 3.0); assert!(selection.front.contains(&selection.selected)); - assert!(selection.front.iter().any(|id| id.starts_with("seed-"))); + assert_eq!(report.pane_width_rounds, vec!["Complete"]); for id in &selection.front { assert!(selection.costs.contains_key(id)); assert!(selection.accepted.contains_key(id)); @@ -1012,6 +1338,259 @@ mod tests { ); } + #[test] + fn progressive_width_rounds_preserve_incumbent_without_global_certificate() { + let (roots, entries, data) = workload(); + let run = run_stage1_to_stage3_with( + roots, + &entries, + &data, + PlanningModels::builtin(), + Stage1ToStage3Search { + max_evaluations: Some(4), + ..Default::default() + }, + ) + .unwrap(); + let selection = run.stage3_plan_selection.unwrap(); + let report = selection.search.candidate_connections.as_ref().unwrap(); + assert!(selection.accepted.contains_key(&selection.selected)); + assert!(report.pane_width_rounds.len() >= 2); + assert!(report.pane_width_rounds[1].starts_with("Expanding")); + assert!(report.pane_widths_pending); + assert!(!report.full_catalog_completed); + assert_eq!(report.lower_bound, None); + assert_eq!(report.relative_gap, None); + assert_ne!(report.status, "optimal"); + assert!(selection.search.candidates_checked <= 4); + } + + #[test] + fn seed_fallback_report_preserves_model_without_inventing_inventory() { + let previous = crate::CandidateConnectionReport { + objective: "monetary_cost".into(), + calibration_scope: "not_required".into(), + calibration_factor: 1.0, + ..Default::default() + }; + let mut interrupted = crate::CandidateConnectionReport { + status: "time_limit".into(), + pane_widths_pending: true, + ..Default::default() + }; + add_phase_metrics(&mut interrupted, &previous); + assert_eq!(interrupted.objective, "monetary_cost"); + assert_eq!(interrupted.calibration_scope, "not_required"); + assert_eq!(interrupted.calibration_factor, 1.0); + assert_eq!(interrupted.inventory_pane_width_domain, None); + assert_eq!(interrupted.physical_configurations, 0); + assert!(!interrupted.full_catalog_completed); + assert!(interrupted.pane_widths_pending); + assert_eq!(interrupted.status, "time_limit"); + assert_eq!(interrupted.lower_bound, None); + assert_eq!(interrupted.relative_gap, None); + } + + #[test] + fn existing_model_metadata_is_not_overwritten_by_older_phases() { + let previous = crate::CandidateConnectionReport { + objective: "monetary_cost".into(), + calibration_scope: "not_required".into(), + calibration_factor: 1.0, + ..Default::default() + }; + let mut current = crate::CandidateConnectionReport { + objective: "ParetoFront".into(), + calibration_scope: "frozen".into(), + calibration_factor: 3.0, + ..Default::default() + }; + add_phase_metrics(&mut current, &previous); + assert_eq!(current.objective, "ParetoFront"); + assert_eq!(current.calibration_scope, "frozen"); + assert_eq!(current.calibration_factor, 3.0); + } + + #[test] + fn interrupted_round_preserves_one_complete_inventory_snapshot() { + let previous = crate::CandidateConnectionReport { + inventory_pane_width_domain: Some("Expanding { max_parts: 32 }".into()), + groups: 7, + logical_candidates: 11, + immediate_candidates: 13, + physical_configurations: 17, + compatible_connections: 101, + supplier_sets: 3, + stored_supplier_ids: 19, + output_references: 2, + structurally_removed_candidates: 5, + structurally_removed_connections: 23, + rounded_down_cost_bounds: 2, + independent_components: 1, + lower_bound: Some(4.0), + relative_gap: Some(0.1), + full_catalog_completed: true, + ..Default::default() + }; + let mut interrupted = crate::CandidateConnectionReport { + status: "pane_width_domain_incomplete".into(), + pane_widths_pending: true, + pane_width_rounds: vec!["Expanding { max_parts: 64 }".into()], + ..Default::default() + }; + add_phase_metrics(&mut interrupted, &previous); + assert_eq!( + interrupted.inventory_pane_width_domain, + previous.inventory_pane_width_domain + ); + assert_eq!( + ( + interrupted.groups, + interrupted.logical_candidates, + interrupted.immediate_candidates + ), + (7, 11, 13) + ); + assert_eq!( + ( + interrupted.physical_configurations, + interrupted.compatible_connections, + interrupted.supplier_sets, + interrupted.stored_supplier_ids + ), + (17, 101, 3, 19) + ); + assert_eq!( + ( + interrupted.output_references, + interrupted.structurally_removed_candidates, + interrupted.structurally_removed_connections + ), + (2, 5, 23) + ); + assert_eq!(interrupted.lower_bound, None); + assert_eq!(interrupted.relative_gap, None); + assert!(!interrupted.full_catalog_completed); + assert!(interrupted.pane_widths_pending); + assert_eq!(interrupted.status, "pane_width_domain_incomplete"); + assert_eq!( + interrupted.pane_width_rounds, + vec!["Expanding { max_parts: 64 }"] + ); + + let mut empty = crate::CandidateConnectionReport { + inventory_pane_width_domain: Some("Complete".into()), + ..Default::default() + }; + add_phase_metrics(&mut empty, &previous); + assert_eq!( + empty.inventory_pane_width_domain.as_deref(), + Some("Complete") + ); + assert_eq!( + ( + empty.physical_configurations, + empty.compatible_connections, + empty.supplier_sets, + empty.stored_supplier_ids + ), + (0, 0, 0, 0) + ); + } + + #[test] + fn repeated_round_ids_preserve_all_cost_and_accuracy_records() { + let make = |price| { + let run = scalar_run( + &crate::PlanSelectionSettings::default(), + Stage1ToStage3Search { + max_evaluations: Some(1), + ..Default::default() + }, + ); + let mut selection = run.stage3_plan_selection.unwrap(); + let physical = run.stage3_selected.unwrap(); + selection + .costs + .get_mut(&selection.selected) + .unwrap() + .cost_per_second = Some(price); + let roots = physical.roots.clone(); + (selection, physical, roots) + }; + let first = make(1.0); + let second = make(2.0); + let initial = make(3.0); + assert_eq!(first.1.id, second.1.id); + assert_eq!(first.1.id, initial.1.id); + let (_, merged) = merge_validated_rounds(first, second, PlanningModels::builtin()); + let (_, merged) = merge_validated_rounds(initial, merged, PlanningModels::builtin()); + assert_eq!(merged.0.costs.len(), 3); + assert_eq!(merged.0.accepted.len(), 3); + let mut prices: Vec<_> = merged + .0 + .costs + .values() + .map(|cost| cost.cost_per_second.unwrap()) + .collect(); + prices.sort_by(f64::total_cmp); + assert_eq!(prices, vec![1.0, 2.0, 3.0]); + assert_eq!(merged.0.selected, merged.1.id); + assert_eq!(merged.0.costs[&merged.1.id].cost_per_second, Some(1.0)); + let settings = crate::PlanSelectionSettings { + cost_model: CostModel::ClusterCapacity(crate::ClusterLimits { + memory_bytes: 1_000_000, + disk_bytes: 1_000_000, + cpu_cores: Some(8), + objective: crate::ClusterObjective::ParetoFront, + }), + ..Default::default() + }; + let models = PlanningModels { + plan_selection: &settings, + ..PlanningModels::builtin() + }; + let (_, merged) = merge_validated_rounds(make(1.0), make(2.0), models); + let (_, merged) = merge_validated_rounds(make(0.5), merged, models); + assert_eq!(merged.0.costs.len(), 3); + assert_eq!(merged.0.front.len(), 3); + assert!(merged.0.front.contains(&merged.1.id)); + } + + #[test] + fn repeated_window_free_computation_searches_one_complete_catalog() { + let (_, entries, data) = workload(); + let roots = vec![( + 0, + QueryRoot::Operator(lower_promql("sum(latency)", AccuracyTarget::exact())), + )]; + let run = run_stage1_to_stage3_with( + roots, + &entries[..1], + &data, + PlanningModels::builtin(), + Stage1ToStage3Search::default(), + ) + .unwrap(); + let selection = run.stage3_plan_selection.unwrap(); + let report = selection.search.candidate_connections.unwrap(); + assert_eq!(report.pane_width_rounds, vec!["Complete"]); + assert_eq!(report.seed_planning_ms, 0.0); + assert!(report.full_catalog_completed); + } + + #[test] + fn literal_workload_does_not_repeat_identical_width_models() { + let run = scalar_run( + &crate::PlanSelectionSettings::default(), + Stage1ToStage3Search::default(), + ); + let selection = run.stage3_plan_selection.unwrap(); + let report = selection.search.candidate_connections.unwrap(); + assert_eq!(report.pane_width_rounds, vec!["Complete"]); + assert!(report.full_catalog_completed); + } + // A failed complete phase must not relabel seed work as initial generation. #[test] fn no_incumbent_report_preserves_seed_and_full_phase_accounting() { @@ -1031,9 +1610,7 @@ mod tests { let (_, entries, data) = workload(); let roots = vec![( 0, - QueryRoot::Scalar(asap_types::ir::ScalarExpr::Literal( - asap_types::ir::scalar::ScalarValue::Float64(1.0), - )), + QueryRoot::Operator(lower_promql("latency[5m]", AccuracyTarget::exact())), )]; let initial = LogicalMemo::from_roots(&roots); let inventory = asap_logical_asap_optimization::per_computation_candidates::generate_per_computation_candidates(roots).unwrap(); @@ -1059,7 +1636,10 @@ mod tests { assert_eq!(report["local_candidate_generation_ms"], 17.0); assert!(report["seed_planning_ms"].as_f64().unwrap() > 0.0); assert!(report["full_planning_ms"].as_f64().unwrap() > 0.0); - assert_eq!(report["refinements"], 2); + assert_eq!( + report["refinements"].as_u64().unwrap() as usize, + report["pane_width_rounds"].as_array().unwrap().len() + ); assert_eq!(report["time_limit_ms"], 30_000); assert_eq!(report["validation_rejection_counts"], serde_json::json!({})); } diff --git a/docs/design_docs/proposals/joint-planning-search.md b/docs/design_docs/proposals/joint-planning-search.md index 469467918..63bbe620d 100644 --- a/docs/design_docs/proposals/joint-planning-search.md +++ b/docs/design_docs/proposals/joint-planning-search.md @@ -510,7 +510,15 @@ Cost normalization must preserve small, meaningful differences relative to solve Search reductions must preserve joint choices. A candidate that costs more alone may enable useful sharing, so local cost dominance is insufficient. Independent components can be solved separately only when dependencies, sharing, and global constraints also separate. Generation caching and sparse compatibility edges reduce overhead without requiring enumeration of complete workload plans. -Bounded fixed-calibration search may first obtain a validated seed from a smaller reader-grid/GCD width domain using the same engine, then spend the remaining budget on the complete domain. Only a definite budget stop may fall back to that validated seed; internal errors remain errors. Exhaustive archives and complete-catalog calibration skip this seed phase. Reports distinguish complete-domain generation from exhaustive assignment traversal, and a seed fallback supplies no full-domain optimum or gap certificate. Generation and model construction cooperate with the deadline; individual evaluator calls may overrun it. +Bounded fixed-calibration search first attempts to obtain a validated seed from a smaller reader-grid/GCD width domain using the same engine, then progressively adds legal widths toward the complete domain. A cached divisor frontier skips rounds that introduce no new width. Logical expansion reuses the pre-state-expansion memo and generates newly introduced window expressions; state expansion, physical inventory, and the MILP are rebuilt for each attempted domain. This is not incremental MILP solving. Only a definite budget stop may fall back to that validated seed; internal errors remain errors. Exhaustive archives and complete-catalog calibration skip this seed phase. Reports distinguish complete-domain generation from exhaustive assignment traversal, and a seed fallback supplies no full-domain optimum or gap certificate. Generation and model construction cooperate with the deadline; individual evaluator calls may overrun it. + +Three complementary reductions control dense pane inventories: + +- **Compatibility memoization and supplier-list sharing.** Cache exact compatibility subproblems and intern identical ordered supplier lists. Separate input positions still have separate connection decisions. This reduces repeated checking and storage, not the logical edge count or MILP variable count by itself. +- **Terminal reader factoring.** Equivalent unary summary estimators and exact-accumulator finalizers used only by query outputs can become one reader template with the union of their compatible suppliers. Equality includes operator parameters, result contracts, timing, materialization, local cost, input contracts, and output eligibility. Each output retains an independent binding through execution scopes; consumers may still choose different producers. Internal readers and pane-merge operators are excluded. In particular, merges with different pane counts are not interchangeable. +- **Progressive width generation.** Spend the shared planning budget on successively larger finite domains and retain the best validated incumbent across rounds. Report attempted domains and pending widths. A restricted-domain lower bound or gap is not a certificate for the complete domain; only a completed full-domain solve can provide that certificate. + +The first two reductions preserve the represented choices under their stated contracts; the third controls work before a timeout and may leave choices unexplored. General cost-based DP compression needs a richer boundary contract covering sharing, resource use, accuracy, and execution demand; it is not implied by compatibility memoization. Fine pane widths cannot be discarded merely because their local cost is high. Within the seed budget, the same MILP can also hold a validated plan's computations and producer connections fixed and try supported query-time materializations. The normal evaluator measures whether retaining a shared result avoids enough repeated upstream work to reduce cost. These limited proposals use the existing solve/evaluation budgets; their bounds and infeasibility apply only to the restricted proposal, never to the full search. This does not create another selector or an optimality claim for the seed. diff --git a/docs/develop_docs/pane-search-compression-evaluation.md b/docs/develop_docs/pane-search-compression-evaluation.md new file mode 100644 index 000000000..dfb5814c5 --- /dev/null +++ b/docs/develop_docs/pane-search-compression-evaluation.md @@ -0,0 +1,117 @@ +# Pane-search compression evaluation + +This report is for planner developers. It separates three changes: storing repeated supplier lists once, factoring equivalent candidate connections, and spending a bounded search budget across progressively larger pane-width domains. Storage compression alone does not reduce the number of MILP variables. Searching fewer widths before a timeout does not prove equivalence to the complete domain. + +## Reproduction + +Use frozen `stage_pipeline` executables; record their SHA-256 hashes. Do not run the four-case comparison concurrently with another benchmark. Both sides use the same input files, AWS pricing, deployment capabilities, enforced accuracy, a 30-second cooperative planning budget, and a 90-second external process timeout. Each case runs sequentially. Whole-process wall time and peak RSS include plan export and validation, so wall time may exceed the cooperative planning budget. + +```bash +python3 tools/benchmark_pane_search.py summarize BEFORE_BINARY BEFORE_DIR +python3 tools/benchmark_pane_search.py run AFTER_BINARY AFTER_DIR +python3 tools/benchmark_pane_search.py compare BEFORE_DIR AFTER_DIR COMPARISON.md +``` + +`summarize` imports existing plan, timing, and `summary.json` artifacts without executing the planner again. Each directory receives `results.json`, with the binary hash, input fingerprints, full search reports, selected cost/usage, and acceptance checks. Comparison rejects mismatched inputs, budgets, or concurrency. Missing old metrics remain **unreported**, rather than zero. + +Broader acceptance uses the same runner: + +```bash +python3 tools/benchmark_pane_search.py run AFTER_BINARY VIEWER_DIR --suite viewer --jobs 2 +python3 tools/benchmark_pane_search.py run AFTER_BINARY ASAP_DIR --suite asap --jobs 2 +``` + +For an archived 30-workload baseline without input fingerprints, compare only cost and acceptance after checking its original pricing configuration: + +```bash +python3 tools/benchmark_pane_search.py compare OLD_ASAP_DIR ASAP_DIR QUALITY.md --quality-only +``` + +This mode checks serialized workloads, deployment capabilities, accuracy modes, Stage 0 DAGs, and objective/calibration settings. It deliberately excludes planning-time and memory comparisons; the old runner must confirm the pricing profile because historical documents do not serialize it. + + +These suites contain the nine viewer workload examples and 30 mixed ASAPQuery workloads, respectively. They check query coverage, enforced accuracy, finite cost, valid cost bounds, no stored-output errors, and no global bound, gap, optimum, or exhaustion claim while pane widths remain pending. Existing Rust/JavaScript tests remain responsible for DAG structure and installation invariants. + +## Baseline + +The existing four-case run is complete. Its frozen executable is `/mydata/asap-final-validation/stage-pipeline-final-composed-budget`; raw artifacts are under `/mydata/pane-compression-validation/before`. All four plans cover every query with enforced guarantees and no stored-output error. + +| Workload | Queries accepted | Wall seconds | Peak RSS KiB | Physical configurations | Compatible connections | Peak MILP variables | Modeled cost/second | +|---|---:|---:|---:|---:|---:|---:|---:| +| mixed seed 862, workload 01 | 9/9 | 36.08 | 510400 | 5785 | 147279 | 159153 | 1.60464455e-6 | +| long mixed seed 862, workload 01 | 9/9 | 32.94 | 415108 | 5785 | 147279 | 159153 | 4.62805740e-6 | +| rangeheavy t8/t5, 1h | 3/3 | 33.67 | 447128 | 17252 | 1705248 | 44034 | 8.33239804e-5 | +| rangeheavy t5, 15m/1h | 2/2 | 33.26 | 949392 | 3505 | 212790 | 390981 | 1.00719775e-5 | + +All four report `time_limit`, with `full_catalog_completed=true`. That flag means configuration generation completed, not that every assignment was evaluated or a global optimum was proved. Peak MILP size can be smaller than the reported final compatibility graph when the deadline prevents constructing a model for that graph. The baseline does not report unique supplier sets or physically stored supplier IDs. + +## Acceptance criteria + +Record candidate counts, logical compatibility connections, unique supplier sets, stored supplier IDs, peak MILP variables/constraints, generation/inventory/model/solve/validation times, whole-process RSS, selected cost, and accuracy. Also record attempted width domains, pending widths, catalog completion, solver status, and gap. Report before/after values together; do not infer a complete-domain model reduction from a smaller intermediate round. + +Independent correctness checks should cover: + +- Supplier-list sharing preserves membership, order, and duplicate input positions. Mutating one shared list must not mutate another list. Cached compatibility includes every semantic/configuration field used by the uncached check. +- Tiny complete-domain problems retain the same feasible assignments and minimum evaluated cost before and after factoring. Include multiple readers, different materializations, duplicate input positions, optional dependencies, and shared upstream producers. Reconstruction must restore every factored operation and preserve costs, accuracy, and query coverage. +- Terminal unary factoring must not absorb a node with additional consumers, observable materialization/state, unsupported error propagation, or different execution demand. Shared producer work remains charged once; reader-specific work remains per use. +- Progressive width domains are nested and add only newly admitted window widths to the logical memo. Each round rebuilds its physical inventory and MILP, so previously considered assignments may be revisited. The domains eventually include the complete legal finite domain when the budget permits. A restricted-domain optimum never becomes a global certificate; timeout preserves the best validated incumbent. +- The four-case benchmark must show measurable changes on the dense rangeheavy cases. Report cost regressions or incomplete-domain effects explicitly. Full 30-workload and nine-viewer acceptance must pass without relaxing accuracy or latency requirements. + +## After-change results + +Final sequential measurements use production revision `ba7dca2c`, including the report-only snapshot fix. The original after-run is retained separately; the table below uses only `after-final` artifacts. + +Before binary SHA-256: `8e182907f9c840c7134876e7d626fdb95223270b4c012ff16735dcf366fefff2`. After binary SHA-256: `52533a57f715f51b889e4f031281f29730607b85a71611ad03e468655fec6b0b`. + +All four workloads passed with the **same selected modeled cost** as their respective baselines and all 23 query requirements accepted. No planner regression was running during these four sequential measurements. + +| Workload | Wall seconds, before → after | RSS KiB, before → after | Peak MILP variables, before → after | Peak MILP constraints, before → after | +|---|---:|---:|---:|---:| +| mixed 01 | 36.08 → 30.69 | 510400 → 207948 | 159153 → 55501 | 306928 → 105689 | +| long mixed 01 | 32.94 → 31.67 | 415108 → 203444 | 159153 → 55501 | 306928 → 105689 | +| rangeheavy t8/t5 | 33.67 → 30.10 | 447128 → 353472 | 44034 → 111171 | 59826 → 332191 | +| rangeheavy 15m/1h | 33.26 → 30.42 | 949392 → 459012 | 390981 → 198129 | 1108599 → 558375 | + +The two-window rangeheavy case used **51.7% less peak RSS** and **49.3% fewer peak MILP variables** in this run. Conversely, the t8/t5 case solved a larger intermediate model: peak variables increased from 44,034 to 111,171 while RSS fell 20.9%. Its baseline generated a dense catalog but did not solve that entire model within the budget. These are same-budget observations, not same-domain compression ratios. This combined-change experiment does not isolate how much of the memory difference comes from supplier storage, factoring, or the progressive domain policy. + +All after-plans report pending widths, no global bound/gap, and no global optimum or exhaustion certificate. The attempted rounds were SeedGrids, then Expanding with maximum parts 4, 8, 16, 32, and 64. The following counters belong to the most recent reported inventory, identified explicitly; a later interrupted preparation round does not overwrite them with zero. + +| Workload | Inventory domain | Physical configurations | Logical compatible connections | Unique supplier lists | Stored supplier IDs | +|---|---|---:|---:|---:|---:| +| mixed 01 | Expanding { max_parts: 16 } | 3324 | 69018 | 2124 | 3159 | +| long mixed 01 | Expanding { max_parts: 16 } | 3324 | 69018 | 2124 | 3159 | +| rangeheavy t8/t5 | Expanding { max_parts: 16 } | 8164 | 395168 | 1265 | 8163 | +| rangeheavy 15m/1h | Expanding { max_parts: 64 } | 2809 | 137982 | 1356 | 2323 | + +Within these after-inventories, stored supplier IDs are 95.4%–98.3% fewer than the logical connection count. This measures list-content storage sharing only; it excludes allocator metadata and does not imply that equivalent MILP connection variables disappeared. The baseline lacks this storage instrumentation. Baseline candidate/connection counts describe complete width catalogs; the after counts above describe intermediate catalogs, so their differences combine factoring and restricted enumeration. + +Reported cumulative phase times, in seconds (before → after): + +| Workload | Local generation | Memo expansion | Inventory | Model construction | Solver | Validation | +|---|---:|---:|---:|---:|---:|---:| +| mixed 01 | 0.007 → 0.007 | 7.175 → 4.314 | 6.743 → 6.178 | 2.834 → 1.011 | 16.716 → 7.423 | 1.182 → 1.040 | +| long mixed 01 | 0.008 → 0.007 | 7.005 → 4.326 | 6.808 → 6.154 | 2.823 → 1.018 | 10.408 → 5.501 | 3.526 → 2.858 | +| rangeheavy t8/t5 | 0.002 → 0.002 | 3.189 → 1.579 | 10.513 → 4.355 | 0.431 → 1.849 | 5.486 → 8.890 | 1.784 → 0.350 | +| rangeheavy 15m/1h | 0.001 → 0.001 | 1.609 → 2.241 | 1.580 → 2.392 | 5.298 → 4.492 | 23.081 → 19.424 | 0.586 → 0.586 | + +These phase values are the emitted report counters. Interrupted preparation rounds may not be fully attributed to individual phases; their sum therefore does not equal whole-process wall time. Wall time and peak RSS cover the complete process. Do not treat every reported phase reduction as a measured reduction in all work of that kind. + +Artifacts: `/mydata/pane-compression-validation/{before,after-final}/results.json`, full plan JSON and GNU time logs in the same directories, and `comparison-final.md` with all before/after counters. Single runs do not establish statistical significance. Costs and accuracy are validated planner-model results, not measured stream-runtime performance. + +### Broader acceptance + +The 30 mixed ASAPQuery workloads passed **30/30**, with **270/270** query requirements accepted. Their selected modeled costs match the previous composed-budget binary **exactly in all 30 cases**, including workload 08. Serialized workloads, deployment capabilities, accuracy modes, and Stage 0 DAGs match the archived baseline. Both runners use the same AWS pricing profile and enforced accuracy; this is a quality comparison, not a planning-time comparison. + +The archived baseline is `/mydata/asap-final-validation/final-workloads-composed-budget/results.json`, using the same before-binary hash listed above. Current results are in `asap-final/results.json`; `asap-quality-comparison.md` contains every case's costs and acceptance counts. Some final interrupted-round reports omit objective/calibration labels; the comparator explicitly marks these unknown and relies on the verified original runner settings rather than interpreting the default calibration value as zero-priced execution. + +The nine viewer workloads also passed **9/9**, with **40/40** query requirements accepted, including the composed-quantile ratio and SQL rollup examples. Their artifacts are in `viewer-final/results.json`. Broad acceptance uses two concurrent workloads and the same 30-second budget, and may overlap regression tests. Its timings are deliberately excluded from the sequential performance comparison. + +### Final regression follow-up + +The regression suite found a genuine optimization regression outside window planning. Two identical SQL percentiles over an at-rest table were sent through progressive pane rounds solely because they repeat every ten minutes. No computation had a pane-eligible window, so the rounds rebuilt the same 40-configuration, 150-connection model and restarted refinement. The 128-evaluation budget expired with two identical unmaterialized KLL builds, costing 0.007678692288679677 per second instead of the baseline's one shared query-materialized build at 0.006000017830006526 (+27.98%). The original pointer-sharing assertion failed both in the suite and in isolation; it was retained unchanged. + +Revision `fe171f64` bypasses pane-domain expansion when the initial computations contain neither a positive range window nor an existing pane chain. It searches the complete candidate domain directly; at-rest status alone does not disable pane planning. The same SQL fixture now passes the original sharing test and selects one shared KLL with exactly the baseline cost. Its report records 58 evaluations, `optimal`, one `Complete` round, and no pending widths. This fix avoids repeated identical-domain searches rather than increasing the budget or relaxing correctness checks. + +The final frozen executable is `/mydata/pane-compression-validation/stage-pipeline-after-gate-final`, SHA-256 `a3234d62d2a6717e3772aba240959f6a224a4e1e031fe9cba27854079f8622f2`, built from `fe171f64`. SQL before/after artifacts are `/tmp/identical-sql-{before,after,gate-fixed}.json`; the unchanged integration test log is `/tmp/sql-sharing-pane-gate-after.log`. + +A fresh 30-second mixed-workload check on this final executable accepted all nine queries and retained the same modeled cost, 0.0000016046445458277792 per second. Its interrupted expansion report preserves `monetary_cost`, calibration scope `not_required`, factor 1, and the inventory source `Expanding { max_parts: 8 }` with 2,733 configurations and 49,897 connections. Pending widths still suppress global optimality certificates. Artifacts are in `/mydata/pane-compression-validation/final-gate-check/`. This targeted follow-up verifies the route/report fixes; the four-case measurements and 39-workload acceptance above remain attributed to their original frozen binary, not this later revision. diff --git a/tools/benchmark_pane_search.py b/tools/benchmark_pane_search.py new file mode 100644 index 000000000..4ed0be9a7 --- /dev/null +++ b/tools/benchmark_pane_search.py @@ -0,0 +1,267 @@ +#!/usr/bin/env python3 +"""Run or summarize frozen-binary pane-search benchmarks without rebuilding. + +Linux /usr/bin/time measures whole-process peak RSS. Search counters describe +reported domains and must not be confused with a complete-catalog optimum. +""" +import argparse +import concurrent.futures +import hashlib +import json +import math +from pathlib import Path +import re +import subprocess +import time + +REPO = Path(__file__).resolve().parents[1] +VIEWER = REPO / "tools/dag-viewer" +FOUR = { + "memo-rangeheavy-t8-t5-1h", "memo-rangeheavy-t5-q09-15m-1h", + "random-mixed-seed-862-workload-01", "random-mixed-seed-862-long-workload-01", +} + + +def digest(path): + h = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def entries(suite): + examples = [e for e in json.loads((VIEWER / "examples.json").read_text()) if "workload" in e] + if suite == "four": + return [e for e in examples if e["id"] in FOUR] + if suite == "viewer": + return examples + result = [] + for entry in examples: + if entry["workload"].startswith("workloads/random-mixed-seed-862"): + folder = (VIEWER / entry["workload"]).parent + for path in sorted(folder.glob("workload-*.json")): + result.append(dict(entry, id=folder.name + "-" + path.stem, + workload=str(path.relative_to(VIEWER)))) + assert len(result) == 30 + return result + + +def inputs(entry, budget): + path = VIEWER / entry["workload"] + tables = path.parent / "tables.json" + # Fingerprint semantics rather than absolute checkout/output paths. + return { + "workload": json.loads(path.read_text()), + "tables": json.loads(tables.read_text()) if tables.exists() else [], + "deployment_capabilities": entry["deployment_capabilities"], + "cost_model": entry["cost_model"], "enforce_accuracy": True, + "budget_seconds": budget, + } + + +def domain_certificate_valid(report): + pending = report.get("pane_widths_pending") is True + if pending and (report.get("lower_bound") is not None or report.get("relative_gap") is not None + or any(report.get(key) is True for key in + ("full_catalog_completed", "catalog_exhausted", "evaluation_complete"))): + return False + incomplete = pending or ("pane_widths_pending" in report and not report.get("full_catalog_completed")) + return not (incomplete and report.get("status") == "optimal") + + +def inspect(entry, output, budget, wall=None, exit_code=None): + name = entry["id"] + specification = inputs(entry, budget) + row = { + "workload": name, "wall_seconds": wall, "exit_code": exit_code, + "input_sha256": hashlib.sha256(json.dumps(specification, sort_keys=True).encode()).hexdigest(), + "artifact": name + ".json", "search": {}, "passed": False, + } + timing = output / (name + ".time") + if timing.exists(): + text = timing.read_text() + rss = re.search(r"Maximum resident set size \(kbytes\): (\d+)", text) + status = re.search(r"Exit status: (\d+)", text) + row["peak_planner_rss_kib"] = int(rss[1]) if rss else None + if exit_code is None and status: + row["exit_code"] = int(status[1]) + workload = specification["workload"]["query_workload"] + row["expected_queries"] = len(workload.get("query_batch") or []) + len(workload.get("repeating_queries") or []) + artifact = output / row["artifact"] + if not artifact.exists(): + row["error"] = "planner wrote no artifact" + return row + try: + document = json.loads(artifact.read_text()) + except (OSError, json.JSONDecodeError) as error: + row["error"] = f"unreadable planner artifact: {error}" + return row + selection = document["stage3_plan_selection"] + selected = selection.get("selected") + row.update(selected=selected, error=selection.get("error"), + stored_outputs_error=selection.get("stored_outputs_error"), + artifact_sha256=digest(artifact), + search=selection.get("search", {}).get("candidate_connections", {})) + accepted = selection.get("accepted", {}).get(selected, []) + row["reported_catalog_metrics_available"] = row["search"].get("physical_configurations", 0) > 0 + row["accepted_queries"] = len(accepted) + row["guarantees_enforced"] = bool(accepted) and all( + not q.get("accepted_regardless_of_guarantee", True) + and q.get("guarantee_check_rejection") is None for q in accepted) + cost = selection.get("costs", {}).get(selected, {}) + value = cost.get("cost_per_second") + row["modeled_cost_per_second"] = value + row["modeled_usage"] = cost.get("usage") + bound = row["search"].get("lower_bound") + row["cost_certificate_valid"] = bool( + isinstance(value, (int, float)) and math.isfinite(value) and value >= 0 + and row["search"].get("status") != "invalid_cost_certificate" + and (bound is None or (math.isfinite(bound) and bound <= value + 1e-8 * max(abs(value), 1e-12)))) + row["domain_certificate_valid"] = domain_certificate_valid(row["search"]) + row["passed"] = bool(row["exit_code"] == 0 and selected and row["guarantees_enforced"] + and len(accepted) == row["expected_queries"] and row["cost_certificate_valid"] + and row["domain_certificate_valid"] and not row["error"] + and not row["stored_outputs_error"]) + return row + + +def run_one(entry, args): + output, binary = args.output.resolve(), args.binary.resolve() + name = entry["id"] + artifact = output / (name + ".json") + if artifact.exists(): + raise FileExistsError(f"Use a new output directory; refusing to overwrite {artifact}") + command = ["/usr/bin/time", "-v", "-o", str(output / (name + ".time")), + "timeout", f"{args.timeout}s", str(binary), + "--workload", str(VIEWER / entry["workload"]), + "--search", "candidate-connections", "--milp-time-limit-s", str(args.budget), + "--cost-model", entry["cost_model"], "--enforce-accuracy-guarantees", + "--deployment-capabilities", json.dumps(entry["deployment_capabilities"]), + "--out", str(artifact)] + for table in inputs(entry, args.budget)["tables"]: + command += ["--table", json.dumps(table)] + started = time.monotonic() + with (output / (name + ".log")).open("w") as log: + code = subprocess.run(command, cwd=REPO, stdout=log, stderr=subprocess.STDOUT).returncode + row = inspect(entry, output, args.budget, time.monotonic() - started, code) + row["command"] = command + print(json.dumps({k: row.get(k) for k in ("workload", "passed", "wall_seconds", "modeled_cost_per_second", "error")}), flush=True) + return row + + +def collect(args): + args.output.mkdir(parents=True, exist_ok=True) + metadata = {"binary_sha256": digest(args.binary), "suite": args.suite, + "budget_seconds": args.budget, "parallel_workloads": args.jobs, + "external_timeout_seconds": args.timeout, "results": []} + cases = entries(args.suite) + if args.command == "summarize": + # Import the original baseline runner's wall times without rerunning it. + old = args.output / "summary.json" + historical = {r["workload"]: r for r in json.loads(old.read_text())} if old.exists() else {} + rows = [inspect(e, args.output, args.budget, + historical.get(e["id"], {}).get("wall_seconds"), + historical.get(e["id"], {}).get("exit_code")) for e in cases] + metadata["results"] = rows + else: + with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as pool: + for row in pool.map(lambda e: run_one(e, args), cases): + metadata["results"].append(row) + (args.output / "results.json").write_text(json.dumps(metadata, indent=2) + "\n") + (args.output / "results.json").write_text(json.dumps(metadata, indent=2) + "\n") + return 0 if all(r["passed"] for r in metadata["results"]) else 1 + + +def compare(args): + before, after = (json.loads((p / "results.json").read_text()) for p in (args.before, args.after)) + if not args.quality_only and (before["budget_seconds"], before["parallel_workloads"]) != (after["budget_seconds"], after["parallel_workloads"]): + raise ValueError("Comparison requires identical planning budget and concurrency") + old = {r["workload"]: r for r in before["results"]} + if set(old) != {r["workload"] for r in after["results"]}: + raise ValueError("Comparison requires the same completed workload set") + lines = ["# Pane-search benchmark comparison", "", + f"Before binary: `{before['binary_sha256']}`. After binary: `{after['binary_sha256']}`.", "", + "Single-run measurements; execution cost and accuracy are planner-model results, not measured stream execution.", "", + "| Workload | Metric | Before | After | Change |", "|---|---|---:|---:|---:|"] + metrics = ["wall_seconds", "peak_planner_rss_kib", "modeled_cost_per_second", "accepted_queries", + "passed", "guarantees_enforced", "cost_certificate_valid", "domain_certificate_valid"] + counters = ["logical_candidates", "physical_configurations", "compatible_connections", "supplier_sets", "stored_supplier_ids", + "peak_milp_variables", "peak_milp_constraints", "local_candidate_generation_ms", "memo_expansion_ms", + "inventory_construction_ms", "model_construction_ms", "solver_ms", "validation_ms", "solve_calls"] + if args.quality_only: + metrics = ["modeled_cost_per_second", "accepted_queries", "expected_queries", "guarantees_enforced", "passed"] + counters = [] + lines.insert(4, "Quality-only comparison: planning times, RSS, and model sizes are deliberately omitted. Legacy inputs are checked against serialized workload, deployment, accuracy mode, and Stage 0 DAG; original pricing settings must additionally be confirmed from the baseline runner.") + for row in after["results"]: + previous = old[row["workload"]] + if "input_sha256" in previous and "input_sha256" in row: + identical_inputs = previous["input_sha256"] == row["input_sha256"] + elif args.quality_only: + left = json.loads((args.before / previous["artifact"]).read_text()) + right = json.loads((args.after / row["artifact"]).read_text()) + identical_inputs = all(left.get(key) == right.get(key) for key in + ("workload", "deployment_capabilities", "accuracy_guarantees", "stage0_query_language_frontends")) + missing_model_metadata = [] + for key in ("objective", "calibration_scope", "calibration_factor"): + left_value, right_value = previous["search"].get(key), row["search"].get(key) + # Default preparation-time reports can omit these labels. + # Zero is not a valid calibration factor; treat it as missing, + # not as evidence that the selected plan was priced at zero. + if not left_value or not right_value: + missing_model_metadata.append(key) + else: + identical_inputs &= left_value == right_value + if missing_model_metadata: + lines.append(f"| {row['workload']} | Unreported model metadata (confirm runner settings) | — | {', '.join(missing_model_metadata)} | — |") + else: + raise ValueError("Legacy results without input fingerprints require --quality-only") + if not identical_inputs: + raise ValueError("Inputs differ: " + row["workload"]) + for key in metrics + counters: + a, b = (previous.get(key), row.get(key)) if key in metrics else (previous["search"].get(key), row["search"].get(key)) + # A final preparation timeout can carry default zero catalog + # counters despite earlier solved rounds. Do not call that 100% + # compression; preserve raw reports in results.json for diagnosis. + catalog_counters = {"logical_candidates", "physical_configurations", "compatible_connections", "supplier_sets", "stored_supplier_ids"} + if key in catalog_counters: + if previous["search"].get("physical_configurations", 0) == 0: + a = None + if row["search"].get("physical_configurations", 0) == 0: + b = None + change = f"{(b / a - 1) * 100:+.2f}%" if type(a) in (int, float) and a != 0 and type(b) in (int, float) else "—" + lines.append(f"| {row['workload']} | {key} | {a if a is not None else 'unreported'} | {b if b is not None else 'unreported'} | {change} |") + domain_keys = ("status", "pane_width_rounds", "inventory_pane_width_domain", "pane_widths_pending", "full_catalog_completed", "catalog_exhausted", "evaluation_complete") + for key in (() if args.quality_only else domain_keys): + lines.append(f"| {row['workload']} | {key} | {previous['search'].get(key, 'unreported')} | {row['search'].get(key, 'unreported')} | — |") + if not args.quality_only: + lines += ["", "Supplier IDs measure storage sharing; compatible connections and MILP variables measure different things. Missing baseline counters are unreported, not zero.", + "Candidate/edge counters describe reported catalogs; peaks and times can span multiple rounds. A smaller restricted domain is not evidence of lossless full-domain compression."] + args.output.write_text("\n".join(lines) + "\n") + return 0 + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + for command in ("run", "summarize"): + p = sub.add_parser(command) + p.add_argument("binary", type=Path) + p.add_argument("output", type=Path) + p.add_argument("--suite", choices=("four", "viewer", "asap"), default="four") + p.add_argument("--budget", type=int, default=30) + p.add_argument("--timeout", type=int, default=90) + p.add_argument("--jobs", type=int, default=1) + p = sub.add_parser("compare") + p.add_argument("before", type=Path) + p.add_argument("after", type=Path) + p.add_argument("output", type=Path) + p.add_argument("--quality-only", action="store_true", help="compare costs/acceptance only; permits legacy artifact input checks") + args = parser.parse_args() + if args.command != "compare" and (args.jobs < 1 or args.budget < 0 or args.timeout < 1): + parser.error("jobs and timeout must be positive; budget must be nonnegative") + return compare(args) if args.command == "compare" else collect(args) + + +if __name__ == "__main__": + raise SystemExit(main())