Compress pane candidate search and progressively expand legal widths

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

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<u64>,
+    complete: Vec<Vec<u64>>,
+    previous: Vec<Vec<u64>>,
+    max_parts: u64,
+    finished: bool,
+}
+
+impl PaneWidthFrontier {
+    pub fn new(
+        grids: impl IntoIterator<Item = u64>,
+        deadline: Option<std::time::Instant>,
+    ) -> Result<Self, super::CandidateGenerationTimeout> {
+        let grids: Vec<_> = grids
+            .into_iter()
+            .filter(|grid| *grid > 0)
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect();
+        let complete = grids
+            .iter()
+            .map(|&grid| pane_width_divisors_before(grid, deadline))
+            .collect::<Result<Vec<_>, _>>()?;
+        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<std::time::Instant>,
+    ) -> Result<Option<PaneWidthDomain>, super::CandidateGenerationTimeout> {
+        while !self.finished {
+            super::check_generation_deadline(deadline)?;
+            let current: Vec<Vec<u64>> = 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 {

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<Instant>,
+) -> 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<PaneWidthDomain>,
+    domain: PaneWidthDomain,
+    deadline: Option<Instant>,
 ) -> 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<Instant>,
+) -> Result<Vec<(usize, ImplementationCandidate)>, CandidateGenerationTimeout> {
+    local_pane_candidates_delta(memo, entries, None, domain, deadline)
+}
+
+fn local_pane_candidates_delta(
+    memo: &Memo,
+    entries: &[QueryWorkloadEntry],
+    previous: Option<PaneWidthDomain>,
+    domain: PaneWidthDomain,
+    deadline: Option<Instant>,
 ) -> Result<Vec<(usize, ImplementationCandidate)>, 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::<BTreeSet<_>>()
+        };
+        assert_eq!(signatures(&incremental), signatures(&complete));
+    }
+
     fn repeated(interval: u32) -> QueryRecurrence {
         QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(interval)))
     }

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<String>,
+    /// Width domains attempted in order; state/physical catalogs are rebuilt per round.
+    #[serde(default)]
+    pub pane_width_rounds: Vec<String>,
+    /// 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)]